vignesh/bottleneckcrm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor lead and opportunity management:

Browse source 
- Updated lead pipeline and task pipeline models to enhance the is_default field's help text for clarity.
- Modified lead and opportunity email tasks to utilize frontend URLs instead of domain names for better link accuracy.
- Removed outdated Celery task tests for leads and opportunities.
- Streamlined permission checks across lead and opportunity views by introducing a unified is_org_admin function.
- Improved HTML templates for email notifications to ensure correct variable usage and link generation.
- Enhanced kanban view filtering logic to ensure proper access control for users based on their roles.
This commit is contained in:
Ashwin 2026-08-05 11:39:52 +05:30
parent e5b6beb4ca
commit 7f3eacdcf2
69 changed files with 1160 additions and 1495 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

112
.github/workflows/tests.yml vendored
View file

@ -91,6 +91,118 @@ jobs:
path: backend/htmlcov/
retention-days: 14
# The job above runs on SQLite, which is fast and blind to a whole class of
# defect: declared varchar lengths, RLS policies, and every test carrying the
# `postgres_only` marker (which the SQLite job deselects, so they had never
# run in CI at all). Two shipped bugs were invisible to a green suite for
# exactly this reason; see crm/test_settings_postgres.py.
backend-tests-postgres:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:16
env:
POSTGRES_PASSWORD: postgres
ports:
- 5432:5432
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
steps:
- uses: actions/checkout@v4
- name: Install uv
uses: astral-sh/setup-uv@v6
with:
enable-cache: true
cache-dependency-glob: 'backend/uv.lock'
- name: Install system dependencies (WeasyPrint)
run: |
sudo apt-get update
sudo apt-get install -y \
libpango-1.0-0 \
libpangocairo-1.0-0 \
libgdk-pixbuf-2.0-0 \
libffi-dev \
shared-mime-info
# A superuser bypasses RLS, and the policies use FORCE ROW LEVEL
# SECURITY so even the owning role is subject to them. Running the
# RLS tests as `postgres` makes every isolation assertion vacuously
# true, the same trap RLS_SETUP.md warns about for local development.
# CREATEDB is what lets pytest-django build the test database.
- name: Create a non-superuser role for the RLS tests
env:
PGPASSWORD: postgres
run: |
psql -h localhost -U postgres -c \
"CREATE ROLE crm_ci WITH LOGIN CREATEDB PASSWORD 'crm_ci' NOSUPERUSER NOBYPASSRLS;"
- name: Install Python dependencies
working-directory: backend
run: uv sync --frozen
# Run 1: the whole suite, for what SQLite cannot see. Declared varchar
# lengths, integrity constraints, real type coercion. It runs as the
# superuser, so RLS is inert here; run 2 is where isolation is tested.
# The reason for the split is that the shared fixtures insert rows with
# no `app.current_org` set, which a policy-enforcing role refuses. That
# is a real gap in the fixtures rather than a CI setting, and closing it
# is a change to almost every test file, so it is deliberately not
# bundled into this one. Until then the split says plainly which run
# covers which.
- name: Run the full suite against PostgreSQL
working-directory: backend
env:
DJANGO_SETTINGS_MODULE: crm.test_settings_postgres
SECRET_KEY: test-secret-key-for-ci
ADMIN_EMAIL: admin@test.com
DBNAME: crm_db
DBUSER: postgres
DBPASSWORD: postgres
DBHOST: localhost
DBPORT: '5432'
# `postgres_only` is deselected here for the same reason it runs in
# the next step instead: those tests assert that RLS hides rows, and
# this run's superuser role is not subject to any policy, so they fail
# here for a reason that is about the role rather than the code.
# No coverage: the SQLite job produces the report and the badge.
run: uv run pytest --no-cov --tb=short -q -m "not postgres_only"
# Run 2: the `postgres_only` tests, as a role that RLS actually binds.
# The SQLite job deselects these, so before this step they had never
# executed in CI at all. They also have to run as `crm_ci` to mean
# anything: as a superuser they pass whether or not a single policy
# exists.
- name: Run the RLS tests as a policy-bound role
working-directory: backend
env:
DJANGO_SETTINGS_MODULE: crm.test_settings_postgres
SECRET_KEY: test-secret-key-for-ci
ADMIN_EMAIL: admin@test.com
DBNAME: crm_db
DBUSER: crm_ci
DBPASSWORD: crm_ci
DBHOST: localhost
DBPORT: '5432'
run: uv run pytest --no-cov --tb=short -q -m postgres_only
# Undefined names and redefinitions only. The wider `ruff check .` has
# 68 pre-existing findings and `black --check` wants 187 files, so a
# blanket gate would be red on arrival and would land a reformat of the
# whole tree in an unrelated change. These two rules are the ones that
# catch defects rather than style: F811 is how `ActivitySerializer` came
# to be declared twice, with the second silently shadowing the first for
# both of its consumers.
- name: Lint for undefined and redefined names
working-directory: backend
run: uvx ruff check --select F811,F821 .
frontend-checks:
runs-on: ubuntu-latest

4
backend/accounts/tasks.py
View file

@ -1,12 +1,12 @@
import logging
from celery import shared_task
from django.conf import settings
from django.core.mail import EmailMessage
from django.template import Context, Template
from django.template.loader import render_to_string
from accounts.models import Account, AccountEmail, AccountEmailLog
from common.links import frontend_url
from common.models import Profile
from common.tasks import set_rls_context
@ -79,7 +79,7 @@ def send_email_to_assigned_user(recipients, account_id, org_id):
if profile:
recipients_list.append(profile.user.email)
context = {}
context["url"] = settings.DOMAIN_NAME
context["url"] = frontend_url(f"/accounts/{account.id}")
context["user"] = profile.user
context["account"] = account
context["created_by"] = created_by

50
backend/accounts/tests_celery_tasks.py
View file

