test(backend): add coverage for services tasks and apis

This commit is contained in:
2026-04-30 12:44:24 +03:30
parent 8774a4d4dc
commit 3152284cf3
15 changed files with 1279 additions and 0 deletions

View File

@@ -0,0 +1 @@

View File

@@ -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)

View File

@@ -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)

View File

@@ -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
)

View File

@@ -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)

View File

@@ -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)

View File

@@ -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"])

View File

@@ -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")

View File

@@ -0,0 +1 @@

View File

@@ -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)

View File

@@ -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)

View File

@@ -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)

View File

@@ -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)

View File

@@ -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")

View File

@@ -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(),
)
)