feat: add comprehensive backend API test suite with 193 tests and CI

Add pytest-based test infrastructure with SQLite for fast isolated testing.
Tests cover all 8 app modules (common, accounts, leads, contacts, opportunity,
cases, tasks, invoices) including CRUD, org isolation, kanban, comments, and
auth endpoints. Achieves 54% coverage with a 50% threshold.

Also fixes 10 backend bugs discovered by the test suite:
- Fix serializer read_only_fields for org (teams, documents, board columns/tasks)
- Fix AccountCommentView.put missing partial=True
- Fix UserDetailView broken reverse relation and missing org on address save
- Fix DocumentDetailView.put partial update and null status bug
- Fix TaskDetailView unhandled DoesNotExist → get_object_or_404
- Fix BoardColumn/BoardTask views missing org on save
- Fix LeadListView.post unhandled IntegrityError for duplicate email
- Fix ContactDetailView.get wrong reverse relation name
- Fix OpportunityDetailView null checks before accessing .org

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.coveragerc

@@ -1,3 +1,17 @@
+[run]
+source = .
+omit =
+    */venv/*
+    */tests/*
+    */test_*
+    */migrations/*
+    */apps.py
+    */admin.py
+    manage.py
+    setup.py
+    crm/wsgi.py
+    crm/asgi.py
+
 [report]
 exclude_lines =
     # lines with this comment are not hit
@@ -20,6 +34,9 @@ exclude_lines =
     except Exception as e:
     except Exception:
 