@ -1,50 +0,0 @@
from datetime import datetime, timedelta
from django.test import TestCase
from django.test.utils import override_settings
from accounts.models import AccountEmail
from accounts.tasks import (
send_email,
send_email_to_assigned_user,
send_scheduled_emails,
)
from accounts.tests import AccountCreateTest
class TestCeleryTasks(AccountCreateTest, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_celery_tasks(self):
org_id = str(self.account.org.id)
email_scheduled = AccountEmail.objects.create(
message_subject="message subject",
message_body="message body",
scheduled_later=True,
timezone="Asia/Kolkata",
from_account=self.account,
scheduled_date_time=(datetime.now() - timedelta(minutes=5)),
from_email="from@email.com",
org=self.account.org,
)
email_scheduled.recipients.add(self.contact.id, self.contact_user1.id)
task = send_scheduled_emails.apply((org_id,))
self.assertEqual("SUCCESS", task.state)
task = send_email.apply((email_scheduled.id, org_id))
self.assertEqual("SUCCESS", task.state)
task = send_email_to_assigned_user.apply(
(
[
self.user.id,
self.user1.id,
],
self.account.id,
org_id,
),
)
self.assertEqual("SUCCESS", task.state)

34
backend/accounts/views.py
View file

@ -27,7 +27,7 @@ from rest_framework.response import Response
from common.custom_fields import validate_payload as validate_custom_fields_payload
from common.lookups import get_scoped_or_404
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from rest_framework.views import APIView
from accounts import swagger_params
@ -194,7 +194,7 @@ class AccountsListView(APIView, LimitOffsetPagination):
queryset = annotate_rollups(
self.model.objects.filter(org=self.request.profile.org)
).order_by("-id")
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
queryset = queryset.filter(
Q(created_by=self.request.profile.user)
| Q(assigned_to=self.request.profile)
@ -436,7 +436,7 @@ class AccountDetailView(APIView):
could watch an account sit in their list, be refused permission to open
it, and still delete it outright.
"""
if self.request.profile.role == "ADMIN" or self.request.profile.is_admin:
if is_org_admin(self.request.profile):
return
if self.request.profile.user_id == account.created_by_id:
return
@ -565,7 +565,7 @@ class AccountDetailView(APIView):
)
def delete(self, request, pk, format=None):
self.object = self.get_object(pk)
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
if self.request.profile.user_id != self.object.created_by_id:
return Response(
{
@ -593,11 +593,10 @@ class AccountDetailView(APIView):
comment_permission = (
self.request.profile.user_id == self.account.created_by_id
or self.request.profile.is_admin
or self.request.profile.role == "ADMIN"
or is_org_admin(self.request.profile)
)
if self.request.profile.is_admin or self.request.profile.role == "ADMIN":
if is_org_admin(self.request.profile):
users_mention = list(
Profile.objects.filter(
is_active=True, org=self.request.profile.org
@ -867,11 +866,7 @@ class AccountCommentView(APIView):
def put(self, request, pk, format=None):
data = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == obj.commented_by
):
if is_org_admin(request.profile) or request.profile == obj.commented_by:
serializer = CommentSerializer(obj, data=data, partial=True)
if data.get("comment"):
if serializer.is_valid():
@ -902,11 +897,7 @@ class AccountCommentView(APIView):
"""Handle partial updates to a comment."""
data = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == obj.commented_by
):
if is_org_admin(request.profile) or request.profile == obj.commented_by:
serializer = CommentSerializer(obj, data=data, partial=True)
if serializer.is_valid():
serializer.save()
@ -929,11 +920,7 @@ class AccountCommentView(APIView):
@extend_schema(tags=["Accounts"], parameters=swagger_params.organization_params)
def delete(self, request, pk, format=None):
self.object = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == self.object.commented_by
):
if is_org_admin(request.profile) or request.profile == self.object.commented_by:
self.object.delete()
return Response(
{"error": False, "message": "Comment Deleted Successfully"},
@ -972,8 +959,7 @@ class AccountAttachmentView(APIView):
except (DjangoValidationError, ValueError):
raise Http404("No such attachment.")
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
is_org_admin(request.profile)
or request.profile.user_id == self.object.created_by_id
):
self.object.delete()

2
backend/cases/analytics_views.py
View file

@ -63,7 +63,7 @@ def _filtered_qs(request) -> tuple[object, datetime, datetime]:
.exclude(status="Duplicate")
)
if profile.role != "ADMIN" and not profile.is_admin:
if not is_org_admin(profile):
qs = qs.filter(
Q(created_by=profile.user) | Q(assigned_to=profile) | Q(watchers=profile)
).distinct()

20
backend/cases/kanban_views.py
View file

@ -22,7 +22,7 @@ from cases.serializer import (
CasePipelineSerializer,
CaseStageSerializer,
)
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.validators import uuid_param
from common.utils import STATUS_CHOICE
@ -99,7 +99,7 @@ class CaseKanbanView(APIView):
)
# Apply permission filtering
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
queryset = queryset.filter(
Q(assigned_to=request.profile) | Q(created_by=request.profile.user)
)
@ -237,7 +237,7 @@ class CaseMoveView(APIView):
case = get_object_or_404(Case, pk=pk, org=org)
# Permission check
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
if not (
request.profile.user == case.created_by
or request.profile in case.assigned_to.all()
@ -391,7 +391,7 @@ class CasePipelineListCreateView(APIView):
"""Create a new pipeline."""
org = request.profile.org
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Only admins can create pipelines"},
status=status.HTTP_403_FORBIDDEN,
@ -476,7 +476,7 @@ class CasePipelineDetailView(APIView):
responses={200: CasePipelineSerializer},
)
def put(self, request, pk):
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -495,7 +495,7 @@ class CasePipelineDetailView(APIView):
@extend_schema(tags=["Case Pipelines"], responses={204: None})
def delete(self, request, pk):
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -527,7 +527,7 @@ class CaseStageCreateView(APIView):
responses={201: CaseStageSerializer},
)
def post(self, request, pipeline_pk):
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -557,7 +557,7 @@ class CaseStageDetailView(APIView):
responses={200: CaseStageSerializer},
)
def put(self, request, pk):
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -576,7 +576,7 @@ class CaseStageDetailView(APIView):
@extend_schema(tags=["Case Stages"], responses={204: None})
def delete(self, request, pk):
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -610,7 +610,7 @@ class CaseStageReorderView(APIView):
)
@transaction.atomic
def post(self, request, pipeline_pk):
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)

4
backend/cases/merge_views.py
View file

@ -23,12 +23,12 @@ from cases.notifications import case_link
from cases.serializer import CaseSerializer
from cases.signals import _create_activity
from common.models import Attachments, Comment
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
def _can_merge(profile, source: Case, target: Case) -> bool:
"""Admin OR the creator of BOTH cases."""
if profile.is_admin or profile.role == "ADMIN":
if is_org_admin(profile):
return True
user = profile.user
return source.created_by_id == user.id and target.created_by_id == user.id

21
backend/cases/migrations/0027_alter_casepipeline_is_default.py Normal file
View file

@ -0,0 +1,21 @@
# Generated by Django 6.0.7 on 2026-08-05 04:54
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("cases", "0026_case_is_sample"),
]
operations = [
migrations.AlterField(
model_name="casepipeline",
name="is_default",
field=models.BooleanField(
default=False,
help_text="Sorts this pipeline first in listings (see Meta.ordering) and is limited to one per org. It does NOT route new records: a case created without a stage keeps stage=NULL, which is the supported status-based kanban mode.",
),
),
]

7
backend/cases/models.py
View file

@ -500,7 +500,12 @@ class CasePipeline(BaseModel):
)
is_default = models.BooleanField(
default=False,
help_text="If true, new cases without explicit pipeline go here",
help_text=(
"Sorts this pipeline first in listings (see Meta.ordering) and is "
"limited to one per org. It does NOT route new records: a case "
"created without a stage keeps stage=NULL, which is the supported "
"status-based kanban mode."
),
)
is_active = models.BooleanField(default=True)

5
backend/cases/serializer.py
View file

@ -16,6 +16,7 @@ from cases.models import (
)
from cases.parent_guards import check_parent_link
from common.models import Profile, Teams
from common.permissions import is_org_admin
from common.serializer import (
OrganizationSerializer,
ProfileSerializer,
@ -621,9 +622,7 @@ class InboundMailboxSerializer(serializers.ModelSerializer):
data = super().to_representation(instance)
request = self.context.get("request")
profile = getattr(request, "profile", None) if request else None
is_admin = bool(
profile and (profile.role == "ADMIN" or getattr(profile, "is_admin", False))
)
is_admin = is_org_admin(profile)
if not is_admin:
data.pop("webhook_secret", None)
data.pop("topic_arn", None)

4
backend/cases/tasks.py
View file

@ -3,7 +3,6 @@ import logging
from datetime import timedelta
from celery import shared_task
from django.conf import settings
from django.core.mail import EmailMessage
from django.core.signing import TimestampSigner
from django.db.models import Q
@ -11,6 +10,7 @@ from django.template.loader import render_to_string
from django.utils import timezone
from cases.models import Case, CsatSurvey, EscalationPolicy, TimeEntry
from cases.notifications import case_link
from cases.workflow import TERMINAL_STATUSES
from common.links import frontend_url
from common.models import Activity, Org, Profile
@ -46,7 +46,7 @@ def send_email_to_assigned_user(recipients, case_id, org_id):
if profile:
recipients_list.append(profile.user.email)
context = {}
context["url"] = settings.DOMAIN_NAME
context["url"] = frontend_url(case_link(case.id))
context["user"] = profile.user
context["case"] = case
context["created_by"] = created_by

26
backend/cases/tests_celery_tasks.py
View file

@ -1,26 +0,0 @@
from django.test import TestCase
from django.test.utils import override_settings
from cases.tasks import send_email_to_assigned_user
from cases.tests import CaseCreation
class TestCeleryTasks(CaseCreation, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_celery_tasks(self):
org_id = str(self.case.org.id)
task = send_email_to_assigned_user.apply(
(
[
self.user.id,
self.user1.id,
],
self.case.id,
org_id,
),
)
self.assertEqual("SUCCESS", task.state)

18
backend/cases/views.py
View file

@ -925,11 +925,7 @@ class CaseCommentView(APIView):
def put(self, request, pk, format=None):
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == obj.commented_by
):
if is_org_admin(request.profile) or request.profile == obj.commented_by:
serializer = CommentSerializer(obj, data=params)
if params.get("comment"):
if serializer.is_valid():
@ -969,11 +965,7 @@ class CaseCommentView(APIView):
"""Handle partial updates to a comment."""
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == obj.commented_by
):
if is_org_admin(request.profile) or request.profile == obj.commented_by:
serializer = CommentSerializer(obj, data=params, partial=True)
if serializer.is_valid():
serializer.save()
@ -1008,11 +1000,7 @@ class CaseCommentView(APIView):
)
def delete(self, request, pk, format=None):
self.object = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == self.object.commented_by
):
if is_org_admin(request.profile) or request.profile == self.object.commented_by:
self.object.delete()
return Response(
{"error": False, "message": "Comment Deleted Successfully"},

1
backend/common/management/commands/seed_data.py
View file

@ -718,7 +718,6 @@ class Command(BaseCommand):
"role": "ADMIN",
"has_sales_access": True,
"has_marketing_access": True,
"is_organization_admin": True,
"is_active": True,
},
)

42
backend/common/migrations/0035_sync_is_organization_admin_with_role.py Normal file
View file

@ -0,0 +1,42 @@
"""Make `is_organization_admin` agree with `role` on every existing row.
The two columns are one binary fact stored twice (`ROLES` is ADMIN/USER and
nothing else) and used to be settable independently. `common.permissions
.is_org_admin` now reads `role` alone and `Profile.save` derives the column, so
this changes nobody's access: it only stops the stored value from contradicting
the role in the three payloads that still return it (`/api/profile/`, the org
list, the login response).
Without this, a row written before the change keeps its stale value until that
profile happens to be saved, and a client reading `is_organization_admin: true`
off a member would show an admin badge for someone the API treats as a member.
The dev database had zero disagreeing rows when this was written. Production
may differ, which is the reason to run it rather than assume.
"""
from django.db import migrations
def sync_flag_to_role(apps, schema_editor):
Profile = apps.get_model("common", "Profile")
Profile.objects.filter(role="ADMIN", is_organization_admin=False).update(
is_organization_admin=True
)
Profile.objects.exclude(role="ADMIN").filter(is_organization_admin=True).update(
is_organization_admin=False
)
class Migration(migrations.Migration):
dependencies = [
("common", "0034_rls_pipeline_tables"),
]
# No reverse: the pre-migration values are not recoverable, and restoring a
# disagreement would restore a lie rather than a capability. Nothing reads
# the column for authorization any more, so leaving it consistent on a
# rollback is correct.
operations = [
migrations.RunPython(sync_flag_to_role, migrations.RunPython.noop),
]

18
backend/common/models.py
View file

@ -241,9 +241,25 @@ class Profile(BaseModel):
def __str__(self):
return f"{self.user.email} <{self.org.name}>"
def save(self, *args, **kwargs):
# `role` and `is_organization_admin` are two columns for one binary
# fact (`ROLES` is ADMIN/USER and nothing else), and they used to be
# settable independently, which let an admin mint a colleague the UI
# showed as a plain user and could never demote. `role` is what every
# surface displays and what `common.permissions.is_org_admin` reads;
# the column is kept because it is in API responses, so it is derived
# here rather than left to drift.
self.is_organization_admin = self.role == "ADMIN"
super().save(*args, **kwargs)
@property
def is_admin(self):
return self.is_organization_admin
"""Alias kept for the API response and for existing callers.
Reads `role`, not the column, so it answers the same as
`is_org_admin` even on an unsaved instance.
"""
return self.role == "ADMIN"
@property
def user_details(self):

25
backend/common/permissions.py
View file

@ -9,7 +9,7 @@ from rest_framework import permissions
def is_org_admin(profile):
"""Whether ``profile`` administers its org.
"""Whether ``profile`` administers its org. ``role`` is the only source.
A plain function and not only the ``IsOrgAdmin`` class below, because most
callers are views that read wide and write narrow: the same endpoint is
@ -19,19 +19,28 @@ def is_org_admin(profile):
That is why this rule had been written out nine times, as a private
``_is_admin`` in nine modules, in three different spellings. Five of them
raised ``AttributeError`` on a ``None`` profile where two returned
``False``, and one reached for ``is_organization_admin`` while the rest
used the ``is_admin`` property that aliases it. All nine agreed for a real
profile, so nothing was broken; one edit that missed eight copies is how
that stops being true.
``False``. All nine agreed for a real profile, so nothing was broken; one
edit that missed eight copies is how that stops being true.
**This deliberately does not consult ``is_organization_admin``.** ``ROLES``
has exactly two members, so that column and ``role == "ADMIN"`` are two
spellings of one binary fact, and the backend used to read them
inconsistently: 82 checks looked only at ``role`` while 43 also accepted
the flag, with both spellings appearing inside the same file. The flag is
the dangerous half of that pair, because no frontend surface displays or
sets it: an admin could ``PATCH`` a colleague to
``{"is_organization_admin": true, "role": "USER"}`` and grant admin over
accounts, contacts, team management and the org record itself, invisibly
and irrevocably from the UI. ``Profile.save`` now derives the column from
``role`` so the two cannot disagree, and ``ProfileSerializer`` no longer
accepts it as an input.
``None`` is not an admin. A view with no org context has no profile, and
answering ``False`` gives it a clean 403 instead of a 500.
"""
if profile is None:
return False
if profile.role == "ADMIN":
return True
return bool(getattr(profile, "is_organization_admin", False))
return getattr(profile, "role", None) == "ADMIN"
class HasOrgContext(permissions.BasePermission):

2
backend/common/search_views.py
View file

@ -58,7 +58,7 @@ def _scope_orgadmin(qs, profile):
def _scope_superuser(qs, profile, user):
"""leads / opportunities / invoices rule: admin = role ADMIN or superuser."""
if profile.role == "ADMIN" or user.is_superuser:
if is_org_admin(profile) or user.is_superuser:
return qs
return qs.filter(_own_filter(profile)).distinct()

25
backend/common/serializer.py
View file

@ -547,13 +547,17 @@ class CreateUserSerializer(serializers.ModelSerializer):
class CreateProfileSerializer(serializers.ModelSerializer):
class Meta:
model = Profile
# `is_organization_admin` is deliberately absent. It is derived from
# `role` in `Profile.save`, and accepting it here was a second, hidden
# way to grant admin: no frontend surface shows or sets the flag, so a
# profile carrying it read as a plain user everywhere a human looked
# while `is_org_admin` answered True. One fact, one input.
fields = (
"role",
"phone",
"alternate_phone",
"has_sales_access",
"has_marketing_access",
"is_organization_admin",
)
# The fields that grant access rather than describe a person. Only an admin
@ -561,7 +565,6 @@ class CreateProfileSerializer(serializers.ModelSerializer):
# notes in common/views/user_views.py.
PRIVILEGED_FIELDS = (
"role",
"is_organization_admin",
"has_sales_access",
"has_marketing_access",
)
@ -620,8 +623,8 @@ class ProfileSerializer(serializers.ModelSerializer):
class ProfileSelfUpdateSerializer(serializers.Serializer):
"""The two fields a user may change on their OWN profile via PATCH /api/profile/.
Deliberately just ``phone`` (on Profile) and ``name`` (on User). Role, org,
and the ``has_*_access`` / ``is_organization_admin`` flags are NOT here:
Deliberately just ``phone`` (on Profile) and ``name`` (on User). Role, org
and the ``has_*_access`` flags are NOT here:
those decide permissions and are an admin's to set. Leaving them out is what
keeps this self-edit endpoint from becoming the same self-serve privilege
escalation ``CreateProfileSerializer`` once was. A field the serializer does
@ -989,8 +992,18 @@ class ActivityUserSerializer(serializers.Serializer):
return obj.user.email.split("@")[0]
class ActivitySerializer(serializers.ModelSerializer):
"""Serializer for recent activities"""
class DashboardActivitySerializer(serializers.ModelSerializer):
"""Recent-activity row for the dashboard feed.
Renamed 2026-08-05. It was a second class also called
`ActivitySerializer`, 580 lines below the first, so it silently shadowed
it and **both** consumers got this shape. `cases/views.py` imports the
other one for the ticket activity timeline, whose frontend component reads
`metadata` (five places) and `created_at`; neither is in this list, so the
timeline rendered no metadata and an invalid date. The dashboard reads
`timestamp` and `humanized_time`, which is why only one of the two broke
visibly.
"""
user = ActivityUserSerializer(read_only=True)
action_display = serializers.CharField(source="get_action_display", read_only=True)

80
backend/common/tasks.py
View file

@ -11,8 +11,8 @@ from django.db import connection
from django.template.loader import render_to_string
from django.utils import timezone
from common.links import frontend_url
from common.models import (
Comment,
MagicLinkToken,
Notification,
Org,
@ -142,76 +142,6 @@ def send_magic_link_email(token_id, raw_code=None):
logger.exception("SES rejected email for magic link token %s", token_id)
@shared_task
def send_email_user_mentions(
comment_id,
called_from,
org_id=None,
):
"""Send Mail To Mentioned Users In The Comment"""
# Set RLS context for org-scoped queries
set_rls_context(org_id)
comment = Comment.objects.filter(id=comment_id).first()
if comment:
comment_text = comment.comment
comment_text_list = comment_text.split()
recipients = []
for comment_text in comment_text_list:
if comment_text.startswith("@"):
if comment_text.strip("@").strip(",") not in recipients:
if User.objects.filter(
username=comment_text.strip("@").strip(","), is_active=True
).exists():
email = (
User.objects.filter(
username=comment_text.strip("@").strip(",")
)
.first()
.email
)
recipients.append(email)
context = {}
context["commented_by"] = comment.commented_by
context["comment_description"] = comment.comment
subject = None
if called_from == "accounts":
subject = "New comment on Account. "
elif called_from == "contacts":
subject = "New comment on Contact. "
elif called_from == "leads":
subject = "New comment on Lead. "
elif called_from == "opportunity":
subject = "New comment on Opportunity. "
elif called_from == "cases":
subject = "New comment on Case. "
elif called_from == "tasks":
subject = "New comment on Task. "
elif called_from == "invoices":
subject = "New comment on Invoice. "
if subject:
context["url"] = settings.DOMAIN_NAME
else:
context["url"] = ""
# subject = 'Django CRM : comment '
if recipients:
for recipient in recipients:
recipients_list = [
recipient,
]
context["mentioned_user"] = recipient
html_content = render_to_string("comment_email.html", context=context)
msg = EmailMessage(
subject,
html_content,
from_email=settings.DEFAULT_FROM_EMAIL,
to=recipients_list,
)
msg.content_subtype = "html"
msg.send()
@shared_task
def send_email_user_status(
user_id,
@ -223,9 +153,11 @@ def send_email_user_status(
context = {}
context["message"] = "deactivated"
context["email"] = user.email
context["url"] = settings.DOMAIN_NAME
if user.has_marketing_access:
context["url"] = context["url"] + "/marketing"
# The web app's root is the dashboard. There is no `/marketing` route,
# and `has_marketing_access` is a profile flag rather than a user one,
# so the branch that appended it could never have been reached from a
# `User` in the first place.
context["url"] = frontend_url("/")
if user.is_active:
context["message"] = "activated"
context["status_changed_user"] = status_changed_user

18
backend/common/templates/comment_email.html
View file

@ -1,18 +0,0 @@
{% extends 'root_email_template_new.html' %}
{% block preheader %}{{ commented_by }} mentioned you in a comment{% endblock preheader %}
{% block heading %}Hi {{ mentioned_user }},{% endblock heading %}
{% block content_body %}
<p style="margin:0 0 8px 0;">{{ commented_by }} mentioned you in a comment.</p>
{% endblock content_body %}
{% block button_link %}
{% if url %}
<a class="btn-link" href="{{ url }}"
style="display:inline-block;padding:14px 32px;background-color:#EA580C;color:#ffffff;text-decoration:none;border-radius:6px;font-size:16px;font-weight:600;line-height:1;">
View Comment
</a>
{% endif %}
{% endblock button_link %}

90
backend/common/tests/test_activity_serializer_shapes.py Normal file
View file

@ -0,0 +1,90 @@
"""Two activity feeds, two shapes, and neither may take the other's name.
`common/serializer.py` defined `ActivitySerializer` twice, 580 lines apart.
Python keeps the second, so both consumers received the dashboard shape:
* the dashboard feed wants `timestamp` and `humanized_time`, and got them
* the ticket activity timeline wants `metadata` and `created_at`, and got
neither, so `TicketActivityTimeline.svelte` (which reads `a.metadata` in
five places and `formatRelativeDate(a.created_at)`) rendered no metadata
and an invalid date
Only one of the pair broke, which is why it survived: the dashboard, the
louder surface, was the one being served correctly. The second class is now
`DashboardActivitySerializer`.
These tests assert the field sets rather than a rendered page, because the
field set is what the two frontends destructure and it is what a rename or a
merge of the two classes would quietly change.
"""
from __future__ import annotations
import pytest
from crum import impersonate
from cases.models import Case
from common.models import Activity
from common.serializer import ActivitySerializer, DashboardActivitySerializer
# What `TicketActivityTimeline.svelte` reads off each row.
TIMELINE_REQUIRED = {"id", "action", "user", "metadata", "created_at"}
# What the dashboard feed renders.
DASHBOARD_REQUIRED = {"id", "action", "action_display", "timestamp", "humanized_time"}
class TestTheTwoSerializersAreDistinct:
def test_they_are_not_the_same_class(self):
assert ActivitySerializer is not DashboardActivitySerializer
def test_the_timeline_serializer_carries_what_the_timeline_reads(self):
fields = set(ActivitySerializer().fields)
missing = TIMELINE_REQUIRED - fields
assert not missing, f"ticket timeline would read undefined fields: {missing}"
def test_the_dashboard_serializer_carries_what_the_dashboard_reads(self):
fields = set(DashboardActivitySerializer().fields)
missing = DASHBOARD_REQUIRED - fields
assert not missing, f"dashboard feed would read undefined fields: {missing}"
def test_the_shapes_genuinely_differ(self):
"""If these ever converge, one class is enough and this file is wrong.
Written as an assertion rather than a comment so that merging them
becomes a deliberate act with a failing test to delete, not something
that happens by accident a second time.
"""
assert set(ActivitySerializer().fields) != set(
DashboardActivitySerializer().fields
)
@pytest.mark.django_db
class TestTheCaseActivityEndpointServesTheTimelineShape:
"""The endpoint, not just the class: `cases/views.py` imports by name."""
def test_activity_rows_carry_metadata_and_created_at(
self, admin_client, admin_user, admin_profile, org_a
):
with impersonate(admin_user):
case = Case.objects.create(
name="Leaking valve", org=org_a, status="New", priority="Normal"
)
Activity.objects.create(
org=org_a,
user=admin_profile,
action="COMMENT",
entity_type="Case",
entity_id=case.id,
entity_name=case.name,
description="left a note",
metadata={"visibility_changed": True, "after": False},
)
response = admin_client.get(f"/api/cases/{case.id}/activities/")
assert response.status_code == 200
row = response.data["activities"][0]
assert row["metadata"] == {"visibility_changed": True, "after": False}
assert row["created_at"] is not None

254
backend/common/tests/test_email_links.py Normal file
View file

@ -0,0 +1,254 @@
"""Every link in an outgoing email points at the web app and at a real record.
These emails had a button whose ``href`` was ``settings.DOMAIN_NAME``, which
names the **API** host, with no path on it at all: the assignee got a "View
Lead" button that opened the Django host's root. `backend/.env` sets
``DOMAIN_NAME=""``, so in development the attribute was the empty string and
the rendered ``href=""`` reloaded whatever page the reader was on. Two other
templates appended a frontend path (``/goals``, ``/opportunities?rotten=true``)
to that backend base, and the second of those is not even a route the frontend
serves; it is ``/pipeline``.
``common.links.frontend_url`` already existed for exactly this, built when the
invoice and estimate portal links had the same defect. The rule these tests
pin: an email link is built from ``FRONTEND_URL`` and carries the path of the
thing the mail is about.
They assert on the rendered HTML rather than the context dict, because the
context key is not the contract, the anchor is: ``leads_assigned.html`` read
``{{ url|default:lead_detail_url }}`` and its two senders each filled a
different one of those names.
"""
from __future__ import annotations
import re
import pytest
from crum import impersonate
from django.core import mail
from accounts.models import Account
from cases.models import Case
from contacts.models import Contact
from leads.models import Lead
from opportunity.models import Opportunity
FRONTEND = "https://app.example.com"
def _hrefs(message):
"""Every anchor target in the message body."""
return re.findall(r'href="([^"]*)"', message.body)
def _button_href(message):
"""The single link the reader is meant to click.
The templates carry one anchor; if that ever stops being true this raises
rather than silently checking the wrong one.
"""
hrefs = [h for h in _hrefs(message) if not h.startswith("mailto:")]
assert len(hrefs) == 1, f"expected one link, found {hrefs}"
return hrefs[0]
@pytest.fixture(autouse=True)
def _outbox_and_frontend(settings):
"""A known `FRONTEND_URL` and an empty outbox for every test here.
`django.test.override_settings` cannot decorate a plain pytest class, so
this is the pytest-django `settings` fixture instead.
"""
settings.FRONTEND_URL = FRONTEND
mail.outbox.clear()
yield
mail.outbox.clear()
@pytest.mark.django_db
class TestAssignmentEmailsLinkToTheRecord:
def test_lead_assignment_links_to_the_lead(self, org_a, admin_user, admin_profile):
from leads.tasks import send_email_to_assigned_user
with impersonate(admin_user):
lead = Lead.objects.create(title="Roof job", org=org_a, status="assigned")
send_email_to_assigned_user([admin_profile.id], lead.id, str(org_a.id))
assert len(mail.outbox) == 1
assert _button_href(mail.outbox[0]) == f"{FRONTEND}/leads/{lead.id}"
def test_contact_assignment_links_to_the_contact(
self, org_a, admin_user, admin_profile
):
from contacts.tasks import send_email_to_assigned_user
with impersonate(admin_user):
contact = Contact.objects.create(first_name="Dana", org=org_a)
send_email_to_assigned_user([admin_profile.id], contact.id, str(org_a.id))
assert len(mail.outbox) == 1
assert _button_href(mail.outbox[0]) == f"{FRONTEND}/contacts/{contact.id}"
def test_account_assignment_links_to_the_account(
self, org_a, admin_user, admin_profile
):
from accounts.tasks import send_email_to_assigned_user
with impersonate(admin_user):
account = Account.objects.create(name="Northwind", org=org_a)
send_email_to_assigned_user([admin_profile.id], account.id, str(org_a.id))
assert len(mail.outbox) == 1
assert _button_href(mail.outbox[0]) == f"{FRONTEND}/accounts/{account.id}"
def test_case_assignment_links_to_the_ticket_path(
self, org_a, admin_user, admin_profile
):
"""`/tickets/<id>`, not `/cases/<id>`.
The model says case and every client routes it at `/tickets`, the same
mismatch `cases.notifications.case_link` exists to absorb. This calls
that function rather than re-deriving the path.
"""
from cases.tasks import send_email_to_assigned_user
with impersonate(admin_user):
case = Case.objects.create(
name="Boiler down", org=org_a, status="New", priority="Normal"
)
send_email_to_assigned_user([admin_profile.id], case.id, str(org_a.id))
assert len(mail.outbox) == 1
assert _button_href(mail.outbox[0]) == f"{FRONTEND}/tickets/{case.id}"
def test_opportunity_assignment_links_to_the_pipeline_path(
self, org_a, admin_user, admin_profile
):
"""`/pipeline/<id>`: the frontend has no `/opportunities` route."""
from opportunity.tasks import send_email_to_assigned_user
with impersonate(admin_user):
deal = Opportunity.objects.create(
name="Retrofit", org=org_a, stage="QUALIFICATION"
)
send_email_to_assigned_user([admin_profile.id], deal.id, str(org_a.id))
assert len(mail.outbox) == 1
assert _button_href(mail.outbox[0]) == f"{FRONTEND}/pipeline/{deal.id}"
@pytest.mark.django_db
class TestTheSecondLeadMailerAgreesWithTheFirst:
"""Two tasks send "you were assigned a lead" from the same template.
They filled different context keys, so the template hedged with
``{{ url|default:lead_detail_url }}``. Both now fill ``url`` with the same
value, which is what lets the template say ``{{ url }}`` like its five
siblings.
"""
def test_the_webhook_mailer_links_to_the_lead(
self, org_a, admin_user, admin_profile
):
"""This one hands the rendered HTML to a second task to send.
`send_email.delay(...)` needs a broker, and the test settings point
Celery at `memory://` without eager mode, so the mail never leaves the
queue. The assertion is on what this task renders and dispatches,
which is the part under test.
"""
from unittest.mock import patch
from leads.tasks import send_lead_assigned_emails
with impersonate(admin_user):
lead = Lead.objects.create(title="Web form", org=org_a, status="assigned")
with patch("leads.tasks.send_email.delay") as dispatched:
send_lead_assigned_emails(lead.id, [admin_profile.id], str(org_a.id))
assert dispatched.call_count == 1
html = dispatched.call_args.kwargs["html_content"]
assert f'href="{FRONTEND}/leads/{lead.id}"' in html
# The render itself is the other half: this template used to raise
# `VariableDoesNotExist` for whichever sender filled the other key.
assert "Web form" in html
def test_it_no_longer_takes_a_host_supplied_base(self):
"""The dropped argument was ``request.META["HTTP_HOST"]``.
Its one caller is the website-lead webhook, so the base of a link in
mail this system sends to its own staff came off the wire. Pinning the
signature keeps a future caller from reintroducing it.
"""
import inspect
from leads.tasks import send_lead_assigned_emails
params = list(
inspect.signature(send_lead_assigned_emails.run).parameters
)
assert params == ["lead_id", "new_assigned_to_list", "org_id"]
@pytest.mark.django_db
class TestAlertEmailsLinkToTheirPage:
def test_stale_deal_alert_links_to_the_rotten_filter(
self, org_a, admin_user, admin_profile
):
from opportunity.tasks import send_stale_deals_alert
with impersonate(admin_user):
deal = Opportunity.objects.create(
name="Stalled", org=org_a, stage="QUALIFICATION"
)
send_stale_deals_alert(org_a, [(deal, 40, 14)])
assert len(mail.outbox) == 1
assert _button_href(mail.outbox[0]) == f"{FRONTEND}/pipeline?rotten=true"
def test_goal_milestone_links_to_goals(self, org_a, admin_user, admin_profile):
from opportunity.models import SalesGoal
from opportunity.tasks import _send_goal_milestone_email
with impersonate(admin_user):
goal = SalesGoal.objects.create(
name="Q3 revenue",
org=org_a,
target_value=10000,
period_start="2026-07-01",
period_end="2026-09-30",
)
_send_goal_milestone_email(admin_profile, goal, "50%", 50, 5000)
assert len(mail.outbox) == 1
assert _button_href(mail.outbox[0]) == f"{FRONTEND}/goals"
@pytest.mark.django_db
def test_a_trailing_slash_on_the_setting_does_not_double_up(
settings, org_a, admin_user, admin_profile
):
"""An operator who sets `FRONTEND_URL=https://app.example.com/` is not wrong.
`frontend_url` strips it, and this is the assertion that keeps the callers
from concatenating by hand again.
"""
settings.FRONTEND_URL = f"{FRONTEND}/"
from leads.tasks import send_email_to_assigned_user
with impersonate(admin_user):
lead = Lead.objects.create(title="Slash", org=org_a, status="assigned")
send_email_to_assigned_user([admin_profile.id], lead.id, str(org_a.id))
assert _button_href(mail.outbox[0]) == f"{FRONTEND}/leads/{lead.id}"

92
backend/common/tests/test_is_org_admin.py
View file

@ -18,12 +18,11 @@ from common.permissions import IsOrgAdmin, is_org_admin
class _FakeProfile:
"""Only the two attributes the rule reads.
"""Only the attributes the rule could read.
A real ``Profile`` needs an org and a user, and nothing here is about the
database. ``is_admin`` is deliberately absent: the rule reads
``is_organization_admin``, which is the column, and the property that
aliases it is what half the old copies used instead.
database. ``is_organization_admin`` is present precisely so the tests can
set it to ``True`` on a member and prove it changes nothing.
"""
def __init__(self, role="USER", is_organization_admin=False):
@ -35,34 +34,36 @@ class TestBothDirections:
def test_an_admin_by_role_is_an_admin(self):
assert is_org_admin(_FakeProfile(role="ADMIN")) is True
def test_an_admin_by_flag_is_an_admin(self):
assert is_org_admin(_FakeProfile(is_organization_admin=True)) is True
def test_a_plain_member_is_not(self):
assert is_org_admin(_FakeProfile()) is False
def test_the_flag_alone_does_not_make_an_admin(self):
"""The hidden-admin case, and the reason ``role`` is the only source.
82 checks in the backend read ``role`` alone and 43 also accepted this
flag, so the same profile was an admin at some endpoints and a member
at others. No frontend surface displays or sets the column, so the
half that accepted it granted a privilege the UI could neither show
nor revoke.
"""
assert is_org_admin(_FakeProfile(is_organization_admin=True)) is False
def test_a_missing_profile_is_not(self):
"""Five of the ten copies raised ``AttributeError`` here.
"""Five of the eleven copies raised ``AttributeError`` here.
A view with no org context has no profile, so this is not theoretical:
it decided whether the caller got a clean 403 or a 500.
"""
assert is_org_admin(None) is False
def test_an_object_without_the_flag_is_judged_on_role_alone(self):
def test_an_object_with_no_role_at_all_is_not_an_admin(self):
"""Not every caller passes a full ``Profile``.
``getattr`` with a default keeps that a ``False`` rather than an
``AttributeError``, which is the behaviour the two None-safe copies had
and the other eight did not.
``AttributeError``, which is the behaviour the two None-safe copies
had and the other nine did not.
"""
class RoleOnly:
role = "USER"
assert is_org_admin(RoleOnly()) is False
RoleOnly.role = "ADMIN"
assert is_org_admin(RoleOnly()) is True
assert is_org_admin(object()) is False
class TestThePermissionClassUsesTheSameRule:
@ -91,17 +92,23 @@ def _backend_root():
def test_no_module_defines_its_own_admin_check():
"""Guard the consolidation with the source, not with a comment.
The ten copies did not appear at once; each was written by somebody who
The eleven copies did not appear at once; each was written by somebody who
needed the rule in a new module and had no reason to know it already
existed eight times. This fails the moment an eleventh is added, which is
the only thing that keeps them from coming back.
existed. This fails the moment a twelfth is added, which is the only thing
that keeps them from coming back.
``is_org_admin`` itself is in the list of names checked. The eleventh copy
lived in ``tasks/access.py`` under exactly that name and so walked past the
first version of this test, which looked only for the private spellings: a
guard that watches for the wrong name is not a guard.
"""
canonical = _backend_root() / "common" / "permissions.py"
offenders = []
for path in _backend_root().rglob("*.py"):
parts = path.parts
if any(p in {".venv", "venv", "migrations", "node_modules"} for p in parts):
continue
if path.name == Path(__file__).name:
if path.name == Path(__file__).name or path == canonical:
continue
try:
tree = ast.parse(path.read_text())
@ -109,7 +116,8 @@ def test_no_module_defines_its_own_admin_check():
continue
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and (
node.name in {"_is_admin", "is_admin_profile", "_is_org_admin"}
node.name
in {"is_org_admin", "_is_admin", "is_admin_profile", "_is_org_admin"}
):
offenders.append(f"{path.relative_to(_backend_root())}:{node.lineno}")
@ -126,4 +134,40 @@ def test_a_real_profile_agrees_with_the_fake_one(admin_profile, user_profile):
assert is_org_admin(user_profile) is False
user_profile.is_organization_admin = True
assert is_org_admin(user_profile) is True
assert is_org_admin(user_profile) is False
@pytest.mark.django_db
class TestTheColumnCannotDisagreeWithTheRole:
"""``Profile.save`` derives the flag, so the pair cannot drift apart.
The column stays because it is in API responses (``/api/profile/``, the
org list, the login payload). A field nothing can write but everything can
read has to be kept true, or it becomes a lie a client acts on.
"""
def test_saving_an_admin_sets_the_flag(self, user_profile):
user_profile.role = "ADMIN"
user_profile.save()
user_profile.refresh_from_db()
assert user_profile.is_organization_admin is True
def test_saving_a_member_clears_the_flag(self, admin_profile):
admin_profile.role = "USER"
admin_profile.save()
admin_profile.refresh_from_db()
assert admin_profile.is_organization_admin is False
def test_a_flag_set_by_hand_does_not_survive_the_save(self, user_profile):
"""The write that used to mint a hidden admin."""
user_profile.is_organization_admin = True
user_profile.save()
user_profile.refresh_from_db()
assert user_profile.is_organization_admin is False
assert is_org_admin(user_profile) is False
def test_the_is_admin_property_reads_the_role(self, user_profile):
assert user_profile.is_admin is False
user_profile.role = "ADMIN"
# Unsaved on purpose: the property must not need a round trip to agree.
assert user_profile.is_admin is True

15
backend/common/tests/test_multitenancy.py
View file

@ -627,8 +627,17 @@ class TestPermissionClasses(MultiTenancyBaseTestCase):
perm = IsOrgAdmin()
self.assertTrue(perm.has_permission(request, None))
def test_is_org_admin_with_org_admin_flag(self):
"""IsOrgAdmin allows organization_admin flag."""
def test_is_org_admin_refuses_the_org_admin_flag_alone(self):
"""IsOrgAdmin reads `role` only, so the flag by itself grants nothing.
This assertion used to be the opposite. `is_organization_admin` and
`role == "ADMIN"` are one binary fact stored twice, and only 43 of the
125 admin checks in the backend consulted the flag, so the same profile
was an admin at some endpoints and a member at others. The flag is the
half no frontend surface shows or sets, which made it a grant an
operator could neither see nor revoke. `Profile.save` now derives it
from `role`; see `common/tests/test_is_org_admin.py`.
"""
from common.permissions import IsOrgAdmin
request = MagicMock()
@ -636,7 +645,7 @@ class TestPermissionClasses(MultiTenancyBaseTestCase):
request.profile.role = "USER"
request.profile.is_organization_admin = True
perm = IsOrgAdmin()
self.assertTrue(perm.has_permission(request, None))
self.assertFalse(perm.has_permission(request, None))
def test_is_org_admin_non_admin(self):
"""IsOrgAdmin denies regular user."""

26
backend/common/tests/test_pat_auth.py
View file

@ -124,16 +124,30 @@ def test_cross_org_pat_cannot_read_other_orgs_leads(org_a, admin_profile):
if connection.vendor != "postgresql":
pytest.skip("RLS isolation is enforced by PostgreSQL; skipping on non-Postgres DB")
from common.tasks import clear_rls_context, set_rls_context
from leads.models import Lead
# Create org B and a lead in it directly, with org explicitly set.
#
# The context has to be set for the INSERT itself. `lead` carries an
# RLS insert-check policy, so a write with `app.current_org` empty is
# refused outright ("new row violates row-level security policy"), and
# this test then failed during its own setup. It went unnoticed because
# the only role that had ever run it was a Postgres superuser, which
# bypasses RLS: the guard was inert exactly where it was meant to bite.
# Cleared again afterwards so the request below is judged on the context
# the PAT's middleware sets, not one left behind here.
org_b = Org.objects.create(name="Cross-Org Isolation Org B")
lead_b = Lead.objects.create(
first_name="Hidden",
last_name="Lead-OrgB",
email="hidden_orgb@test.com",
org=org_b,
)
set_rls_context(org_b.id)
try:
lead_b = Lead.objects.create(
first_name="Hidden",
last_name="Lead-OrgB",
email="hidden_orgb@test.com",
org=org_b,
)
finally:
clear_rls_context()
# Mint a PAT for the org-A admin profile and call the leads list as that token.
raw, _ = PersonalAccessToken.generate(profile=admin_profile, name="cross-org-cli")

73
backend/common/tests/test_suite_collects_everything.py Normal file
View file

@ -0,0 +1,73 @@
"""A test file pytest never collects is not a test, and it reads like one.
`pytest.ini` sets ``python_files = tests.py test_*.py *_tests.py``. Nine files
matched none of those patterns, because they were named ``tests_*.py``: the
plural ``tests_`` prefix is one character away from the ``test_`` that gets
collected and a full word away from the ``*_tests.py`` suffix that also does.
They were not merely idle. Every one of them raised ``ImportError`` on import,
naming base classes and helpers that had been deleted (``AccountCreateTest``,
``CaseCreation``, ``ObjectsCreation``) or, in one case, the ``Company`` model
that became ``Org``. So the tree contained 29 test functions that could not run
and had not run for a long time, and `grep` still found "a celery task test"
for anyone who went looking for coverage of the mail tasks. That is worse than
an empty directory, which is why they were deleted rather than renamed.
This test guards the naming, not the deletion: it fails on a new file that
looks like a test and would be skipped by the collector.
"""
from __future__ import annotations
import fnmatch
from pathlib import Path
# Kept in sync with `python_files` in pytest.ini by hand. There is no public
# API to read it back from a running session, and duplicating three globs is
# cheaper than the failure this prevents.
COLLECTED_PATTERNS = ("tests.py", "test_*.py", "*_tests.py")
SKIP_DIRS = {".venv", "venv", "node_modules", "migrations", "htmlcov", "staticfiles"}
def _backend_root() -> Path:
return Path(__file__).resolve().parents[2]
def _looks_like_tests(name: str) -> bool:
"""Files a reader would take for a test suite."""
return name.startswith("test") or "_test" in name
def test_no_test_file_is_invisible_to_the_collector():
stranded = []
for path in _backend_root().rglob("*.py"):
if any(part in SKIP_DIRS for part in path.parts):
continue
name = path.name
if not _looks_like_tests(name):
continue
if any(fnmatch.fnmatch(name, pattern) for pattern in COLLECTED_PATTERNS):
continue
stranded.append(str(path.relative_to(_backend_root())))
assert stranded == [], (
"These files look like tests but pytest will not collect them "
f"(python_files = {' '.join(COLLECTED_PATTERNS)}). Rename to test_*.py "
"or delete them: " + ", ".join(stranded)
)
def test_the_sweep_actually_looks_at_something():
"""A guard that silently walks zero files passes forever.
The same trap as `common/tests/test_org_index_coverage.py`: without this,
a wrong root or an over-broad skip list turns the check above into a
green light that means nothing.
"""
seen = [
p
for p in _backend_root().rglob("test_*.py")
if not any(part in SKIP_DIRS for part in p.parts)
]
assert len(seen) > 50, f"only found {len(seen)} test files, the walk is wrong"

25
backend/common/tests/test_user_name_fallback.py
View file

@ -35,7 +35,24 @@ class TestUserNameAutoFill:
user.refresh_from_db()
assert user.name == ""
def test_long_email_local_part_is_truncated_to_255(self):
long_local = "a" * 300
user = User.objects.create(email=f"{long_local}@example.com")
assert len(user.name) == 255
def test_the_longest_storable_email_still_fits_the_name_column(self):
"""The `[:255]` in `User.save` is a belt, and it can never be reached.
`email` is an `EmailField`, so `varchar(254)`, which caps the local
part at 242 characters once `@example.com` is accounted for. `name` is
`varchar(255)`. A local part long enough to need truncating cannot be
stored in the first place.
This test used to assert the truncation directly, with a 300-character
local part and a 312-character email. SQLite does not enforce declared
string lengths, so it passed there and passed in CI; against
PostgreSQL the same line raises `StringDataRightTruncation` before
`name` is ever considered. It was pinning behaviour reachable only on
a database this project does not run in production.
"""
longest_local = "a" * (254 - len("@example.com"))
user = User.objects.create(email=f"{longest_local}@example.com")
assert len(user.name) == 242
user.refresh_from_db()
assert user.name == longest_local

44
backend/common/tests/test_users.py
View file

@ -8,6 +8,7 @@ import pytest
from rest_framework import status
from common.models import Address, Profile
from common.permissions import is_org_admin
@pytest.mark.django_db
@ -702,3 +703,46 @@ class TestProfilePrivilegeEscalation:
assert response.status_code == status.HTTP_200_OK
admin_profile.refresh_from_db()
assert admin_profile.role == "ADMIN"
def test_admin_cannot_mint_a_hidden_admin_via_the_flag(
self, admin_client, regular_user, user_profile
):
"""The grant that no screen could show and no screen could take back.
`is_organization_admin` was a writable privileged field and 43 checks
across the backend accepted it in place of `role`. No frontend surface
reads or writes the column, so a member carrying it appeared, on every
page an operator can open, to be an ordinary member, while accounts,
contacts, documents, team management and the org record treated them
as an admin. `role` is now the only input.
"""
response = admin_client.patch(
self._url(regular_user.id),
{"is_organization_admin": True},
format="json",
)
assert response.status_code == status.HTTP_200_OK
user_profile.refresh_from_db()
assert user_profile.is_organization_admin is False
assert user_profile.role == "USER"
def test_the_flag_grants_nothing_even_when_written_directly(
self, user_client, user_profile
):
"""Defence in depth for rows written before the field was closed.
A database may already hold `is_organization_admin=True` on a member,
from the old serializer or a direct write. Reading it back must not
reopen the endpoints the flag used to unlock, so this bypasses the
model entirely with a queryset update.
"""
Profile.objects.filter(pk=user_profile.pk).update(is_organization_admin=True)
# The roster is the check that mattered: admin-only, and one of the 43
# that used to read the flag.
assert user_client.get("/api/users/").status_code == status.HTTP_403_FORBIDDEN
user_profile.refresh_from_db()
assert user_profile.is_organization_admin is True
assert is_org_admin(user_profile) is False

268
backend/common/tests_celery_tasks.py
View file

@ -1,268 +0,0 @@
from django.test import TestCase
from django.test.utils import override_settings
from accounts.tests import AccountCreateTest
from cases.tests import CaseCreation
from common.models import User
from common.tasks import (
send_email_user_delete,
send_email_user_mentions,
send_email_user_status,
)
from common.tests import ObjectsCreation
from contacts.tests import ContactObjectsCreation
from invoices.tests import InvoiceCreateTest
from leads.tests import TestLeadModel
from opportunity.tests import OpportunityModel
from tasks.tests import TaskCreateTest
class TestCeleryTasks(ObjectsCreation, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_celery_tasks(self):
task = send_email_user_status.apply(
(
self.user1.id,
self.user.id,
),
)
self.assertEqual("SUCCESS", task.state)
self.user1.is_active = False
self.user1.has_sales_access = False
self.user1.has_marketing_access = True
self.user1.save()
task = send_email_user_status.apply(
(self.user1.id,),
)
self.assertEqual("SUCCESS", task.state)
self.user1.is_active = True
self.user1.has_sales_access = False
self.user1.has_marketing_access = False
self.user1.save()
task = send_email_user_status.apply(
(self.user1.id,),
)
self.assertEqual("SUCCESS", task.state)
task = send_email_user_delete.apply(
(self.user1.email,),
)
self.assertEqual("SUCCESS", task.state)
class TestUserMentionsForAccountComments(AccountCreateTest, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_user_mentions_for_account_comment(self):
self.user_comment = User.objects.create(
first_name="johnComment",
username="johnDoeComment",
email="johnDoeComment@example.com",
role="ADMIN",
)
self.user_comment.set_password("password")
self.user_comment.save()
self.comment.comment = f"content @{self.user_comment.username}"
self.comment.account = self.account
self.comment.save()
task = send_email_user_mentions.apply(
(
self.comment.id,
"accounts",
str(self.comment.org.id),
),
)
self.assertEqual("SUCCESS", task.state)
class TestUserMentionsForContactsComments(ContactObjectsCreation, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_user_mentions_for_contacts_comments(self):
self.user_comment = User.objects.create(
first_name="johnComment",
username="johnDoeComment",
email="johnDoeComment@example.com",
role="ADMIN",
)
self.user_comment.set_password("password")
self.user_comment.save()
self.comment.comment = f"content @{self.user_comment.username}"
self.comment.contact = self.contact
self.comment.save()
task = send_email_user_mentions.apply(
(
self.comment.id,
"contacts",
str(self.comment.org.id),
),
)
self.assertEqual("SUCCESS", task.state)
class TestUserMentionsForLeadsComments(TestLeadModel, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_user_mentions_for_leads_comments(self):
self.user_comment = User.objects.create(
first_name="johnComment",
username="johnDoeComment",
email="johnDoeComment@example.com",
role="ADMIN",
)
self.user_comment.set_password("password")
self.user_comment.save()
self.comment.comment = f"content @{self.user_comment.username}"
self.comment.lead = self.lead
self.comment.save()
task = send_email_user_mentions.apply(
(
self.comment.id,
"leads",
str(self.comment.org.id),
),
)
self.assertEqual("SUCCESS", task.state)
class TestUserMentionsForOpportunityComments(OpportunityModel, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_user_mentions_for_opportunity_comments(self):
self.user_comment = User.objects.create(
first_name="johnComment",
username="johnDoeComment",
email="johnDoeComment@example.com",
role="ADMIN",
)
self.user_comment.set_password("password")
self.user_comment.save()
self.comment.comment = f"content @{self.user_comment.username}"
self.comment.opportunity = self.opportunity
self.comment.save()
task = send_email_user_mentions.apply(
(
self.comment.id,
"opportunity",
str(self.comment.org.id),
),
)
self.assertEqual("SUCCESS", task.state)
class TestUserMentionsForCasesComments(CaseCreation, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_user_mentions_for_cases_comments(self):
self.user_comment = User.objects.create(
first_name="johnComment",
username="johnDoeComment",
email="johnDoeComment@example.com",
role="ADMIN",
)
self.user_comment.set_password("password")
self.user_comment.save()
self.comment.comment = f"content @{self.user_comment.username}"
self.comment.case = self.case
self.comment.save()
task = send_email_user_mentions.apply(
(
self.comment.id,
"cases",
str(self.comment.org.id),
),
)
self.assertEqual("SUCCESS", task.state)
class TestUserMentionsForTasksComments(TaskCreateTest, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_user_mentions_for_tasks_comments(self):
self.user_comment = User.objects.create(
first_name="johnComment",
username="johnDoeComment",
email="johnDoeComment@example.com",
role="ADMIN",
)
self.user_comment.set_password("password")
self.user_comment.save()
self.comment.comment = f"content @{self.user_comment.username}"
self.comment.task = self.task
self.comment.save()
task = send_email_user_mentions.apply(
(
self.comment.id,
"tasks",
str(self.comment.org.id),
),
)
self.assertEqual("SUCCESS", task.state)
class TestUserMentionsForInvoiceComments(InvoiceCreateTest, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_user_mentions_for_invoice_comments(self):
self.user_comment = User.objects.create(
first_name="johnComment",
username="johnDoeComment",
email="johnDoeComment@example.com",
role="ADMIN",
)
self.user_comment.set_password("password")
self.user_comment.save()
self.comment.comment = f"content @{self.user_comment.username}"
self.comment.invoice = self.invoice
self.comment.save()
task = send_email_user_mentions.apply(
(
self.comment.id,
"invoices",
str(self.comment.org.id),
),
)
self.assertEqual("SUCCESS", task.state)

14
backend/common/views/dashboard_views.py
View file

@ -14,7 +14,7 @@ from accounts.serializer import AccountSerializer
from cases.models import Case
from common import serializer, swagger_params
from common.models import Activity
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.utils import STAGES
from contacts.models import Contact
from contacts.serializer import ContactSerializer
@ -96,7 +96,7 @@ class ApiHomeView(APIView):
opportunities = Opportunity.objects.filter(org=org)
tasks = Task.objects.filter(org=org)
is_admin = profile.role == "ADMIN" or request.user.is_superuser
is_admin = is_org_admin(profile) or request.user.is_superuser
if not is_admin:
accounts = accounts.filter(
@ -291,7 +291,7 @@ class ApiHomeView(APIView):
.select_related("user", "user__user")
.order_by("-created_at")[:10]
)
context["activities"] = serializer.ActivitySerializer(
context["activities"] = serializer.DashboardActivitySerializer(
activities, many=True
).data
@ -348,7 +348,7 @@ class ApiTodayView(APIView):
today = now.date()
yesterday = today - timedelta(days=1)
week_end = today + timedelta(days=7)
is_admin = profile.role == "ADMIN" or request.user.is_superuser
is_admin = is_org_admin(profile) or request.user.is_superuser
org_currency = org.default_currency or "USD"
def mine(qs):
@ -607,7 +607,7 @@ class ActivityListView(APIView):
description="Filter by entity type (Account, Lead, Contact, etc.)",
),
],
responses={200: serializer.ActivitySerializer(many=True)},
responses={200: serializer.DashboardActivitySerializer(many=True)},
)
def get(self, request, *args, **kwargs):
if not request.profile:
@ -631,7 +631,9 @@ class ActivityListView(APIView):
activities = queryset.select_related("user", "user__user")[:limit]
# Serialize
activities_data = serializer.ActivitySerializer(activities, many=True).data
activities_data = serializer.DashboardActivitySerializer(
activities, many=True
).data
return Response(
{

12
backend/common/views/document_views.py
View file

@ -10,7 +10,7 @@ from rest_framework.views import APIView
from common import swagger_params
from common.models import Document, Profile, Teams
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.validators import payload_id_list
from common.serializer import (
DocumentCreateSerializer,
@ -64,7 +64,7 @@ class DocumentListView(APIView, LimitOffsetPagination):
queryset = self.model.objects.filter(org=self.request.profile.org).order_by(
"-id"
)
if not (self.request.user.is_superuser or self.request.profile.role == "ADMIN"):
if not (self.request.user.is_superuser or is_org_admin(self.request.profile)):
queryset = queryset.filter(_visible_to(self.request.profile)).distinct()
request_post = params
@ -86,7 +86,7 @@ class DocumentListView(APIView, LimitOffsetPagination):
profile_list = Profile.objects.filter(
is_active=True, org=self.request.profile.org
)
if self.request.profile.role == "ADMIN" or self.request.profile.is_admin:
if is_org_admin(self.request.profile):
profiles = profile_list.order_by("user__email")
else:
profiles = profile_list.filter(role="ADMIN").order_by("user__email")
@ -234,7 +234,7 @@ class DocumentDetailView(APIView):
literally the same predicate the list uses. Doing it by hand was how
the two drifted apart in the first place.
"""
if self.request.profile.role == "ADMIN" or self.request.user.is_superuser:
if is_org_admin(self.request.profile) or self.request.user.is_superuser:
return True
return (
Document.objects.filter(pk=document.pk)
@ -250,7 +250,7 @@ class DocumentDetailView(APIView):
who uploaded it, or an admin. `created_by` is a User FK. Comparing it
to a Profile is the bug that made this branch unreachable.
"""
if self.request.profile.role == "ADMIN" or self.request.user.is_superuser:
if is_org_admin(self.request.profile) or self.request.user.is_superuser:
return True
return document.created_by_id == self.request.profile.user_id
@ -302,7 +302,7 @@ class DocumentDetailView(APIView):
if not self._may_read(self.object):
return self._forbidden()
profile_list = Profile.objects.filter(org=self.request.profile.org)
if request.profile.role == "ADMIN" or request.user.is_superuser:
if is_org_admin(request.profile) or request.user.is_superuser:
profiles = profile_list.order_by("user__email")
else:
profiles = profile_list.filter(role="ADMIN").order_by("user__email")

3
backend/common/views/org_settings_views.py
View file

@ -3,6 +3,7 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from common.permissions import is_org_admin
from common.serializer import OrgSettingsSerializer
@ -34,7 +35,7 @@ class OrgSettingsView(APIView):
{"error": "Organization context required"},
status=status.HTTP_400_BAD_REQUEST,
)
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Only admins can update organization settings"},
status=status.HTTP_403_FORBIDDEN,

15
backend/common/views/organization_views.py
View file

@ -9,6 +9,7 @@ from rest_framework_simplejwt.authentication import JWTAuthentication
from common import serializer, swagger_params
from common.models import Org, Profile
from common.permissions import is_org_admin
from common.serializer import (
CreateProfileSerializer,
OrganizationSerializer,
@ -41,8 +42,8 @@ class OrgProfileCreateView(APIView):
# now creating the profile
profile_obj = self.model2.objects.create(user=request.user, org=org_obj)
# now the current user is the admin of the newly created organisation
profile_obj.is_organization_admin = True
# now the current user is the admin of the newly created organisation.
# `is_organization_admin` is derived from `role` in `Profile.save`.
profile_obj.role = "ADMIN"
profile_obj.save()
@ -113,10 +114,7 @@ class OrgUpdateView(APIView):
status=status.HTTP_403_FORBIDDEN,
)
if verb == "update" and (
request.profile.role != "ADMIN"
and not request.profile.is_organization_admin
):
if verb == "update" and not is_org_admin(request.profile):
return None, Response(
{
"error": True,
@ -228,10 +226,7 @@ class OrgApiKeyView(APIView):
status=status.HTTP_400_BAD_REQUEST,
)
if (
request.profile.role != "ADMIN"
and not request.profile.is_organization_admin
):
if not is_org_admin(request.profile):
return None, Response(
{
"error": True,

10
backend/common/views/tags_views.py
View file

@ -13,7 +13,7 @@ from cases.models import Case
from common import swagger_params
from common.lookups import get_scoped_or_404
from common.models import APISettings, Tags
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.serializer import TagsSerializer
from contacts.models import Contact
from leads.models import Lead
@ -165,7 +165,7 @@ class TagsListView(APIView, LimitOffsetPagination):
def post(self, request, *args, **kwargs):
"""Create a new tag (admin only)."""
# Admin only for create
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": True, "errors": "Only admins can create tags"},
status=status.HTTP_403_FORBIDDEN,
@ -283,7 +283,7 @@ class TagsDetailView(APIView):
def put(self, request, pk, *args, **kwargs):
"""Update a tag (admin only)."""
# Admin only
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": True, "errors": "Only admins can update tags"},
status=status.HTTP_403_FORBIDDEN,
@ -365,7 +365,7 @@ class TagsDetailView(APIView):
def delete(self, request, pk, **kwargs):
"""Archive a tag - soft delete (admin only)."""
# Admin only
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": True, "errors": "Only admins can archive tags"},
status=status.HTTP_403_FORBIDDEN,
@ -413,7 +413,7 @@ class TagsRestoreView(APIView):
def post(self, request, pk, **kwargs):
"""Restore an archived tag (admin only)."""
# Admin only
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": True, "errors": "Only admins can restore tags"},
status=status.HTTP_403_FORBIDDEN,

13
backend/common/views/team_views.py
View file

@ -9,6 +9,7 @@ from rest_framework.views import APIView
from common import swagger_params
from common.lookups import get_scoped_or_404
from common.models import Profile, Teams
from common.permissions import is_org_admin
from common.serializer import (
TeamCreateSerializer,
TeamsSerializer,
@ -81,7 +82,7 @@ class TeamsListView(APIView, LimitOffsetPagination):
},
)
def get(self, *args, **kwargs):
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
return Response(
{
"error": True,
@ -108,7 +109,7 @@ class TeamsListView(APIView, LimitOffsetPagination):
},
)
def post(self, request, *args, **kwargs):
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
return Response(
{
"error": True,
@ -156,7 +157,7 @@ class TeamsDetailView(APIView):
},
)
def get(self, request, pk, **kwargs):
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
return Response(
{
"error": True,
@ -185,7 +186,7 @@ class TeamsDetailView(APIView):
},
)
def put(self, request, pk, *args, **kwargs):
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
return Response(
{
"error": True,
@ -243,7 +244,7 @@ class TeamsDetailView(APIView):
)
def patch(self, request, pk, *args, **kwargs):
"""Handle partial updates to a team."""
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
return Response(
{
"error": True,
@ -282,7 +283,7 @@ class TeamsDetailView(APIView):
},
)
def delete(self, request, pk, **kwargs):
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
return Response(
{
"error": True,

26
backend/common/views/user_views.py
View file

@ -98,7 +98,10 @@ class UsersListView(APIView, LimitOffsetPagination):
},
)
def post(self, request, format=None):
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
return Response(
{"error": True, "errors": "Permission Denied"},
status=status.HTTP_403_FORBIDDEN,
@ -192,7 +195,10 @@ class UsersListView(APIView, LimitOffsetPagination):
status=status.HTTP_400_BAD_REQUEST,
)
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
return Response(
{"error": True, "errors": "Permission Denied"},
status=status.HTTP_403_FORBIDDEN,
@ -313,8 +319,7 @@ class UserDetailView(APIView):
def get(self, request, pk, format=None):
profile_obj = self.get_object(pk)
if (
self.request.profile.role != "ADMIN"
and not self.request.profile.is_admin
not is_org_admin(self.request.profile)
and self.request.profile.id != profile_obj.id
):
return Response(
@ -368,7 +373,7 @@ class UserDetailView(APIView):
profile = self.get_object(pk)
address_obj = profile.address
if (
self.request.profile.role != "ADMIN"
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
and self.request.profile.id != profile.id
):
@ -445,7 +450,7 @@ class UserDetailView(APIView):
params = request.data
profile = self.get_object(pk)
if (
self.request.profile.role != "ADMIN"
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
and self.request.profile.id != profile.id
):
@ -504,7 +509,7 @@ class UserDetailView(APIView):
},
)
def delete(self, request, pk, format=None):
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
return Response(
{"error": True, "errors": "Permission Denied"},
status=status.HTTP_403_FORBIDDEN,
@ -543,7 +548,10 @@ class UserStatusView(APIView):
},
)
def post(self, request, pk, format=None):
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
return Response(
{
"error": True,
@ -570,7 +578,7 @@ class UserStatusView(APIView):
target_is_admin = is_org_admin(profile)
if target_is_admin and not (
profiles.filter(is_active=True)
.filter(Q(role="ADMIN") | Q(is_organization_admin=True))
.filter(role="ADMIN")
.exclude(pk=profile.pk)
.exists()
):

4
backend/contacts/tasks.py
View file

@ -1,8 +1,8 @@
from celery import shared_task
from django.conf import settings
from django.core.mail import EmailMessage
from django.template.loader import render_to_string
from common.links import frontend_url
from common.models import Profile
from common.tasks import set_rls_context
from contacts.models import Contact
@ -20,7 +20,7 @@ def send_email_to_assigned_user(recipients, contact_id, org_id):
if profile:
recipients_list.append(profile.user.email)
context = {}
context["url"] = settings.DOMAIN_NAME
context["url"] = frontend_url(f"/contacts/{contact.id}")
context["user"] = profile.user
context["contact"] = contact
context["created_by"] = created_by

26
backend/contacts/tests_celery_tasks.py
View file

@ -1,26 +0,0 @@
from django.test import TestCase
from django.test.utils import override_settings
from contacts.tasks import send_email_to_assigned_user
from contacts.tests import ContactObjectsCreation
class TestCeleryTasks(ContactObjectsCreation, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_celery_tasks(self):
org_id = str(self.contact.org.id)
task = send_email_to_assigned_user.apply(
(
[
self.user.id,
self.user_contacts_mp.id,
],
self.contact.id,
org_id,
),
)
self.assertEqual("SUCCESS", task.state)

32
backend/contacts/views.py
View file

@ -15,7 +15,7 @@ from rest_framework.pagination import LimitOffsetPagination
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from rest_framework.views import APIView
from common.custom_fields import validate_payload as validate_custom_fields_payload
@ -62,7 +62,7 @@ class ContactsListView(APIView, LimitOffsetPagination):
.select_related("account")
.prefetch_related("account_contacts", "assigned_to__user", "teams", "tags")
)
if self.request.profile.role != "ADMIN" and not self.request.profile.is_admin:
if not is_org_admin(self.request.profile):
queryset = queryset.filter(
Q(assigned_to__in=[self.request.profile])
| Q(created_by=self.request.profile.user)
@ -319,7 +319,7 @@ class ContactDetailView(APIView):
that is true for viewing but false for editing.
"""
profile = self.request.profile
if profile.role == "ADMIN" or profile.is_admin:
if is_org_admin(profile):
return
if profile.user_id == contact.created_by_id:
return
@ -532,7 +532,7 @@ class ContactDetailView(APIView):
assigned_dict["name"] = each.user.email
assigned_data.append(assigned_dict)
if self.request.profile.is_admin or self.request.profile.role == "ADMIN":
if is_org_admin(self.request.profile):
users_mention = list(
Profile.objects.filter(is_active=True, org=request.profile.org).values(
"user__email"
@ -631,8 +631,7 @@ class ContactDetailView(APIView):
# work on a contact, only an admin or the person who entered it may
# destroy the record. This comparison was already the right one.
if (
self.request.profile.role != "ADMIN"
and not self.request.profile.is_admin
not is_org_admin(self.request.profile)
and self.request.profile.user_id != self.object.created_by_id
):
return Response(
@ -857,11 +856,7 @@ class ContactCommentView(APIView):
def put(self, request, pk, format=None):
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == obj.commented_by
):
if is_org_admin(request.profile) or request.profile == obj.commented_by:
serializer = CommentSerializer(obj, data=params)
if serializer.is_valid():
serializer.save()
@ -900,11 +895,7 @@ class ContactCommentView(APIView):
"""Handle partial updates to a comment."""
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == obj.commented_by
):
if is_org_admin(request.profile) or request.profile == obj.commented_by:
serializer = CommentSerializer(obj, data=params, partial=True)
if serializer.is_valid():
serializer.save()
@ -939,11 +930,7 @@ class ContactCommentView(APIView):
)
def delete(self, request, pk, format=None):
self.object = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == self.object.commented_by
):
if is_org_admin(request.profile) or request.profile == self.object.commented_by:
self.object.delete()
return Response(
{"error": False, "message": "Comment Deleted Successfully"},
@ -994,8 +981,7 @@ class ContactAttachmentView(APIView):
except (DjangoValidationError, ValueError):
raise Http404("No such attachment.")
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
is_org_admin(request.profile)
or request.profile.user_id == self.object.created_by_id
):
self.object.delete()

45
backend/crm/test_settings_postgres.py Normal file
View file

@ -0,0 +1,45 @@
"""Test settings that run the suite against a real PostgreSQL.
`crm.test_settings` uses in-memory SQLite, which is fast and is what CI has
always run. It also cannot see an entire class of defect, and that class keeps
producing real bugs:
* an org name longer than `varchar(100)` raised `DataError` and an uncaught
500 in production, while all 3113 SQLite tests passed (SQLite ignores
declared string lengths entirely)
* six pipeline tables shipped with no RLS policy, invisible because RLS is a
PostgreSQL feature and every test skipped it
* the `postgres_only` marker exists precisely for tests that need the real
database, and CI ran with `-m "not postgres_only"`, so none of them had ever
executed there
This module changes only the database. Everything else, the eager-broker
Celery config and the cheap password hasher, is inherited from
`crm.test_settings` so the two runs differ in exactly one variable.
Connection details come from the same `DBNAME`/`DBUSER`/`DBHOST`/`DBPORT`
environment variables the application settings read, so this needs no
CI-specific values baked in.
**The role must not be a superuser.** A PostgreSQL superuser bypasses RLS
outright, and `get_enable_policy_sql` sets `FORCE ROW LEVEL SECURITY` so even
the table owner is subject to its policies. Running these tests as `postgres`
would leave every isolation assertion vacuously true, which is the failure mode
described in `RLS_SETUP.md`.
"""
import os
from crm.test_settings import * # noqa: F401, F403
DATABASES = {
"default": {
"ENGINE": "django.db.backends.postgresql",
"NAME": os.environ.get("DBNAME", "crm_db"),
"USER": os.environ.get("DBUSER", "postgres"),
"PASSWORD": os.environ.get("DBPASSWORD", "postgres"),
"HOST": os.environ.get("DBHOST", "localhost"),
"PORT": os.environ.get("DBPORT", "5432"),
"CONN_MAX_AGE": 0,
}
}

41
backend/invoices/api_views.py
View file

@ -19,7 +19,7 @@ from rest_framework.views import APIView
from common.custom_fields import validate_payload as validate_custom_fields_payload
from common.models import Attachments, Comment, CustomFieldDefinition
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.validators import uuid_param
from common.serializer import (
AttachmentsSerializer,
@ -87,7 +87,6 @@ class InvoiceListView(APIView, LimitOffsetPagination):
def get_queryset(self):
"""Get invoices filtered by org and user permissions"""
org = self.request.profile.org
role = self.request.profile.role
queryset = (
self.model.objects.filter(org=org)
@ -96,7 +95,10 @@ class InvoiceListView(APIView, LimitOffsetPagination):
)
# Non-admin users can only see their own or assigned invoices
if role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
queryset = queryset.filter(
Q(created_by=self.request.profile.user)
| Q(assigned_to=self.request.profile)
@ -885,7 +887,7 @@ class ProductListView(APIView, LimitOffsetPagination):
# need it to build line items), but changing it is an admin act, the
# same posture Organization and Team settings take. A non-admin who
# curls this endpoint is refused here, not just hidden from in the UI.
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{
"error": True,
@ -934,7 +936,7 @@ class ProductDetailView(APIView):
def _require_admin(self, request):
"""Editing the shared catalog is admin-only; see ProductListView.post."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{
"error": True,
@ -1007,13 +1009,15 @@ class EstimateListView(APIView, LimitOffsetPagination):
def get_queryset(self):
org = self.request.profile.org
role = self.request.profile.role
queryset = Estimate.objects.filter(org=org).select_related(
"account", "contact", "opportunity", "created_by", "converted_to_invoice"
)
if role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
# created_by is a User FK; comparing it to a Profile does not just
# silently fail here, it reaches the DB as Q(created_by=<Profile>)
# and raised ValueError -- a 500 on every non-admin estimate list.
@ -1349,7 +1353,7 @@ class RecurringInvoiceListView(APIView, LimitOffsetPagination):
# same scoping as the invoice and estimate lists. created_by is a User
# FK, so it is matched against request.profile.user; assigned_to holds
# Profiles, matched against request.profile.
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
queryset = queryset.filter(
Q(created_by=request.profile.user) | Q(assigned_to=request.profile)
).distinct()
@ -1548,7 +1552,7 @@ def _forbid_non_admin_template(
reason, so it is the same gate rather than a second copy of the role check
that could drift away from this one.
"""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": True, "message": message},
status=status.HTTP_403_FORBIDDEN,
@ -1778,8 +1782,9 @@ class InvoiceCommentDetailView(APIView):
)
# Only comment author or admin can edit
role = request.profile.role
if comment.commented_by != request.profile and role != "ADMIN":
if comment.commented_by != request.profile and not is_org_admin(
request.profile
):
return Response(
{"error": True, "message": "Permission denied"},
status=status.HTTP_403_FORBIDDEN,
@ -1805,8 +1810,9 @@ class InvoiceCommentDetailView(APIView):
status=status.HTTP_404_NOT_FOUND,
)
role = request.profile.role
if comment.commented_by != request.profile and role != "ADMIN":
if comment.commented_by != request.profile and not is_org_admin(
request.profile
):
return Response(
{"error": True, "message": "Permission denied"},
status=status.HTTP_403_FORBIDDEN,
@ -1873,8 +1879,9 @@ class InvoiceAttachmentDetailView(APIView):
# created_by is a User FK, so it must be compared against request.user --
# comparing it to request.profile is always unequal and locks the
# uploader out of their own attachment.
role = request.profile.role
if attachment.created_by_id != request.user.id and role != "ADMIN":
if attachment.created_by_id != request.user.id and not is_org_admin(
request.profile
):
return Response(
{"error": True, "message": "Permission denied"},
status=status.HTTP_403_FORBIDDEN,
@ -1904,7 +1911,7 @@ def _forbid_non_admin_reports(request):
admin-only (Django superusers included), the same bar as the org and team
settings.
"""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{
"error": True,
@ -2303,7 +2310,7 @@ class InvoiceFromOpportunityView(APIView):
)
# Check permission
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
if not (
(request.profile.user == opportunity.created_by)
or (request.profile in opportunity.assigned_to.all())

4
backend/invoices/permissions.py
View file

@ -21,6 +21,8 @@ invoice or estimate endpoint through :func:`get_invoice_or_error` /
from rest_framework import status
from rest_framework.response import Response
from common.permissions import is_org_admin
from invoices.models import Estimate, Invoice, RecurringInvoice
INVOICE_NOT_FOUND = "Invoice not found"
@ -49,7 +51,7 @@ def has_object_access(request, obj):
"""
profile = request.profile
if profile.role == "ADMIN" or request.user.is_superuser:
if is_org_admin(profile) or request.user.is_superuser:
return True
if obj.created_by_id and obj.created_by_id == request.user.id:

25
backend/invoices/tests_celery_tasks.py
View file

@ -1,25 +0,0 @@
from django.test import TestCase
from django.test.utils import override_settings
from invoices.tasks import send_email, send_invoice_email, send_invoice_email_cancel
from invoices.tests import InvoiceCreateTest
class TestSendMailOnInvoiceCreationTask(InvoiceCreateTest, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_send_mail_on_invoice_creation_task(self):
org_id = str(self.invoice.org.id)
task = send_email.apply(
(self.invoice.id, [self.user.id, self.user1.id], org_id)
)
self.assertEqual("SUCCESS", task.state)
task = send_invoice_email.apply((self.invoice.id, org_id))
self.assertEqual("SUCCESS", task.state)
task = send_invoice_email_cancel.apply((self.invoice.id, org_id))
self.assertEqual("SUCCESS", task.state)

521
backend/invoices/tests_legacy.py
View file

@ -1,521 +0,0 @@
from datetime import datetime, timedelta
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import TestCase
from django.urls import reverse
from accounts.models import Account
from common.models import Address, Attachments, Comment, Company, Teams, User
from invoices.models import Invoice, InvoiceHistory
class InvoiceCreateTest:
def setUp(self):
self.company, _ = Company.objects.get_or_create(
name="test company",
address="IN",
sub_domain="test",
country="IN",
)
self.user = User.objects.create(
first_name="johnInvoice",
username="johnDoeInvoice",
email="johnDoeInvoice@example.com",
role="ADMIN",
company=self.company,
)
self.user.set_password("password")
self.user.save()
self.user1 = User.objects.create(
first_name="janeInvoice",
username="janeDoeInvoice",
email="janeDoeInvoice@example.com",
role="USER",
has_sales_access=True,
company=self.company,
)
self.user1.set_password("password")
self.user1.save()
self.user2 = User.objects.create(
first_name="joeInvoice",
username="joeInvoice",
email="joeInvoice@example.com",
role="USER",
has_sales_access=True,
company=self.company,
)
self.user2.set_password("password")
self.user2.save()
self.team_dev = Teams.objects.create(name="invoices teams")
self.team_dev.users.add(self.user2.id)
self.account = Account.objects.create(
name="john invoice",
email="johnDoeInvoice@example.com",
phone="123456789",
billing_address_line="",
billing_street="street name",
billing_city="city name",
billing_state="state",
billing_postcode="1234",
billing_country="US",
website="www.example.como",
created_by=self.user,
status="open",
industry="SOFTWARE",
description="Testing",
)
self.account.assigned_to.add(self.user1.id)
self.from_address = Address.objects.create(
street="from street number",
city="from city",
state="from state",
postcode=12346,
country="IN",
)
self.to_address = Address.objects.create(
street="to street number",
city="to city",
state="to state",
postcode=12346,
country="IN",
)
self.invoice = Invoice.objects.create(
invoice_title="invoice title",
invoice_number="invoice number",
currency="USD",
email="invoiceTitle@email.com",
due_date=(datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"),
total_amount="1000",
created_by=self.user,
from_address=self.from_address,
to_address=self.to_address,
)
self.invoice.assigned_to.add(self.user1.id)
self.invoice.accounts.add(self.account.id)
self.invoice_history = InvoiceHistory.objects.create(
invoice_title="invoice title",
invoice_number="invoice number",
currency="USD",
email="invoiceTitle@email.com",
due_date=(datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"),
total_amount="1000",
from_address=self.from_address,
to_address=self.to_address,
invoice=self.invoice,
)
self.invoice_1 = Invoice.objects.create(
invoice_title="invoice title",
invoice_number="invoice_1 number",
currency="USD",
email="invoiceTitle@email.com",
due_date=(datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"),
total_amount="1000",
created_by=self.user1,
)
self.invoice_1.assigned_to.add(self.user2.id)
self.invoice_1.assigned_to.add(self.user1.id)
self.invoice_1.accounts.add(self.account.id)
self.comment = Comment.objects.create(
comment="test comment", invoice=self.invoice, commented_by=self.user
)
self.attachment = Attachments.objects.create(
attachment="image.png", invoice=self.invoice, created_by=self.user
)
class InvoiceListTestCase(InvoiceCreateTest, TestCase):
def test_invoices_list(self):
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.get(reverse("invoices:invoices_list"))
self.assertEqual(response.status_code, 200)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.get(reverse("invoices:invoices_list"))
self.assertEqual(response.status_code, 200)
data = {
"invoice_title_number": "title",
"created_by": self.user.id,
"assigned_to": self.user1.id,
"status": "Draft",
"total_amount": "1000",
}
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:invoices_list"), data)
self.assertEqual(response.status_code, 200)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:invoices_list"), data)
self.assertEqual(response.status_code, 200)
self.assertEqual(str(self.invoice), "invoice number")
self.assertEqual(str(self.invoice.formatted_rate()), "0 USD")
self.assertEqual(str(self.invoice.formatted_total_quantity()), "0 Hours")
self.assertEqual(str(self.invoice_history), "invoice number")
self.assertEqual(str(self.invoice_history.formatted_rate()), "0 USD")
self.assertEqual(
str(self.invoice_history.formatted_total_quantity()), "0 Hours"
)
self.assertEqual(str(self.invoice_history.formatted_total_amount()), "USD 1000")
# self.assertTrue(
# self.invoice_history.created_on_arrow in ["just now" or "seconds ago"]
# )
class InvoiceAddTestCase(InvoiceCreateTest, TestCase):
def test_invoices_create(self):
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.get(reverse("invoices:invoices_create"))
self.assertEqual(response.status_code, 200)
data = {
"invoice_title": "invoice title create",
"status": "Draft",
"invoice_number": "invoice number",
"currency": "INR",
"email": "invoice@example.com",
"due_data": (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"),
"total_amount": "1234",
"teams": self.team_dev.id,
"accounts": self.account.id,
"assigned_to": self.user1.id,
}
response = self.client.post(reverse("invoices:invoices_create"), data)
self.assertEqual(response.status_code, 200)
data = {
"invoice_title": "invoice title",
"status": "Draft",
"invoice_number": "INV123",
"currency": "INR",
"email": "invoice@example.com",
"due_date": (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"),
"total_amount": "1234",
"teams": self.team_dev.id,
"accounts": self.account.id,
"assigned_to": self.user1.id,
"quantity": 0,
}
response = self.client.post(reverse("invoices:invoices_create"), data)
self.assertEqual(response.status_code, 200)
data = {
"invoice_title": "invoice title",
"status": "Draft",
"invoice_number": "INV1234",
"currency": "INR",
"email": "invoice@example.com",
"due_date": (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"),
"total_amount": "1234",
"teams": self.team_dev.id,
"accounts": self.account.id,
"assigned_to": self.user1.id,
"quantity": 0,
"from_account": self.account.id,
}
response = self.client.post(reverse("invoices:invoices_create"), data)
self.assertEqual(response.status_code, 200)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.get(reverse("invoices:invoices_list"))
self.assertEqual(response.status_code, 200)
class InvoiceDetailTestCase(InvoiceCreateTest, TestCase):
def test_invoices_detail(self):
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_details", args=(self.invoice.id,))
)
self.assertEqual(response.status_code, 200)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_details", args=(self.invoice.id,))
)
self.assertEqual(response.status_code, 200)
response = self.client.get(
reverse("invoices:invoice_details", args=(self.invoice_1.id,))
)
self.assertEqual(response.status_code, 200)
self.client.login(email="joeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_details", args=(self.invoice.id,))
)
self.assertEqual(response.status_code, 403)
class InvoiceEditTestCase(InvoiceCreateTest, TestCase):
def test_invoices_edit(self):
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_edit", args=(self.invoice.id,))
)
self.assertEqual(response.status_code, 200)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_edit", args=(self.invoice_1.id,))
)
self.assertEqual(response.status_code, 200)
self.client.login(email="joeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_edit", args=(self.invoice.id,))
)
self.assertEqual(response.status_code, 403)
data = {
"invoice_title": "invoice title",
"status": "Draft",
"invoice_number": "INV1234",
"currency": "INR",
"email": "invoice@example.com",
"due_date": (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"),
"total_amount": "1234",
"teams": self.team_dev.id,
"accounts": self.account.id,
"assigned_to": self.user1.id,
"quantity": 0,
"from_account": self.account.id,
"from-address_line": "",
"from-street": "",
"from-city": "",
"from-state": "",
"from-postcode": "",
"from-country": "",
"to-address_line": "",
"to-street": "",
"to-city": "",
"to-state": "",
"to-postcode": "",
"to-country": "",
}
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.post(
reverse("invoices:invoice_edit", args=(self.invoice.id,)), data
)
self.assertEqual(response.status_code, 200)
data.pop("from_account")
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.post(
reverse("invoices:invoice_edit", args=(self.invoice.id,)), data
)
self.assertEqual(response.status_code, 200)
data.pop("invoice_number")
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.post(
reverse("invoices:invoice_edit", args=(self.invoice.id,)), data
)
self.assertEqual(response.status_code, 200)
class InvoiceSendMailTestCase(InvoiceCreateTest, TestCase):
def test_invoices_send_mail(self):
self.client.login(email="joeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_send_mail", args=(self.invoice_1.id,))
)
self.assertEqual(response.status_code, 403)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_send_mail", args=(self.invoice_1.id,))
)
self.assertEqual(response.status_code, 302)
class InvoiceChangeStatusPaidTestCase(InvoiceCreateTest, TestCase):
def test_invoices_change_status_to_paid(self):
self.client.login(email="joeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_change_status_paid", args=(self.invoice_1.id,))
)
self.assertEqual(response.status_code, 403)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_change_status_paid", args=(self.invoice_1.id,))
)
self.assertEqual(response.status_code, 302)
class InvoiceChangeStatusCancelledTestCase(InvoiceCreateTest, TestCase):
def test_invoices_change_status_to_cancelled(self):
self.client.login(email="joeInvoice@example.com", password="password")
response = self.client.get(
reverse(
"invoices:invoice_change_status_cancelled", args=(self.invoice_1.id,)
)
)
self.assertEqual(response.status_code, 403)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.get(
reverse(
"invoices:invoice_change_status_cancelled", args=(self.invoice_1.id,)
)
)
self.assertEqual(response.status_code, 302)
class InvoiceDownloadTestCase(InvoiceCreateTest, TestCase):
def test_invoices_download(self):
self.client.login(email="joeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_download", args=(self.invoice.id,))
)
self.assertEqual(response.status_code, 403)
# self.client.login(email='johnDoeInvoice@example.com',
# password='password')
# response = self.client.get(
# reverse('invoices:invoice_download', args=(self.invoice_1.id,)))
# self.assertEqual(response.status_code, 200)
class AddCommentTestCase(InvoiceCreateTest, TestCase):
def test_invoice_add_comment(self):
self.client.login(email="johnDoeInvoice@example.com", password="password")
data = {
"comment": "",
"invoice_id": self.invoice.id,
}
response = self.client.post(reverse("invoices:add_comment"), data)
self.assertEqual(response.status_code, 200)
data = {
"comment": "test comment invoice",
"invoice_id": self.invoice.id,
}
response = self.client.post(reverse("invoices:add_comment"), data)
self.assertEqual(response.status_code, 200)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:add_comment"), data)
self.assertEqual(response.status_code, 200)
class UpdateCommentTestCase(InvoiceCreateTest, TestCase):
def test_invoice_update_comment(self):
self.client.login(email="johnDoeInvoice@example.com", password="password")
data = {
"commentid": self.comment.id,
"invoice_id": self.invoice.id,
"comment": "",
}
response = self.client.post(reverse("invoices:edit_comment"), data)
self.assertEqual(response.status_code, 200)
data = {
"comment": "test comment",
"commentid": self.comment.id,
"invoice_id": self.invoice.id,
}
response = self.client.post(reverse("invoices:edit_comment"), data)
self.assertEqual(response.status_code, 200)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:edit_comment"), data)
self.assertEqual(response.status_code, 200)
class DeleteCommentTestCase(InvoiceCreateTest, TestCase):
def test_invoice_delete_comment(self):
data = {
"comment_id": self.comment.id,
}
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:remove_comment"), data)
self.assertEqual(response.status_code, 200)
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:remove_comment"), data)
self.assertEqual(response.status_code, 200)
class AddAttachmentTestCase(InvoiceCreateTest, TestCase):
def test_invoice_add_attachment(self):
data = {
"attachment": SimpleUploadedFile(
"file_name.txt", bytes("file contents.", "utf-8")
),
"invoice_id": self.invoice.id,
}
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:add_attachment"), data)
self.assertEqual(response.status_code, 200)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:add_attachment"), data)
self.assertEqual(response.status_code, 200)
data = {"attachment": "", "invoice_id": self.invoice.id}
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:add_attachment"), data)
self.assertEqual(response.status_code, 200)
class DeleteAttachmentTestCase(InvoiceCreateTest, TestCase):
def test_invoice_delete_attachment(self):
data = {
"attachment_id": self.attachment.id,
}
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:remove_attachment"), data)
self.assertEqual(response.status_code, 200)
self.client.login(email="johnDoeInvoice@example.com", password="password")
response = self.client.post(reverse("invoices:remove_attachment"), data)
self.assertEqual(response.status_code, 200)
class InvoiceDeleteTestCase(InvoiceCreateTest, TestCase):
def test_invoices_delete(self):
self.client.login(email="joeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_delete", args=(self.invoice_1.id,))
)
self.assertEqual(response.status_code, 403)
self.client.login(email="janeDoeInvoice@example.com", password="password")
response = self.client.get(
reverse("invoices:invoice_delete", args=(self.invoice_1.id,))
)
self.assertEqual(response.status_code, 302)
self.client.login(email="johnDoeInvoice@example.com", password="password")
self.invoice.status = "Sent"
self.invoice.is_email_sent = False
self.invoice.save()
self.assertEqual(self.invoice.is_sent(), True)
self.invoice.status = "Sent"
self.invoice.is_email_sent = True
self.invoice.save()
self.assertEqual(self.invoice.is_resent(), True)
self.assertEqual(self.invoice.is_draft(), False)
self.invoice.status = "Paid"
self.invoice.save()
self.assertEqual(self.invoice.is_paid_or_cancelled(), True)
response = self.client.get(
reverse("invoices:invoice_delete", args=(self.invoice.id,))
+ f"?view_account={self.account.id}"
)
self.assertEqual(response.status_code, 302)

21
backend/leads/migrations/0016_alter_leadpipeline_is_default.py Normal file
View file

@ -0,0 +1,21 @@
# Generated by Django 6.0.7 on 2026-08-05 04:54
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("leads", "0015_alter_lead_is_sample"),
]
operations = [
migrations.AlterField(
model_name="leadpipeline",
name="is_default",
field=models.BooleanField(
default=False,
help_text="Sorts this pipeline first in listings (see Meta.ordering) and is limited to one per org. It does NOT route new records: a lead created without a stage keeps stage=NULL, which is the supported status-based kanban mode.",
),
),
]

7
backend/leads/models.py
View file

@ -239,7 +239,12 @@ class LeadPipeline(BaseModel):
)
is_default = models.BooleanField(
default=False,
help_text="If true, new leads without explicit pipeline go here",
help_text=(
"Sorts this pipeline first in listings (see Meta.ordering) and is "
"limited to one per org. It does NOT route new records: a lead "
"created without a stage keeps stage=NULL, which is the supported "
"status-based kanban mode."
),
)
is_active = models.BooleanField(default=True)

