From 3152284cf3965bce12809ca21ddeb2807bb07340 Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Thu, 30 Apr 2026 12:44:24 +0330 Subject: [PATCH] test(backend): add coverage for services tasks and apis --- apps/clients/tests/__init__.py | 1 + apps/clients/tests/test_services.py | 67 ++++++ apps/clients/tests/test_views.py | 125 +++++++++++ apps/notifications/tests/test_tasks.py | 20 ++ apps/projects/tests/test_permissions.py | 34 +++ apps/projects/tests/test_services.py | 97 +++++++++ apps/reports/tests/test_api_views.py | 105 +++++++++ apps/reports/tests/test_exporters.py | 104 +++++++++ apps/tags/tests/__init__.py | 1 + apps/tags/tests/test_services.py | 59 ++++++ apps/tags/tests/test_views.py | 136 ++++++++++++ apps/users/tests/test_api_views.py | 199 ++++++++++++++++++ apps/users/tests/test_auth_services.py | 149 +++++++++++++ apps/users/tests/test_utils.py | 36 ++++ apps/workspaces/tests/test_api_permissions.py | 146 +++++++++++++ 15 files changed, 1279 insertions(+) create mode 100644 apps/clients/tests/__init__.py create mode 100644 apps/clients/tests/test_services.py create mode 100644 apps/clients/tests/test_views.py create mode 100644 apps/notifications/tests/test_tasks.py create mode 100644 apps/projects/tests/test_permissions.py create mode 100644 apps/projects/tests/test_services.py create mode 100644 apps/reports/tests/test_api_views.py create mode 100644 apps/reports/tests/test_exporters.py create mode 100644 apps/tags/tests/__init__.py create mode 100644 apps/tags/tests/test_services.py create mode 100644 apps/tags/tests/test_views.py create mode 100644 apps/users/tests/test_api_views.py create mode 100644 apps/users/tests/test_auth_services.py create mode 100644 apps/users/tests/test_utils.py create mode 100644 apps/workspaces/tests/test_api_permissions.py diff --git a/apps/clients/tests/__init__.py b/apps/clients/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/apps/clients/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/apps/clients/tests/test_services.py b/apps/clients/tests/test_services.py new file mode 100644 index 0000000..b4db019 --- /dev/null +++ b/apps/clients/tests/test_services.py @@ -0,0 +1,67 @@ +from django.test import TestCase +from rest_framework.exceptions import ValidationError + +from apps.clients.models import Client +from apps.clients.services.clients import create_client, update_client +from apps.users.models import User +from apps.workspaces.models import Workspace, WorkspaceMembership + + +class ClientServiceTests(TestCase): + @classmethod + def setUpTestData(cls): + cls.owner = User.objects.create_user(mobile="09120000001", password="secret123") + cls.member = User.objects.create_user(mobile="09120000002", password="secret123") + cls.outsider = User.objects.create_user(mobile="09120000003", password="secret123") + cls.workspace = Workspace.objects.create(name="Clients", owner=cls.owner) + WorkspaceMembership.objects.create( + workspace=cls.workspace, + user=cls.member, + role=WorkspaceMembership.Role.MEMBER, + is_active=True, + ) + + def test_create_client_creates_record_for_workspace_member(self): + client = create_client( + self.member, + self.workspace.id, + "Acme", + notes="Priority account", + ) + + self.assertEqual(client.name, "Acme") + self.assertEqual(client.notes, "Priority account") + self.assertEqual(client.workspace, self.workspace) + self.assertEqual(client.created_by, self.member) + + def test_create_client_rejects_non_member(self): + with self.assertRaises(ValidationError) as exc: + create_client(self.outsider, self.workspace.id, "Acme") + + self.assertIn("workspace", exc.exception.detail) + + def test_create_client_rejects_duplicate_name_in_workspace(self): + Client.objects.create(workspace=self.workspace, name="Acme") + + with self.assertRaises(ValidationError) as exc: + create_client(self.owner, self.workspace.id, "Acme") + + self.assertIn("name", exc.exception.detail) + + def test_update_client_updates_name_and_notes(self): + client = Client.objects.create(workspace=self.workspace, name="Acme", notes="Old") + + updated = update_client(client, name="Globex", notes="New") + + self.assertEqual(updated.name, "Globex") + self.assertEqual(updated.notes, "New") + + def test_update_client_rejects_duplicate_new_name(self): + Client.objects.create(workspace=self.workspace, name="Globex") + client = Client.objects.create(workspace=self.workspace, name="Acme") + + with self.assertRaises(ValidationError) as exc: + update_client(client, name="Globex") + + self.assertIn("name", exc.exception.detail) + diff --git a/apps/clients/tests/test_views.py b/apps/clients/tests/test_views.py new file mode 100644 index 0000000..4d02aaa --- /dev/null +++ b/apps/clients/tests/test_views.py @@ -0,0 +1,125 @@ +from rest_framework.test import APITestCase + +from apps.clients.models import Client +from apps.users.models import User +from apps.workspaces.models import Workspace, WorkspaceMembership + + +class ClientViewTests(APITestCase): + @classmethod + def setUpTestData(cls): + cls.owner = User.objects.create_user(mobile="09120000011", password="secret123") + cls.admin = User.objects.create_user(mobile="09120000012", password="secret123") + cls.second_admin = User.objects.create_user(mobile="09120000013", password="secret123") + cls.member = User.objects.create_user(mobile="09120000014", password="secret123") + cls.guest = User.objects.create_user(mobile="09120000015", password="secret123") + cls.outsider = User.objects.create_user(mobile="09120000016", password="secret123") + + cls.workspace = Workspace.objects.create(name="Clients API", owner=cls.owner) + for user, role in ( + (cls.admin, WorkspaceMembership.Role.ADMIN), + (cls.second_admin, WorkspaceMembership.Role.ADMIN), + (cls.member, WorkspaceMembership.Role.MEMBER), + (cls.guest, WorkspaceMembership.Role.GUEST), + ): + WorkspaceMembership.objects.create( + workspace=cls.workspace, + user=user, + role=role, + is_active=True, + ) + + cls.other_workspace = Workspace.objects.create(name="Other", owner=cls.outsider) + cls.visible_client = Client.objects.create(workspace=cls.workspace, name="Visible") + cls.hidden_client = Client.objects.create(workspace=cls.other_workspace, name="Hidden") + cls.admin_owned_client = Client.objects.create( + workspace=cls.workspace, + name="Admin Owned", + created_by=cls.admin, + updated_by=cls.admin, + ) + + def test_list_only_returns_clients_for_member_workspaces(self): + self.client.force_authenticate(user=self.member) + + response = self.client.get("/api/clients/") + + self.assertEqual(response.status_code, 200) + results = ( + response.data + if isinstance(response.data, list) + else response.data.get("results") + or response.data.get("items") + or response.data.get("notifications") + or [] + ) + names = {item["name"] for item in results} + self.assertIn("Visible", names) + self.assertNotIn("Hidden", names) + + def test_owner_can_create_client(self): + self.client.force_authenticate(user=self.owner) + + response = self.client.post( + "/api/clients/", + { + "workspace_id": str(self.workspace.id), + "name": "Created", + "notes": "Important", + }, + format="json", + ) + + self.assertEqual(response.status_code, 201) + self.assertEqual(response.data["name"], "Created") + + def test_member_cannot_create_client(self): + self.client.force_authenticate(user=self.member) + + response = self.client.post( + "/api/clients/", + { + "workspace_id": str(self.workspace.id), + "name": "Created", + }, + format="json", + ) + + self.assertEqual(response.status_code, 403) + + def test_admin_can_update_client(self): + self.client.force_authenticate(user=self.admin) + + response = self.client.patch( + f"/api/clients/{self.visible_client.id}/", + {"name": "Renamed"}, + format="json", + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data["name"], "Renamed") + + def test_admin_can_delete_only_client_they_created(self): + self.client.force_authenticate(user=self.second_admin) + + forbidden = self.client.delete(f"/api/clients/{self.admin_owned_client.id}/") + self.assertEqual(forbidden.status_code, 403) + + self.client.force_authenticate(user=self.admin) + allowed = self.client.delete(f"/api/clients/{self.admin_owned_client.id}/") + self.assertEqual(allowed.status_code, 204) + self.assertTrue(Client.all_objects.get(id=self.admin_owned_client.id).is_deleted) + + def test_owner_can_delete_any_client(self): + client = Client.objects.create( + workspace=self.workspace, + name="Owner Delete", + created_by=self.admin, + updated_by=self.admin, + ) + self.client.force_authenticate(user=self.owner) + + response = self.client.delete(f"/api/clients/{client.id}/") + + self.assertEqual(response.status_code, 204) + self.assertTrue(Client.all_objects.get(id=client.id).is_deleted) diff --git a/apps/notifications/tests/test_tasks.py b/apps/notifications/tests/test_tasks.py new file mode 100644 index 0000000..8084a5f --- /dev/null +++ b/apps/notifications/tests/test_tasks.py @@ -0,0 +1,20 @@ +from unittest.mock import patch + +from django.conf import settings +from django.test import TestCase + +from apps.notifications.tasks import cleanup_redis_notifications + + +class NotificationTaskTests(TestCase): + @patch("apps.notifications.tasks.RedisNotificationStore.cleanup_expired") + def test_cleanup_redis_notifications_uses_settings_retention_days(self, cleanup_expired): + cleanup_expired.return_value = 7 + + removed = cleanup_redis_notifications() + + self.assertEqual(removed, 7) + cleanup_expired.assert_called_once_with( + retention_days=settings.NOTIFICATION_RETENTION_DAYS + ) + diff --git a/apps/projects/tests/test_permissions.py b/apps/projects/tests/test_permissions.py new file mode 100644 index 0000000..4acc80d --- /dev/null +++ b/apps/projects/tests/test_permissions.py @@ -0,0 +1,34 @@ +from django.test import SimpleTestCase + +from apps.projects.api.permissions import IsProjectManager, IsProjectMember, get_project_from_obj + + +class DummyWorkspace: + pass + + +class DummyProject: + def __init__(self): + self.workspace = DummyWorkspace() + + +class DummyRelatedObject: + def __init__(self): + self.project = DummyProject() + + +class ProjectPermissionHelperTests(SimpleTestCase): + def test_get_project_from_obj_returns_project_for_project_like_object(self): + project = DummyProject() + + self.assertIs(get_project_from_obj(project), project) + + def test_get_project_from_obj_returns_related_project(self): + related = DummyRelatedObject() + + self.assertIs(get_project_from_obj(related), related.project) + + def test_permission_messages_remain_defined(self): + self.assertTrue(IsProjectMember.message) + self.assertTrue(IsProjectManager.message) + diff --git a/apps/projects/tests/test_services.py b/apps/projects/tests/test_services.py new file mode 100644 index 0000000..7353124 --- /dev/null +++ b/apps/projects/tests/test_services.py @@ -0,0 +1,97 @@ +from django.test import TestCase +from rest_framework.exceptions import PermissionDenied, ValidationError + +from apps.clients.models import Client +from apps.projects.models import Project +from apps.projects.services.projects import ( + create_project, + toggle_project_archive, + update_project, +) +from apps.users.models import User +from apps.workspaces.models import Workspace, WorkspaceMembership + + +class ProjectServiceTests(TestCase): + @classmethod + def setUpTestData(cls): + cls.owner = User.objects.create_user(mobile="09120000041", password="secret123") + cls.member = User.objects.create_user(mobile="09120000042", password="secret123") + cls.outsider = User.objects.create_user(mobile="09120000043", password="secret123") + cls.workspace = Workspace.objects.create(name="Projects Services", owner=cls.owner) + WorkspaceMembership.objects.create( + workspace=cls.workspace, + user=cls.member, + role=WorkspaceMembership.Role.MEMBER, + is_active=True, + ) + cls.account_client = Client.objects.create(workspace=cls.workspace, name="Acme") + + def test_create_project_creates_workspace_shared_project(self): + project = create_project( + user=self.member, + workspace=self.workspace, + name="Alpha", + client=self.account_client, + description="Desc", + color="#123456", + ) + + self.assertEqual(project.name, "Alpha") + self.assertEqual(project.client, self.account_client) + self.assertEqual(project.description, "Desc") + + def test_create_project_rejects_non_member(self): + with self.assertRaises(PermissionDenied): + create_project(self.outsider, self.workspace, "Alpha") + + def test_create_project_rejects_duplicate_name(self): + Project.objects.create(workspace=self.workspace, name="Alpha") + + with self.assertRaises(ValidationError) as exc: + create_project(self.owner, self.workspace, "Alpha") + + self.assertIn("name", exc.exception.detail) + + def test_update_project_updates_client_and_fields(self): + second_client = Client.objects.create(workspace=self.workspace, name="Globex") + project = Project.objects.create( + workspace=self.workspace, + name="Alpha", + client=self.account_client, + ) + + updated = update_project( + project, + name="Beta", + client=str(second_client.id), + description="Updated", + color="#abcdef", + ) + + self.assertEqual(updated.name, "Beta") + self.assertEqual(updated.client, second_client) + self.assertEqual(updated.description, "Updated") + self.assertEqual(updated.color, "#abcdef") + + def test_update_project_rejects_duplicate_name(self): + Project.objects.create(workspace=self.workspace, name="Beta") + project = Project.objects.create( + workspace=self.workspace, + name="Alpha", + client=self.account_client, + ) + + with self.assertRaises(ValidationError) as exc: + update_project(project, name="Beta", client=str(self.account_client.id)) + + self.assertIn("name", exc.exception.detail) + + def test_toggle_project_archive_flips_state(self): + project = Project.objects.create(workspace=self.workspace, name="Alpha") + + toggle_project_archive(project) + self.assertTrue(Project.objects.get(id=project.id).is_archived) + + toggle_project_archive(project) + self.assertFalse(Project.objects.get(id=project.id).is_archived) diff --git a/apps/reports/tests/test_api_views.py b/apps/reports/tests/test_api_views.py new file mode 100644 index 0000000..4242cf0 --- /dev/null +++ b/apps/reports/tests/test_api_views.py @@ -0,0 +1,105 @@ +from datetime import date +from types import SimpleNamespace +from unittest.mock import patch + +from django.core.files.base import ContentFile +from django.test import TestCase +from rest_framework.test import APIClient + +from apps.reports.models import ReportExportJob +from apps.users.models import User +from apps.workspaces.models import Workspace, WorkspaceMembership + + +class ReportExportApiTests(TestCase): + @classmethod + def setUpTestData(cls): + cls.owner = User.objects.create_user( + mobile="09126660001", + password="secret123", + first_name="Owner", + ) + cls.admin = User.objects.create_user( + mobile="09126660002", + password="secret123", + first_name="Admin", + ) + cls.workspace = Workspace.objects.create(name="Exports", owner=cls.owner) + WorkspaceMembership.objects.create( + workspace=cls.workspace, + user=cls.admin, + role=WorkspaceMembership.Role.ADMIN, + is_active=True, + ) + + def setUp(self): + self.client = APIClient() + + def test_create_export_job_enqueues_background_task(self): + self.client.force_authenticate(user=self.owner) + + filters = SimpleNamespace( + workspace=self.workspace, + period="this_month", + from_date=date(2026, 4, 1), + to_date=date(2026, 4, 30), + user_id=None, + client_id=None, + project_id=None, + tag_ids=[], + ) + with patch("apps.reports.api.views.load_report_filters", return_value=filters): + with patch("apps.reports.api.views.generate_report_export_task.delay") as delay: + response = self.client.post( + "/api/reports/exports/", + { + "workspace": str(self.workspace.id), + "period": "this_month", + "export_type": "excel", + "language": "en", + }, + format="json", + ) + + self.assertEqual(response.status_code, 202) + self.assertEqual(ReportExportJob.objects.count(), 1) + delay.assert_called_once() + + def test_list_only_returns_requesting_users_jobs(self): + own_job = ReportExportJob.objects.create( + requesting_user=self.owner, + workspace=self.workspace, + export_type=ReportExportJob.ExportType.EXCEL, + filters={"workspace": str(self.workspace.id)}, + ) + ReportExportJob.objects.create( + requesting_user=self.admin, + workspace=self.workspace, + export_type=ReportExportJob.ExportType.PDF, + filters={"workspace": str(self.workspace.id)}, + ) + self.client.force_authenticate(user=self.owner) + + response = self.client.get("/api/reports/exports/") + + self.assertEqual(response.status_code, 200) + self.assertEqual(len(response.data), 1) + self.assertEqual(response.data[0]["id"], str(own_job.id)) + + def test_download_returns_completed_file(self): + job = ReportExportJob.objects.create( + requesting_user=self.owner, + workspace=self.workspace, + export_type=ReportExportJob.ExportType.EXCEL, + status=ReportExportJob.Status.COMPLETED, + filters={"workspace": str(self.workspace.id)}, + file_name="report.xlsx", + ) + job.file.save("reports/exports/report.xlsx", ContentFile(b"content"), save=False) + job.save(update_fields=["file", "updated_at"]) + self.client.force_authenticate(user=self.owner) + + response = self.client.get(f"/api/reports/exports/{job.id}/download/") + + self.assertEqual(response.status_code, 200) + self.assertIn("attachment; filename=", response["Content-Disposition"]) diff --git a/apps/reports/tests/test_exporters.py b/apps/reports/tests/test_exporters.py new file mode 100644 index 0000000..aff24f1 --- /dev/null +++ b/apps/reports/tests/test_exporters.py @@ -0,0 +1,104 @@ +from io import BytesIO + +from django.test import TestCase +from openpyxl import load_workbook + +from apps.reports.services.export_i18n import build_export_locale +from apps.reports.services.exporters import build_excel_report, build_pdf_report + + +def make_report_data(*, user_name="Owner User", mobile="09129990001", hourly_rate=None): + return { + "scope": { + "workspace": {"name": "Exports", "thumbnail_path": None}, + "period": "this_month", + "from_date": "2026-04-01", + "to_date": "2026-04-30", + "user": {"name": user_name, "mobile": mobile} if user_name else None, + }, + "summary": { + "total_duration": "02:00:00", + "billable_duration": "02:00:00", + "non_billable_duration": "00:00:00", + "income_totals": [{"amount": "30.00", "currency": "USD"}], + }, + "days": [ + { + "date": "2026-04-12", + "billable_duration": "02:00:00", + "non_billable_duration": "00:00:00", + "total_duration": "02:00:00", + "latest_hourly_rate": hourly_rate, + "income_totals": [{"amount": "30.00", "currency": "USD"}], + } + ], + "clients": [ + { + "name": "Acme", + "billable_duration": "02:00:00", + "non_billable_duration": "00:00:00", + "total_duration": "02:00:00", + "income_totals": [{"amount": "30.00", "currency": "USD"}], + } + ], + "projects": [], + "tags": [], + } + + +class ReportExporterTests(TestCase): + def test_excel_export_adds_per_user_sheets_and_daily_rate_column(self): + locale = build_export_locale("en") + report_data = make_report_data( + hourly_rate={"amount": "15.00", "currency": "USD"}, + ) + per_user_reports = [ + make_report_data(user_name="Owner User", mobile="09129990001"), + make_report_data(user_name="Team Mate", mobile="09129990002"), + ] + + workbook = load_workbook( + BytesIO( + build_excel_report( + report_data=report_data, + locale=locale, + per_user_reports=per_user_reports, + ) + ) + ) + + self.assertEqual(workbook.sheetnames[0], "Overall Report") + self.assertIn("Owner User", workbook.sheetnames[1]) + self.assertIn("Team Mate", workbook.sheetnames[2]) + + worksheet = workbook.active + values = list(worksheet.iter_rows(values_only=True)) + + self.assertTrue(any(row[:2] == ("User", "Owner User") for row in values if row)) + self.assertTrue(any(row[:2] == ("Mobile", "09129990001") for row in values if row)) + + daily_header = next(row[:6] for row in values if row and row[0] == "Date") + self.assertEqual( + daily_header, + ( + "Date", + "Billable hours", + "Non-billable hours", + "Total hours", + "Hourly rate", + "Income", + ), + ) + + daily_row = next(row[:6] for row in values if row and row[0] == "2026/04/12") + self.assertEqual(daily_row[4], "15 USD") + + def test_pdf_export_supports_persian_locale(self): + locale = build_export_locale("fa") + report_data = make_report_data( + hourly_rate={"amount": "15.00", "currency": "USD"}, + ) + + content = build_pdf_report(report_data=report_data, locale=locale) + + self.assertEqual(content[:4], b"%PDF") diff --git a/apps/tags/tests/__init__.py b/apps/tags/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/apps/tags/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/apps/tags/tests/test_services.py b/apps/tags/tests/test_services.py new file mode 100644 index 0000000..afe37eb --- /dev/null +++ b/apps/tags/tests/test_services.py @@ -0,0 +1,59 @@ +from django.test import TestCase +from rest_framework.exceptions import PermissionDenied, ValidationError + +from apps.tags.models import Tag +from apps.tags.services.tags import create_tag, update_tag +from apps.users.models import User +from apps.workspaces.models import Workspace, WorkspaceMembership + + +class TagServiceTests(TestCase): + @classmethod + def setUpTestData(cls): + cls.owner = User.objects.create_user(mobile="09120000021", password="secret123") + cls.member = User.objects.create_user(mobile="09120000022", password="secret123") + cls.outsider = User.objects.create_user(mobile="09120000023", password="secret123") + cls.workspace = Workspace.objects.create(name="Tags", owner=cls.owner) + WorkspaceMembership.objects.create( + workspace=cls.workspace, + user=cls.member, + role=WorkspaceMembership.Role.MEMBER, + is_active=True, + ) + + def test_create_tag_creates_record_for_workspace_member(self): + tag = create_tag(self.member, self.workspace.id, "Urgent", "#111111") + + self.assertEqual(tag.name, "Urgent") + self.assertEqual(tag.color, "#111111") + self.assertEqual(tag.created_by, self.member) + + def test_create_tag_rejects_non_member(self): + with self.assertRaises(PermissionDenied): + create_tag(self.outsider, self.workspace.id, "Urgent") + + def test_create_tag_rejects_duplicate_name(self): + Tag.objects.create(workspace=self.workspace, name="Urgent") + + with self.assertRaises(ValidationError) as exc: + create_tag(self.owner, self.workspace.id, "Urgent") + + self.assertIn("name", exc.exception.detail) + + def test_update_tag_changes_requested_fields(self): + tag = Tag.objects.create(workspace=self.workspace, name="Urgent", color="#000000") + + updated = update_tag(tag, name="Later", color="#222222") + + self.assertEqual(updated.name, "Later") + self.assertEqual(updated.color, "#222222") + + def test_update_tag_rejects_duplicate_name(self): + Tag.objects.create(workspace=self.workspace, name="Existing") + tag = Tag.objects.create(workspace=self.workspace, name="Urgent") + + with self.assertRaises(ValidationError) as exc: + update_tag(tag, name="Existing") + + self.assertIn("name", exc.exception.detail) + diff --git a/apps/tags/tests/test_views.py b/apps/tags/tests/test_views.py new file mode 100644 index 0000000..c5526a4 --- /dev/null +++ b/apps/tags/tests/test_views.py @@ -0,0 +1,136 @@ +from rest_framework.test import APITestCase + +from apps.tags.models import Tag +from apps.users.models import User +from apps.workspaces.models import Workspace, WorkspaceMembership + + +class TagViewTests(APITestCase): + @classmethod + def setUpTestData(cls): + cls.owner = User.objects.create_user(mobile="09120000031", password="secret123") + cls.admin = User.objects.create_user(mobile="09120000032", password="secret123") + cls.second_admin = User.objects.create_user(mobile="09120000033", password="secret123") + cls.member = User.objects.create_user(mobile="09120000034", password="secret123") + cls.guest = User.objects.create_user(mobile="09120000035", password="secret123") + cls.outsider = User.objects.create_user(mobile="09120000036", password="secret123") + + cls.workspace = Workspace.objects.create(name="Tags API", owner=cls.owner) + for user, role in ( + (cls.admin, WorkspaceMembership.Role.ADMIN), + (cls.second_admin, WorkspaceMembership.Role.ADMIN), + (cls.member, WorkspaceMembership.Role.MEMBER), + (cls.guest, WorkspaceMembership.Role.GUEST), + ): + WorkspaceMembership.objects.create( + workspace=cls.workspace, + user=user, + role=role, + is_active=True, + ) + + cls.other_workspace = Workspace.objects.create(name="Elsewhere", owner=cls.outsider) + cls.visible_tag = Tag.objects.create(workspace=cls.workspace, name="Visible", color="#111111") + cls.hidden_tag = Tag.objects.create(workspace=cls.other_workspace, name="Hidden", color="#222222") + cls.admin_owned_tag = Tag.objects.create( + workspace=cls.workspace, + name="Admin Tag", + color="#333333", + created_by=cls.admin, + updated_by=cls.admin, + ) + + def test_list_only_returns_tags_for_member_workspaces(self): + self.client.force_authenticate(user=self.member) + + response = self.client.get("/api/tags/") + + self.assertEqual(response.status_code, 200) + results = ( + response.data + if isinstance(response.data, list) + else response.data.get("results") + or response.data.get("items") + or [] + ) + names = {item["name"] for item in results} + self.assertIn("Visible", names) + self.assertNotIn("Hidden", names) + + def test_member_can_create_tag(self): + self.client.force_authenticate(user=self.member) + + response = self.client.post( + "/api/tags/", + { + "workspace_id": str(self.workspace.id), + "name": "Created", + "color": "#123456", + }, + format="json", + ) + + self.assertEqual(response.status_code, 201) + self.assertEqual(response.data["name"], "Created") + + def test_guest_cannot_create_tag(self): + self.client.force_authenticate(user=self.guest) + + response = self.client.post( + "/api/tags/", + { + "workspace_id": str(self.workspace.id), + "name": "Created", + }, + format="json", + ) + + self.assertEqual(response.status_code, 403) + + def test_member_cannot_update_tag(self): + self.client.force_authenticate(user=self.member) + + response = self.client.patch( + f"/api/tags/{self.visible_tag.id}/", + {"name": "Updated"}, + format="json", + ) + + self.assertEqual(response.status_code, 403) + + def test_admin_can_update_tag(self): + self.client.force_authenticate(user=self.admin) + + response = self.client.patch( + f"/api/tags/{self.visible_tag.id}/", + {"name": "Updated"}, + format="json", + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data["name"], "Updated") + + def test_admin_can_delete_only_tag_they_created(self): + self.client.force_authenticate(user=self.second_admin) + + forbidden = self.client.delete(f"/api/tags/{self.admin_owned_tag.id}/") + self.assertEqual(forbidden.status_code, 403) + + self.client.force_authenticate(user=self.admin) + allowed = self.client.delete(f"/api/tags/{self.admin_owned_tag.id}/") + self.assertEqual(allowed.status_code, 204) + self.assertTrue(Tag.all_objects.get(id=self.admin_owned_tag.id).is_deleted) + + def test_owner_can_delete_any_tag(self): + tag = Tag.objects.create( + workspace=self.workspace, + name="Owner Delete", + created_by=self.admin, + updated_by=self.admin, + ) + self.client.force_authenticate(user=self.owner) + + response = self.client.delete(f"/api/tags/{tag.id}/") + + self.assertEqual(response.status_code, 204) + self.assertTrue(Tag.all_objects.get(id=tag.id).is_deleted) diff --git a/apps/users/tests/test_api_views.py b/apps/users/tests/test_api_views.py new file mode 100644 index 0000000..47ad56a --- /dev/null +++ b/apps/users/tests/test_api_views.py @@ -0,0 +1,199 @@ +from unittest.mock import patch + +from rest_framework.test import APIRequestFactory +from rest_framework import status +from rest_framework.test import APITestCase + +from apps.users.api.views import RegisterWithPasswordView +from apps.users.models import User + + +class UserApiViewTests(APITestCase): + @classmethod + def setUpTestData(cls): + cls.user = User.objects.create_user( + mobile="09123330001", + password="secret123", + first_name="Ali", + last_name="Test", + ) + cls.other_user = User.objects.create_user( + mobile="09123330002", + password="secret123", + first_name="Sara", + last_name="Search", + ) + + @patch("apps.users.api.views.register_user_with_password") + def test_register_with_password_view_returns_tokens(self, register_user_with_password): + register_user_with_password.return_value = { + "access": "access-token", + "refresh": "refresh-token", + } + request = APIRequestFactory().post( + "/api/users/register/password/", + {"mobile": "09123330009", "password": "secret123"}, + format="json", + ) + + response = RegisterWithPasswordView.as_view()(request) + + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + self.assertEqual(response.data["access"], "access-token") + register_user_with_password.assert_called_once_with("09123330009", "secret123") + + def test_register_with_password_requires_mobile_and_password(self): + request = APIRequestFactory().post("/api/users/register/password/", {}, format="json") + response = RegisterWithPasswordView.as_view()(request) + + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + @patch("apps.users.api.views.generate_and_send_otp") + def test_send_otp_view_validates_and_dispatches(self, generate_and_send_otp): + response = self.client.post( + "/api/users/otp/send/", + {"mobile": "09123330009", "mode": "login"}, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + generate_and_send_otp.assert_called_once_with( + mobile="09123330009", + mode="login", + ) + + @patch("apps.users.api.views.login_with_password") + def test_login_view_returns_tokens(self, login_with_password): + login_with_password.return_value = {"access": "a", "refresh": "r"} + + response = self.client.post( + "/api/users/login/", + {"mobile": "09123330001", "password": "secret123"}, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["refresh"], "r") + login_with_password.assert_called_once() + + @patch("apps.users.api.views.login_with_otp") + def test_login_otp_view_returns_tokens(self, login_with_otp): + login_with_otp.return_value = {"access": "a", "refresh": "r"} + + response = self.client.post( + "/api/users/otp/login/", + {"mobile": "09123330001", "code": "123456"}, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["access"], "a") + + @patch("apps.users.api.views.reset_password_with_otp") + def test_reset_password_view_calls_service(self, reset_password_with_otp): + response = self.client.post( + "/api/users/password/reset/", + { + "mobile": "09123330001", + "code": "123456", + "password": "new-secret123", + "re_password": "new-secret123", + }, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + reset_password_with_otp.assert_called_once_with( + mobile="09123330001", + code="123456", + password="new-secret123", + ) + + @patch("apps.users.api.views.change_password") + def test_change_password_view_requires_auth_and_calls_service(self, change_password): + self.client.force_authenticate(user=self.user) + + response = self.client.patch( + "/api/users/password/change/", + { + "old_password": "secret123", + "new_password": "new-secret123", + "re_password": "new-secret123", + }, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + change_password.assert_called_once_with( + user=self.user, + old_password="secret123", + new_password="new-secret123", + ) + + @patch("apps.users.api.views.logout_user") + def test_logout_view_calls_service(self, logout_user): + self.client.force_authenticate(user=self.user) + + response = self.client.post( + "/api/users/logout/", + {"refresh": "refresh-token"}, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_205_RESET_CONTENT) + logout_user.assert_called_once_with("refresh-token") + + def test_user_list_and_profile_views_require_authentication(self): + list_response = self.client.get("/api/users/list/") + me_response = self.client.get("/api/users/me/") + + self.assertEqual(list_response.status_code, status.HTTP_401_UNAUTHORIZED) + self.assertEqual(me_response.status_code, status.HTTP_401_UNAUTHORIZED) + + def test_user_list_returns_users_for_authenticated_request(self): + self.client.force_authenticate(user=self.user) + + response = self.client.get("/api/users/list/") + + self.assertEqual(response.status_code, status.HTTP_200_OK) + items = ( + response.data + if isinstance(response.data, list) + else response.data.get("results") + or response.data.get("items") + or [] + ) + mobiles = {item["mobile"] for item in items} + self.assertIn(self.user.mobile, mobiles) + self.assertIn(self.other_user.mobile, mobiles) + + def test_user_me_retrieve_and_patch_work(self): + self.client.force_authenticate(user=self.user) + + retrieve_response = self.client.get("/api/users/me/") + self.assertEqual(retrieve_response.status_code, status.HTTP_200_OK) + self.assertEqual(retrieve_response.data["mobile"], self.user.mobile) + + patch_response = self.client.patch( + "/api/users/me/", + {"first_name": "Updated", "description": "Bio"}, + format="json", + ) + + self.assertEqual(patch_response.status_code, status.HTTP_200_OK) + self.assertEqual(patch_response.data["first_name"], "Updated") + self.user.refresh_from_db() + self.assertEqual(self.user.description, "Bio") + + def test_user_search_handles_missing_mobile_not_found_and_success(self): + self.client.force_authenticate(user=self.user) + + missing = self.client.get("/api/users/search/") + self.assertEqual(missing.status_code, status.HTTP_400_BAD_REQUEST) + + not_found = self.client.get("/api/users/search/?mobile=09129999999") + self.assertEqual(not_found.status_code, status.HTTP_404_NOT_FOUND) + + success = self.client.get(f"/api/users/search/?mobile={self.other_user.mobile}") + self.assertEqual(success.status_code, status.HTTP_200_OK) + self.assertEqual(success.data["mobile"], self.other_user.mobile) diff --git a/apps/users/tests/test_auth_services.py b/apps/users/tests/test_auth_services.py new file mode 100644 index 0000000..d7864e7 --- /dev/null +++ b/apps/users/tests/test_auth_services.py @@ -0,0 +1,149 @@ +from unittest.mock import Mock, patch + +from django.test import TestCase +from rest_framework.exceptions import ValidationError +from rest_framework_simplejwt.tokens import RefreshToken + +from apps.users.models import LoginAttempt, User +from apps.users.services.auth import ( + change_password, + generate_and_send_otp, + login_with_otp, + login_with_password, + logout_user, + register_user_with_otp, + register_user_with_password, + reset_password_with_otp, +) + + +class FakeRedisConnection: + def __init__(self): + self.store = {} + + def get(self, key): + return self.store.get(key) + + def setex(self, key, timeout, value): + self.store[key] = str(value).encode("utf-8") + + def delete(self, key): + self.store.pop(key, None) + + +class AuthServiceTests(TestCase): + def test_register_user_with_password_creates_user_and_tokens(self): + tokens = register_user_with_password("09120000001", "secret123") + + self.assertTrue(User.objects.filter(mobile="09120000001").exists()) + self.assertIn("access", tokens) + self.assertIn("refresh", tokens) + + @patch("apps.users.services.auth.send_verification_sms.delay") + @patch("apps.users.services.auth.get_redis_connection") + def test_generate_and_send_otp_stores_code_and_schedules_sms( + self, + get_redis_connection, + send_delay, + ): + User.objects.create_user(mobile="09120000002", password="secret123") + fake_redis = FakeRedisConnection() + get_redis_connection.return_value = fake_redis + + generate_and_send_otp("09120000002", "login") + + self.assertIn("verification_code:09120000002", fake_redis.store) + send_delay.assert_called_once() + + @patch("apps.users.services.auth.record_login_attempt") + def test_login_with_password_records_success(self, record_login_attempt): + user = User.objects.create_user(mobile="09120000003", password="secret123") + + tokens = login_with_password("09120000003", "secret123", request=None) + + self.assertIn("access", tokens) + record_login_attempt.assert_called_once_with( + None, + user, + LoginAttempt.StatusType.SUCCESS, + ) + + @patch("apps.users.services.auth.record_login_attempt") + def test_login_with_password_rejects_invalid_password(self, record_login_attempt): + User.objects.create_user(mobile="09120000004", password="secret123") + + with self.assertRaises(ValidationError): + login_with_password("09120000004", "wrong-password", request=None) + + record_login_attempt.assert_called_once() + + @patch("apps.users.services.auth.record_login_attempt") + @patch("apps.users.services.auth.get_redis_connection") + def test_login_with_otp_creates_user_and_consumes_code( + self, + get_redis_connection, + record_login_attempt, + ): + fake_redis = FakeRedisConnection() + fake_redis.setex("verification_code:09120000005", 120, "12345") + get_redis_connection.return_value = fake_redis + + tokens = login_with_otp("09120000005", "12345", request=None) + + self.assertTrue(User.objects.filter(mobile="09120000005").exists()) + self.assertIn("access", tokens) + self.assertNotIn("verification_code:09120000005", fake_redis.store) + record_login_attempt.assert_called_once() + + @patch("apps.users.services.auth.get_redis_connection") + def test_register_user_with_otp_verifies_code_and_marks_user_verified( + self, + get_redis_connection, + ): + fake_redis = FakeRedisConnection() + fake_redis.setex("verification_code:09120000006", 120, "12345") + get_redis_connection.return_value = fake_redis + + tokens = register_user_with_otp( + mobile="09120000006", + code="12345", + password="secret123", + first_name="OTP", + last_name="User", + ) + + user = User.objects.get(mobile="09120000006") + self.assertTrue(user.is_verified) + self.assertTrue(user.check_password("secret123")) + self.assertIn("refresh", tokens) + + @patch("apps.users.services.auth.get_redis_connection") + def test_reset_password_with_otp_updates_password(self, get_redis_connection): + user = User.objects.create_user(mobile="09120000007", password="oldsecret") + fake_redis = FakeRedisConnection() + fake_redis.setex("verification_code:09120000007", 120, "12345") + get_redis_connection.return_value = fake_redis + + reset_password_with_otp("09120000007", "12345", "newsecret") + + user.refresh_from_db() + self.assertTrue(user.check_password("newsecret")) + self.assertNotIn("verification_code:09120000007", fake_redis.store) + + def test_change_password_updates_existing_user_password(self): + user = User.objects.create_user(mobile="09120000008", password="oldsecret") + + change_password(user, "oldsecret", "newsecret") + + user.refresh_from_db() + self.assertTrue(user.check_password("newsecret")) + self.assertIsNotNone(user.password_updated_at) + + def test_logout_user_blacklists_refresh_token(self): + user = User.objects.create_user(mobile="09120000009", password="secret123") + refresh = str(RefreshToken.for_user(user)) + + logout_user(refresh) + + with self.assertRaises(ValidationError): + logout_user(refresh) diff --git a/apps/users/tests/test_utils.py b/apps/users/tests/test_utils.py new file mode 100644 index 0000000..bffb00c --- /dev/null +++ b/apps/users/tests/test_utils.py @@ -0,0 +1,36 @@ +from django.test import RequestFactory, TestCase + +from apps.users.models import LoginAttempt, User +from apps.users.utils import _get_ip, record_login_attempt + + +class UserUtilsTests(TestCase): + @classmethod + def setUpTestData(cls): + cls.user = User.objects.create_user(mobile="09120000051", password="secret123") + + def setUp(self): + self.factory = RequestFactory() + + def test_get_ip_returns_none_without_request(self): + self.assertIsNone(_get_ip(None)) + + def test_get_ip_prefers_forwarded_header(self): + request = self.factory.get("/", HTTP_X_FORWARDED_FOR="1.1.1.1, 2.2.2.2") + + self.assertEqual(_get_ip(request), "1.1.1.1") + + def test_get_ip_falls_back_to_remote_addr(self): + request = self.factory.get("/", REMOTE_ADDR="3.3.3.3") + + self.assertEqual(_get_ip(request), "3.3.3.3") + + def test_record_login_attempt_persists_attempt(self): + request = self.factory.get("/", REMOTE_ADDR="4.4.4.4") + + record_login_attempt(request, user=self.user, status=LoginAttempt.StatusType.SUCCESS) + + attempt = LoginAttempt.objects.get() + self.assertEqual(attempt.user, self.user) + self.assertEqual(attempt.status, LoginAttempt.StatusType.SUCCESS) + self.assertEqual(attempt.ip_address, "4.4.4.4") diff --git a/apps/workspaces/tests/test_api_permissions.py b/apps/workspaces/tests/test_api_permissions.py new file mode 100644 index 0000000..a3ed6ec --- /dev/null +++ b/apps/workspaces/tests/test_api_permissions.py @@ -0,0 +1,146 @@ +from types import SimpleNamespace + +from django.test import TestCase + +from apps.users.models import User +from apps.workspaces.api.permissions import ( + CanWorkspaceManageMembers, + IsWorkspaceAdmin, + IsWorkspaceMember, + IsWorkspaceOwner, +) +from apps.workspaces.models import Workspace, WorkspaceMembership + + +class WorkspacePermissionTests(TestCase): + @classmethod + def setUpTestData(cls): + cls.owner = User.objects.create_user(mobile="09127770021", password="secret123") + cls.admin = User.objects.create_user(mobile="09127770022", password="secret123") + cls.member = User.objects.create_user(mobile="09127770023", password="secret123") + cls.guest = User.objects.create_user(mobile="09127770024", password="secret123") + cls.outsider = User.objects.create_user(mobile="09127770025", password="secret123") + + cls.workspace = Workspace.objects.create(name="Workspace Perms", owner=cls.owner) + cls.admin_membership = WorkspaceMembership.objects.create( + workspace=cls.workspace, + user=cls.admin, + role=WorkspaceMembership.Role.ADMIN, + is_active=True, + ) + cls.member_membership = WorkspaceMembership.objects.create( + workspace=cls.workspace, + user=cls.member, + role=WorkspaceMembership.Role.MEMBER, + is_active=True, + ) + cls.guest_membership = WorkspaceMembership.objects.create( + workspace=cls.workspace, + user=cls.guest, + role=WorkspaceMembership.Role.GUEST, + is_active=True, + ) + cls.workspace_note = SimpleNamespace(workspace=cls.workspace) + + def _request_for(self, user): + return SimpleNamespace(user=user) + + def test_is_workspace_owner_handles_workspace_and_membership_objects(self): + permission = IsWorkspaceOwner() + + self.assertTrue( + permission.has_object_permission( + self._request_for(self.owner), + None, + self.workspace, + ) + ) + self.assertTrue( + permission.has_object_permission( + self._request_for(self.owner), + None, + self.admin_membership, + ) + ) + self.assertFalse( + permission.has_object_permission( + self._request_for(self.admin), + None, + self.workspace, + ) + ) + + def test_is_workspace_admin_accepts_workspace_related_objects(self): + permission = IsWorkspaceAdmin() + + self.assertTrue( + permission.has_object_permission( + self._request_for(self.admin), + None, + self.workspace, + ) + ) + self.assertTrue( + permission.has_object_permission( + self._request_for(self.admin), + None, + self.workspace_note, + ) + ) + self.assertFalse( + permission.has_object_permission( + self._request_for(self.member), + None, + self.workspace, + ) + ) + + def test_is_workspace_member_allows_active_guest_but_not_outsider(self): + permission = IsWorkspaceMember() + + self.assertTrue( + permission.has_object_permission( + self._request_for(self.guest), + None, + self.workspace, + ) + ) + self.assertFalse( + permission.has_object_permission( + self._request_for(self.outsider), + None, + self.workspace, + ) + ) + + def test_can_workspace_manage_members_only_allows_owner_and_admin(self): + permission = CanWorkspaceManageMembers() + + self.assertTrue( + permission.has_object_permission( + self._request_for(self.owner), + None, + self.workspace, + ) + ) + self.assertTrue( + permission.has_object_permission( + self._request_for(self.admin), + None, + self.admin_membership, + ) + ) + self.assertFalse( + permission.has_object_permission( + self._request_for(self.member), + None, + self.workspace, + ) + ) + self.assertFalse( + permission.has_object_permission( + self._request_for(self.owner), + None, + object(), + ) + )