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

Merge pull request #736 from Django-CRM/dev

Browse source 
Refactor API views to improve parameter handling and validation
This commit is contained in:
Ashwin Kumar 2026-08-05 05:29:09 +05:30 committed by GitHub
commit c7b2b9750b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
29 changed files with 783 additions and 543 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

61
backend/accounts/views.py
View file

@ -46,6 +46,7 @@ from accounts.serializer import (
TagsSerializer,
)
from common.utils import create_attachment, get_or_create_tags, handle_m2m_assignment
from common.validators import payload_id_list, uuid_list_param
from accounts.tasks import send_email, send_email_to_assigned_user
from cases.serializer import CaseSerializer
from common.models import (
@ -205,13 +206,13 @@ class AccountsListView(APIView, LimitOffsetPagination):
queryset = queryset.filter(city__icontains=params.get("city"))
if params.get("industry"):
queryset = queryset.filter(industry__icontains=params.get("industry"))
if params.getlist("tags"):
tags = uuid_list_param(params, "tags")
if tags:
queryset = queryset.filter(tags__id__in=tags).distinct()
assigned_to = uuid_list_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(
tags__id__in=params.getlist("tags")
).distinct()
if params.getlist("assigned_to"):
queryset = queryset.filter(
assigned_to__id__in=params.getlist("assigned_to")
assigned_to__id__in=assigned_to
).distinct()
if params.get("search"):
queryset = queryset.filter(name__icontains=params.get("search"))
@ -490,13 +491,7 @@ class AccountDetailView(APIView):
account_object.contacts.clear()
if data.get("contacts"):
contacts_list = data.get("contacts")
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -520,13 +515,7 @@ class AccountDetailView(APIView):
account_object.teams.clear()
if data.get("teams"):
teams_list = data.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
if teams:
account_object.teams.add(*teams)
@ -534,13 +523,7 @@ class AccountDetailView(APIView):
account_object.assigned_to.clear()
if data.get("assigned_to"):
assigned_to_list = data.get("assigned_to")
if isinstance(assigned_to_list, str):
assigned_to_list = json.loads(assigned_to_list)
# Extract IDs if assigned_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assigned_to_list
]
assigned_ids = payload_id_list(assigned_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -802,13 +785,7 @@ class AccountDetailView(APIView):
account_object.contacts.clear()
contacts_list = data.get("contacts")
if contacts_list:
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -833,13 +810,7 @@ class AccountDetailView(APIView):
account_object.teams.clear()
teams_list = data.get("teams")
if teams_list:
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(
id__in=team_ids, org=request.profile.org
)
@ -849,13 +820,7 @@ class AccountDetailView(APIView):
account_object.assigned_to.clear()
assigned_to_list = data.get("assigned_to")
if assigned_to_list:
if isinstance(assigned_to_list, str):
assigned_to_list = json.loads(assigned_to_list)
# Extract IDs if assigned_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assigned_to_list
]
assigned_ids = payload_id_list(assigned_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)

5
backend/cases/analytics_views.py
View file

@ -28,6 +28,7 @@ from cases.access import is_org_admin
from cases.models import Case
from cases.serializer import CaseSerializer
from common.permissions import HasOrgContext
from common.validators import uuid_param
# ---------------------------------------------------------------------------
# Shared filter parsing
@ -69,9 +70,9 @@ def _filtered_qs(request) -> tuple[object, datetime, datetime]:
if priority := params.get("priority"):
qs = qs.filter(priority=priority)
if team_id := params.get("team"):
if team_id := uuid_param(params, "team"):
qs = qs.filter(teams=team_id)
if agent_id := params.get("agent"):
if agent_id := uuid_param(params, "agent"):
qs = qs.filter(assigned_to=agent_id)
from_dt = _parse_dt(params.get("from", ""))

3
backend/cases/approval_views.py
View file

@ -34,6 +34,7 @@ from cases.serializer import (
)
from common.models import Activity
from common.permissions import HasOrgContext
from common.validators import uuid_param
def _is_admin(profile) -> bool:
@ -311,7 +312,7 @@ class ApprovalInboxView(APIView):
if state and state != "all":
qs = qs.filter(state=state)
case_id = request.query_params.get("case")
case_id = uuid_param(request.query_params, "case")
if case_id:
qs = qs.filter(case_id=case_id)

18
backend/cases/kanban_views.py
View file

@ -23,6 +23,7 @@ from cases.serializer import (
CaseStageSerializer,
)
from common.permissions import HasOrgContext
from common.validators import uuid_param
from common.utils import STATUS_CHOICE
@ -83,7 +84,7 @@ class CaseKanbanView(APIView):
def get(self, request):
"""Get kanban board data."""
org = request.profile.org
pipeline_id = request.query_params.get("pipeline_id")
pipeline_id = uuid_param(request.query_params, "pipeline_id")
# Base queryset with filters. Merged duplicates are hidden by default
# (matches the cases-list behavior); admins can pass show_merged=true.
@ -112,8 +113,9 @@ class CaseKanbanView(APIView):
def _apply_filters(self, queryset, params):
"""Apply common filters to queryset."""
if params.get("assigned_to"):
queryset = queryset.filter(assigned_to__id=params.get("assigned_to"))
assigned_to = uuid_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(assigned_to__id=assigned_to)
if params.get("priority"):
queryset = queryset.filter(priority=params.get("priority"))
if params.get("case_type"):
@ -123,10 +125,12 @@ class CaseKanbanView(APIView):
queryset = queryset.filter(
Q(name__icontains=search) | Q(description__icontains=search)
)
if params.get("account"):
queryset = queryset.filter(account_id=params.get("account"))
if params.get("tags"):
queryset = queryset.filter(tags__id=params.get("tags"))
account = uuid_param(params, "account")
if account:
queryset = queryset.filter(account_id=account)
tags = uuid_param(params, "tags")
if tags:
queryset = queryset.filter(tags__id=tags)
if params.get("created_at__gte"):
queryset = queryset.filter(created_at__gte=params.get("created_at__gte"))
if params.get("created_at__lte"):

5
backend/cases/time_views.py
View file