24
backend/leads/tasks.py
View file

@ -11,6 +11,7 @@ from django.template.loader import render_to_string
logger = logging.getLogger(__name__)
from accounts.models import Account
from common.links import frontend_url
from common.models import Org, Profile
from common.tasks import set_rls_context
from leads.models import Lead
@ -58,7 +59,7 @@ def send_email(
@shared_task
def send_lead_assigned_emails(lead_id, new_assigned_to_list, site_address, org_id):
def send_lead_assigned_emails(lead_id, new_assigned_to_list, org_id):
set_rls_context(org_id)
lead_instance = Lead.objects.filter(
~Q(status="converted"), pk=lead_id, is_active=True
@ -71,12 +72,21 @@ def send_lead_assigned_emails(lead_id, new_assigned_to_list, site_address, org_i
from_email = settings.DEFAULT_FROM_EMAIL
template_name = "assigned_to/leads_assigned.html"
url = site_address
url += "/leads/" + str(lead_instance.id) + "/view/"
# The `site_address` argument this used to take was built by its one caller
# from `request.META["HTTP_HOST"]`, on an endpoint an unauthenticated web
# form posts to. That named the API host, where `/leads/<id>` does not
# exist, and it put a client-supplied value into the link of an email this
# system sends to its own users.
context = {
"lead_instance": lead_instance,
"lead_detail_url": url,
# `lead`, not `lead_instance`: the template's two senders used
# different names for the same object and it hedged with
# `{{ lead.title|default:lead_instance }}`. Django resolves a filter's
# argument eagerly, so that expression raised `VariableDoesNotExist`
# out of `render_to_string` for whichever sender supplied `lead`,
# which is the one that runs on an ordinary assignment. That email
# never rendered, let alone sent.
"lead": lead_instance,
"url": frontend_url(f"/leads/{lead_instance.id}"),
}
mail_kwargs = {"subject": subject, "from_email": from_email}
for profile in users:
@ -101,7 +111,7 @@ def send_email_to_assigned_user(recipients, lead_id, org_id, source=""):
if profile:
recipients_list.append(profile.user.email)
context = {}
context["url"] = settings.DOMAIN_NAME
context["url"] = frontend_url(f"/leads/{lead.id}")
context["user"] = profile.user
context["lead"] = lead
context["created_by"] = created_by

138
backend/leads/tests_celery_tasks.py
View file

@ -1,138 +0,0 @@
from django.test import TestCase
from django.test.utils import override_settings
from leads.tasks import (
create_lead_from_file,
send_email,
send_email_to_assigned_user,
send_lead_assigned_emails,
)
from leads.tests import TestLeadModel
class TestCeleryTasks(TestLeadModel, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_celery_tasks(self):
org_id = str(self.lead.org.id)
task = send_email_to_assigned_user.apply(
(
[
self.user.id,
self.user1.id,
],
self.lead.id,
org_id,
),
)
self.assertEqual("SUCCESS", task.state)
task = send_lead_assigned_emails.apply(
(
self.lead.id,
[
self.user.id,
self.user1.id,
self.user2.id,
],
"https://www.example.com",
org_id,
),
)
self.assertEqual("SUCCESS", task.state)
task = send_email.apply(
(
"mail subject",
"html content",
),
{
"recipients": [
self.user.id,
self.user1.id,
self.user2.id,
],
},
)
self.assertEqual("SUCCESS", task.state)
org_id1 = str(self.lead1.org.id)
task = send_lead_assigned_emails.apply(
(
self.lead1.id,
[
self.user.id,
self.user1.id,
self.user2.id,
],
"https://www.example.com",
org_id1,
),
)
self.assertEqual("SUCCESS", task.state)
valid_rows = [
{
"title": "lead1 csv",
"first name": "john",
"last name": "doe",
"website": "www.example.com",
"phone": "911234567890",
"email": "user1@email.com",
"address": "address for lead1",
},
{
"title": "lead2 csv",
"first name": "jane",
"last name": "doe",
"website": "www.website.com",
"phone": "911234567891",
"email": "user2@email.com",
"address": "address for lead2",
},
{
"title": "lead3 csv",
"first name": "joe",
"last name": "doe",
"website": "www.test.com",
"phone": "911234567892",
"email": "user3@email.com",
"address": "address for lead3",
},
{
"title": "lead4 csv",
"first name": "john",
"last name": "doe",
"website": "www.sample.com",
"phone": "911234567893",
"email": "user4@email.com",
"address": "address for lead4",
},
]
invalid_rows = [
{
"title": "lead5 csv",
"first name": "joe",
"last name": "doe",
"website": "www.test.com",
"phone": "911234567892",
"email": "useremail.com",
"address": "address for lead3",
},
{
"title": "lead6 csv",
"first name": "john",
"last name": "doe",
"website": "www.sample.com",
"phone": "911234567893",
"email": "user4@email",
"address": "address for lead4",
},
]
task = create_lead_from_file.apply(
(valid_rows, invalid_rows, self.user.id, "example.com", org_id),
)
self.assertEqual("SUCCESS", task.state)

20
backend/leads/views/kanban_views.py
View file

@ -14,7 +14,7 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.validators import uuid_param
from common.utils import LEAD_STATUS
from leads.models import Lead, LeadPipeline, LeadStage
@ -83,7 +83,7 @@ class LeadKanbanView(APIView):
)
# Apply permission filtering
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
queryset = queryset.filter(
Q(assigned_to=request.profile) | Q(created_by=request.profile.user)
)
@ -224,7 +224,7 @@ class LeadMoveView(APIView):
lead = get_object_or_404(Lead, pk=pk, org=org)
# Permission check
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
if not (
request.profile.user == lead.created_by
or request.profile in lead.assigned_to.all()
@ -386,7 +386,7 @@ class LeadPipelineListCreateView(APIView):
org = request.profile.org
# Only admins can create pipelines
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Only admins can create pipelines"},
status=status.HTTP_403_FORBIDDEN,
@ -479,7 +479,7 @@ class LeadPipelineDetailView(APIView):
)
def put(self, request, pk):
"""Update pipeline."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -499,7 +499,7 @@ class LeadPipelineDetailView(APIView):
@extend_schema(tags=["Lead Pipelines"], responses={204: None})
def delete(self, request, pk):
"""Delete pipeline (soft delete by setting is_active=False)."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -533,7 +533,7 @@ class LeadStageCreateView(APIView):
)
def post(self, request, pipeline_pk):
"""Add a new stage to pipeline."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -564,7 +564,7 @@ class LeadStageDetailView(APIView):
)
def put(self, request, pk):
"""Update stage."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -584,7 +584,7 @@ class LeadStageDetailView(APIView):
@extend_schema(tags=["Lead Stages"], responses={204: None})
def delete(self, request, pk):
"""Delete stage."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -620,7 +620,7 @@ class LeadStageReorderView(APIView):
@transaction.atomic
def post(self, request, pipeline_pk):
"""Reorder stages by providing ordered list of stage IDs."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)

18
backend/leads/views/lead_interactions.py
View file

@ -6,7 +6,7 @@ from rest_framework.views import APIView
from common.lookups import get_scoped_or_404
from common.models import APISettings, Attachments, Comment
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.serializer import LeadCommentSerializer
from contacts.models import Contact
from leads import swagger_params
@ -119,7 +119,7 @@ class LeadCommentView(APIView):
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
is_org_admin(request.profile)
or request.user.is_superuser
or request.profile == obj.commented_by
):
@ -162,7 +162,7 @@ class LeadCommentView(APIView):
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
is_org_admin(request.profile)
or request.user.is_superuser
or request.profile == obj.commented_by
):
@ -201,7 +201,7 @@ class LeadCommentView(APIView):
def delete(self, request, pk, format=None):
self.object = self.get_object(pk)
if (
request.profile.role == "ADMIN"
is_org_admin(request.profile)
or request.user.is_superuser
or request.profile == self.object.commented_by
):
@ -245,7 +245,7 @@ class LeadAttachmentView(APIView):
# contacts, cases, tasks, tags and teams equivalents.
self.object = get_scoped_or_404(self.model, pk, request.profile.org)
if (
request.profile.role == "ADMIN"
is_org_admin(request.profile)
or request.user.is_superuser
or request.profile.user == self.object.created_by
):
@ -312,10 +312,12 @@ class CreateLeadFromSite(APIView):
org=api_setting.org,
)
lead.assigned_to.add(user)
# Send Email to Assigned Users
site_address = request.scheme + "://" + request.META["HTTP_HOST"]
# Send Email to Assigned Users. The task builds its own link from
# FRONTEND_URL; it used to be handed a base derived from this
# request's Host header, which named the API rather than the web
# app and let a caller of this endpoint influence the link.
send_lead_assigned_emails.delay(
lead.id, [user.id], site_address, str(api_setting.org.id)
lead.id, [user.id], str(api_setting.org.id)
)
# Create Contact
try:

25
backend/leads/views/lead_views.py
View file

@ -25,7 +25,7 @@ from common.models import (
Teams,
User,
)
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.serializer import (
AttachmentsSerializer,
CustomFieldDefinitionSerializer,
@ -97,7 +97,10 @@ class LeadListView(APIView, LimitOffsetPagination):
"assigned_to",
)
).order_by("-id")
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
queryset = queryset.filter(
Q(assigned_to__in=[self.request.profile])
| Q(created_by=self.request.profile.user)
@ -422,7 +425,7 @@ class LeadDetailView(APIView):
from in there was wrapped in a second one and rendered as a 500 instead
of the intended 403.
"""
if self.request.profile.role == "ADMIN" or self.request.user.is_superuser:
if is_org_admin(self.request.profile) or self.request.user.is_superuser:
return
allowed = {profile.id for profile in self.lead_obj.assigned_to.all()}
if self.request.profile.user_id == self.lead_obj.created_by_id:
@ -448,7 +451,7 @@ class LeadDetailView(APIView):
assigned_dict["name"] = each.user.email
assigned_data.append(assigned_dict)
if self.request.user.is_superuser or self.request.profile.role == "ADMIN":
if self.request.user.is_superuser or is_org_admin(self.request.profile):
users_mention = list(
Profile.objects.filter(
is_active=True, org=self.request.profile.org
@ -486,7 +489,7 @@ class LeadDetailView(APIView):
object_id=self.lead_obj.id,
org=self.request.profile.org,
).order_by("-id")
if self.request.profile.role == "ADMIN" or self.request.user.is_superuser:
if is_org_admin(self.request.profile) or self.request.user.is_superuser:
users = Profile.objects.filter(
is_active=True, org=self.request.profile.org
).order_by("user__email")
@ -584,7 +587,10 @@ class LeadDetailView(APIView):
{"error": True, "errors": "User company doesnot match with header...."},
status=status.HTTP_403_FORBIDDEN,
)
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
if not (
(self.request.profile.user == self.lead_obj.created_by)
or (self.request.profile in self.lead_obj.assigned_to.all())
@ -846,7 +852,10 @@ class LeadDetailView(APIView):
status=status.HTTP_403_FORBIDDEN,
)
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
if not (
(self.request.profile.user == self.lead_obj.created_by)
or (self.request.profile in self.lead_obj.assigned_to.all())
@ -1063,7 +1072,7 @@ class LeadDetailView(APIView):
def delete(self, request, pk, **kwargs):
self.object = self.get_object(pk)
if (
request.profile.role == "ADMIN"
is_org_admin(request.profile)
or request.user.is_superuser
or request.profile.user == self.object.created_by
) and self.object.org == request.profile.org:

7
backend/opportunity/tasks.py
View file

@ -9,6 +9,7 @@ from django.db import connection
from django.template.loader import render_to_string
from django.utils import timezone
from common.links import frontend_url
from common.models import Org, Profile
from opportunity.models import Opportunity, SalesGoal, StageAgingConfig
@ -35,7 +36,7 @@ def send_email_to_assigned_user(recipients, opportunity_id, org_id):
if profile:
recipients_list.append(profile.user.email)
context = {}
context["url"] = settings.DOMAIN_NAME
context["url"] = frontend_url(f"/pipeline/{opportunity.id}")
context["user"] = profile.user
context["opportunity"] = opportunity
context["created_by"] = created_by
@ -120,7 +121,7 @@ def send_stale_deals_alert(org, stale_opps):
}
for opp, days, expected in deals
],
"url": settings.DOMAIN_NAME,
"url": frontend_url("/pipeline?rotten=true"),
"deal_count": len(deals),
}
subject = f"[BottleCRM] {len(deals)} stale deal{'s' if len(deals) > 1 else ''} need attention"
@ -226,7 +227,7 @@ def _send_goal_milestone_email(profile, goal, milestone_label, percent, achieved
"milestone": milestone_label,
"percent": percent,
"achieved": achieved,
"url": settings.DOMAIN_NAME,
"url": frontend_url("/goals"),
}
subject = f"[BottleCRM] Goal '{goal.name}' reached {milestone_label}!"
html_content = render_to_string(

26
backend/opportunity/tests_celery_tasks.py
View file

@ -1,26 +0,0 @@
from django.test import TestCase
from django.test.utils import override_settings
from opportunity.tasks import send_email_to_assigned_user
from opportunity.tests import OpportunityModel
class TestCeleryTasks(OpportunityModel, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_celery_tasks(self):
org_id = str(self.opportunity.org.id)
task = send_email_to_assigned_user.apply(
(
[
self.user.id,
self.user1.id,
],
self.opportunity.id,
org_id,
),
)
self.assertEqual("SUCCESS", task.state)

4
backend/opportunity/views/aging_views.py
View file

@ -3,7 +3,7 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.utils import STAGES
from opportunity.models import StageAgingConfig
from opportunity.serializer import StageAgingConfigSerializer
@ -39,7 +39,7 @@ class StageAgingConfigView(APIView):
def put(self, request):
"""Bulk upsert stage aging configs (admin only)."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": True, "errors": "Only admins can update aging config"},
status=status.HTTP_403_FORBIDDEN,