+show_missing = True
+fail_under = 50
+
 omit =
     *tests*
     ./settings/*

.github/workflows/tests.yml

@@ -0,0 +1,77 @@
+name: Backend Tests
+
+on:
+  push:
+    branches: [master]
+    paths:
+      - 'backend/**'
+      - '.coveragerc'
+      - '.github/workflows/tests.yml'
+  pull_request:
+    branches: [master]
+    paths:
+      - 'backend/**'
+      - '.coveragerc'
+      - '.github/workflows/tests.yml'
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python 3.12
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+          cache: 'pip'
+          cache-dependency-path: backend/requirements.txt
+
+      - 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
+
+      - name: Install Python dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r backend/requirements.txt
+
+      - name: Run tests with coverage
+        working-directory: backend
+        env:
+          DJANGO_SETTINGS_MODULE: crm.test_settings
+          SECRET_KEY: test-secret-key-for-ci
+          ADMIN_EMAIL: admin@test.com
+        run: |
+          pytest -m "not postgres_only" --tb=short -v
+
+      - name: Generate coverage badge
+        if: github.ref == 'refs/heads/master' && github.event_name == 'push'
+        working-directory: backend
+        run: |
+          pip install coverage-badge
+          coverage-badge -o ../coverage-badge.svg -f
+
+      - name: Commit coverage badge
+        if: github.ref == 'refs/heads/master' && github.event_name == 'push'
+        run: |
+          git config --local user.email "github-actions[bot]@users.noreply.github.com"
+          git config --local user.name "github-actions[bot]"
+          git add coverage-badge.svg || true
+          git diff --staged --quiet || git commit -m "Update coverage badge [skip ci]"
+          git push || true
+
+      - name: Upload coverage report
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-report
+          path: backend/htmlcov/
+          retention-days: 14

.gitignore

@@ -37,6 +37,8 @@ ENV/
 .coverage
 .pytest_cache/
 htmlcov/
+coverage.xml
+coverage-badge.svg
 
 # Documentation
 docs/build/

README.md

@@ -7,6 +7,7 @@ A modern, open-source CRM platform built with Django REST Framework and SvelteKi
 ![Django](https://img.shields.io/badge/django-5.x-green.svg)
 ![SvelteKit](https://img.shields.io/badge/sveltekit-2.x-orange.svg)
 ![Svelte](https://img.shields.io/badge/svelte-5-orange.svg)
+![Coverage](./coverage-badge.svg)
 
 ## Overview
 
@@ -210,12 +211,31 @@ python manage.py manage_rls --test
 
 ## Development
 
+### Testing
+
+```bash
+cd backend
+
+# Run all tests with coverage
+pytest
+
+# Run tests without coverage (faster)
+pytest --no-cov -x
+
+# Run a specific module's tests
+pytest accounts/tests/
+pytest leads/tests/test_leads_kanban.py
+
+# Run tests matching a keyword
+pytest -k "test_login"
+
+# View coverage report in browser
+open htmlcov/index.html
+```
+
 ### Backend Commands
 
 ```bash
-# Run tests
-cd backend && pytest
-
 # Format code
 black . && isort .
 

backend/accounts/tests/test_accounts_api.py

@@ -0,0 +1,144 @@
+import pytest
+from django.contrib.contenttypes.models import ContentType
+from django.core.files.uploadedfile import SimpleUploadedFile
+from rest_framework.exceptions import PermissionDenied
+
+from accounts.models import Account
+from common.models import Attachments, Comment, Tags
+
+
+@pytest.mark.django_db
+class TestAccountListView:
+    """Tests for GET /api/accounts/ and POST /api/accounts/."""
+
+    def test_list_accounts(self, admin_client, org_a):
+        Account.objects.create(name="Test Account", org=org_a)
+        response = admin_client.get("/api/accounts/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "active_accounts" in data
+
+    def test_create_account(self, admin_client, org_a):
+        response = admin_client.post("/api/accounts/", {"name": "New Account"})
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+        assert Account.objects.filter(name="New Account", org=org_a).exists()
+
+    def test_create_account_with_tags(self, admin_client, org_a):
+        tag = Tags.objects.create(name="VIP", org=org_a)
+        response = admin_client.post(
+            "/api/accounts/",
+            {"name": "Tagged Account", "tags": [str(tag.id)]},
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+
+    def test_create_account_invalid_data(self, admin_client):
+        response = admin_client.post("/api/accounts/", {})
+        assert response.status_code == 400
+        data = response.json()
+        assert data["error"] is True
+
+    def test_create_account_unauthenticated(self, unauthenticated_client):
+        with pytest.raises(PermissionDenied):
+            unauthenticated_client.post(
+                "/api/accounts/", {"name": "Test"}
+            )
+
+
+@pytest.mark.django_db
+class TestAccountDetailView:
+    """Tests for GET/PUT/DELETE /api/accounts/<pk>/."""
+
+    def test_get_account_detail(self, admin_client, org_a):
+        account = Account.objects.create(name="Detail Account", org=org_a)
+        response = admin_client.get(f"/api/accounts/{account.id}/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "account_obj" in data
+
+    def test_update_account(self, admin_client, org_a):
+        account = Account.objects.create(name="Old Name", org=org_a)
+        response = admin_client.put(
+            f"/api/accounts/{account.id}/",
+            {"name": "New Name"},
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+
+    def test_delete_account(self, admin_client, org_a):
+        account = Account.objects.create(name="Delete Me", org=org_a)
+        response = admin_client.delete(f"/api/accounts/{account.id}/")
+        assert response.status_code == 200
+        assert not Account.objects.filter(id=account.id).exists()
+
+    def test_get_account_cross_org(self, org_b_client, org_a):
+        account = Account.objects.create(name="Org A Account", org=org_a)
+        response = org_b_client.get(f"/api/accounts/{account.id}/")
+        assert response.status_code == 404
+
+
+@pytest.mark.django_db
+class TestAccountCommentView:
+    """Tests for PUT/DELETE /api/accounts/comment/<pk>/."""
+
+    def test_edit_comment(self, admin_client, admin_profile, org_a):
+        account = Account.objects.create(name="Comment Account", org=org_a)
+        ct = ContentType.objects.get_for_model(Account)
+        comment = Comment.objects.create(
+            content_type=ct,
+            object_id=account.id,
+            comment="Test comment",
+            commented_by=admin_profile,
+            org=org_a,
+            created_by=admin_profile.user,
+            updated_by=admin_profile.user,
+        )
+        response = admin_client.put(
+            f"/api/accounts/comment/{comment.id}/",
+            {"comment": "Updated"},
+            format="json",
+        )
+        assert response.status_code == 200
+
+    def test_delete_comment(self, admin_client, admin_profile, org_a):
+        account = Account.objects.create(name="Comment Account", org=org_a)
+        ct = ContentType.objects.get_for_model(Account)
+        comment = Comment.objects.create(
+            content_type=ct,
+            object_id=account.id,
+            comment="Delete me",
+            commented_by=admin_profile,
+            org=org_a,
+            created_by=admin_profile.user,
+            updated_by=admin_profile.user,
+        )
+        response = admin_client.delete(
+            f"/api/accounts/comment/{comment.id}/"
+        )
+        assert response.status_code == 200
+
+
+@pytest.mark.django_db
+class TestAccountAttachmentView:
+    """Tests for DELETE /api/accounts/attachment/<pk>/."""
+
+    def test_delete_attachment(self, admin_client, admin_profile, org_a):
+        account = Account.objects.create(name="Attachment Account", org=org_a)
+        ct = ContentType.objects.get_for_model(Account)
+        attachment = Attachments.objects.create(
+            content_type=ct,
+            object_id=account.id,
+            file_name="test.txt",
+            attachment=SimpleUploadedFile("test.txt", b"content"),
+            org=org_a,
+            created_by=admin_profile.user,
+        )
+        response = admin_client.delete(
+            f"/api/accounts/attachment/{attachment.id}/"
+        )
+        assert response.status_code == 200

backend/accounts/views.py

@@ -711,7 +711,7 @@ class AccountCommentView(APIView):
             or request.profile.is_admin
             or request.profile == obj.commented_by
         ):
-            serializer = CommentSerializer(obj, data=data)
+            serializer = CommentSerializer(obj, data=data, partial=True)
             if data.get("comment"):
                 if serializer.is_valid():
                     serializer.save()

backend/cases/tests/test_cases_api.py

@@ -0,0 +1,166 @@
+"""
+Tests for the Case list and detail API views.
+
+Run with: pytest cases/tests/test_cases_api.py -v
+"""
+
+from unittest.mock import patch
+
+import pytest
+from django.db import connection
+from rest_framework import status
+from rest_framework.exceptions import PermissionDenied
+
+from cases.models import Case
+
+
+CASES_LIST_URL = "/api/cases/"
+
+
+def _detail_url(pk):
+    return f"/api/cases/{pk}/"
+
+
+def _set_rls(org):
+    """Set PostgreSQL RLS context so direct ORM writes are allowed.
+    No-op on SQLite (used in tests).
+    """
+    if connection.vendor != "postgresql":
+        return
+    with connection.cursor() as cursor:
+        cursor.execute(
+            "SELECT set_config('app.current_org', %s, false)", [str(org.id)]
+        )
+
+
+@pytest.fixture
+def case_a(admin_user, org_a):
+    """A case belonging to org_a, created by admin_user."""
+    _set_rls(org_a)
+    return Case.objects.create(
+        name="Bug in login page",
+        status="New",
+        priority="High",
+        case_type="Bug",
+        description="Login page crashes on submit.",
+        created_by=admin_user,
+        org=org_a,
+    )
+
+
+@pytest.fixture
+def case_b(user_b, org_b):
+    """A case belonging to org_b, created by user_b."""
+    _set_rls(org_b)
+    return Case.objects.create(
+        name="Feature request from Org B",
+        status="New",
+        priority="Low",
+        created_by=user_b,
+        org=org_b,
+    )
+
+
+@pytest.mark.django_db
+class TestCaseListView:
+    """Tests for GET /api/cases/ and POST /api/cases/"""
+
+    def test_list_cases(self, admin_client, case_a):
+        response = admin_client.get(CASES_LIST_URL)
+        assert response.status_code == status.HTTP_200_OK
+        assert "cases" in response.data
+        assert response.data["cases_count"] >= 1
+
+    @patch("cases.views.send_email_to_assigned_user")
+    def test_create_case(self, mock_email, admin_client):
+        response = admin_client.post(
+            CASES_LIST_URL,
+            {
+                "name": "New support ticket",
+                "status": "New",
+                "priority": "Normal",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        assert "id" in response.data
+
+    def test_create_case_unauthenticated(self, unauthenticated_client):
+        with pytest.raises(PermissionDenied):
+            unauthenticated_client.post(
+                CASES_LIST_URL,
+                {
+                    "name": "Should fail",
+                    "status": "New",
+                    "priority": "Normal",
+                },
+                format="json",
+            )
+
+    def test_org_isolation(self, org_b_client, case_a):
+        """org_b_client must not see cases belonging to org_a."""
+        response = org_b_client.get(CASES_LIST_URL)
+        assert response.status_code == status.HTTP_200_OK
+        case_ids = [c["id"] for c in response.data["cases"]]
+        assert str(case_a.id) not in case_ids
+
+    @patch("cases.views.send_email_to_assigned_user")
+    def test_create_case_invalid(self, mock_email, admin_client):
+        """POST without the required 'name' field should return 400."""
+        response = admin_client.post(
+            CASES_LIST_URL,
+            {
+                "status": "New",
+                "priority": "Normal",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.data["error"] is True
+
+
+@pytest.mark.django_db
+class TestCaseDetailView:
+    """Tests for GET/PUT/DELETE/POST /api/cases/<pk>/"""
+
+    def test_get_detail(self, admin_client, case_a):
+        response = admin_client.get(_detail_url(case_a.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["cases_obj"]["id"] == str(case_a.id)
+
+    @patch("cases.views.send_email_to_assigned_user")
+    def test_update_case(self, mock_email, admin_client, case_a):
+        response = admin_client.put(
+            _detail_url(case_a.pk),
+            {
+                "name": "Updated case name",
+                "status": "Assigned",
+                "priority": "Urgent",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        case_a.refresh_from_db()
+        assert case_a.name == "Updated case name"
+
+    def test_delete_case(self, admin_client, case_a):
+        response = admin_client.delete(_detail_url(case_a.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        assert not Case.objects.filter(pk=case_a.pk).exists()
+
+    def test_cross_org_detail_returns_404(self, org_b_client, case_a):
+        """A client from org_b cannot retrieve a case from org_a."""
+        response = org_b_client.get(_detail_url(case_a.pk))
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+    def test_add_comment(self, admin_client, case_a):
+        response = admin_client.post(
+            _detail_url(case_a.pk),
+            {"comment": "This is a test comment."},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert "comments" in response.data

backend/cases/tests/test_cases_kanban.py

@@ -0,0 +1,192 @@
+"""
+Tests for Case Kanban, Pipeline, and Stage API views.
+
+Run with: pytest cases/tests/test_cases_kanban.py -v
+"""
+
+import pytest
+from django.db import connection
+from rest_framework import status
+
+from cases.models import Case, CasePipeline, CaseStage
+
+
+PIPELINES_URL = "/api/cases/pipelines/"
+KANBAN_URL = "/api/cases/kanban/"
+
+
+def _pipeline_detail_url(pk):
+    return f"/api/cases/pipelines/{pk}/"
+
+
+def _stage_create_url(pipeline_pk):
+    return f"/api/cases/pipelines/{pipeline_pk}/stages/"
+
+
+def _stage_reorder_url(pipeline_pk):
+    return f"/api/cases/pipelines/{pipeline_pk}/stages/reorder/"
+
+
+def _stage_detail_url(pk):
+    return f"/api/cases/stages/{pk}/"
+
+
+def _set_rls(org):
+    """Set PostgreSQL RLS context so direct ORM writes are allowed.
+    No-op on SQLite (used in tests).
+    """
+    if connection.vendor != "postgresql":
+        return
+    with connection.cursor() as cursor:
+        cursor.execute(
+            "SELECT set_config('app.current_org', %s, false)", [str(org.id)]
+        )
+
+
+@pytest.fixture
+def pipeline(admin_user, org_a):
+    """A case pipeline belonging to org_a with no default stages."""
+    _set_rls(org_a)
+    return CasePipeline.objects.create(
+        name="Support Pipeline",
+        org=org_a,
+        created_by=admin_user,
+    )
+
+
+@pytest.fixture
+def stage_open(pipeline, admin_user, org_a):
+    """An 'open' stage inside the test pipeline."""
+    _set_rls(org_a)
+    return CaseStage.objects.create(
+        pipeline=pipeline,
+        name="New",
+        order=0,
+        color="#3B82F6",
+        stage_type="open",
+        org=org_a,
+        created_by=admin_user,
+    )
+
+
+@pytest.fixture
+def stage_closed(pipeline, admin_user, org_a):
+    """A 'closed' stage inside the test pipeline."""
+    _set_rls(org_a)
+    return CaseStage.objects.create(
+        pipeline=pipeline,
+        name="Resolved",
+        order=1,
+        color="#22C55E",
+        stage_type="closed",
+        org=org_a,
+        created_by=admin_user,
+    )
+
+
+@pytest.mark.django_db
+class TestCasePipeline:
+    """Tests for pipeline CRUD: /api/cases/pipelines/"""
+
+    def test_create_pipeline(self, admin_client):
+        response = admin_client.post(
+            PIPELINES_URL,
+            {
+                "name": "Engineering Pipeline",
+                "create_default_stages": False,
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.data["name"] == "Engineering Pipeline"
+
+    def test_list_pipelines(self, admin_client, pipeline):
+        response = admin_client.get(PIPELINES_URL)
+        assert response.status_code == status.HTTP_200_OK
+        assert "pipelines" in response.data
+        names = [p["name"] for p in response.data["pipelines"]]
+        assert "Support Pipeline" in names
+
+    def test_get_pipeline_detail(self, admin_client, pipeline):
+        response = admin_client.get(_pipeline_detail_url(pipeline.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["name"] == "Support Pipeline"
+
+    def test_delete_empty_pipeline(self, admin_client, pipeline):
+        """Deleting a pipeline with no cases should succeed with 204."""
+        response = admin_client.delete(_pipeline_detail_url(pipeline.pk))
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+
+@pytest.mark.django_db
+class TestCaseStage:
+    """Tests for stage CRUD and reorder."""
+
+    def test_create_stage(self, admin_client, pipeline):
+        response = admin_client.post(
+            _stage_create_url(pipeline.pk),
+            {
+                "name": "In Progress",
+                "order": 2,
+                "color": "#F59E0B",
+                "stage_type": "open",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.data["name"] == "In Progress"
+
+    def test_update_stage(self, admin_client, stage_open):
+        response = admin_client.put(
+            _stage_detail_url(stage_open.pk),
+            {
+                "name": "Triaged",
+                "color": "#8B5CF6",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        stage_open.refresh_from_db()
+        assert stage_open.name == "Triaged"
+
+    def test_delete_empty_stage(self, admin_client, stage_open):
+        """Deleting a stage with no cases should succeed with 204."""
+        response = admin_client.delete(_stage_detail_url(stage_open.pk))
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert not CaseStage.objects.filter(pk=stage_open.pk).exists()
+
+    def test_reorder_stages(self, admin_client, pipeline, stage_open, stage_closed):
+        """Reorder stages so that 'Resolved' comes before 'New'."""
+        response = admin_client.post(
+            _stage_reorder_url(pipeline.pk),
+            {
+                "stage_ids": [str(stage_closed.pk), str(stage_open.pk)],
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        stage_closed.refresh_from_db()
+        stage_open.refresh_from_db()
+        assert stage_closed.order < stage_open.order
+
+
+@pytest.mark.django_db
+class TestCaseKanban:
+    """Tests for the kanban board view: GET /api/cases/kanban/"""
+
+    def test_kanban_view(self, admin_client, org_a, admin_user):
+        """The default status-based kanban view should return columns."""
+        # Create a case so there is data (RLS context set by _set_rls)
+        _set_rls(org_a)
+        Case.objects.create(
+            name="Kanban test case",
+            status="New",
+            priority="Normal",
+            created_by=admin_user,
+            org=org_a,
+        )
+        response = admin_client.get(KANBAN_URL)
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["mode"] == "status"
+        assert "columns" in response.data
+        assert response.data["total_cases"] >= 1

backend/cases/tests/test_solutions.py

@@ -0,0 +1,129 @@
+"""
+Tests for the Solution (Knowledge Base) API views.
+
+Run with: pytest cases/tests/test_solutions.py -v
+"""
+
+import pytest
+from django.db import connection
+from rest_framework import status
+
+from cases.models import Solution
+
+
+SOLUTIONS_LIST_URL = "/api/cases/solutions/"
+
+
+def _detail_url(pk):
+    return f"/api/cases/solutions/{pk}/"
+
+
+def _publish_url(pk):
+    return f"/api/cases/solutions/{pk}/publish/"
+
+
+def _unpublish_url(pk):
+    return f"/api/cases/solutions/{pk}/unpublish/"
+
+
+def _set_rls(org):
+    """Set PostgreSQL RLS context so direct ORM writes are allowed.
+    No-op on SQLite (used in tests).
+    """
+    if connection.vendor != "postgresql":
+        return
+    with connection.cursor() as cursor:
+        cursor.execute(
+            "SELECT set_config('app.current_org', %s, false)", [str(org.id)]
+        )
+
+
+@pytest.fixture
+def solution_a(admin_user, org_a):
+    """A draft solution belonging to org_a."""
+    _set_rls(org_a)
+    return Solution.objects.create(
+        title="How to reset password",
+        description="Step-by-step guide for password reset.",
+        status="draft",
+        org=org_a,
+        created_by=admin_user,
+    )
+
+
+@pytest.fixture
+def approved_solution(admin_user, org_a):
+    """An approved (but unpublished) solution belonging to org_a."""
+    _set_rls(org_a)
+    return Solution.objects.create(
+        title="Troubleshooting login issues",
+        description="Common login issues and resolutions.",
+        status="approved",
+        is_published=False,
+        org=org_a,
+        created_by=admin_user,
+    )
+
+
+@pytest.mark.django_db
+class TestSolutionListView:
+    """Tests for GET /api/cases/solutions/ and POST /api/cases/solutions/"""
+
+    def test_list_solutions(self, admin_client, solution_a):
+        response = admin_client.get(SOLUTIONS_LIST_URL)
+        assert response.status_code == status.HTTP_200_OK
+        assert "results" in response.data
+
+    def test_create_solution(self, admin_client):
+        response = admin_client.post(
+            SOLUTIONS_LIST_URL,
+            {
+                "title": "New Knowledge Base Article",
+                "description": "Helpful description.",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.data["title"] == "New Knowledge Base Article"
+        assert response.data["status"] == "draft"
+
+
+@pytest.mark.django_db
+class TestSolutionDetailView:
+    """Tests for GET/PUT/DELETE /api/cases/solutions/<pk>/"""
+
+    def test_get_solution(self, admin_client, solution_a):
+        response = admin_client.get(_detail_url(solution_a.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["title"] == solution_a.title
+
+    def test_update_solution(self, admin_client, solution_a):
+        response = admin_client.put(
+            _detail_url(solution_a.pk),
+            {
+                "title": "Updated title",
+                "description": "Updated description.",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        solution_a.refresh_from_db()
+        assert solution_a.title == "Updated title"
+
+    def test_publish_solution(self, admin_client, approved_solution):
+        """Publishing an approved solution should succeed."""
+        response = admin_client.post(_publish_url(approved_solution.pk))
+        assert response.status_code == status.HTTP_200_OK
+        approved_solution.refresh_from_db()
+        assert approved_solution.is_published is True
+
+    def test_unpublish_solution(self, admin_client, approved_solution):
+        """Unpublishing a published solution should succeed."""
+        # First publish it
+        approved_solution.publish()
+        assert approved_solution.is_published is True
+
+        response = admin_client.post(_unpublish_url(approved_solution.pk))
+        assert response.status_code == status.HTTP_200_OK
+        approved_solution.refresh_from_db()
+        assert approved_solution.is_published is False

backend/cases/urls.py

@@ -36,12 +36,7 @@ urlpatterns = [
         kanban_views.CaseStageDetailView.as_view(),
         name="stage_detail",
     ),
-    # Case detail routes (must be after specific routes due to pk pattern)
-    path("<str:pk>/", views.CaseDetailView.as_view()),
-    path("<str:pk>/move/", kanban_views.CaseMoveView.as_view(), name="case_move"),
-    path("comment/<str:pk>/", views.CaseCommentView.as_view()),
-    path("attachment/<str:pk>/", views.CaseAttachmentView.as_view()),
-    # Solutions (Knowledge Base) endpoints
+    # Solutions (Knowledge Base) endpoints (must be before <str:pk>/ to avoid conflicts)
     path(
         "solutions/", solution_views.SolutionListView.as_view(), name="solutions_list"
     ),
@@ -60,4 +55,9 @@ urlpatterns = [
         solution_views.SolutionUnpublishView.as_view(),
         name="solution_unpublish",
     ),
+    # Case detail routes (must be after all named routes due to <str:pk> pattern)
+    path("<str:pk>/", views.CaseDetailView.as_view()),
+    path("<str:pk>/move/", kanban_views.CaseMoveView.as_view(), name="case_move"),
+    path("comment/<str:pk>/", views.CaseCommentView.as_view()),
+    path("attachment/<str:pk>/", views.CaseAttachmentView.as_view()),
 ]

backend/common/serializer.py

@@ -463,13 +463,15 @@ class DocumentCreateSerializer(serializers.ModelSerializer):
                 raise serializers.ValidationError(
                     "Document with this Title already exists"
                 )
-        if Document.objects.filter(title__iexact=title, org=self.org).exists():
-            raise serializers.ValidationError("Document with this Title already exists")
+        else:
+            if Document.objects.filter(title__iexact=title, org=self.org).exists():
+                raise serializers.ValidationError("Document with this Title already exists")
         return title
 
     class Meta:
         model = Document
         fields = ["title", "document_file", "status", "org"]
+        read_only_fields = ["org"]
 
 
 def find_urls(string):
@@ -759,7 +761,7 @@ class TeamCreateSerializer(serializers.ModelSerializer):
             ):
                 raise serializers.ValidationError("Team already exists with this name")
         else:
-            if Teams.objects.filter(name__iexact=name).exists():
+            if Teams.objects.filter(name__iexact=name, org=self.org).exists():
                 raise serializers.ValidationError("Team already exists with this name")
         return name
 
@@ -772,6 +774,7 @@ class TeamCreateSerializer(serializers.ModelSerializer):
             "created_by",
             "org",
         )
+        read_only_fields = ("created_at", "created_by", "org")
 
 
 class TeamswaggerCreateSerializer(serializers.ModelSerializer):

backend/common/tests/test_auth.py

@@ -0,0 +1,215 @@
+"""
+Tests for authentication views: login, register, me, token refresh, org switch.
+
+Run with: pytest common/tests/test_auth.py -v
+"""
+
+import pytest
+from rest_framework import status
+
+from common.models import Org, Profile, User
+from common.serializer import OrgAwareRefreshToken
+
+
+@pytest.mark.django_db
+class TestLoginView:
+    """Tests for POST /api/auth/login/"""
+
+    url = "/api/auth/login/"
+
+    def test_login_success(self, unauthenticated_client, admin_user, admin_profile):
+        response = unauthenticated_client.post(
+            self.url,
+            {"email": "admin@test.com", "password": "testpass123"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert "access_token" in response.data
+        assert "refresh_token" in response.data
+        assert "current_org" in response.data
+
+    def test_login_wrong_password(self, unauthenticated_client, admin_user, admin_profile):
+        response = unauthenticated_client.post(
+            self.url,
+            {"email": "admin@test.com", "password": "wrongpassword"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    def test_login_nonexistent_user(self, unauthenticated_client):
+        response = unauthenticated_client.post(
+            self.url,
+            {"email": "nobody@test.com", "password": "testpass123"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    def test_login_returns_tokens_and_org(
+        self, unauthenticated_client, admin_user, admin_profile, org_a
+    ):
+        response = unauthenticated_client.post(
+            self.url,
+            {"email": "admin@test.com", "password": "testpass123"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.data
+        assert "access_token" in data
+        assert "refresh_token" in data
+        assert "user" in data
+        assert "current_org" in data
+        assert data["current_org"]["id"] == str(org_a.id)
+
+    def test_login_with_specific_org(self, unauthenticated_client, admin_user, org_b):
+        # Give admin_user access to org_b
+        profile_b = Profile.objects.create(
+            user=admin_user, org=org_b, role="USER", is_active=True
+        )
+        response = unauthenticated_client.post(
+            self.url,
+            {
+                "email": "admin@test.com",
+                "password": "testpass123",
+                "org_id": str(org_b.id),
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["current_org"]["id"] == str(org_b.id)
+
+    def test_login_unauthorized_org(
+        self, unauthenticated_client, admin_user, admin_profile, org_b
+    ):
+        # admin_user does NOT have a profile in org_b
+        response = unauthenticated_client.post(
+            self.url,
+            {
+                "email": "admin@test.com",
+                "password": "testpass123",
+                "org_id": str(org_b.id),
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+@pytest.mark.django_db
+class TestRegisterView:
+    """Tests for POST /api/auth/register/"""
+
+    url = "/api/auth/register/"
+
+    def test_register_success(self, unauthenticated_client):
+        response = unauthenticated_client.post(
+            self.url,
+            {
+                "email": "newuser@test.com",
+                "password": "securepass123",
+                "confirm_password": "securepass123",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert "user_id" in response.data
+        assert response.data["email"] == "newuser@test.com"
+
+    def test_register_duplicate_email(self, unauthenticated_client, admin_user):
+        response = unauthenticated_client.post(
+            self.url,
+            {
+                "email": "admin@test.com",
+                "password": "securepass123",
+                "confirm_password": "securepass123",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    def test_register_missing_fields(self, unauthenticated_client):
+        response = unauthenticated_client.post(
+            self.url,
+            {"email": "newuser@test.com"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
+@pytest.mark.django_db
+class TestMeView:
+    """Tests for GET /api/auth/me/"""
+
+    url = "/api/auth/me/"
+
+    def test_me_authenticated(self, admin_client, admin_user):
+        response = admin_client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["email"] == admin_user.email
+
+    def test_me_unauthenticated(self, unauthenticated_client):
+        response = unauthenticated_client.get(self.url)
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+@pytest.mark.django_db
+class TestTokenRefreshView:
+    """Tests for POST /api/auth/refresh-token/"""
+
+    url = "/api/auth/refresh-token/"
+
+    def test_refresh_valid_token(
+        self, unauthenticated_client, admin_user, org_a, admin_profile
+    ):
+        token = OrgAwareRefreshToken.for_user_and_org(admin_user, org_a, admin_profile)
+        response = unauthenticated_client.post(
+            self.url,
+            {"refresh": str(token)},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert "access" in response.data
+        assert "refresh" in response.data
+
+    def test_refresh_invalid_token(self, unauthenticated_client):
+        response = unauthenticated_client.post(
+            self.url,
+            {"refresh": "invalid-token-string"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+@pytest.mark.django_db
+class TestOrgSwitchView:
+    """Tests for POST /api/auth/switch-org/"""
+
+    url = "/api/auth/switch-org/"
+
+    def test_switch_org_success(self, admin_client, admin_user, org_b):
+        # Give admin_user access to org_b
+        Profile.objects.create(
+            user=admin_user, org=org_b, role="USER", is_active=True
+        )
+        response = admin_client.post(
+            self.url,
+            {"org_id": str(org_b.id)},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["current_org"]["id"] == str(org_b.id)
+
+    def test_switch_to_unauthorized_org(self, admin_client, org_b):
+        # admin_user does NOT have a profile in org_b
+        response = admin_client.post(
+            self.url,
+            {"org_id": str(org_b.id)},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_switch_org_missing_org_id(self, admin_client):
+        response = admin_client.post(
+            self.url,
+            {},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST

backend/common/tests/test_documents.py

@@ -0,0 +1,86 @@
+"""
+Tests for document management views: list, create, detail, update, delete.
+
+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
+
+
+@pytest.mark.django_db
+class TestDocumentListView:
+    """Tests for GET/POST /api/documents/"""
+
+    url = "/api/documents/"
+
+    def test_list_documents(self, admin_client):
+        response = admin_client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+        assert "documents_active" in response.data
+
+    def test_create_document(self, admin_client):
+        test_file = SimpleUploadedFile(
+            "test.txt", b"file content", content_type="text/plain"
+        )
+        response = admin_client.post(
+            self.url,
+            {"title": "Test Document", "document_file": test_file},
+            format="multipart",
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.data["error"] is False
+
+    def test_unauthenticated(self, unauthenticated_client):
+        with pytest.raises(PermissionDenied):
+            unauthenticated_client.get(self.url)
+
+
+@pytest.mark.django_db
+class TestDocumentDetailView:
+    """Tests for GET/PUT/DELETE /api/documents/<pk>/"""
+
+    def _url(self, pk):
+        return f"/api/documents/{pk}/"
+
+    def _create_doc(self, org, user):
+        """Helper to create a document directly in the database."""
+        return Document.objects.create(
+            title="Sample Doc",
+            document_file="docs/sample.txt",
+            org=org,
+            created_by=user,
+        )
+
+    def test_get_document(self, admin_client, org_a, admin_user):
+        doc = self._create_doc(org_a, admin_user)
+        response = admin_client.get(self._url(doc.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert "doc_obj" in response.data
+
+    def test_update_document(self, admin_client, org_a, admin_user):
+        doc = self._create_doc(org_a, admin_user)
+        response = admin_client.put(
+            self._url(doc.pk),
+            {"title": "Updated Title"},
+            format="multipart",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+
+    def test_delete_document(self, admin_client, org_a, admin_user):
+        doc = self._create_doc(org_a, admin_user)
+        response = admin_client.delete(self._url(doc.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        assert not Document.objects.filter(pk=doc.pk).exists()
+
+    def test_org_isolation(self, org_b_client, org_a, admin_user, admin_profile):
+        """org_b_client cannot access a document belonging to org_a."""
+        doc = self._create_doc(org_a, admin_user)
+        response = org_b_client.get(self._url(doc.pk))
+        assert response.status_code == status.HTTP_404_NOT_FOUND

backend/common/tests/test_tags.py

@@ -0,0 +1,86 @@
+"""
+Tests for tag management views: list, create, detail, update, soft-delete, restore.
+
+Run with: pytest common/tests/test_tags.py -v
+"""
+
+import pytest
+from rest_framework import status
+from rest_framework.exceptions import PermissionDenied
+
+from common.models import Tags
+
+
+@pytest.mark.django_db
+class TestTagsListView:
+    """Tests for GET/POST /api/tags/"""
+
+    url = "/api/tags/"
+
+    def test_create_tag(self, admin_client, org_a):
+        response = admin_client.post(
+            self.url,
+            {"name": "Important"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.data["error"] is False
+        assert response.data["tag"]["name"] == "Important"
+
+    def test_list_tags(self, admin_client, org_a):
+        Tags.objects.create(name="Urgent", slug="urgent", org=org_a)
+        response = admin_client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+        assert "tags" in response.data
+        assert response.data["tags_count"] >= 1
+
+    def test_unauthenticated(self, unauthenticated_client):
+        with pytest.raises(PermissionDenied):
+            unauthenticated_client.get(self.url)
+
+
+@pytest.mark.django_db
+class TestTagsDetailView:
+    """Tests for GET/PUT/DELETE /api/tags/<pk>/ and POST /api/tags/<pk>/restore/"""
+
+    def _url(self, pk):
+        return f"/api/tags/{pk}/"
+
+    def _restore_url(self, pk):
+        return f"/api/tags/{pk}/restore/"
+
+    def test_get_tag(self, admin_client, org_a):
+        tag = Tags.objects.create(name="Feature", slug="feature", org=org_a)
+        response = admin_client.get(self._url(tag.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert "tag" in response.data
+
+    def test_update_tag(self, admin_client, org_a):
+        tag = Tags.objects.create(name="Old Tag", slug="old-tag", org=org_a)
+        response = admin_client.put(
+            self._url(tag.pk),
+            {"name": "New Tag"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        assert response.data["tag"]["name"] == "New Tag"
+
+    def test_soft_delete_tag(self, admin_client, org_a):
+        tag = Tags.objects.create(name="Deletable", slug="deletable", org=org_a)
+        response = admin_client.delete(self._url(tag.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        # Tag still exists in the database but is inactive
+        tag.refresh_from_db()
+        assert tag.is_active is False
+
+    def test_restore_tag(self, admin_client, org_a):
+        tag = Tags.objects.create(
+            name="Archived", slug="archived", org=org_a, is_active=False
+        )
+        response = admin_client.post(self._restore_url(tag.pk), format="json")
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        tag.refresh_from_db()
+        assert tag.is_active is True

backend/common/tests/test_teams.py

@@ -0,0 +1,67 @@
+"""
+Tests for team management views: list, create, detail, update, delete teams.
+
+Run with: pytest common/tests/test_teams.py -v
+"""
+
+import pytest
+from rest_framework import status
+
+from common.models import Teams
+
+
+@pytest.mark.django_db
+class TestTeamsListView:
+    """Tests for GET/POST /api/teams/"""
+
+    url = "/api/teams/"
+
+    def test_create_team(self, admin_client, org_a):
+        response = admin_client.post(
+            self.url,
+            {"name": "Engineering", "description": "Engineering team"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+
+    def test_list_teams(self, admin_client, org_a):
+        Teams.objects.create(name="Sales", org=org_a)
+        response = admin_client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+        assert "teams" in response.data
+
+    def test_list_teams_non_admin(self, user_client):
+        response = user_client.get(self.url)
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+@pytest.mark.django_db
+class TestTeamsDetailView:
+    """Tests for GET/PUT/DELETE /api/teams/<pk>/"""
+
+    def _url(self, pk):
+        return f"/api/teams/{pk}/"
+
+    def test_get_team_detail(self, admin_client, org_a):
+        team = Teams.objects.create(name="Support", org=org_a)
+        response = admin_client.get(self._url(team.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert "team" in response.data
+
+    def test_update_team(self, admin_client, org_a):
+        team = Teams.objects.create(name="Old Name", org=org_a)
+        response = admin_client.put(
+            self._url(team.pk),
+            {"name": "New Name", "description": "Updated"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+
+    def test_delete_team(self, admin_client, org_a):
+        team = Teams.objects.create(name="To Delete", org=org_a)
+        response = admin_client.delete(self._url(team.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        assert not Teams.objects.filter(pk=team.pk).exists()

backend/common/tests/test_users.py

@@ -0,0 +1,71 @@
+"""
+Tests for user management views: list users, user detail, update user.
+
+Run with: pytest common/tests/test_users.py -v
+"""
+
+import pytest
+from rest_framework import status
+from rest_framework.exceptions import PermissionDenied
+
+
+@pytest.mark.django_db
+class TestUsersListView:
+    """Tests for GET /api/users/"""
+
+    url = "/api/users/"
+
+    def test_list_users_as_admin(self, admin_client):
+        response = admin_client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+        assert "active_users" in response.data
+
+    def test_list_users_unauthenticated(self, unauthenticated_client):
+        with pytest.raises(PermissionDenied):
+            unauthenticated_client.get(self.url)
+
+    def test_list_users_no_cross_org_leak(
+        self, org_b_client, admin_user, admin_profile
+    ):
+        """org_b_client should not see org_a's users in its listing."""
+        response = org_b_client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+        # Collect all user emails across active and inactive lists
+        active = response.data["active_users"]["active_users"]
+        inactive = response.data["inactive_users"]["inactive_users"]
+        all_emails = [
+            p["user_details"]["email"] for p in active
+        ] + [
+            p["user_details"]["email"] for p in inactive
+        ]
+        assert admin_user.email not in all_emails
+
+
+@pytest.mark.django_db
+class TestUserDetailView:
+    """Tests for GET/PUT /api/user/<pk>/"""
+
+    def _url(self, user_id):
+        return f"/api/user/{user_id}/"
+
+    def test_get_user_detail(self, admin_client, admin_user):
+        response = admin_client.get(self._url(admin_user.id))
+        assert response.status_code == status.HTTP_200_OK
+        assert "data" in response.data
+
+    def test_update_user(self, admin_client, admin_user):
+        response = admin_client.put(
+            self._url(admin_user.id),
+            {
+                "email": admin_user.email,
+                "role": "ADMIN",
+                "phone": "+1234567890",
+            },
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+    def test_get_user_wrong_org(self, org_b_client, admin_user, admin_profile):
+        """org_b_client cannot view a user that belongs to org_a."""
+        response = org_b_client.get(self._url(admin_user.id))
+        assert response.status_code == status.HTTP_404_NOT_FOUND

backend/common/views/document_views.py

@@ -334,14 +334,17 @@ class DocumentDetailView(APIView):
                     status=status.HTTP_403_FORBIDDEN,
                 )
         serializer = DocumentCreateSerializer(
-            data=params, instance=self.object, request_obj=request
+            data=params, instance=self.object, request_obj=request, partial=True
         )
         if serializer.is_valid():
-            doc = serializer.save(
-                document_file=request.FILES.get("document_file"),
-                status=params.get("status"),
-                org=request.profile.org,
-            )
+            save_kwargs = {
+                "org": request.profile.org,
+            }
+            if request.FILES.get("document_file"):
+                save_kwargs["document_file"] = request.FILES.get("document_file")
+            if params.get("status"):
+                save_kwargs["status"] = params.get("status")
+            doc = serializer.save(**save_kwargs)
             doc.shared_to.clear()
             if params.get("shared_to"):
                 assinged_to_list = params.get("shared_to")

backend/common/views/user_views.py

@@ -12,7 +12,7 @@ from rest_framework.views import APIView
 from cases.models import Case
 from cases.serializer import CaseSerializer
 from common import swagger_params
-from common.models import Profile, Teams
+from common.models import Comment, Profile, Teams
 from common.serializer import (
     BillingAddressSerializer,
     CommentSerializer,
@@ -261,7 +261,7 @@ class UserDetailView(APIView):
         cases = Case.objects.filter(assigned_to=profile_obj, org=request.profile.org)
         context["cases"] = CaseSerializer(cases, many=True).data
         context["assigned_data"] = assigned_data
-        comments = profile_obj.user_comments.all()
+        comments = Comment.objects.filter(commented_by=profile_obj)
         context["comments"] = CommentSerializer(comments, many=True).data
         context["countries"] = COUNTRIES
         return Response(
@@ -321,7 +321,7 @@ class UserDetailView(APIView):
                 status=status.HTTP_400_BAD_REQUEST,
             )
         if address_serializer.is_valid():
-            address_obj = address_serializer.save()
+            address_obj = address_serializer.save(org=request.profile.org)
             user = serializer.save()
             user.email = user.email
             user.save()

backend/conftest.py

@@ -0,0 +1,106 @@
+"""
+Shared pytest fixtures for Django CRM backend tests.
+
+Provides authenticated API clients, test organizations, users, and profiles
+that can be reused across all test modules.
+"""
+
+import pytest
+from django.db import connection
+from rest_framework.test import APIClient
+
+from common.models import Org, Profile, User
+from common.serializer import OrgAwareRefreshToken
+
+
+def set_rls_context(org):
+    """Set the PostgreSQL RLS session variable for direct ORM operations in tests."""
+    with connection.cursor() as cursor:
+        cursor.execute(
+            "SELECT set_config('app.current_org', %s, false)", [str(org.id)]
+        )
+
+
+def clear_rls_context():
+    """Clear the PostgreSQL RLS session variable."""
+    with connection.cursor() as cursor:
+        cursor.execute("SELECT set_config('app.current_org', '', false)")
+
+
+@pytest.fixture(autouse=True)
+def _use_db(db):
+    """Ensure all tests have database access."""
+
+
+@pytest.fixture
+def org_a():
+    return Org.objects.create(name="Test Organization A")
+
+
+@pytest.fixture
+def org_b():
+    return Org.objects.create(name="Test Organization B")
+
+
+@pytest.fixture
+def admin_user():
+    return User.objects.create_user(email="admin@test.com", password="testpass123")
+
+
+@pytest.fixture
+def regular_user():
+    return User.objects.create_user(email="user@test.com", password="testpass123")
+
+
+@pytest.fixture
+def user_b():
+    return User.objects.create_user(email="userb@test.com", password="testpass123")
+
+
+@pytest.fixture
+def admin_profile(admin_user, org_a):
+    return Profile.objects.create(
+        user=admin_user, org=org_a, role="ADMIN", is_active=True
+    )
+
+
+@pytest.fixture
+def user_profile(regular_user, org_a):
+    return Profile.objects.create(
+        user=regular_user, org=org_a, role="USER", is_active=True
+    )
+
+
+@pytest.fixture
+def profile_b(user_b, org_b):
+    return Profile.objects.create(
+        user=user_b, org=org_b, role="ADMIN", is_active=True
+    )
+
+
+def _make_authenticated_client(user, org, profile):
+    """Create an APIClient with JWT auth for the given user/org."""
+    client = APIClient()
+    token = OrgAwareRefreshToken.for_user_and_org(user, org, profile)
+    client.credentials(HTTP_AUTHORIZATION=f"Bearer {token.access_token}")
+    return client
+
+
+@pytest.fixture
+def admin_client(admin_user, org_a, admin_profile):
+    return _make_authenticated_client(admin_user, org_a, admin_profile)
+
+
+@pytest.fixture
+def user_client(regular_user, org_a, user_profile):
+    return _make_authenticated_client(regular_user, org_a, user_profile)
+
+
+@pytest.fixture
+def org_b_client(user_b, org_b, profile_b):
+    return _make_authenticated_client(user_b, org_b, profile_b)
+
+
+@pytest.fixture
+def unauthenticated_client():
+    return APIClient()

backend/contacts/tests/test_contacts_api.py

@@ -0,0 +1,175 @@
+"""
+Tests for Contacts API endpoints.
+
+Covers:
+- ContactsListView (GET list, POST create)
+- ContactDetailView (GET detail, PUT update, DELETE, POST comment)
+- Organization isolation between tenants
+- Duplicate email validation per org
+
+Run with: pytest contacts/tests/test_contacts_api.py -v
+"""
+
+from unittest.mock import patch
+
+import pytest
+from django.db import connection
+from rest_framework import status
+from rest_framework.exceptions import PermissionDenied
+
+from contacts.models import Contact
+
+
+CONTACTS_LIST_URL = "/api/contacts/"
+
+
+def _detail_url(pk):
+    return f"/api/contacts/{pk}/"
+
+
+def _set_rls(org):
+    """Set PostgreSQL RLS context so direct ORM writes are allowed.
+    No-op on SQLite (used in tests).
+    """
+    if connection.vendor != "postgresql":
+        return
+    with connection.cursor() as cursor:
+        cursor.execute(
+            "SELECT set_config('app.current_org', %s, false)", [str(org.id)]
+        )
+
+
+@pytest.fixture
+def contact_a(admin_user, org_a):
+    """A contact belonging to org_a, created by admin_user."""
+    _set_rls(org_a)
+    return Contact.objects.create(
+        first_name="Alice",
+        last_name="Smith",
+        email="alice@example.com",
+        org=org_a,
+        created_by=admin_user,
+    )
+
+
+@pytest.fixture
+def contact_b(user_b, org_b):
+    """A contact belonging to org_b, created by user_b."""
+    _set_rls(org_b)
+    return Contact.objects.create(
+        first_name="Bob",
+        last_name="Jones",
+        email="bob@example.com",
+        org=org_b,
+        created_by=user_b,
+    )
+
+
+@pytest.mark.django_db
+class TestContactListView:
+    """Tests for GET /api/contacts/ and POST /api/contacts/"""
+
+    @patch("contacts.views.send_email_to_assigned_user.delay")
+    def test_list_contacts(self, mock_email, admin_client, contact_a):
+        """Admin can list contacts in their org."""
+        response = admin_client.get(CONTACTS_LIST_URL)
+        assert response.status_code == status.HTTP_200_OK
+        assert "contact_obj_list" in response.data
+        assert response.data["contacts_count"] >= 1
+
+    @patch("contacts.views.send_email_to_assigned_user.delay")
+    def test_create_contact(self, mock_email, admin_client):
+        """Admin can create a new contact via POST."""
+        payload = {
+            "first_name": "Charlie",
+            "last_name": "Brown",
+            "email": "charlie@example.com",
+        }
+        response = admin_client.post(CONTACTS_LIST_URL, payload, format="json")
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+
+    def test_create_contact_unauthenticated(self, unauthenticated_client):
+        """Unauthenticated requests are rejected."""
+        payload = {
+            "first_name": "Nope",
+            "last_name": "User",
+            "email": "nope@example.com",
+        }
+        with pytest.raises(PermissionDenied):
+            unauthenticated_client.post(
+                CONTACTS_LIST_URL, payload, format="json"
+            )
+
+    @patch("contacts.views.send_email_to_assigned_user.delay")
+    def test_org_isolation(
+        self, mock_email, admin_client, org_b_client, contact_a, contact_b
+    ):
+        """Contacts created in org_a must not appear in org_b's list."""
+        # org_b_client should only see org_b contacts
+        response = org_b_client.get(CONTACTS_LIST_URL)
+        assert response.status_code == status.HTTP_200_OK
+        emails = [c["email"] for c in response.data["contact_obj_list"]]
+        assert "alice@example.com" not in emails
+        assert "bob@example.com" in emails
+
+
+@pytest.mark.django_db
+class TestContactDetailView:
+    """Tests for GET/PUT/DELETE/POST on /api/contacts/<pk>/"""
+
+    def test_get_contact_detail(self, admin_client, contact_a):
+        """Admin can retrieve a single contact's detail."""
+        response = admin_client.get(_detail_url(contact_a.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert "contact_obj" in response.data
+        assert response.data["contact_obj"]["id"] == str(contact_a.pk)
+
+    @patch("contacts.views.send_email_to_assigned_user.delay")
+    def test_update_contact(self, mock_email, admin_client, contact_a):
+        """Admin can update a contact via PUT."""
+        payload = {
+            "first_name": "Updated",
+            "last_name": "Name",
+            "email": "alice@example.com",
+        }
+        response = admin_client.put(
+            _detail_url(contact_a.pk), payload, format="json"
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        contact_a.refresh_from_db()
+        assert contact_a.first_name == "Updated"
+
+    def test_delete_contact(self, admin_client, contact_a):
+        """Admin can delete a contact."""
+        response = admin_client.delete(_detail_url(contact_a.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        assert not Contact.objects.filter(pk=contact_a.pk).exists()
+
+    def test_get_contact_cross_org(self, org_b_client, contact_a):
+        """Accessing a contact from another org returns 404."""
+        response = org_b_client.get(_detail_url(contact_a.pk))
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+    def test_add_comment_to_contact(self, admin_client, contact_a):
+        """Admin can add a comment to a contact via POST to detail endpoint."""
+        payload = {"comment": "This is a test comment"}
+        response = admin_client.post(
+            _detail_url(contact_a.pk), payload, format="json"
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert "comments" in response.data
+
+    @patch("contacts.views.send_email_to_assigned_user.delay")
+    def test_duplicate_email_same_org(self, mock_email, admin_client, contact_a):
+        """Creating a contact with a duplicate email in the same org returns 400."""
+        payload = {
+            "first_name": "Duplicate",
+            "last_name": "Email",
+            "email": "alice@example.com",
+        }
+        response = admin_client.post(CONTACTS_LIST_URL, payload, format="json")
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.data["error"] is True

backend/contacts/views.py

@@ -433,7 +433,7 @@ class ContactDetailView(APIView):
                 "attachments": AttachmentsSerializer(attachments, many=True).data,
                 "assigned_data": assigned_data,
                 "tasks": TaskSerializer(
-                    contact_obj.contacts_tasks.all(), many=True
+                    contact_obj.task_contacts.all(), many=True
                 ).data,
                 "users_mention": users_mention,
             }

backend/crm/test_settings.py

@@ -0,0 +1,15 @@
+"""
+Test settings for Django CRM backend.
+
+Uses SQLite in-memory database instead of PostgreSQL for fast, isolated tests.
+RLS (Row-Level Security) is PostgreSQL-only and is skipped on SQLite.
+"""
+
+from crm.settings import *  # noqa: F401, F403
+
+DATABASES = {
+    "default": {
+        "ENGINE": "django.db.backends.sqlite3",
+        "NAME": ":memory:",
+    }
+}

backend/invoices/tests/test_invoices_api.py

@@ -0,0 +1,308 @@
+import pytest
+from unittest.mock import patch
+from rest_framework.exceptions import PermissionDenied
+
+from accounts.models import Account
+from contacts.models import Contact
+from invoices.models import Invoice, InvoiceLineItem, Payment
+
+
+# ---------------------------------------------------------------------------
+# Local fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def account_for_invoice(org_a):
+    return Account.objects.create(name="Invoice Test Account", org=org_a)
+
+
+@pytest.fixture
+def contact_for_invoice(org_a):
+    return Contact.objects.create(
+        first_name="Test",
+        last_name="Contact",
+        email="testcontact@example.com",
+        org=org_a,
+    )
+
+
+@pytest.fixture
+def account_org_b(org_b):
+    return Account.objects.create(name="Org B Account", org=org_b)
+
+
+@pytest.fixture
+def contact_org_b(org_b):
+    return Contact.objects.create(
+        first_name="OrgB",
+        last_name="Contact",
+        email="orgb@example.com",
+        org=org_b,
+    )
+
+
+@pytest.fixture
+def invoice(account_for_invoice, org_a):
+    return Invoice.objects.create(
+        invoice_title="Test Invoice",
+        account=account_for_invoice,
+        currency="USD",
+        org=org_a,
+    )
+
+
+@pytest.fixture
+def invoice_org_b(account_org_b, org_b):
+    return Invoice.objects.create(
+        invoice_title="Org B Invoice",
+        account=account_org_b,
+        currency="USD",
+        org=org_b,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Invoice List / Create
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.django_db
+class TestInvoiceListView:
+    """Tests for GET /api/invoices/ and POST /api/invoices/."""
+
+    def test_list_invoices(self, admin_client, invoice):
+        response = admin_client.get("/api/invoices/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "results" in data
+        assert data["count"] >= 1
+
+    @patch("invoices.api_views.create_invoice_history.delay")
+    def test_create_invoice(
+        self,
+        mock_history,
+        admin_client,
+        account_for_invoice,
+        contact_for_invoice,
+    ):
+        response = admin_client.post(
+            "/api/invoices/",
+            {
+                "invoice_title": "New Invoice",
+                "account_id": str(account_for_invoice.id),
+                "contact_id": str(contact_for_invoice.id),
+                "currency": "USD",
+            },
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["error"] is False
+        assert data["message"] == "Invoice created successfully"
+        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",
+            )
+
+    def test_org_isolation(self, org_b_client, invoice):
+        """org_b_client should not see invoices belonging to org_a."""
+        response = org_b_client.get("/api/invoices/")
+        assert response.status_code == 200
+        data = response.json()
+        ids = [r["id"] for r in data["results"]]
+        assert str(invoice.id) not in ids
+
+
+# ---------------------------------------------------------------------------
+# Invoice Detail / Update / Delete
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.django_db
+class TestInvoiceDetailView:
+    """Tests for GET/PUT/DELETE /api/invoices/<pk>/."""
+
+    def test_get_detail(self, admin_client, invoice):
+        response = admin_client.get(f"/api/invoices/{invoice.id}/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "invoice" in data
+        assert data["invoice"]["id"] == str(invoice.id)
+
+    @patch("invoices.api_views.create_invoice_history.delay")
+    def test_update_invoice(self, mock_history, admin_client, invoice):
+        response = admin_client.put(
+            f"/api/invoices/{invoice.id}/",
+            {"invoice_title": "Updated Title"},
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+        assert data["message"] == "Invoice updated successfully"
+
+    def test_delete_invoice_not_supported(self, admin_client, invoice):
+        """InvoiceDetailView does not implement DELETE; expect 405."""
+        response = admin_client.delete(f"/api/invoices/{invoice.id}/")
+        assert response.status_code == 405
+
+    def test_cross_org(self, org_b_client, invoice):
+        """org_b_client must not access an invoice belonging to org_a."""
+        response = org_b_client.get(f"/api/invoices/{invoice.id}/")
+        assert response.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# Invoice Actions (send, mark-paid, duplicate, cancel)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.django_db
+class TestInvoiceActions:
+    """Tests for invoice action endpoints."""
+
+    @patch("invoices.api_views.send_invoice_to_client.delay")
+    def test_send_invoice(self, mock_send, admin_client, invoice):
+        response = admin_client.post(f"/api/invoices/{invoice.id}/send/")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+        assert data["message"] == "Invoice sent successfully"
+        invoice.refresh_from_db()
+        assert invoice.status == "Sent"
+
+    def test_mark_paid(self, admin_client, invoice):
+        response = admin_client.post(
+            f"/api/invoices/{invoice.id}/mark-paid/",
+            {
+                "amount": "500.00",
+                "payment_method": "BANK_TRANSFER",
+                "payment_date": "2026-01-15",
+            },
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+        assert data["message"] == "Payment recorded successfully"
+        assert "invoice" in data
+
+    def test_duplicate_invoice(self, admin_client, invoice):
+        response = admin_client.post(f"/api/invoices/{invoice.id}/duplicate/")
+        assert response.status_code == 201
+        data = response.json()
+        assert data["error"] is False
+        assert data["message"] == "Invoice duplicated successfully"
+        assert "invoice" in data
+        assert data["invoice"]["invoice_title"].startswith("Copy of")
+
+    def test_cancel_invoice(self, admin_client, invoice):
+        response = admin_client.post(f"/api/invoices/{invoice.id}/cancel/")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+        assert data["message"] == "Invoice cancelled successfully"
+        invoice.refresh_from_db()
+        assert invoice.status == "Cancelled"
+
+    def test_cancel_already_cancelled(self, admin_client, invoice):
+        invoice.status = "Cancelled"
+        invoice.save()
+        response = admin_client.post(f"/api/invoices/{invoice.id}/cancel/")
+        assert response.status_code == 400
+        data = response.json()
+        assert data["error"] is True
+
+    def test_cancel_paid_invoice(self, admin_client, invoice):
+        invoice.status = "Paid"
+        invoice.save()
+        response = admin_client.post(f"/api/invoices/{invoice.id}/cancel/")
+        assert response.status_code == 400
+        data = response.json()
+        assert data["error"] is True
+
+
+# ---------------------------------------------------------------------------
+# Invoice Line Items
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.django_db
+class TestInvoiceLineItems:
+    """Tests for /api/invoices/<invoice_id>/line-items/."""
+
+    def test_create_line_item(self, admin_client, invoice):
+        response = admin_client.post(
+            f"/api/invoices/{invoice.id}/line-items/",
+            {
+                "name": "Consulting Hours",
+                "quantity": "10.00",
+                "unit_price": "150.00",
+            },
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["error"] is False
+        assert data["message"] == "Line item added"
+        assert "line_item" in data
+
+    def test_list_line_items(self, admin_client, invoice):
+        InvoiceLineItem.objects.create(
+            invoice=invoice,
+            name="Widget A",
+            quantity=5,
+            unit_price=20,
+            org=invoice.org,
+        )
+        response = admin_client.get(f"/api/invoices/{invoice.id}/line-items/")
+        assert response.status_code == 200
+        data = response.json()
+        assert isinstance(data, list)
+        assert len(data) >= 1
+
+
+# ---------------------------------------------------------------------------
+# Invoice Payments
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.django_db
+class TestInvoicePayments:
+    """Tests for /api/invoices/<invoice_id>/payments/."""
+
+    def test_create_payment(self, admin_client, invoice):
+        response = admin_client.post(
+            f"/api/invoices/{invoice.id}/payments/",
+            {
+                "amount": "250.00",
+                "payment_date": "2026-02-01",
+                "payment_method": "CREDIT_CARD",
+            },
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["error"] is False
+        assert data["message"] == "Payment recorded"
+        assert "payment" in data
+
+    def test_list_payments(self, admin_client, invoice):
+        Payment.objects.create(
+            invoice=invoice,
+            amount=100,
+            payment_date="2026-01-10",
+            payment_method="CASH",
+            org=invoice.org,
+        )
+        response = admin_client.get(f"/api/invoices/{invoice.id}/payments/")
+        assert response.status_code == 200
+        data = response.json()
+        assert isinstance(data, list)
+        assert len(data) >= 1

backend/leads/tests/test_leads_api.py

@@ -0,0 +1,198 @@
+import pytest
+from rest_framework.exceptions import PermissionDenied
+
+from leads.models import Lead
+
+
+@pytest.mark.django_db
+class TestLeadListView:
+    """Tests for GET /api/leads/ and POST /api/leads/."""
+
+    def test_list_leads(self, admin_client, admin_user, org_a):
+        Lead.objects.create(
+            first_name="John",
+            last_name="Doe",
+            email="john@example.com",
+            created_by=admin_user,
+            org=org_a,
+        )
+        response = admin_client.get("/api/leads/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "open_leads" in data
+
+    def test_create_lead(self, admin_client, org_a):
+        response = admin_client.post(
+            "/api/leads/",
+            {
+                "first_name": "Jane",
+                "last_name": "Smith",
+                "email": "jane@example.com",
+            },
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+        assert Lead.objects.filter(email="jane@example.com", org=org_a).exists()
+
+    def test_create_lead_invalid_data(self, admin_client):
+        response = admin_client.post(
+            "/api/leads/",
+            {"email": "not-a-valid-email"},
+        )
+        assert response.status_code == 400
+        data = response.json()
+        assert data["error"] is True
+
+    def test_create_lead_unauthenticated(self, unauthenticated_client):
+        with pytest.raises(PermissionDenied):
+            unauthenticated_client.post(
+                "/api/leads/",
+                {"first_name": "Test", "last_name": "Lead", "email": "t@t.com"},
+            )
+
+    def test_org_isolation(self, org_b_client, admin_user, org_a):
+        Lead.objects.create(
+            first_name="Private",
+            last_name="Lead",
+            email="private@org-a.com",
+            created_by=admin_user,
+            org=org_a,
+        )
+        response = org_b_client.get("/api/leads/")
+        assert response.status_code == 200
+        data = response.json()
+        open_leads_list = data["open_leads"]["open_leads"]
+        emails = [lead["email"] for lead in open_leads_list]
+        assert "private@org-a.com" not in emails
+
+
+@pytest.mark.django_db
+class TestLeadDetailView:
+    """Tests for GET/PUT/DELETE /api/leads/<pk>/."""
+
+    def test_get_lead_detail(self, admin_client, admin_user, org_a):
+        lead = Lead.objects.create(
+            first_name="Detail",
+            last_name="Lead",
+            email="detail@example.com",
+            created_by=admin_user,
+            org=org_a,
+        )
+        response = admin_client.get(f"/api/leads/{lead.id}/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "lead_obj" in data
+
+    def test_update_lead(self, admin_client, admin_user, org_a):
+        lead = Lead.objects.create(
+            first_name="Old",
+            last_name="Name",
+            email="update@example.com",
+            created_by=admin_user,
+            org=org_a,
+        )
+        response = admin_client.put(
+            f"/api/leads/{lead.id}/",
+            {
+                "first_name": "New",
+                "last_name": "Name",
+                "email": "update@example.com",
+            },
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+
+    def test_delete_lead(self, admin_client, admin_user, org_a):
+        lead = Lead.objects.create(
+            first_name="Delete",
+            last_name="Me",
+            email="delete@example.com",
+            created_by=admin_user,
+            org=org_a,
+        )
+        response = admin_client.delete(f"/api/leads/{lead.id}/")
+        assert response.status_code == 200
+        assert not Lead.objects.filter(id=lead.id).exists()
+
+    def test_get_lead_cross_org(self, org_b_client, admin_user, org_a):
+        lead = Lead.objects.create(
+            first_name="Cross",
+            last_name="Org",
+            email="cross@org-a.com",
+            created_by=admin_user,
+            org=org_a,
+        )
+        response = org_b_client.get(f"/api/leads/{lead.id}/")
+        assert response.status_code == 404
+
+
+@pytest.mark.django_db
+class TestLeadCommentView:
+    """Tests for POST /api/leads/<pk>/ (adding comments)."""
+
+    def test_add_comment_to_lead(self, admin_client, admin_user, org_a):
+        lead = Lead.objects.create(
+            first_name="Comment",
+            last_name="Lead",
+            email="comment@example.com",
+            created_by=admin_user,
+            org=org_a,
+        )
+        response = admin_client.post(
+            f"/api/leads/{lead.id}/",
+            {"comment": "This is a test comment"},
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert "lead_obj" in data
+        assert "comments" in data
+
+
+@pytest.mark.django_db
+class TestLeadUniqueEmail:
+    """Tests for unique email constraint per organization."""
+
+    def test_duplicate_email_same_org(self, admin_client, org_a):
+        admin_client.post(
+            "/api/leads/",
+            {
+                "first_name": "First",
+                "last_name": "Lead",
+                "email": "dupe@example.com",
+            },
+        )
+        response = admin_client.post(
+            "/api/leads/",
+            {
+                "first_name": "Second",
+                "last_name": "Lead",
+                "email": "dupe@example.com",
+            },
+        )
+        assert response.status_code == 400
+
+    def test_same_email_different_org(
+        self, admin_client, org_b_client, admin_user, user_b, org_a, org_b
+    ):
+        resp1 = admin_client.post(
+            "/api/leads/",
+            {
+                "first_name": "Org A",
+                "last_name": "Lead",
+                "email": "shared@example.com",
+            },
+        )
+        assert resp1.status_code == 200
+
+        resp2 = org_b_client.post(
+            "/api/leads/",
+            {
+                "first_name": "Org B",
+                "last_name": "Lead",
+                "email": "shared@example.com",
+            },
+        )
+        assert resp2.status_code == 200

backend/leads/tests/test_leads_kanban.py

@@ -0,0 +1,179 @@
+import pytest
+
+from leads.models import LeadPipeline, LeadStage
+
+
+@pytest.mark.django_db
+class TestLeadPipelineListCreateView:
+    """Tests for GET/POST /api/leads/pipelines/."""
+
+    def test_create_pipeline(self, admin_client, org_a):
+        response = admin_client.post(
+            "/api/leads/pipelines/",
+            {"name": "Sales Pipeline"},
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["name"] == "Sales Pipeline"
+        assert LeadPipeline.objects.filter(
+            name="Sales Pipeline", org=org_a
+        ).exists()
+
+    def test_list_pipelines(self, admin_client, admin_user, org_a):
+        LeadPipeline.objects.create(
+            name="Pipeline A", org=org_a, created_by=admin_user
+        )
+        response = admin_client.get("/api/leads/pipelines/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "pipelines" in data
+
+    def test_create_pipeline_non_admin(self, user_client):
+        response = user_client.post(
+            "/api/leads/pipelines/",
+            {"name": "Blocked Pipeline"},
+            format="json",
+        )
+        assert response.status_code == 403
+
+
+@pytest.mark.django_db
+class TestLeadPipelineDetailView:
+    """Tests for GET/PUT/DELETE /api/leads/pipelines/<pk>/."""
+
+    def test_get_pipeline_detail(self, admin_client, admin_user, org_a):
+        pipeline = LeadPipeline.objects.create(
+            name="Detail Pipeline", org=org_a, created_by=admin_user
+        )
+        response = admin_client.get(f"/api/leads/pipelines/{pipeline.id}/")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["name"] == "Detail Pipeline"
+
+    def test_update_pipeline(self, admin_client, admin_user, org_a):
+        pipeline = LeadPipeline.objects.create(
+            name="Old Name", org=org_a, created_by=admin_user
+        )
+        response = admin_client.put(
+            f"/api/leads/pipelines/{pipeline.id}/",
+            {"name": "New Name"},
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["name"] == "New Name"
+
+    def test_delete_empty_pipeline(self, admin_client, admin_user, org_a):
+        pipeline = LeadPipeline.objects.create(
+            name="Empty Pipeline", org=org_a, created_by=admin_user
+        )
+        response = admin_client.delete(
+            f"/api/leads/pipelines/{pipeline.id}/"
+        )
+        assert response.status_code == 204
+
+
+@pytest.mark.django_db
+class TestLeadStageViews:
+    """Tests for stage creation, update, delete, and reorder."""
+
+    def test_create_stage(self, admin_client, admin_user, org_a):
+        pipeline = LeadPipeline.objects.create(
+            name="Stage Pipeline",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.post(
+            f"/api/leads/pipelines/{pipeline.id}/stages/",
+            {
+                "name": "Qualification",
+                "order": 1,
+                "color": "#FF5733",
+                "stage_type": "open",
+            },
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["name"] == "Qualification"
+
+    def test_update_stage(self, admin_client, admin_user, org_a):
+        pipeline = LeadPipeline.objects.create(
+            name="Update Stage Pipeline",
+            org=org_a,
+            created_by=admin_user,
+        )
+        stage = LeadStage.objects.create(
+            pipeline=pipeline,
+            name="Initial",
+            order=0,
+            color="#000000",
+            stage_type="open",
+            org=org_a,
+        )
+        response = admin_client.put(
+            f"/api/leads/stages/{stage.id}/",
+            {"name": "Renamed Stage", "color": "#FFFFFF"},
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["name"] == "Renamed Stage"
+
+    def test_delete_empty_stage(self, admin_client, admin_user, org_a):
+        pipeline = LeadPipeline.objects.create(
+            name="Delete Stage Pipeline",
+            org=org_a,
+            created_by=admin_user,
+        )
+        stage = LeadStage.objects.create(
+            pipeline=pipeline,
+            name="Disposable",
+            order=0,
+            color="#000000",
+            stage_type="open",
+            org=org_a,
+        )
+        response = admin_client.delete(f"/api/leads/stages/{stage.id}/")
+        assert response.status_code == 204
+
+    def test_reorder_stages(self, admin_client, admin_user, org_a):
+        pipeline = LeadPipeline.objects.create(
+            name="Reorder Pipeline",
+            org=org_a,
+            created_by=admin_user,
+        )
+        stage_a = LeadStage.objects.create(
+            pipeline=pipeline,
+            name="Stage A",
+            order=0,
+            org=org_a,
+        )
+        stage_b = LeadStage.objects.create(
+            pipeline=pipeline,
+            name="Stage B",
+            order=1,
+            org=org_a,
+        )
+        # Reverse the order
+        response = admin_client.post(
+            f"/api/leads/pipelines/{pipeline.id}/stages/reorder/",
+            {"stage_ids": [str(stage_b.id), str(stage_a.id)]},
+            format="json",
+        )
+        assert response.status_code == 200
+        stage_a.refresh_from_db()
+        stage_b.refresh_from_db()
+        assert stage_b.order < stage_a.order
+
+
+@pytest.mark.django_db
+class TestLeadKanbanView:
+    """Tests for GET /api/leads/kanban/."""
+
+    def test_kanban_view(self, admin_client, org_a):
+        response = admin_client.get("/api/leads/kanban/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "columns" in data

backend/leads/views/lead_views.py

@@ -1,6 +1,7 @@
 import json
 
 from django.contrib.contenttypes.models import ContentType
+from django.db import IntegrityError
 from django.db.models import Q
 from django.shortcuts import get_object_or_404
 from drf_spectacular.utils import extend_schema, inline_serializer
@@ -215,9 +216,22 @@ class LeadListView(APIView, LimitOffsetPagination):
         data = request.data
         serializer = LeadCreateSerializer(data=data, request_obj=request)
         if serializer.is_valid():
-            lead_obj = serializer.save(
-                created_by=request.profile.user, org=request.profile.org
-            )
+            try:
+                lead_obj = serializer.save(
+                    created_by=request.profile.user, org=request.profile.org
+                )
+            except IntegrityError:
+                return Response(
+                    {
+                        "error": True,
+                        "errors": {
+                            "email": [
+                                "A lead with this email already exists in your organization."
+                            ]
+                        },
+                    },
+                    status=status.HTTP_400_BAD_REQUEST,
+                )
             if data.get("tags", None):
                 tags = data.get("tags")
                 tag_objs = Tags.objects.filter(

backend/opportunity/tests/test_opportunities_api.py

@@ -0,0 +1,280 @@
+"""
+Tests for Opportunities API endpoints.
+
+Covers:
+- OpportunityListView (GET list, POST create)
+- OpportunityDetailView (GET detail, PUT update, DELETE, POST comment)
+- OpportunityLineItemListView / DetailView (CRUD for line items)
+- Organization isolation between tenants
+
+Run with: pytest opportunity/tests/test_opportunities_api.py -v
+"""
+
+from unittest.mock import patch
+
+import pytest
+from django.db import connection
+from rest_framework import status
+from rest_framework.exceptions import PermissionDenied
+
+from accounts.models import Account
+from opportunity.models import Opportunity, OpportunityLineItem
+
+
+OPPORTUNITIES_LIST_URL = "/api/opportunities/"
+
+
+def _detail_url(pk):
+    return f"/api/opportunities/{pk}/"
+
+
+def _line_items_url(opportunity_pk):
+    return f"/api/opportunities/{opportunity_pk}/line-items/"
+
+
+def _line_item_detail_url(opportunity_pk, line_item_pk):
+    return f"/api/opportunities/{opportunity_pk}/line-items/{line_item_pk}/"
+
+
+def _set_rls(org):
+    """Set PostgreSQL RLS context so direct ORM writes are allowed.
+    No-op on SQLite (used in tests).
+    """
+    if connection.vendor != "postgresql":
+        return
+    with connection.cursor() as cursor:
+        cursor.execute(
+            "SELECT set_config('app.current_org', %s, false)", [str(org.id)]
+        )
+
+
+@pytest.fixture
+def opp_a(admin_user, org_a):
+    """An opportunity belonging to org_a, created by admin_user."""
+    _set_rls(org_a)
+    return Opportunity.objects.create(
+        name="Org A Deal",
+        stage="QUALIFICATION",
+        org=org_a,
+        created_by=admin_user,
+    )
+
+
+@pytest.fixture
+def opp_b(user_b, org_b):
+    """An opportunity belonging to org_b, created by user_b."""
+    _set_rls(org_b)
+    return Opportunity.objects.create(
+        name="Org B Deal",
+        stage="QUALIFICATION",
+        org=org_b,
+        created_by=user_b,
+    )
+
+
+@pytest.mark.django_db
+class TestOpportunityListView:
+    """Tests for GET /api/opportunities/ and POST /api/opportunities/"""
+
+    @patch("opportunity.views.opportunity_views.send_email_to_assigned_user.delay")
+    def test_list_opportunities(self, mock_email, admin_client, opp_a):
+        """Admin can list opportunities in their org."""
+        response = admin_client.get(OPPORTUNITIES_LIST_URL)
+        assert response.status_code == status.HTTP_200_OK
+        assert "opportunities" in response.data
+        assert response.data["opportunities_count"] >= 1
+
+    @patch("opportunity.views.opportunity_views.send_email_to_assigned_user.delay")
+    def test_create_opportunity(self, mock_email, admin_client):
+        """Admin can create an opportunity via POST."""
+        payload = {
+            "name": "New Deal",
+            "stage": "QUALIFICATION",
+        }
+        response = admin_client.post(
+            OPPORTUNITIES_LIST_URL, payload, format="json"
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+
+    @patch("opportunity.views.opportunity_views.send_email_to_assigned_user.delay")
+    def test_create_opportunity_with_account(
+        self, mock_email, admin_client, org_a, admin_user
+    ):
+        """Admin can create an opportunity linked to an account."""
+        _set_rls(org_a)
+        account = Account.objects.create(
+            name="Acme Corp",
+            org=org_a,
+            created_by=admin_user,
+        )
+        payload = {
+            "name": "Acme Deal",
+            "stage": "QUALIFICATION",
+            "account": str(account.pk),
+        }
+        response = admin_client.post(
+            OPPORTUNITIES_LIST_URL, payload, format="json"
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+
+    def test_create_opportunity_unauthenticated(self, unauthenticated_client):
+        """Unauthenticated requests are rejected."""
+        payload = {
+            "name": "Unauthorized Deal",
+            "stage": "QUALIFICATION",
+        }
+        with pytest.raises(PermissionDenied):
+            unauthenticated_client.post(
+                OPPORTUNITIES_LIST_URL, payload, format="json"
+            )
+
+    @patch("opportunity.views.opportunity_views.send_email_to_assigned_user.delay")
+    def test_org_isolation(
+        self, mock_email, admin_client, org_b_client, opp_a, opp_b
+    ):
+        """Opportunities in org_a must not appear in org_b's list."""
+        response = org_b_client.get(OPPORTUNITIES_LIST_URL)
+        assert response.status_code == status.HTTP_200_OK
+        names = [o["name"] for o in response.data["opportunities"]]
+        assert "Org A Deal" not in names
+        assert "Org B Deal" in names
+
+
+@pytest.mark.django_db
+class TestOpportunityDetailView:
+    """Tests for GET/PUT/DELETE/POST on /api/opportunities/<pk>/"""
+
+    @pytest.fixture()
+    def opportunity(self, org_a, admin_user):
+        _set_rls(org_a)
+        return Opportunity.objects.create(
+            name="Detail Opp",
+            stage="QUALIFICATION",
+            org=org_a,
+            created_by=admin_user,
+        )
+
+    def test_get_detail(self, admin_client, opportunity):
+        """Admin can retrieve a single opportunity's detail."""
+        response = admin_client.get(_detail_url(opportunity.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert "opportunity_obj" in response.data
+        assert response.data["opportunity_obj"]["id"] == str(opportunity.pk)
+
+    @patch("opportunity.views.opportunity_views.send_email_to_assigned_user.delay")
+    def test_update(self, mock_email, admin_client, opportunity):
+        """Admin can update an opportunity via PUT."""
+        payload = {
+            "name": "Updated Opp",
+            "stage": "PROPOSAL",
+        }
+        response = admin_client.put(
+            _detail_url(opportunity.pk), payload, format="json"
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        opportunity.refresh_from_db()
+        assert opportunity.name == "Updated Opp"
+
+    def test_delete(self, admin_client, opportunity):
+        """Admin can delete an opportunity."""
+        response = admin_client.delete(_detail_url(opportunity.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        assert not Opportunity.objects.filter(pk=opportunity.pk).exists()
+
+    def test_cross_org(self, org_b_client, opportunity):
+        """Accessing an opportunity from another org returns an error."""
+        response = org_b_client.get(_detail_url(opportunity.pk))
+        # The detail view uses .filter().first() which returns None,
+        # then accessing .org on None raises AttributeError -> 500,
+        # or RLS hides the row entirely. Either way, data is not returned.
+        assert response.status_code != status.HTTP_200_OK or (
+            response.data.get("error") is True
+        )
+
+    def test_add_comment(self, admin_client, opportunity):
+        """Admin can add a comment to an opportunity via POST to detail."""
+        payload = {"comment": "Looks like a promising deal"}
+        response = admin_client.post(
+            _detail_url(opportunity.pk), payload, format="json"
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert "comments" in response.data
+
+
+@pytest.mark.django_db
+class TestOpportunityLineItems:
+    """Tests for line item CRUD under /api/opportunities/<pk>/line-items/"""
+
+    @pytest.fixture()
+    def opportunity(self, org_a, admin_user):
+        _set_rls(org_a)
+        return Opportunity.objects.create(
+            name="LineItem Opp",
+            stage="QUALIFICATION",
+            org=org_a,
+            created_by=admin_user,
+        )
+
+    @pytest.fixture()
+    def line_item(self, opportunity, org_a):
+        _set_rls(org_a)
+        return OpportunityLineItem.objects.create(
+            opportunity=opportunity,
+            name="Widget A",
+            quantity=2,
+            unit_price=50,
+            org=org_a,
+        )
+
+    def test_create_line_item(self, admin_client, opportunity):
+        """Admin can create a line item for an opportunity."""
+        payload = {
+            "name": "Consulting Hours",
+            "quantity": 10,
+            "unit_price": "150.00",
+        }
+        response = admin_client.post(
+            _line_items_url(opportunity.pk), payload, format="json"
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.data["error"] is False
+        assert response.data["line_item"]["name"] == "Consulting Hours"
+
+    def test_list_line_items(self, admin_client, opportunity, line_item):
+        """Admin can list line items for an opportunity."""
+        response = admin_client.get(_line_items_url(opportunity.pk))
+        assert response.status_code == status.HTTP_200_OK
+        assert "line_items" in response.data
+        assert len(response.data["line_items"]) == 1
+        assert response.data["line_items"][0]["name"] == "Widget A"
+
+    def test_update_line_item(self, admin_client, opportunity, line_item):
+        """Admin can update an existing line item."""
+        payload = {
+            "name": "Widget A (Revised)",
+            "quantity": 5,
+            "unit_price": "75.00",
+        }
+        response = admin_client.put(
+            _line_item_detail_url(opportunity.pk, line_item.pk),
+            payload,
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        line_item.refresh_from_db()
+        assert line_item.name == "Widget A (Revised)"
+        assert line_item.quantity == 5
+
+    def test_delete_line_item(self, admin_client, opportunity, line_item):
+        """Admin can delete a line item."""
+        response = admin_client.delete(
+            _line_item_detail_url(opportunity.pk, line_item.pk)
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["error"] is False
+        assert not OpportunityLineItem.objects.filter(pk=line_item.pk).exists()

backend/opportunity/views/opportunity_views.py

@@ -292,6 +292,11 @@ class OpportunityDetailView(APIView):
     def put(self, request, pk, format=None):
         params = request.data
         opportunity_object = self.get_object(pk=pk)
+        if not opportunity_object:
+            return Response(
+                {"error": True, "errors": "Opportunity not found."},
+                status=status.HTTP_404_NOT_FOUND,
+            )
         if opportunity_object.org != request.profile.org:
             return Response(
                 {"error": True, "errors": "User company doesnot match with header...."},
@@ -473,6 +478,11 @@ class OpportunityDetailView(APIView):
     )
     def get(self, request, pk, format=None):
         self.opportunity = self.get_object(pk=pk)
+        if not self.opportunity:
+            return Response(
+                {"error": True, "errors": "Opportunity not found."},
+                status=status.HTTP_404_NOT_FOUND,
+            )
         context = {}
         context["opportunity_obj"] = OpportunitySerializer(self.opportunity).data
         if self.opportunity.org != request.profile.org:

backend/pytest.ini

@@ -1,4 +1,15 @@
 [pytest]
-DJANGO_SETTINGS_MODULE=crm.settings
+DJANGO_SETTINGS_MODULE = crm.test_settings
 python_files = tests.py test_*.py *_tests.py
-
+addopts =
+    --strict-markers
+    --tb=short
+    -v
+    --cov=.
+    --cov-config=../.coveragerc
+    --cov-report=term-missing
+    --cov-report=html:htmlcov
+    --cov-report=xml:coverage.xml
+markers =
+    slow: marks tests as slow
+    postgres_only: marks tests that require PostgreSQL

backend/requirements.txt

@@ -24,4 +24,9 @@ weasyprint==68.1
 cairocffi==1.7.1
 
 # Production server
-gunicorn==25.1.0
+gunicorn==25.1.0
+
+# Testing
+pytest==9.0.2
+pytest-django==4.12.0
+pytest-cov==7.0.0

backend/tasks/serializer.py

@@ -46,7 +46,7 @@ class BoardTaskSerializer(serializers.ModelSerializer):
     class Meta:
         model = BoardTask
         fields = "__all__"
-        read_only_fields = ("id", "created_at", "updated_at", "completed_at")
+        read_only_fields = ("id", "created_at", "updated_at", "completed_at", "org", "column")
 
 
 class BoardColumnSerializer(serializers.ModelSerializer):
@@ -58,7 +58,7 @@ class BoardColumnSerializer(serializers.ModelSerializer):
     class Meta:
         model = BoardColumn
         fields = "__all__"
-        read_only_fields = ("id", "created_at", "updated_at", "board")
+        read_only_fields = ("id", "created_at", "updated_at", "board", "org")
 
     @extend_schema_field(int)
     def get_task_count(self, obj):

backend/tasks/tests/test_boards.py

@@ -0,0 +1,187 @@
+import pytest
+
+from tasks.models import Board, BoardColumn, BoardMember, BoardTask
+
+
+@pytest.mark.django_db
+class TestBoardListView:
+    """Tests for GET /api/boards/ and POST /api/boards/."""
+
+    def test_list_boards(self, admin_client, admin_profile, org_a):
+        board = Board.objects.create(
+            name="Test Board", owner=admin_profile, org=org_a
+        )
+        BoardMember.objects.create(board=board, profile=admin_profile, role="owner")
+        response = admin_client.get("/api/boards/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "results" in data
+        assert data["count"] >= 1
+
+    def test_create_board(self, admin_client, org_a):
+        response = admin_client.post(
+            "/api/boards/",
+            {"name": "New Board", "create_default_columns": False},
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["name"] == "New Board"
+        assert Board.objects.filter(name="New Board", org=org_a).exists()
+
+
+@pytest.mark.django_db
+class TestBoardDetailView:
+    """Tests for GET/PUT/DELETE /api/boards/<pk>/."""
+
+    def test_get_board(self, admin_client, admin_profile, admin_user, org_a):
+        board = Board.objects.create(
+            name="Detail Board",
+            owner=admin_profile,
+            org=org_a,
+            created_by=admin_user,
+        )
+        BoardMember.objects.create(board=board, profile=admin_profile, role="owner")
+        response = admin_client.get(f"/api/boards/{board.id}/")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["name"] == "Detail Board"
+
+    def test_update_board(self, admin_client, admin_profile, admin_user, org_a):
+        board = Board.objects.create(
+            name="Old Board Name",
+            owner=admin_profile,
+            org=org_a,
+            created_by=admin_user,
+        )
+        BoardMember.objects.create(board=board, profile=admin_profile, role="owner")
+        response = admin_client.put(
+            f"/api/boards/{board.id}/",
+            {"name": "Updated Board Name"},
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["name"] == "Updated Board Name"
+
+    def test_delete_board(self, admin_client, admin_profile, admin_user, org_a):
+        board = Board.objects.create(
+            name="Delete Board",
+            owner=admin_profile,
+            org=org_a,
+            created_by=admin_user,
+        )
+        BoardMember.objects.create(board=board, profile=admin_profile, role="owner")
+        response = admin_client.delete(f"/api/boards/{board.id}/")
+        assert response.status_code == 204
+        assert not Board.objects.filter(id=board.id).exists()
+
+
+@pytest.mark.django_db
+class TestBoardColumns:
+    """Tests for GET/POST /api/boards/<board_pk>/columns/."""
+
+    def test_create_column(self, admin_client, admin_profile, admin_user, org_a):
+        board = Board.objects.create(
+            name="Column Board",
+            owner=admin_profile,
+            org=org_a,
+            created_by=admin_user,
+        )
+        BoardMember.objects.create(board=board, profile=admin_profile, role="owner")
+        response = admin_client.post(
+            f"/api/boards/{board.id}/columns/",
+            {"name": "Backlog", "order": 0, "color": "#6B7280"},
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["name"] == "Backlog"
+        assert BoardColumn.objects.filter(board=board, name="Backlog").exists()
+
+    def test_list_columns(self, admin_client, admin_profile, admin_user, org_a):
+        board = Board.objects.create(
+            name="List Column Board",
+            owner=admin_profile,
+            org=org_a,
+            created_by=admin_user,
+        )
+        BoardMember.objects.create(board=board, profile=admin_profile, role="owner")
+        BoardColumn.objects.create(board=board, name="To Do", order=1, org=org_a)
+        BoardColumn.objects.create(board=board, name="Done", order=2, org=org_a)
+        response = admin_client.get(f"/api/boards/{board.id}/columns/")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data) == 2
+
+
+@pytest.mark.django_db
+class TestBoardTasks:
+    """Tests for board task endpoints."""
+
+    def _create_board_with_column(self, admin_profile, admin_user, org_a):
+        """Helper to create a board with one column and owner membership."""
+        board = Board.objects.create(
+            name="Task Board",
+            owner=admin_profile,
+            org=org_a,
+            created_by=admin_user,
+        )
+        BoardMember.objects.create(board=board, profile=admin_profile, role="owner")
+        column = BoardColumn.objects.create(
+            board=board, name="To Do", order=1, org=org_a
+        )
+        return board, column
+
+    def test_create_board_task(
+        self, admin_client, admin_profile, admin_user, org_a
+    ):
+        _board, column = self._create_board_with_column(
+            admin_profile, admin_user, org_a
+        )
+        response = admin_client.post(
+            f"/api/boards/columns/{column.id}/tasks/",
+            {"title": "New Card", "priority": "medium"},
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["title"] == "New Card"
+        assert BoardTask.objects.filter(column=column, title="New Card").exists()
+
+    def test_get_board_task(
+        self, admin_client, admin_profile, admin_user, org_a
+    ):
+        """Verify a board task appears in the column task list."""
+        _board, column = self._create_board_with_column(
+            admin_profile, admin_user, org_a
+        )
+        BoardTask.objects.create(
+            column=column,
+            title="Existing Card",
+            priority="high",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.get(f"/api/boards/columns/{column.id}/tasks/")
+        assert response.status_code == 200
+        data = response.json()
+        titles = [t["title"] for t in data]
+        assert "Existing Card" in titles
+
+    def test_delete_board_task(
+        self, admin_client, admin_profile, admin_user, org_a
+    ):
+        _board, column = self._create_board_with_column(
+            admin_profile, admin_user, org_a
+        )
+        task = BoardTask.objects.create(
+            column=column,
+            title="Delete Card",
+            priority="low",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.delete(f"/api/boards/tasks/{task.id}/")
+        assert response.status_code == 204
+        assert not BoardTask.objects.filter(id=task.id).exists()

backend/tasks/tests/test_tasks_api.py

@@ -0,0 +1,164 @@
+import pytest
+from rest_framework.exceptions import PermissionDenied
+
+from tasks.models import Task
+
+
+@pytest.mark.django_db
+class TestTaskListView:
+    """Tests for GET /api/tasks/ and POST /api/tasks/."""
+
+    def test_list_tasks(self, admin_client, admin_user, org_a):
+        Task.objects.create(
+            title="Sample Task",
+            status="New",
+            priority="Medium",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.get("/api/tasks/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "tasks" in data
+        assert data["tasks_count"] >= 1
+
+    def test_create_task(self, admin_client, org_a):
+        response = admin_client.post(
+            "/api/tasks/",
+            {"title": "New Task", "status": "New", "priority": "High"},
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+        assert Task.objects.filter(title="New Task", org=org_a).exists()
+
+    def test_create_task_unauthenticated(self, unauthenticated_client):
+        with pytest.raises(PermissionDenied):
+            unauthenticated_client.post(
+                "/api/tasks/",
+                {"title": "Unauthorized Task", "status": "New", "priority": "Low"},
+                format="json",
+            )
+
+    def test_org_isolation(self, admin_client, admin_user, org_a, org_b):
+        """Tasks from another org should not appear in the task list."""
+        Task.objects.create(
+            title="Org A Task",
+            status="New",
+            priority="Low",
+            org=org_a,
+            created_by=admin_user,
+        )
+        # Create a task in org_b that should not be visible
+        from common.models import User
+
+        other_user = User.objects.create_user(
+            email="other@example.com", password="testpass123"
+        )
+        Task.objects.create(
+            title="Org B Task",
+            status="New",
+            priority="Low",
+            org=org_b,
+            created_by=other_user,
+        )
+        response = admin_client.get("/api/tasks/")
+        assert response.status_code == 200
+        data = response.json()
+        titles = [t["title"] for t in data["tasks"]]
+        assert "Org A Task" in titles
+        assert "Org B Task" not in titles
+
+
+@pytest.mark.django_db
+class TestTaskDetailView:
+    """Tests for GET/PUT/DELETE /api/tasks/<pk>/."""
+
+    def test_get_detail(self, admin_client, admin_user, org_a):
+        task = Task.objects.create(
+            title="Detail Task",
+            status="New",
+            priority="Medium",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.get(f"/api/tasks/{task.id}/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "task_obj" in data
+        assert data["task_obj"]["title"] == "Detail Task"
+
+    def test_update_task(self, admin_client, admin_user, org_a):
+        task = Task.objects.create(
+            title="Old Title",
+            status="New",
+            priority="Low",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.put(
+            f"/api/tasks/{task.id}/",
+            {"title": "Updated Title", "status": "In Progress", "priority": "High"},
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+        task.refresh_from_db()
+        assert task.title == "Updated Title"
+        assert task.status == "In Progress"
+
+    def test_delete_task(self, admin_client, admin_user, org_a):
+        task = Task.objects.create(
+            title="Delete Me",
+            status="New",
+            priority="Low",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.delete(f"/api/tasks/{task.id}/")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["error"] is False
+        assert not Task.objects.filter(id=task.id).exists()
+
+    def test_cross_org(self, org_b_client, admin_user, org_a):
+        """A client from org_b must not access a task belonging to org_a."""
+        task = Task.objects.create(
+            title="Org A Only",
+            status="New",
+            priority="Low",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = org_b_client.get(f"/api/tasks/{task.id}/")
+        assert response.status_code == 404
+
+    def test_add_comment(self, admin_client, admin_user, org_a):
+        task = Task.objects.create(
+            title="Comment Task",
+            status="New",
+            priority="Medium",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.post(
+            f"/api/tasks/{task.id}/",
+            {"comment": "This is a test comment"},
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert "comments" in data
+
+    def test_create_task_invalid(self, admin_client):
+        """POST without a title should return 400."""
+        response = admin_client.post(
+            "/api/tasks/",
+            {"status": "New", "priority": "Low"},
+            format="json",
+        )
+        assert response.status_code == 400
+        data = response.json()
+        assert data["error"] is True

backend/tasks/tests/test_tasks_kanban.py

@@ -0,0 +1,147 @@
+import pytest
+
+from tasks.models import Task, TaskPipeline, TaskStage
+
+
+@pytest.mark.django_db
+class TestTaskPipeline:
+    """Tests for /api/tasks/pipelines/ endpoints."""
+
+    def test_create_pipeline(self, admin_client, org_a):
+        response = admin_client.post(
+            "/api/tasks/pipelines/",
+            {"name": "Development Pipeline", "create_default_stages": False},
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["name"] == "Development Pipeline"
+        assert TaskPipeline.objects.filter(name="Development Pipeline", org=org_a).exists()
+
+    def test_list_pipelines(self, admin_client, admin_user, org_a):
+        TaskPipeline.objects.create(
+            name="Pipeline 1", org=org_a, created_by=admin_user
+        )
+        response = admin_client.get("/api/tasks/pipelines/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "pipelines" in data
+        assert len(data["pipelines"]) >= 1
+
+    def test_get_pipeline_detail(self, admin_client, admin_user, org_a):
+        pipeline = TaskPipeline.objects.create(
+            name="Detail Pipeline", org=org_a, created_by=admin_user
+        )
+        response = admin_client.get(f"/api/tasks/pipelines/{pipeline.id}/")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["name"] == "Detail Pipeline"
+
+    def test_delete_empty_pipeline(self, admin_client, admin_user, org_a):
+        pipeline = TaskPipeline.objects.create(
+            name="Empty Pipeline", org=org_a, created_by=admin_user
+        )
+        response = admin_client.delete(f"/api/tasks/pipelines/{pipeline.id}/")
+        assert response.status_code == 204
+
+
+@pytest.mark.django_db
+class TestTaskStage:
+    """Tests for /api/tasks/pipelines/<pk>/stages/ and /api/tasks/stages/<pk>/."""
+
+    def test_create_stage(self, admin_client, admin_user, org_a):
+        pipeline = TaskPipeline.objects.create(
+            name="Stage Pipeline", org=org_a, created_by=admin_user
+        )
+        response = admin_client.post(
+            f"/api/tasks/pipelines/{pipeline.id}/stages/",
+            {
+                "name": "To Do",
+                "order": 1,
+                "color": "#3B82F6",
+                "stage_type": "open",
+            },
+            format="json",
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["name"] == "To Do"
+        assert TaskStage.objects.filter(
+            pipeline=pipeline, name="To Do"
+        ).exists()
+
+    def test_update_stage(self, admin_client, admin_user, org_a):
+        pipeline = TaskPipeline.objects.create(
+            name="Update Pipeline", org=org_a, created_by=admin_user
+        )
+        stage = TaskStage.objects.create(
+            pipeline=pipeline,
+            name="Old Stage",
+            order=1,
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.put(
+            f"/api/tasks/stages/{stage.id}/",
+            {"name": "Renamed Stage", "order": 2},
+            format="json",
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["name"] == "Renamed Stage"
+
+    def test_delete_empty_stage(self, admin_client, admin_user, org_a):
+        pipeline = TaskPipeline.objects.create(
+            name="Delete Pipeline", org=org_a, created_by=admin_user
+        )
+        stage = TaskStage.objects.create(
+            pipeline=pipeline,
+            name="Delete Me",
+            order=1,
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.delete(f"/api/tasks/stages/{stage.id}/")
+        assert response.status_code == 204
+        assert not TaskStage.objects.filter(id=stage.id).exists()
+
+    def test_reorder_stages(self, admin_client, admin_user, org_a):
+        pipeline = TaskPipeline.objects.create(
+            name="Reorder Pipeline", org=org_a, created_by=admin_user
+        )
+        stage_a = TaskStage.objects.create(
+            pipeline=pipeline, name="A", order=0, org=org_a, created_by=admin_user
+        )
+        stage_b = TaskStage.objects.create(
+            pipeline=pipeline, name="B", order=1, org=org_a, created_by=admin_user
+        )
+        # Reverse the order: B first, then A
+        response = admin_client.post(
+            f"/api/tasks/pipelines/{pipeline.id}/stages/reorder/",
+            {"stage_ids": [str(stage_b.id), str(stage_a.id)]},
+            format="json",
+        )
+        assert response.status_code == 200
+        stage_a.refresh_from_db()
+        stage_b.refresh_from_db()
+        assert stage_b.order < stage_a.order
+
+
+@pytest.mark.django_db
+class TestTaskKanban:
+    """Tests for GET /api/tasks/kanban/."""
+
+    def test_kanban_view(self, admin_client, admin_user, org_a):
+        Task.objects.create(
+            title="Kanban Task",
+            status="New",
+            priority="Medium",
+            org=org_a,
+            created_by=admin_user,
+        )
+        response = admin_client.get("/api/tasks/kanban/")
+        assert response.status_code == 200
+        data = response.json()
+        assert "columns" in data
+        assert data["mode"] == "status"
+        assert data["total_tasks"] >= 1

backend/tasks/views/board_views.py

@@ -330,7 +330,7 @@ class BoardColumnListCreateView(APIView):
                 status=status.HTTP_400_BAD_REQUEST,
             )
 
-        column = serializer.save(board=board, created_by=request.user)
+        column = serializer.save(board=board, org=org, created_by=request.user)
         return Response(
             BoardColumnSerializer(column).data, status=status.HTTP_201_CREATED
         )
@@ -402,7 +402,7 @@ class BoardTaskListCreateView(APIView):
                 status=status.HTTP_400_BAD_REQUEST,
             )
 
-        task = serializer.save(column=column, created_by=request.user)
+        task = serializer.save(column=column, org=org, created_by=request.user)
 
         # Handle assigned_to
         if "assigned_to_ids" in data:

backend/tasks/views/task_views.py

@@ -2,6 +2,7 @@ import json
 
 from django.contrib.contenttypes.models import ContentType
 from django.db.models import Q
+from django.shortcuts import get_object_or_404
 from drf_spectacular.utils import extend_schema, inline_serializer
 from rest_framework import serializers, status
 from rest_framework.pagination import LimitOffsetPagination
@@ -263,7 +264,7 @@ class TaskDetailView(APIView):
     permission_classes = (IsAuthenticated, HasOrgContext)
 
     def get_object(self, pk):
-        return Task.objects.get(pk=pk, org=self.request.profile.org)
+        return get_object_or_404(Task, pk=pk, org=self.request.profile.org)
 
     def get_context_data(self, **kwargs):
         context = {}