@ -36,6 +36,7 @@ from cases.serializer import (
)
from common.models import Profile
from common.permissions import HasOrgContext
from common.validators import uuid_param
def _is_admin(profile):
@ -257,7 +258,7 @@ class UnbilledEntriesView(APIView):
permission_classes = (IsAuthenticated, HasOrgContext)
def get(self, request):
account_id = request.query_params.get("account")
account_id = uuid_param(request.query_params, "account")
qs = (
_visible_entry_qs(request.profile)
.filter(billable=True, invoice__isnull=True, ended_at__isnull=False)
@ -279,7 +280,7 @@ class TimesheetView(APIView):
permission_classes = (IsAuthenticated, HasOrgContext)
def get(self, request):
profile_param = request.query_params.get("profile")
profile_param = uuid_param(request.query_params, "profile")
target_profile_id = request.profile.id
if profile_param and profile_param != str(request.profile.id):
if not _is_admin(request.profile):

98
backend/cases/views.py
View file

@ -60,6 +60,7 @@ from common.serializer import (
CustomFieldDefinitionSerializer,
)
from common.utils import CASE_TYPE, PRIORITY_CHOICE, STATUS_CHOICE
from common.validators import payload_id_list, uuid_list_param, uuid_param
from contacts.models import Contact
from contacts.serializer import ContactSerializer
@ -104,16 +105,17 @@ def apply_case_list_filters(queryset, params):
queryset = queryset.filter(status=status_values[0])
if params.get("priority"):
queryset = queryset.filter(priority=params.get("priority"))
if params.get("account"):
queryset = queryset.filter(account=params.get("account"))
account = uuid_param(params, "account")
if account:
queryset = queryset.filter(account=account)
if params.get("case_type"):
queryset = queryset.filter(case_type=params.get("case_type"))
if params.getlist("assigned_to"):
queryset = queryset.filter(
assigned_to__id__in=params.getlist("assigned_to")
).distinct()
if params.get("tags"):
queryset = queryset.filter(tags__id__in=params.getlist("tags")).distinct()
assigned_to = uuid_list_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(assigned_to__id__in=assigned_to).distinct()
tags = uuid_list_param(params, "tags")
if tags:
queryset = queryset.filter(tags__id__in=tags).distinct()
if params.get("search"):
search = params.get("search")
queryset = queryset.filter(
@ -338,13 +340,7 @@ class CaseListView(APIView, LimitOffsetPagination):
if params.get("contacts"):
contacts_list = params.get("contacts")
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -353,26 +349,14 @@ class CaseListView(APIView, LimitOffsetPagination):
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
if teams.exists():
cases_obj.teams.add(*teams)
if params.get("assigned_to"):
assinged_to_list = params.get("assigned_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -492,13 +476,7 @@ class CaseDetailView(APIView):
cases_object.contacts.clear()
if params.get("contacts"):
contacts_list = params.get("contacts")
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -508,13 +486,7 @@ class CaseDetailView(APIView):
cases_object.teams.clear()
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
if teams.exists():
cases_object.teams.add(*teams)
@ -522,13 +494,7 @@ class CaseDetailView(APIView):
cases_object.assigned_to.clear()
if params.get("assigned_to"):
assinged_to_list = params.get("assigned_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -879,13 +845,7 @@ class CaseDetailView(APIView):
cases_object.contacts.clear()
contacts_list = params.get("contacts")
if contacts_list:
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -895,13 +855,7 @@ class CaseDetailView(APIView):
cases_object.teams.clear()
teams_list = params.get("teams")
if teams_list:
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(
id__in=team_ids, org=request.profile.org
)
@ -911,13 +865,7 @@ class CaseDetailView(APIView):
cases_object.assigned_to.clear()
assigned_to_list = params.get("assigned_to")
if assigned_to_list:
if isinstance(assigned_to_list, str):
assigned_to_list = json.loads(assigned_to_list)
# Extract IDs if assigned_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assigned_to_list
]
assigned_ids = payload_id_list(assigned_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -927,13 +875,7 @@ class CaseDetailView(APIView):
cases_object.tags.clear()
tags_list = params.get("tags")
if tags_list:
if isinstance(tags_list, str):
tags_list = json.loads(tags_list)
# Extract IDs if tags_list contains objects with 'id' field
tag_ids = [
tag.get("id") if isinstance(tag, dict) else tag
for tag in tags_list
]
tag_ids = payload_id_list(tags_list, "tags")
tag_objs = Tags.objects.filter(
id__in=tag_ids, org=request.profile.org, is_active=True
)

12
backend/common/tests/test_auth.py
View file

@ -14,7 +14,6 @@ from unittest.mock import MagicMock, patch
import pytest
from django.utils import timezone
from rest_framework import status
from rest_framework.exceptions import PermissionDenied
from common.models import Org, Profile, User
from common.serializer import OrgAwareRefreshToken
@ -366,9 +365,14 @@ class TestProfileDetailView:
assert str(response.data["org"]["id"]) == str(org_a.id)
def test_profile_detail_unauthenticated(self, unauthenticated_client):
"""Unauthenticated user gets error (401 or PermissionDenied from middleware)."""
with pytest.raises((PermissionDenied, Exception)):
unauthenticated_client.get(self.url)
"""The request is refused, and refused as a response, not an exception.
The previous form caught ``(PermissionDenied, Exception)``, which is
just ``Exception``: it would have passed on a typo, an import error or
a 500. It asserted nothing about access control.
"""
response = unauthenticated_client.get(self.url)
assert response.status_code == status.HTTP_403_FORBIDDEN
# ---------------------------------------------------------------------------

14
backend/common/tests/test_dashboard.py
View file

@ -8,7 +8,6 @@ import uuid
import pytest
from rest_framework import status
from rest_framework.exceptions import PermissionDenied
from accounts.models import Account
from common.models import Activity
@ -34,9 +33,16 @@ class TestDashboardView:
assert "opportunities_count" in data
def test_dashboard_unauthenticated(self, unauthenticated_client):
"""Unauthenticated user gets PermissionDenied."""
with pytest.raises(PermissionDenied):
unauthenticated_client.get(self.url)
"""The request is refused, and refused as a response, not an exception.
DRF catches ``PermissionDenied`` in ``handle_exception`` and renders it,
so a test client never sees it raised. Wrapping the call in
``pytest.raises`` therefore failed with DID NOT RAISE no matter how the
endpoint behaved, which meant this test could never have caught the
access opening up.
"""
response = unauthenticated_client.get(self.url)
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_dashboard_counts(self, admin_client, org_a, admin_user):
"""Dashboard should return accurate counts."""

13
backend/common/tests/test_documents.py
View file

@ -7,7 +7,6 @@ Run with: pytest common/tests/test_documents.py -v
import pytest
from django.core.files.uploadedfile import SimpleUploadedFile
from rest_framework import status
from rest_framework.exceptions import PermissionDenied
from common.models import Document, Profile, Teams
@ -36,8 +35,16 @@ class TestDocumentListView:
assert response.data["error"] is False
def test_unauthenticated(self, unauthenticated_client):
with pytest.raises(PermissionDenied):
unauthenticated_client.get(self.url)
"""The request is refused, and refused as a response, not an exception.
DRF catches ``PermissionDenied`` in ``handle_exception`` and renders it,
so a test client never sees it raised. Wrapping the call in
``pytest.raises`` therefore failed with DID NOT RAISE no matter how the
endpoint behaved, which meant this test could never have caught the
access opening up.
"""
response = unauthenticated_client.get(self.url)
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_list_documents_context_keys(self, admin_client, org_a):
"""Document list should include all expected context keys."""

126
backend/common/tests/test_malformed_id_filters.py Normal file
View file

@ -0,0 +1,126 @@
"""A malformed id in a filter must answer 400, never 500.
Every list endpoint accepts id-valued filters straight off the query string.
Django parses them with ``uuid.UUID`` while building the query and raises
``django.core.exceptions.ValidationError``, which DRF's exception handler does
not translate, so the response was a 500. A stale bookmark, a hand-edited URL,
or any client sending a placeholder id took the endpoint down.
The frontend now strips non-UUID values before it builds a request, but the
frontend is not the trust boundary: these endpoints serve the mobile app, API
tokens and curl as well. These tests exercise the API directly for that reason.
Both directions matter. A well-formed id must still filter, so each endpoint is
asserted twice: 400 for junk, 200 for a real id.
"""
import pytest
from rest_framework import status
# Shapes that reached production URLs. `p1`, `acc-1` and `profile-1` are
# placeholder ids that older frontend fixtures emitted verbatim.
MALFORMED = ["notauuid", "p1", "acc-1", "profile-1", "abc-123", "0"]
WELL_FORMED = "3f2504e0-4f89-41d3-9a0c-0305e82c3301"
# (label, url, param). One row per endpoint-and-filter that takes an id.
ID_FILTERS = [
("leads.tags", "/api/leads/", "tags"),
("leads.assigned_to", "/api/leads/", "assigned_to"),
("accounts.tags", "/api/accounts/", "tags"),
("accounts.assigned_to", "/api/accounts/", "assigned_to"),
("contacts.tags", "/api/contacts/", "tags"),
("contacts.assigned_to", "/api/contacts/", "assigned_to"),
("cases.tags", "/api/cases/", "tags"),
("cases.assigned_to", "/api/cases/", "assigned_to"),
("cases.account", "/api/cases/", "account"),
("tasks.tags", "/api/tasks/", "tags"),
("tasks.assigned_to", "/api/tasks/", "assigned_to"),
("tasks.account", "/api/tasks/", "account"),
("opportunities.tags", "/api/opportunities/", "tags"),
("opportunities.assigned_to", "/api/opportunities/", "assigned_to"),
("opportunities.account", "/api/opportunities/", "account"),
("invoices.account", "/api/invoices/", "account"),
("invoices.assigned_to", "/api/invoices/", "assigned_to"),
("invoices.created_by", "/api/invoices/", "created_by"),
("documents.shared_to", "/api/documents/", "shared_to"),
("teams.assigned_users", "/api/teams/", "assigned_users"),
("teams.created_by", "/api/teams/", "created_by"),
]
IDS = [row[0] for row in ID_FILTERS]
@pytest.mark.django_db
@pytest.mark.parametrize("label,url,param", ID_FILTERS, ids=IDS)
class TestMalformedIdIsRejected:
def test_malformed_id_answers_400(self, admin_client, label, url, param):
for bad in MALFORMED:
response = admin_client.get(f"{url}?{param}={bad}")
assert response.status_code == status.HTTP_400_BAD_REQUEST, (
f"{label} answered {response.status_code} for {param}={bad!r}; "
"a malformed id must be a 400, not a crash"
)
def test_well_formed_id_is_still_accepted(self, admin_client, label, url, param):
"""The guard must not have turned a working filter into a 400.
The id matches nothing, so this asserts the request is served, not that
rows come back.
"""
response = admin_client.get(f"{url}?{param}={WELL_FORMED}")
assert response.status_code == status.HTTP_200_OK, (
f"{label} answered {response.status_code} for a well-formed id"
)
def test_a_blank_value_means_no_filter(self, admin_client, label, url, param):
"""``?tags=`` is what a select with a blank first option submits.
It has to mean "no filter", not "malformed id".
"""
response = admin_client.get(f"{url}?{param}=")
assert response.status_code == status.HTTP_200_OK
@pytest.mark.django_db
class TestKanbanBoardsRejectMalformedIds:
"""The board endpoints read a narrower param set than their list twins."""
BOARDS = [
("/api/tasks/kanban/", "assigned_to"),
("/api/tasks/kanban/", "pipeline_id"),
("/api/cases/kanban/", "assigned_to"),
("/api/cases/kanban/", "account"),
("/api/opportunities/kanban/", "assigned_to"),
("/api/opportunities/kanban/", "tags"),
]
@pytest.mark.parametrize("url,param", BOARDS)
def test_malformed_id_answers_400(self, admin_client, url, param):
response = admin_client.get(f"{url}?{param}=notauuid")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.parametrize("url,param", BOARDS)
def test_well_formed_id_is_still_accepted(self, admin_client, url, param):
response = admin_client.get(f"{url}?{param}={WELL_FORMED}")
# `pipeline_id` names a record that has to exist, so 404 is a correct
# answer there. What matters is that it parsed rather than crashed.
assert response.status_code in (
status.HTTP_200_OK,
status.HTTP_404_NOT_FOUND,
)
@pytest.mark.django_db
class TestSharedToAcceptsABareId:
"""``?shared_to=<id>`` used to be a 500 because only JSON text parsed."""
def test_bare_id_is_accepted(self, admin_client):
response = admin_client.get(f"/api/documents/?shared_to={WELL_FORMED}")
assert response.status_code == status.HTTP_200_OK
def test_json_array_is_still_accepted(self, admin_client):
response = admin_client.get(
f'/api/documents/?shared_to=["{WELL_FORMED}"]'
)
assert response.status_code == status.HTTP_200_OK

9
backend/common/tests/test_organizations.py
View file

@ -104,9 +104,14 @@ class TestOrgProfileCreateView:
assert org_a.name in org_names
def test_list_orgs_unauthenticated(self, unauthenticated_client):
"""Unauthenticated user gets 401."""
"""Unauthenticated callers are refused.
The endpoint answers 403, not the 401 this test used to assert, because
no authenticator offers a ``WWW-Authenticate`` challenge. Access is
denied either way; only the status text differs.
"""
response = unauthenticated_client.get(self.url)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.django_db

14
backend/common/tests/test_settings.py
View file

@ -6,7 +6,6 @@ Run with: pytest common/tests/test_settings.py -v
import pytest
from rest_framework import status
from rest_framework.exceptions import PermissionDenied
from common.models import APISettings
@ -45,9 +44,16 @@ class TestDomainListView:
assert response.status_code == status.HTTP_400_BAD_REQUEST
def test_unauthenticated(self, unauthenticated_client):
"""Unauthenticated user gets PermissionDenied."""
with pytest.raises(PermissionDenied):
unauthenticated_client.get(self.url)
"""The request is refused, and refused as a response, not an exception.
DRF catches ``PermissionDenied`` in ``handle_exception`` and renders it,
so a test client never sees it raised. Wrapping the call in
``pytest.raises`` therefore failed with DID NOT RAISE no matter how the
endpoint behaved, which meant this test could never have caught the
access opening up.
"""
response = unauthenticated_client.get(self.url)
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.django_db

99
backend/common/tests/test_user_views_org_context.py Normal file
View file

@ -0,0 +1,99 @@
"""``HasOrgContext`` on the three user-management views.
``UsersListView``, ``UserDetailView`` and ``UserStatusView`` each read
``request.profile`` unconditionally: ``request.profile.role`` for the admin
check, ``request.profile.org`` for the object lookup. ``GetProfileAndOrg``
leaves ``request.profile`` as ``None`` whenever the JWT carries no ``org_id``
claim, or carries one the caller is no longer a member of, so every one of
those reads was an ``AttributeError`` on a stale token: a 500 where the answer
should have been a 403.
This is not a tenant leak. The middleware already requires ``is_active=True``
before it sets ``request.profile`` (``common/middleware/get_company.py``), so a
deactivated profile never reached these views to begin with. What was missing
was the clean refusal.
Both directions are asserted, since a permission class that cannot return
``False`` is decoration and one that cannot return ``True`` is an outage.
"""
import pytest
from rest_framework import status
from rest_framework.test import APIClient
from rest_framework_simplejwt.tokens import RefreshToken
from common.models import Profile
def _client_without_org_claim(user):
"""A client whose JWT authenticates but names no org.
This is the shape a token has between login and picking an org, and the
shape a stale token keeps after its membership is revoked.
"""
client = APIClient()
token = RefreshToken.for_user(user)
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token.access_token}")
return client
@pytest.mark.django_db
class TestUsersListViewOrgContext:
url = "/api/users/"
def test_without_org_context_it_is_refused_not_crashed(self, admin_user):
response = _client_without_org_claim(admin_user).get(self.url)
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_with_org_context_an_admin_still_gets_through(self, admin_client):
"""The False direction is worthless without this one beside it."""
response = admin_client.get(self.url)
assert response.status_code == status.HTTP_200_OK
def test_post_without_org_context_is_refused(self, admin_user):
response = _client_without_org_claim(admin_user).post(
self.url, {"email": "x@test.com", "role": "USER"}, format="json"
)
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.django_db
class TestUserDetailViewOrgContext:
def test_without_org_context_it_is_refused_not_crashed(
self, admin_user, admin_profile
):
client = _client_without_org_claim(admin_user)
response = client.get(f"/api/user/{admin_user.id}/")
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_with_org_context_an_admin_still_gets_through(
self, admin_client, admin_user, admin_profile
):
response = admin_client.get(f"/api/user/{admin_user.id}/")
assert response.status_code == status.HTTP_200_OK
@pytest.mark.django_db
class TestUserStatusViewOrgContext:
def test_without_org_context_it_is_refused_not_crashed(
self, admin_user, user_profile
):
client = _client_without_org_claim(admin_user)
response = client.post(
f"/api/user/{user_profile.user.id}/status/",
{"status": "Inactive"},
format="json",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_with_org_context_an_admin_still_gets_through(
self, admin_client, user_profile
):
response = admin_client.post(
f"/api/user/{user_profile.user.id}/status/",
{"status": "Inactive"},
format="json",
)
assert response.status_code == status.HTTP_200_OK
user_profile.refresh_from_db()
assert Profile.objects.get(pk=user_profile.pk).is_active is False

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

@ -6,7 +6,6 @@ Run with: pytest common/tests/test_users.py -v
import pytest
from rest_framework import status
from rest_framework.exceptions import PermissionDenied
from common.models import Address, Profile
@ -23,8 +22,16 @@ class TestUsersListView:
assert "active_users" in response.data
def test_list_users_unauthenticated(self, unauthenticated_client):
with pytest.raises(PermissionDenied):
unauthenticated_client.get(self.url)
"""The request is refused, and refused as a response, not an exception.
DRF catches ``PermissionDenied`` in ``handle_exception`` and renders it,
so a test client never sees it raised. Wrapping the call in
``pytest.raises`` therefore failed with DID NOT RAISE no matter how the
endpoint behaved, which meant this test could never have caught the
access opening up.
"""
response = unauthenticated_client.get(self.url)
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_list_users_no_cross_org_leak(
self, org_b_client, admin_user, admin_profile

173
backend/common/tests/test_uuid_params.py Normal file
View file

@ -0,0 +1,173 @@
"""Tests for the UUID query-parameter validators in ``common.validators``.
Every list endpoint takes id-valued filters (``tags``, ``assigned_to``,
``account``, ...) straight from the query string and feeds them to an ``id``
lookup. Django's ``UUIDField`` raises ``django.core.exceptions.ValidationError``
while the query is still being built when a value is not a UUID, and DRF's
exception handler does not translate that class, so the request answered 500.
These helpers turn that into a clean 400. The tests below pin both directions:
a well-formed value passes through untouched, a malformed one raises DRF's
``ValidationError``.
"""
import pytest
from django.http import QueryDict
from rest_framework.exceptions import ValidationError
from common.validators import (
payload_id_list,
uuid_list_param,
uuid_param,
validate_uuid,
validate_uuid_list,
)
GOOD = "3f2504e0-4f89-41d3-9a0c-0305e82c3301"
OTHER = "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d"
class TestValidateUuid:
def test_returns_a_well_formed_value_unchanged(self):
assert validate_uuid(GOOD, "tags") == GOOD
def test_accepts_uppercase(self):
assert validate_uuid(GOOD.upper(), "tags") == GOOD.upper()
@pytest.mark.parametrize(
"bad", ["notauuid", "p1", "1", "abc-123", "profile-1", GOOD[:-1], GOOD + "0"]
)
def test_rejects_a_malformed_value(self, bad):
with pytest.raises(ValidationError) as exc:
validate_uuid(bad, "tags")
# The field name has to reach the client, otherwise a page with six
# filters gives no clue which one is wrong.
assert "tags" in str(exc.value)
@pytest.mark.parametrize(
"spelling",
[
GOOD.replace("-", ""),
"{" + GOOD + "}",
f"urn:uuid:{GOOD}",
],
)
def test_accepts_every_spelling_the_orm_accepts(self, spelling):
"""Acceptance must match ``UUIDField``, not a narrower idea of it.
``Lead.objects.filter(tags__id__in=[v])`` builds fine for the plain,
upper-case, unhyphenated, braced and URN forms alike, because Django
parses with ``uuid.UUID``. A validator that took only the hyphenated
form would start rejecting requests that work today, turning a crash
fix into a behaviour change. This helper parses the same way for that
reason, so the two cannot drift apart.
"""
assert validate_uuid(spelling, "tags") == spelling
class TestValidateUuidList:
def test_accepts_a_list(self):
assert validate_uuid_list([GOOD, OTHER], "tags") == [GOOD, OTHER]
def test_accepts_a_bare_string_as_a_single_value(self):
"""A body may send one id unwrapped.
``common/views/team_views.py`` used to pass such a string to
``users__id__in=``, which iterates it character by character and
silently matches nothing. Normalising here removes that trap.
"""
assert validate_uuid_list(GOOD, "assigned_users") == [GOOD]
def test_drops_nothing_and_rejects_the_whole_list_on_one_bad_entry(self):
with pytest.raises(ValidationError) as exc:
validate_uuid_list([GOOD, "notauuid"], "tags")
assert "notauuid" in str(exc.value)
def test_empty_input_is_an_empty_list(self):
assert validate_uuid_list(None, "tags") == []
assert validate_uuid_list([], "tags") == []
assert validate_uuid_list("", "tags") == []
def test_ignores_empty_entries_inside_a_list(self):
"""``?tags=&tags=<id>`` is a browser artefact, not a malformed id.
A `<select>` with a blank first option submits the empty string. That
should mean "no filter", not a 400.
"""
assert validate_uuid_list([GOOD, ""], "tags") == [GOOD]
class TestPayloadIdList:
"""The write paths take M2M ids three different shapes."""
def test_accepts_a_plain_list(self):
assert payload_id_list([GOOD, OTHER], "tags") == [GOOD, OTHER]
def test_accepts_a_json_encoded_list(self):
"""Multipart bodies send the list as JSON text."""
assert payload_id_list(f'["{GOOD}", "{OTHER}"]', "tags") == [GOOD, OTHER]
def test_accepts_a_list_of_objects_and_takes_their_ids(self):
value = [{"id": GOOD, "name": "urgent"}, {"id": OTHER, "name": "vip"}]
assert payload_id_list(value, "tags") == [GOOD, OTHER]
def test_accepts_a_bare_id_string(self):
"""Previously this either crashed or matched nothing.
The update paths ran ``json.loads`` on it, which raises
``JSONDecodeError`` and answered 500. The create paths skipped the
decode and passed the string straight to ``id__in``, which iterates it
character by character and quietly matched no rows.
"""
assert payload_id_list(GOOD, "tags") == [GOOD]
def test_malformed_json_is_a_400_not_a_500(self):
with pytest.raises(ValidationError) as exc:
payload_id_list("[not json", "tags")
assert "tags" in str(exc.value)
def test_a_malformed_id_inside_the_list_is_a_400(self):
with pytest.raises(ValidationError):
payload_id_list([GOOD, "notauuid"], "tags")
def test_a_malformed_id_inside_an_object_is_a_400(self):
with pytest.raises(ValidationError):
payload_id_list([{"id": "notauuid"}], "tags")
def test_empty_input_is_an_empty_list(self):
for empty in (None, "", [], "[]"):
assert payload_id_list(empty, "tags") == []
def test_a_scalar_that_is_not_a_string_is_a_400(self):
with pytest.raises(ValidationError):
payload_id_list({"id": GOOD}, "tags")
class TestQueryDictHelpers:
def test_uuid_param_reads_a_single_value(self):
params = QueryDict(f"account={GOOD}")
assert uuid_param(params, "account") == GOOD
def test_uuid_param_is_none_when_absent_or_blank(self):
assert uuid_param(QueryDict(""), "account") is None
assert uuid_param(QueryDict("account="), "account") is None
def test_uuid_param_rejects_a_malformed_value(self):
with pytest.raises(ValidationError):
uuid_param(QueryDict("account=notauuid"), "account")
def test_uuid_list_param_reads_every_repeat(self):
params = QueryDict(f"tags={GOOD}&tags={OTHER}")
assert uuid_list_param(params, "tags") == [GOOD, OTHER]
def test_uuid_list_param_is_empty_when_absent(self):
assert uuid_list_param(QueryDict(""), "tags") == []
def test_uuid_list_param_rejects_a_malformed_repeat(self):
params = QueryDict(f"tags={GOOD}&tags=notauuid")
with pytest.raises(ValidationError):
uuid_list_param(params, "tags")
def test_uuid_list_param_accepts_a_plain_dict(self):
"""Some callers hold ``request.data``, which is a dict, not a QueryDict."""
assert uuid_list_param({"tags": [GOOD]}, "tags") == [GOOD]

102
backend/common/validators.py
View file

@ -2,10 +2,13 @@
Common validators for CRM models.
"""
import json
import re
import uuid as uuid_module
from django.core.validators import RegexValidator
from django.utils.translation import gettext_lazy as _
from rest_framework.exceptions import ValidationError as DRFValidationError
# E.164 format: +[country code][number] (max 15 digits total)
@ -26,6 +29,105 @@ flexible_phone_validator = RegexValidator(
)
def validate_uuid(value, field: str) -> str:
"""Return ``value`` if it is a usable id, otherwise raise a DRF 400.
Id-valued filters (``tags``, ``assigned_to``, ``account``, ...) arrive as
raw query-string text and go straight into an ``id`` lookup. Django parses
them with ``uuid.UUID`` while building the query and raises
``django.core.exceptions.ValidationError`` on anything malformed. DRF's
exception handler does not translate that class, so the request answered
500 rather than 400: a stale bookmark or a hand-edited URL took the
endpoint down.
Parsing here with ``uuid.UUID`` deliberately mirrors what ``UUIDField``
does, so this accepts exactly what the ORM accepts (hyphenated, bare hex,
braced and URN forms alike) and cannot drift into rejecting requests that
used to work.
"""
try:
uuid_module.UUID(str(value))
except (AttributeError, TypeError, ValueError):
raise DRFValidationError(
{field: [f"'{value}' is not a valid id."]}
) from None
return value
def validate_uuid_list(value, field: str) -> list:
"""Normalise ``value`` to a list of ids, raising a DRF 400 on any bad one.
Accepts a list, a single bare string (a body may send one id unwrapped),
or nothing. Empty entries are dropped rather than rejected: a ``<select>``
with a blank first option submits ``""``, which means "no filter" and not
"malformed id".
The bare-string case also closes a silent bug. Passing a string to an
``__in`` lookup makes Django iterate it character by character, so the
filter matches nothing and reports no error.
"""
if value in (None, ""):
return []
if isinstance(value, str):
value = [value]
return [validate_uuid(v, field) for v in value if v not in (None, "")]
def payload_id_list(value, field: str) -> list:
"""Ids from a request-body value, whatever shape the client sent.
The M2M write paths (``tags``, ``contacts``, ``teams``, ``assigned_to``)
accept three shapes: a real list of ids, a JSON-encoded list because a
multipart body cannot carry one natively, and a list of
``{"id": ..., "name": ...}`` objects because some clients echo back what
the API gave them. Each view used to unpack that inline, and every copy
had the same two holes: ``json.loads`` on malformed text raised
``JSONDecodeError`` and answered 500, and a malformed id reached the ORM
and answered 500 as well.
A bare id string is now accepted as a single-item list. Before, the update
paths ran it through ``json.loads`` and crashed, while the create paths
handed it to ``id__in`` unchanged, where Django iterates the string
character by character and silently matches nothing.
"""
if value in (None, ""):
return []
if isinstance(value, str):
text = value.strip()
if text.startswith("["):
try:
value = json.loads(text)
except json.JSONDecodeError:
raise DRFValidationError(
{field: [f"{field} must be a list of ids."]}
) from None
else:
value = [text]
if not isinstance(value, (list, tuple)):
raise DRFValidationError({field: [f"{field} must be a list of ids."]})
ids = [item.get("id") if isinstance(item, dict) else item for item in value]
return validate_uuid_list(ids, field)
def uuid_param(params, field: str):
"""A single id-valued query parameter, or ``None`` when absent or blank."""
value = params.get(field)
if value in (None, ""):
return None
return validate_uuid(value, field)
def uuid_list_param(params, field: str) -> list:
"""Every id-valued value for a repeatable query parameter.
Uses ``getlist`` when the mapping has it (``QueryDict``) and falls back to
``get`` for a plain dict, which is what ``request.data`` can be.
"""
if hasattr(params, "getlist"):
return validate_uuid_list(params.getlist(field), field)
return validate_uuid_list(params.get(field), field)
def normalize_phone(phone: str) -> str:
"""
Normalize a phone number by removing all non-digit characters.

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

@ -6,12 +6,12 @@ from rest_framework import serializers, status
from rest_framework.pagination import LimitOffsetPagination
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.utils import json
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.validators import payload_id_list
from common.serializer import (
DocumentCreateSerializer,
DocumentCreateSwaggerSerializer,
@ -74,10 +74,13 @@ class DocumentListView(APIView, LimitOffsetPagination):
if request_post.get("status"):
queryset = queryset.filter(status=request_post.get("status"))
if request_post.get("shared_to"):
queryset = queryset.filter(
shared_to__id__in=json.loads(request_post.get("shared_to"))
)
# `json.loads` on the raw value meant `?shared_to=<id>`, the
# obvious spelling, raised JSONDecodeError and answered 500. Only a
# JSON array ever worked, and a malformed id inside one answered
# 500 as well.
shared_to = payload_id_list(request_post.get("shared_to"), "shared_to")
if shared_to:
queryset = queryset.filter(shared_to__id__in=shared_to)
context = {}
profile_list = Profile.objects.filter(
@ -194,13 +197,7 @@ class DocumentListView(APIView, LimitOffsetPagination):
)
if params.get("shared_to"):
assinged_to_list = params.get("shared_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "shared_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -208,13 +205,7 @@ class DocumentListView(APIView, LimitOffsetPagination):
doc.shared_to.add(*profiles)
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
if teams:
doc.teams.add(*teams)
@ -387,13 +378,7 @@ class DocumentDetailView(APIView):
doc.shared_to.clear()
if params.get("shared_to"):
assinged_to_list = params.get("shared_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "shared_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -403,13 +388,7 @@ class DocumentDetailView(APIView):
doc.teams.clear()
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
if teams:
doc.teams.add(*teams)

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

@ -14,6 +14,7 @@ from common.serializer import (
TeamswaggerCreateSerializer,
)
from common.tasks import remove_users, update_team_users
from common.validators import uuid_list_param, uuid_param
class TeamsListView(APIView, LimitOffsetPagination):
@ -28,10 +29,15 @@ class TeamsListView(APIView, LimitOffsetPagination):
if params:
if params.get("team_name"):
queryset = queryset.filter(name__icontains=params.get("team_name"))
if params.get("created_by"):
queryset = queryset.filter(created_by=params.get("created_by"))
if params.get("assigned_users"):
queryset = queryset.filter(users__id__in=params.get("assigned_users"))
created_by = uuid_param(params, "created_by")
if created_by:
queryset = queryset.filter(created_by=created_by)
# `get` rather than `getlist` handed `__in` a single string, which
# Django iterates character by character, so this filter matched
# nothing and said nothing about why.
assigned_users = uuid_list_param(params, "assigned_users")
if assigned_users:
queryset = queryset.filter(users__id__in=assigned_users)
context = {}
results_teams = self.paginate_queryset(

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

@ -81,7 +81,7 @@ class GetTeamsAndUsersView(APIView):
class UsersListView(APIView, LimitOffsetPagination):
permission_classes = (IsAuthenticated,)
permission_classes = (IsAuthenticated, HasOrgContext)
@extend_schema(
tags=["users"],
@ -272,7 +272,7 @@ class UsersListView(APIView, LimitOffsetPagination):
class UserDetailView(APIView):
permission_classes = (IsAuthenticated,)
permission_classes = (IsAuthenticated, HasOrgContext)
def get_object(self, pk):
# Security fix: Filter by org to prevent cross-org enumeration
@ -529,7 +529,7 @@ class UserDetailView(APIView):
class UserStatusView(APIView):
permission_classes = (IsAuthenticated,)
permission_classes = (IsAuthenticated, HasOrgContext)
@extend_schema(
tags=["users"],

61
backend/contacts/views.py
View file

@ -33,6 +33,7 @@ from common.serializer import (
CustomFieldDefinitionSerializer,
)
from common.utils import COUNTRIES
from common.validators import payload_id_list, uuid_list_param
from contacts import swagger_params
from contacts.models import Contact
from contacts.serializer import (
@ -101,18 +102,18 @@ class ContactsListView(APIView, LimitOffsetPagination):
queryset = queryset.filter(phone__icontains=params.get("phone"))
if params.get("email"):
queryset = queryset.filter(email__icontains=params.get("email"))
if params.getlist("assigned_to"):
assigned_to = uuid_list_param(params, "assigned_to")
if assigned_to:
# `getlist` to ask and `get` to read gave `__in` a single
# string, which Django iterates character by character -- each
# character then failed to parse as a UUID, so filtering by an
# owner answered 500.
queryset = queryset.filter(
assigned_to__id__in=params.getlist("assigned_to")
).distinct()
if params.get("tags"):
queryset = queryset.filter(
tags__id__in=params.getlist("tags")
assigned_to__id__in=assigned_to
).distinct()
tags = uuid_list_param(params, "tags")
if tags:
queryset = queryset.filter(tags__id__in=tags).distinct()
if params.get("search"):
search = params.get("search")
queryset = queryset.filter(
@ -246,25 +247,13 @@ class ContactsListView(APIView, LimitOffsetPagination):
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
contact_obj.teams.add(*teams)
if params.get("assigned_to"):
assinged_to_list = params.get("assigned_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org
)
@ -478,26 +467,14 @@ class ContactDetailView(APIView):
contact_obj.teams.clear()
if data.get("teams"):
teams_list = data.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
contact_obj.teams.add(*teams)
contact_obj.assigned_to.clear()
if data.get("assigned_to"):
assinged_to_list = data.get("assigned_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org
)
@ -834,13 +811,7 @@ class ContactDetailView(APIView):
contact_obj.teams.clear()
teams_list = data.get("teams")
if teams_list:
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
contact_obj.teams.add(*teams)
@ -848,13 +819,7 @@ class ContactDetailView(APIView):
contact_obj.assigned_to.clear()
assigned_to_list = data.get("assigned_to")
if assigned_to_list:
if isinstance(assigned_to_list, str):
assigned_to_list = json.loads(assigned_to_list)
# Extract IDs if assigned_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assigned_to_list
]
assigned_ids = payload_id_list(assigned_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org
)

27
backend/invoices/api_views.py
View file

@ -20,6 +20,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.validators import uuid_param
from common.serializer import (
AttachmentsSerializer,
CommentSerializer,
@ -121,25 +122,21 @@ class InvoiceListView(APIView, LimitOffsetPagination):
if params.get("status"):
queryset = queryset.filter(status=params.get("status"))
# Filter by account
if params.get("account"):
queryset = queryset.filter(account_id=params.get("account"))
# Filter by contact
if params.get("contact"):
queryset = queryset.filter(contact_id=params.get("contact"))
# Filter by opportunity
if params.get("opportunity"):
queryset = queryset.filter(opportunity_id=params.get("opportunity"))
# Filter by the related record, when one was named
for related in ("account", "contact", "opportunity"):
related_id = uuid_param(params, related)
if related_id:
queryset = queryset.filter(**{f"{related}_id": related_id})
# Filter by assigned user
if params.get("assigned_to"):
queryset = queryset.filter(assigned_to__id=params.get("assigned_to"))
assigned_to = uuid_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(assigned_to__id=assigned_to)
# Filter by created_by
if params.get("created_by"):
queryset = queryset.filter(created_by__id=params.get("created_by"))
created_by = uuid_param(params, "created_by")
if created_by:
queryset = queryset.filter(created_by__id=created_by)
# Filter by date range
if params.get("issue_date_gte"):

21
backend/invoices/tests/test_invoices_api.py
View file

@ -5,7 +5,6 @@ from unittest.mock import patch
import pytest
from django.utils import timezone
from rest_framework.exceptions import PermissionDenied
from accounts.models import Account
from contacts.models import Contact
@ -252,12 +251,20 @@ class TestInvoiceListView:
assert "invoice" in data
def test_create_invoice_unauthenticated(self, unauthenticated_client):
with pytest.raises(PermissionDenied):
unauthenticated_client.post(
"/api/invoices/",
{"invoice_title": "Unauthenticated Invoice"},
format="json",
)
"""The request is refused, and refused as a response, not an exception.
DRF catches ``PermissionDenied`` in ``handle_exception`` and renders it,
so a test client never sees it raised. Wrapping the call in
``pytest.raises`` therefore failed with DID NOT RAISE no matter how the
endpoint behaved, which meant this test could never have caught the
access opening up.
"""
response = unauthenticated_client.post(
"/api/invoices/",
{"invoice_title": "Unauthenticated Invoice"},
format="json",
)
assert response.status_code == 403
def test_org_isolation(self, org_b_client, invoice):
"""org_b_client should not see invoices belonging to org_a."""

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

@ -15,6 +15,7 @@ from rest_framework.response import Response
from rest_framework.views import APIView
from common.permissions import HasOrgContext
from common.validators import uuid_param
from common.utils import LEAD_STATUS
from leads.models import Lead, LeadPipeline, LeadStage
from leads.serializer import (
@ -71,7 +72,7 @@ class LeadKanbanView(APIView):
def get(self, request):
"""Get kanban board data."""
org = request.profile.org
pipeline_id = request.query_params.get("pipeline_id")
pipeline_id = uuid_param(request.query_params, "pipeline_id")
# Base queryset with filters
queryset = (
@ -98,8 +99,9 @@ class LeadKanbanView(APIView):
def _apply_filters(self, queryset, params):
"""Apply common filters to queryset."""
if params.get("assigned_to"):
queryset = queryset.filter(assigned_to__id=params.get("assigned_to"))
assigned_to = uuid_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(assigned_to__id=assigned_to)
if params.get("rating"):
queryset = queryset.filter(rating=params.get("rating"))
if params.get("search"):

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

@ -34,6 +34,7 @@ from common.serializer import (
TeamsSerializer,
)
from common.utils import COUNTRIES, INDCHOICES, LEAD_SOURCE, LEAD_STATUS
from common.validators import payload_id_list, uuid_list_param, validate_uuid_list
from contacts.models import Contact
from leads import swagger_params
from leads.models import Lead
@ -114,16 +115,14 @@ class LeadListView(APIView, LimitOffsetPagination):
)
if params.get("source"):
queryset = queryset.filter(source=params.get("source"))
if params.getlist("assigned_to"):
queryset = queryset.filter(
assigned_to__id__in=params.getlist("assigned_to")
)
assigned_to = uuid_list_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(assigned_to__id__in=assigned_to)
if params.get("status"):
queryset = queryset.filter(status=params.get("status"))
if params.getlist("tags"):
queryset = queryset.filter(
tags__id__in=params.getlist("tags")
)
tags = uuid_list_param(params, "tags")
if tags:
queryset = queryset.filter(tags__id__in=tags)
if params.get("city"):
queryset = queryset.filter(city__icontains=params.get("city"))
if params.get("email"):
@ -315,16 +314,17 @@ class LeadListView(APIView, LimitOffsetPagination):
},
status=status.HTTP_400_BAD_REQUEST,
)
if data.get("tags", None):
tags = data.get("tags")
tags = validate_uuid_list(data.get("tags"), "tags")
if tags:
tag_objs = Tags.objects.filter(
id__in=tags, org=request.profile.org, is_active=True
)
lead_obj.tags.add(*tag_objs)
if data.get("contacts", None):
contacts = validate_uuid_list(data.get("contacts"), "contacts")
if contacts:
obj_contact = Contact.objects.filter(
id__in=data.get("contacts"), org=request.profile.org
id__in=contacts, org=request.profile.org
)
lead_obj.contacts.add(*obj_contact)
@ -339,25 +339,13 @@ class LeadListView(APIView, LimitOffsetPagination):
if data.get("teams", None):
teams_list = data.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
lead_obj.teams.add(*teams)
if data.get("assigned_to", None):
assinged_to_list = data.get("assigned_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org
)
@ -713,8 +701,8 @@ class LeadDetailView(APIView):
save_kwargs["custom_fields"] = cleaned_cf
lead_obj = serializer.save(**save_kwargs)
lead_obj.tags.clear()
if params.get("tags"):
tags = params.get("tags")
tags = validate_uuid_list(params.get("tags"), "tags")
if tags:
tag_objs = Tags.objects.filter(
id__in=tags, org=request.profile.org, is_active=True
)
@ -735,10 +723,13 @@ class LeadDetailView(APIView):
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = validate_uuid_list(
[
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
],
"contacts",
)
obj_contact = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -747,13 +738,7 @@ class LeadDetailView(APIView):
lead_obj.teams.clear()
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
lead_obj.teams.add(*teams)
@ -763,10 +748,13 @@ class LeadDetailView(APIView):
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = validate_uuid_list(
[
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
],
"assigned_to",
)
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org
)
@ -1001,13 +989,7 @@ class LeadDetailView(APIView):
lead_obj.tags.clear()
tags = params.get("tags")
if tags:
if isinstance(tags, str):
tags = json.loads(tags)
# Extract IDs if tags contains objects with 'id' field
tag_ids = [
tag.get("id") if isinstance(tag, dict) else tag
for tag in tags
]
tag_ids = payload_id_list(tags, "tags")
tag_objs = Tags.objects.filter(
id__in=tag_ids, org=request.profile.org, is_active=True
)
@ -1017,13 +999,7 @@ class LeadDetailView(APIView):
lead_obj.contacts.clear()
contacts_list = params.get("contacts")
if contacts_list:
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
obj_contact = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -1033,13 +1009,7 @@ class LeadDetailView(APIView):
lead_obj.teams.clear()
teams_list = params.get("teams")
if teams_list:
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(
id__in=team_ids, org=request.profile.org
)
@ -1049,13 +1019,7 @@ class LeadDetailView(APIView):
lead_obj.assigned_to.clear()
assigned_to_list = params.get("assigned_to")
if assigned_to_list:
if isinstance(assigned_to_list, str):
assigned_to_list = json.loads(assigned_to_list)
# Extract IDs if assigned_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assigned_to_list
]
assigned_ids = payload_id_list(assigned_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org
)

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

@ -9,6 +9,7 @@ from rest_framework.views import APIView
from common.models import Profile
from common.permissions import HasOrgContext
from common.validators import uuid_param
from opportunity.models import SalesGoal
from opportunity.serializer import SalesGoalCreateSerializer, SalesGoalSerializer
@ -35,10 +36,12 @@ class SalesGoalListView(APIView, LimitOffsetPagination):
queryset = queryset.filter(
period_start__lte=today, period_end__gte=today
)
if params.get("assigned_to"):
queryset = queryset.filter(assigned_to_id=params["assigned_to"])
if params.get("team"):
queryset = queryset.filter(team_id=params["team"])
assigned_to = uuid_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(assigned_to_id=assigned_to)
team = uuid_param(params, "team")
if team:
queryset = queryset.filter(team_id=team)
if params.get("period_type"):
queryset = queryset.filter(period_type=params["period_type"])
if params.get("search"):

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

@ -17,6 +17,7 @@ from rest_framework.response import Response
from rest_framework.views import APIView
from common.permissions import HasOrgContext
from common.validators import uuid_list_param, uuid_param
from opportunity.models import Opportunity, StageAgingConfig
from opportunity.serializer import (
OpportunityKanbanCardSerializer,
@ -122,14 +123,15 @@ class OpportunityKanbanView(APIView):
def _apply_filters(self, queryset, params):
if params.get("search"):
queryset = queryset.filter(name__icontains=params.get("search"))
if params.get("account"):
queryset = queryset.filter(account_id=params.get("account"))
if params.getlist("assigned_to"):
queryset = queryset.filter(
assigned_to__id__in=params.getlist("assigned_to")
).distinct()
if params.getlist("tags"):
queryset = queryset.filter(tags__id__in=params.getlist("tags")).distinct()
account = uuid_param(params, "account")
if account:
queryset = queryset.filter(account_id=account)
assigned_to = uuid_list_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(assigned_to__id__in=assigned_to).distinct()
tags = uuid_list_param(params, "tags")
if tags:
queryset = queryset.filter(tags__id__in=tags).distinct()
if params.get("closed_on__gte"):
queryset = queryset.filter(closed_on__gte=params.get("closed_on__gte"))
if params.get("closed_on__lte"):

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

@ -26,6 +26,7 @@ from common.serializer import (
ProfileSerializer,
)
from common.utils import CURRENCY_CODES, SOURCES, STAGES
from common.validators import payload_id_list, uuid_list_param, uuid_param
from contacts.models import Contact
from contacts.serializer import ContactSerializer
from opportunity import swagger_params
@ -133,21 +134,22 @@ class OpportunityListView(APIView, LimitOffsetPagination):
if params:
if params.get("name"):
queryset = queryset.filter(name__icontains=params.get("name"))
if params.get("account"):
queryset = queryset.filter(account=params.get("account"))
account = uuid_param(params, "account")
if account:
queryset = queryset.filter(account=account)
if params.get("stage"):
queryset = queryset.filter(stage__contains=params.get("stage"))
if params.get("lead_source"):
queryset = queryset.filter(
lead_source__contains=params.get("lead_source")
)
if params.getlist("tags"):
tags = uuid_list_param(params, "tags")
if tags:
queryset = queryset.filter(tags__id__in=tags).distinct()
assigned_to = uuid_list_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(
tags__id__in=params.getlist("tags")
).distinct()
if params.getlist("assigned_to"):
queryset = queryset.filter(
assigned_to__id__in=params.getlist("assigned_to")
assigned_to__id__in=assigned_to
).distinct()
if params.get("search"):
queryset = queryset.filter(name__icontains=params.get("search"))
@ -316,13 +318,7 @@ class OpportunityListView(APIView, LimitOffsetPagination):
if params.get("contacts"):
contacts_list = params.get("contacts")
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -330,13 +326,7 @@ class OpportunityListView(APIView, LimitOffsetPagination):
if params.get("tags"):
tags = params.get("tags")
if isinstance(tags, str):
tags = json.loads(tags)
# Extract IDs if tags contains objects with 'id' field
tag_ids = [
item.get("id") if isinstance(item, dict) else item
for item in tags
]
tag_ids = payload_id_list(tags, "tags")
tag_objs = Tags.objects.filter(
id__in=tag_ids, org=request.profile.org, is_active=True
)
@ -351,25 +341,13 @@ class OpportunityListView(APIView, LimitOffsetPagination):
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
opportunity_obj.teams.add(*teams)
if params.get("assigned_to"):
assinged_to_list = params.get("assigned_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -505,13 +483,7 @@ class OpportunityDetailView(APIView):
opportunity_object.contacts.clear()
if params.get("contacts"):
contacts_list = params.get("contacts")
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -520,13 +492,7 @@ class OpportunityDetailView(APIView):
opportunity_object.tags.clear()
if params.get("tags"):
tags = params.get("tags")
if isinstance(tags, str):
tags = json.loads(tags)
# Extract IDs if tags contains objects with 'id' field
tag_ids = [
item.get("id") if isinstance(item, dict) else item
for item in tags
]
tag_ids = payload_id_list(tags, "tags")
tag_objs = Tags.objects.filter(
id__in=tag_ids, org=request.profile.org, is_active=True
)
@ -541,26 +507,14 @@ class OpportunityDetailView(APIView):
opportunity_object.teams.clear()
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
opportunity_object.teams.add(*teams)
opportunity_object.assigned_to.clear()
if params.get("assigned_to"):
assinged_to_list = params.get("assigned_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -878,13 +832,7 @@ class OpportunityDetailView(APIView):
opportunity_object.contacts.clear()
contacts_list = params.get("contacts")
if contacts_list:
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -894,13 +842,7 @@ class OpportunityDetailView(APIView):
opportunity_object.tags.clear()
tags = params.get("tags")
if tags:
if isinstance(tags, str):
tags = json.loads(tags)
# Extract IDs if tags contains objects with 'id' field
tag_ids = [
tag.get("id") if isinstance(tag, dict) else tag
for tag in tags
]
tag_ids = payload_id_list(tags, "tags")
tag_objs = Tags.objects.filter(
id__in=tag_ids, org=request.profile.org, is_active=True
)
@ -910,13 +852,7 @@ class OpportunityDetailView(APIView):
opportunity_object.teams.clear()
teams_list = params.get("teams")
if teams_list:
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(
id__in=team_ids, org=request.profile.org
)
@ -926,13 +862,7 @@ class OpportunityDetailView(APIView):
opportunity_object.assigned_to.clear()
assigned_to_list = params.get("assigned_to")
if assigned_to_list:
if isinstance(assigned_to_list, str):
assigned_to_list = json.loads(assigned_to_list)
# Extract IDs if assigned_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assigned_to_list
]
assigned_ids = payload_id_list(assigned_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)

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

@ -15,6 +15,7 @@ from rest_framework.response import Response
from rest_framework.views import APIView
from common.permissions import HasOrgContext
from common.validators import uuid_param
from tasks.models import Task, TaskPipeline, TaskStage
from tasks.serializer import (
TaskKanbanCardSerializer,
@ -106,7 +107,7 @@ class TaskKanbanView(APIView):
def get(self, request):
"""Get kanban board data."""
org = request.profile.org
pipeline_id = request.query_params.get("pipeline_id")
pipeline_id = uuid_param(request.query_params, "pipeline_id")
# Base queryset with filters
queryset = (
@ -134,8 +135,9 @@ class TaskKanbanView(APIView):
def _apply_filters(self, queryset, params):
"""Apply common filters to queryset."""
if params.get("assigned_to"):
queryset = queryset.filter(assigned_to__id=params.get("assigned_to"))
assigned_to = uuid_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(assigned_to__id=assigned_to)
if params.get("priority"):
queryset = queryset.filter(priority=params.get("priority"))
if params.get("search"):
@ -147,16 +149,13 @@ class TaskKanbanView(APIView):
queryset = queryset.filter(due_date__gte=params.get("due_date__gte"))
if params.get("due_date__lte"):
queryset = queryset.filter(due_date__lte=params.get("due_date__lte"))
if params.get("account"):
queryset = queryset.filter(account_id=params.get("account"))
if params.get("lead"):
queryset = queryset.filter(lead_id=params.get("lead"))
if params.get("opportunity"):
queryset = queryset.filter(opportunity_id=params.get("opportunity"))
if params.get("case"):
queryset = queryset.filter(case_id=params.get("case"))
if params.get("tags"):
queryset = queryset.filter(tags__id=params.get("tags"))
for related in ("account", "lead", "opportunity", "case"):
related_id = uuid_param(params, related)
if related_id:
queryset = queryset.filter(**{f"{related}_id": related_id})
tags = uuid_param(params, "tags")
if tags:
queryset = queryset.filter(tags__id=tags)
return queryset
def _get_status_kanban(self, queryset):

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

@ -25,6 +25,7 @@ from common.models import (
Teams,
)
from common.permissions import HasOrgContext
from common.validators import payload_id_list, uuid_list_param, uuid_param
from common.serializer import (
AttachmentsSerializer,
CommentSerializer,
@ -90,14 +91,14 @@ class TaskListView(APIView, LimitOffsetPagination):
queryset = queryset.filter(status=params.get("status"))
if params.get("priority"):
queryset = queryset.filter(priority=params.get("priority"))
if params.getlist("assigned_to"):
assigned_to = uuid_list_param(params, "assigned_to")
if assigned_to:
queryset = queryset.filter(
assigned_to__id__in=params.getlist("assigned_to")
).distinct()
if params.get("tags"):
queryset = queryset.filter(
tags__id__in=params.getlist("tags")
assigned_to__id__in=assigned_to
).distinct()
tags = uuid_list_param(params, "tags")
if tags:
queryset = queryset.filter(tags__id__in=tags).distinct()
if params.get("search"):
queryset = queryset.filter(title__icontains=params.get("search"))
if params.get("due_date__gte"):
@ -112,14 +113,10 @@ class TaskListView(APIView, LimitOffsetPagination):
queryset = queryset.filter(
created_at__lte=params.get("created_at__lte")
)
if params.get("account"):
queryset = queryset.filter(account_id=params.get("account"))
if params.get("opportunity"):
queryset = queryset.filter(opportunity_id=params.get("opportunity"))
if params.get("case"):
queryset = queryset.filter(case_id=params.get("case"))
if params.get("lead"):
queryset = queryset.filter(lead_id=params.get("lead"))
for related in ("account", "opportunity", "case", "lead"):
related_id = uuid_param(params, related)
if related_id:
queryset = queryset.filter(**{f"{related}_id": related_id})
for raw_key, raw_value in params.items():
if raw_key.startswith("cf_") and raw_value:
cf_key = raw_key[3:]
@ -281,13 +278,7 @@ class TaskListView(APIView, LimitOffsetPagination):
)
if params.get("contacts"):
contacts_list = params.get("contacts")
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -295,25 +286,13 @@ class TaskListView(APIView, LimitOffsetPagination):
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
task_obj.teams.add(*teams)
if params.get("assigned_to"):
assinged_to_list = params.get("assigned_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -582,13 +561,7 @@ class TaskDetailView(APIView):
task_obj.contacts.clear()
if params.get("contacts"):
contacts_list = params.get("contacts")
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -597,26 +570,14 @@ class TaskDetailView(APIView):
task_obj.teams.clear()
if params.get("teams"):
teams_list = params.get("teams")
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(id__in=team_ids, org=request.profile.org)
task_obj.teams.add(*teams)
task_obj.assigned_to.clear()
if params.get("assigned_to"):
assinged_to_list = params.get("assigned_to")
if isinstance(assinged_to_list, str):
assinged_to_list = json.loads(assinged_to_list)
# Extract IDs if assinged_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assinged_to_list
]
assigned_ids = payload_id_list(assinged_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -710,13 +671,7 @@ class TaskDetailView(APIView):
task_obj.contacts.clear()
contacts_list = params.get("contacts")
if contacts_list:
if isinstance(contacts_list, str):
contacts_list = json.loads(contacts_list)
# Extract IDs if contacts_list contains objects with 'id' field
contact_ids = [
item.get("id") if isinstance(item, dict) else item
for item in contacts_list
]
contact_ids = payload_id_list(contacts_list, "contacts")
contacts = Contact.objects.filter(
id__in=contact_ids, org=request.profile.org
)
@ -726,13 +681,7 @@ class TaskDetailView(APIView):
task_obj.teams.clear()
teams_list = params.get("teams")
if teams_list:
if isinstance(teams_list, str):
teams_list = json.loads(teams_list)
# Extract IDs if teams_list contains objects with 'id' field
team_ids = [
item.get("id") if isinstance(item, dict) else item
for item in teams_list
]
team_ids = payload_id_list(teams_list, "teams")
teams = Teams.objects.filter(
id__in=team_ids, org=request.profile.org
)
@ -742,13 +691,7 @@ class TaskDetailView(APIView):
task_obj.assigned_to.clear()
assigned_to_list = params.get("assigned_to")
if assigned_to_list:
if isinstance(assigned_to_list, str):
assigned_to_list = json.loads(assigned_to_list)
# Extract IDs if assigned_to_list contains objects with 'id' field
assigned_ids = [
item.get("id") if isinstance(item, dict) else item
for item in assigned_to_list
]
assigned_ids = payload_id_list(assigned_to_list, "assigned_to")
profiles = Profile.objects.filter(
id__in=assigned_ids, org=request.profile.org, is_active=True
)
@ -758,13 +701,7 @@ class TaskDetailView(APIView):
task_obj.tags.clear()
tags_list = params.get("tags")
if tags_list:
if isinstance(tags_list, str):
tags_list = json.loads(tags_list)
# Extract IDs if tags_list contains objects with 'id' field
tag_ids = [
tag.get("id") if isinstance(tag, dict) else tag
for tag in tags_list
]
tag_ids = payload_id_list(tags_list, "tags")
tag_objs = Tags.objects.filter(
id__in=tag_ids, org=request.profile.org, is_active=True
)