12
backend/opportunity/views/goal_views.py
View file

@ -8,7 +8,7 @@ from rest_framework.response import Response
from rest_framework.views import APIView
from common.models import Profile
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.validators import uuid_param
from opportunity.models import SalesGoal
from opportunity.serializer import SalesGoalCreateSerializer, SalesGoalSerializer
@ -21,7 +21,7 @@ class SalesGoalListView(APIView, LimitOffsetPagination):
org = request.profile.org
queryset = SalesGoal.objects.filter(org=org)
is_admin = request.profile.role == "ADMIN" or request.user.is_superuser
is_admin = is_org_admin(request.profile) or request.user.is_superuser
if not is_admin:
queryset = queryset.filter(
Q(assigned_to=request.profile)
@ -70,7 +70,7 @@ class SalesGoalListView(APIView, LimitOffsetPagination):
)
def post(self, request, *args, **kwargs):
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": True, "errors": "Only admins can create goals."},
status=status.HTTP_403_FORBIDDEN,
@ -109,7 +109,7 @@ class SalesGoalDetailView(APIView):
{"error": True, "errors": "Goal not found."},
status=status.HTTP_404_NOT_FOUND,
)
is_admin = request.profile.role == "ADMIN" or request.user.is_superuser
is_admin = is_org_admin(request.profile) or request.user.is_superuser
if not is_admin:
if goal.assigned_to != request.profile and (
not goal.team or goal.team not in request.profile.user_teams.all()
@ -122,7 +122,7 @@ class SalesGoalDetailView(APIView):
return Response(serializer.data)
def put(self, request, pk, *args, **kwargs):
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": True, "errors": "Only admins can update goals."},
status=status.HTTP_403_FORBIDDEN,
@ -148,7 +148,7 @@ class SalesGoalDetailView(APIView):
)
def delete(self, request, pk, *args, **kwargs):
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": True, "errors": "Only admins can delete goals."},
status=status.HTTP_403_FORBIDDEN,

6
backend/opportunity/views/kanban_views.py
View file

@ -16,7 +16,7 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.validators import uuid_list_param, uuid_param
from opportunity.models import Opportunity, StageAgingConfig
from opportunity.serializer import (
@ -66,7 +66,7 @@ class OpportunityKanbanView(APIView):
# Match the list view's RBAC scoping so users only see opps they own
# or are assigned to. Kanban shouldn't reveal more than the table.
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
queryset = queryset.filter(
Q(created_by=request.profile.user)
| Q(assigned_to=request.profile)
@ -160,7 +160,7 @@ class OpportunityMoveView(APIView):
opportunity = get_object_or_404(Opportunity, pk=pk, org=org)
# Match Opportunity update/delete RBAC: admin OR creator OR assignee.
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
is_owner = request.profile == opportunity.created_by
is_assignee = request.profile in opportunity.assigned_to.all()
if not (is_owner or is_assignee):

12
backend/opportunity/views/line_item_views.py
View file

@ -4,7 +4,7 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from invoices.models import Product
from invoices.serializer import ProductSerializer
from opportunity.models import Opportunity, OpportunityLineItem
@ -34,7 +34,10 @@ class OpportunityLineItemListView(APIView):
{"error": True, "message": "Opportunity not found"},
status=status.HTTP_404_NOT_FOUND,
)
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
if not (
(self.request.profile.user == opportunity.created_by)
or (self.request.profile in opportunity.assigned_to.all())
@ -167,7 +170,10 @@ class OpportunityLineItemDetailView(APIView):
{"error": True, "message": "Opportunity not found"},
status=status.HTTP_404_NOT_FOUND,
)
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
if not (
(self.request.profile.user == opportunity.created_by)
or (self.request.profile in opportunity.assigned_to.all())

10
backend/opportunity/views/opportunity_interactions.py
View file

@ -6,7 +6,7 @@ from rest_framework.views import APIView
from common.lookups import get_scoped_or_404
from common.models import Attachments, Comment
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.serializer import CommentSerializer
from opportunity import swagger_params
from opportunity.serializer import OpportunityCommentEditSwaggerSerializer
@ -37,7 +37,7 @@ class OpportunityCommentView(APIView):
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
is_org_admin(request.profile)
or request.user.is_superuser
or request.profile == obj.commented_by
):
@ -81,7 +81,7 @@ class OpportunityCommentView(APIView):
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
is_org_admin(request.profile)
or request.user.is_superuser
or request.profile == obj.commented_by
):
@ -120,7 +120,7 @@ class OpportunityCommentView(APIView):
def delete(self, request, pk, format=None):
self.object = self.get_object(pk)
if (
request.profile.role == "ADMIN"
is_org_admin(request.profile)
or request.user.is_superuser
or request.profile == self.object.commented_by
):
@ -172,7 +172,7 @@ class OpportunityAttachmentView(APIView):
# the fix looking complete while the branch stayed unreachable.
self.object = get_scoped_or_404(self.model, pk, request.profile.org)
if (
request.profile.role == "ADMIN"
is_org_admin(request.profile)
or request.user.is_superuser
or request.profile.user == self.object.created_by
):

18
backend/opportunity/views/opportunity_views.py
View file

@ -18,7 +18,7 @@ from accounts.models import Account
from accounts.serializer import AccountSerializer, TagsSerializer
from common.custom_fields import validate_payload as validate_custom_fields_payload
from common.models import Attachments, Comment, CustomFieldDefinition, Profile, Tags, Teams
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.serializer import (
AttachmentsSerializer,
CommentSerializer,
@ -117,7 +117,10 @@ class OpportunityListView(APIView, LimitOffsetPagination):
)
accounts = Account.objects.filter(org=self.request.profile.org)
contacts = Contact.objects.filter(org=self.request.profile.org)
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
queryset = queryset.filter(
Q(created_by=self.request.profile.user)
| Q(assigned_to=self.request.profile)
@ -414,7 +417,7 @@ class OpportunityDetailView(APIView):
Raising beats returning a Response: a returned Response from a helper
gets wrapped in `Response(...)` by the caller and renders as a 500.
"""
if self.request.profile.role == "ADMIN" or self.request.user.is_superuser:
if is_org_admin(self.request.profile) or self.request.user.is_superuser:
return
if self.request.profile.user_id == opportunity.created_by_id:
return
@ -575,7 +578,10 @@ class OpportunityDetailView(APIView):
{"error": True, "errors": "User company doesnot match with header...."},
status=status.HTTP_403_FORBIDDEN,
)
if self.request.profile.role != "ADMIN" and not self.request.user.is_superuser:
if (
not is_org_admin(self.request.profile)
and not self.request.user.is_superuser
):
if self.request.profile.user != self.object.created_by:
return Response(
{
@ -627,10 +633,10 @@ class OpportunityDetailView(APIView):
comment_permission = (
self.request.profile.user_id == self.opportunity.created_by_id
or self.request.user.is_superuser
or self.request.profile.role == "ADMIN"
or is_org_admin(self.request.profile)
)
if self.request.user.is_superuser or self.request.profile.role == "ADMIN":
if self.request.user.is_superuser or is_org_admin(self.request.profile):
users_mention = list(
Profile.objects.filter(
is_active=True, org=self.request.profile.org

11
backend/tasks/access.py
View file

@ -32,14 +32,15 @@ from rest_framework.exceptions import PermissionDenied
from tasks.models import Task
# Re-exported, not redefined, the same arrangement `cases.access` uses. This
# was an eleventh copy of the admin rule and the only one that had taken the
# canonical name, so the guard test looking for private spellings walked past
# it. It read `role == "ADMIN" or is_admin`, which is now one fact and not two.
from common.permissions import is_org_admin # noqa: F401
_DENIED = "You do not have Permission to perform this action"
def is_org_admin(profile):
"""True for the two ways this codebase spells "admin"."""
return profile.role == "ADMIN" or bool(getattr(profile, "is_admin", False))
def visible_tasks_qs(profile):
"""Tasks ``profile`` is allowed to open, the queryset form of `access`."""
qs = Task.objects.filter(org=profile.org)

21
backend/tasks/migrations/0013_alter_taskpipeline_is_default.py Normal file
View file

@ -0,0 +1,21 @@
# Generated by Django 6.0.7 on 2026-08-05 04:54
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("tasks", "0012_task_is_sample"),
]
operations = [
migrations.AlterField(
model_name="taskpipeline",
name="is_default",
field=models.BooleanField(
default=False,
help_text="Sorts this pipeline first in listings (see Meta.ordering) and is limited to one per org. It does NOT route new records: a task created without a stage keeps stage=NULL, which is the supported status-based kanban mode.",
),
),
]

7
backend/tasks/models.py
View file

@ -207,7 +207,12 @@ class TaskPipeline(BaseModel):
)
is_default = models.BooleanField(
default=False,
help_text="If true, new tasks without explicit pipeline go here",
help_text=(
"Sorts this pipeline first in listings (see Meta.ordering) and is "
"limited to one per org. It does NOT route new records: a task "
"created without a stage keeps stage=NULL, which is the supported "
"status-based kanban mode."
),
)
is_active = models.BooleanField(default=True)

26
backend/tasks/tests_celery_tasks.py
View file

@ -1,26 +0,0 @@
from django.test import TestCase
from django.test.utils import override_settings
from tasks.celery_tasks import send_email
from tasks.tests import TaskCreateTest
class TestEventCeleryTasks(TaskCreateTest, TestCase):
@override_settings(
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND="memory",
)
def test_event_celery_tasks(self):
org_id = str(self.task.org.id)
task = send_email.apply(
(
self.task.id,
[
self.user.id,
self.user1.id,
],
org_id,
)
)
self.assertEqual("SUCCESS", task.state)

20
backend/tasks/views/kanban_views.py
View file

@ -14,7 +14,7 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from common.permissions import HasOrgContext
from common.permissions import HasOrgContext, is_org_admin
from common.validators import uuid_param
from tasks.models import Task, TaskPipeline, TaskStage
from tasks.serializer import (
@ -119,7 +119,7 @@ class TaskKanbanView(APIView):
)
# Apply permission filtering
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
queryset = queryset.filter(
Q(assigned_to=request.profile) | Q(created_by=request.profile.user)
)
@ -256,7 +256,7 @@ class TaskMoveView(APIView):
task = get_object_or_404(Task, pk=pk, org=org)
# Permission check
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
if not (
request.profile.user == task.created_by
or request.profile in task.assigned_to.all()
@ -414,7 +414,7 @@ class TaskPipelineListCreateView(APIView):
org = request.profile.org
# Only admins can create pipelines
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Only admins can create pipelines"},
status=status.HTTP_403_FORBIDDEN,
@ -490,7 +490,7 @@ class TaskPipelineDetailView(APIView):
)
def put(self, request, pk):
"""Update pipeline."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -510,7 +510,7 @@ class TaskPipelineDetailView(APIView):
@extend_schema(tags=["Task Pipelines"], responses={204: None})
def delete(self, request, pk):
"""Delete pipeline (soft delete by setting is_active=False)."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -544,7 +544,7 @@ class TaskStageCreateView(APIView):
)
def post(self, request, pipeline_pk):
"""Add a new stage to pipeline."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -575,7 +575,7 @@ class TaskStageDetailView(APIView):
)
def put(self, request, pk):
"""Update stage."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -595,7 +595,7 @@ class TaskStageDetailView(APIView):
@extend_schema(tags=["Task Stages"], responses={204: None})
def delete(self, request, pk):
"""Delete stage."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)
@ -631,7 +631,7 @@ class TaskStageReorderView(APIView):
@transaction.atomic
def post(self, request, pipeline_pk):
"""Reorder stages by providing ordered list of stage IDs."""
if request.profile.role != "ADMIN" and not request.user.is_superuser:
if not is_org_admin(request.profile) and not request.user.is_superuser:
return Response(
{"error": "Permission denied"}, status=status.HTTP_403_FORBIDDEN
)

18
backend/tasks/views/task_views.py
View file

@ -780,11 +780,7 @@ class TaskCommentView(APIView):
def put(self, request, pk, format=None):
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == obj.commented_by
):
if is_org_admin(request.profile) or request.profile == obj.commented_by:
serializer = CommentSerializer(obj, data=params)
if params.get("comment"):
if serializer.is_valid():
@ -825,11 +821,7 @@ class TaskCommentView(APIView):
"""Handle partial updates to a comment."""
params = request.data
obj = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == obj.commented_by
):
if is_org_admin(request.profile) or request.profile == obj.commented_by:
serializer = CommentSerializer(obj, data=params, partial=True)
if serializer.is_valid():
serializer.save()
@ -864,11 +856,7 @@ class TaskCommentView(APIView):
)
def delete(self, request, pk, format=None):
self.object = self.get_object(pk)
if (
request.profile.role == "ADMIN"
or request.profile.is_admin
or request.profile == self.object.commented_by
):
if is_org_admin(request.profile) or request.profile == self.object.commented_by:
self.object.delete()
return Response(
{"error": False, "message": "Comment Deleted Successfully"},

4
backend/templates/assigned_to/leads_assigned.html
View file

@ -17,7 +17,7 @@
<td class="metadata-label" style="padding:12px 16px 4px 16px;font-size:13px;color:#757575;font-weight:500;">Lead</td>
</tr>
<tr>
<td class="metadata-value" style="padding:0 16px 12px 16px;font-size:15px;color:#212121;">{{ lead.title|default:lead_instance }}</td>
<td class="metadata-value" style="padding:0 16px 12px 16px;font-size:15px;color:#212121;">{{ lead.title }}</td>
</tr>
{% if source %}
<tr>
@ -39,7 +39,7 @@
{% endblock metadata %}
{% block button_link %}
<a class="btn-link" href="{{ url|default:lead_detail_url }}"
<a class="btn-link" href="{{ url }}"
style="display:inline-block;padding:14px 32px;background-color:#EA580C;color:#ffffff;text-decoration:none;border-radius:6px;font-size:16px;font-weight:600;line-height:1;">
View Lead
</a>

2
backend/templates/opportunity/goal_milestone.html
View file

@ -43,7 +43,7 @@ Congratulations, {{ user.get_username }}!
{% block button_link %}
<div style="margin-bottom:20px">
<a href="{{url}}/goals"
<a href="{{ url }}"
style="display:block;width:200px;background:#16a34a;padding:10px;text-align:center;color:#fff;font-size:1rem;font-weight:600;margin:0px auto;margin-bottom:20px;border-radius:5px;text-decoration:none">View Goals</a>
</div>
{% endblock button_link %}

2
backend/templates/opportunity/stale_deals_alert.html
View file

@ -31,7 +31,7 @@ Hello {{ user.get_username }}
{% block button_link %}
<div style="margin-bottom:20px">
<a href="{{url}}/opportunities?rotten=true"
<a href="{{ url }}"
style="display:block;width:200px;background:#dc2626;padding:10px;text-align:center;color:#fff;font-size:1rem;font-weight:600;margin:0px auto;margin-bottom:20px;border-radius:5px;text-decoration:none">View Stale Deals</a>
</div>
{% endblock button_link %}