Files
qlockify-backend-deployment/apps/workspaces/tests/test_capabilities.py
Amirhossein Khalili 95f5e85e44
Some checks failed
Backend CI/CD / test (push) Has been cancelled
Backend CI/CD / deploy (push) Has been cancelled
feat(workspaces): add bulk member import endpoints
2026-06-18 22:53:34 +03:30

528 lines
19 KiB
Python

from datetime import timedelta
from decimal import Decimal
from django.utils import timezone
from rest_framework.test import APITestCase
from apps.clients.models import Client
from apps.projects.models import Project
from apps.tags.models import Tag
from apps.users.models import User
from apps.workspaces.models import (
HourlyRateHistory,
PriceUnit,
Workspace,
WorkspaceMembership,
WorkspaceUserRate,
)
class WorkspaceCapabilityTests(APITestCase):
@classmethod
def setUpTestData(cls):
cls.owner = cls._user(1)
cls.admin = cls._user(2)
cls.member = cls._user(3)
cls.guest = cls._user(4)
cls.extra_owner = cls._user(5)
cls.workspace = Workspace.objects.create(name="Ops", description="", owner=cls.owner)
WorkspaceMembership.objects.create(
workspace=cls.workspace,
user=cls.admin,
role=WorkspaceMembership.Role.ADMIN,
is_active=True,
)
WorkspaceMembership.objects.create(
workspace=cls.workspace,
user=cls.member,
role=WorkspaceMembership.Role.MEMBER,
is_active=True,
)
WorkspaceMembership.objects.create(
workspace=cls.workspace,
user=cls.guest,
role=WorkspaceMembership.Role.GUEST,
is_active=True,
)
cls.project = Project.objects.create(
workspace=cls.workspace,
name="Alpha",
description="",
)
@staticmethod
def _user(index):
return User.objects.create_user(
mobile=f"091255500{index:02d}",
password="secret123",
first_name=f"User{index}",
)
def test_member_is_read_only_for_clients_and_projects(self):
client = Client.objects.create(
workspace=self.workspace,
name="Existing Client",
notes="",
)
self.client.force_authenticate(user=self.member)
client_response = self.client.post(
"/api/clients/",
{
"workspace_id": str(self.workspace.id),
"name": "Acme",
"notes": "",
},
format="json",
)
update_client_response = self.client.patch(
f"/api/clients/{client.id}/",
{"name": "Updated"},
format="json",
)
delete_client_response = self.client.delete(f"/api/clients/{client.id}/")
project_response = self.client.post(
"/api/projects/",
{
"workspace": str(self.workspace.id),
"name": "Beta",
"description": "",
"client": None,
},
format="json",
)
update_project_response = self.client.patch(
f"/api/projects/{self.project.id}/",
{"description": "Blocked edit"},
format="json",
)
archive_project_response = self.client.post(
f"/api/projects/{self.project.id}/archive/"
)
delete_project_response = self.client.delete(f"/api/projects/{self.project.id}/")
self.assertEqual(client_response.status_code, 403)
self.assertEqual(update_client_response.status_code, 403)
self.assertEqual(delete_client_response.status_code, 403)
self.assertEqual(project_response.status_code, 403)
self.assertEqual(update_project_response.status_code, 403)
self.assertEqual(archive_project_response.status_code, 403)
self.assertEqual(delete_project_response.status_code, 403)
def test_member_can_create_tags_and_manage_own_time_entries(self):
tag = Tag.objects.create(
workspace=self.workspace,
name="Existing",
color="#000000",
)
self.client.force_authenticate(user=self.member)
create_tag_response = self.client.post(
"/api/tags/",
{
"workspace_id": str(self.workspace.id),
"name": "New Tag",
"color": "#ffffff",
},
format="json",
)
update_tag_response = self.client.patch(
f"/api/tags/{tag.id}/",
{"name": "Changed"},
format="json",
)
delete_tag_response = self.client.delete(f"/api/tags/{tag.id}/")
now = timezone.now()
create_entry_response = self.client.post(
"/api/time-entries/",
{
"workspace_id": str(self.workspace.id),
"start_time": now.isoformat(),
"end_time": (now + timedelta(hours=1)).isoformat(),
"description": "Focus block",
},
format="json",
)
self.assertEqual(create_tag_response.status_code, 201)
self.assertEqual(update_tag_response.status_code, 403)
self.assertEqual(delete_tag_response.status_code, 403)
self.assertEqual(create_entry_response.status_code, 201)
entry_id = create_entry_response.data["id"]
update_entry_response = self.client.patch(
f"/api/time-entries/{entry_id}/",
{"description": "Updated focus block"},
format="json",
)
delete_entry_response = self.client.delete(f"/api/time-entries/{entry_id}/")
self.assertEqual(update_entry_response.status_code, 200)
self.assertEqual(delete_entry_response.status_code, 204)
def test_guest_is_read_only_for_workspace_resources(self):
Client.objects.create(workspace=self.workspace, name="Visible Client", notes="")
Tag.objects.create(workspace=self.workspace, name="Visible Tag", color="#123456")
self.client.force_authenticate(user=self.guest)
list_clients_response = self.client.get(
f"/api/clients/?workspace={self.workspace.id}"
)
list_projects_response = self.client.get(
f"/api/projects/?workspace={self.workspace.id}"
)
create_tag_response = self.client.post(
"/api/tags/",
{
"workspace_id": str(self.workspace.id),
"name": "Blocked",
"color": "#ffffff",
},
format="json",
)
create_entry_response = self.client.post(
"/api/time-entries/",
{
"workspace_id": str(self.workspace.id),
"start_time": timezone.now().isoformat(),
"description": "Blocked guest entry",
},
format="json",
)
edit_project_response = self.client.patch(
f"/api/projects/{self.project.id}/",
{"description": "Blocked"},
format="json",
)
self.assertEqual(list_clients_response.status_code, 200)
self.assertEqual(list_projects_response.status_code, 200)
self.assertEqual(create_tag_response.status_code, 403)
self.assertEqual(create_entry_response.status_code, 403)
self.assertEqual(edit_project_response.status_code, 403)
def test_member_cannot_edit_project(self):
self.client.force_authenticate(user=self.member)
response = self.client.patch(
f"/api/projects/{self.project.id}/",
{"description": "Still blocked"},
format="json",
)
self.assertEqual(response.status_code, 403)
def test_member_can_list_workspace_members_with_restricted_user_fields(self):
self.client.force_authenticate(user=self.member)
response = self.client.get(
f"/api/workspace-memberships/?workspace={self.workspace.id}"
)
self.assertEqual(response.status_code, 200)
payload = (
response.data.get("items", response.data)
if isinstance(response.data, dict)
else response.data
)
self.assertGreaterEqual(len(payload), 1)
first_user = payload[0]["user"]
self.assertNotIn("mobile", first_user)
self.assertNotIn("email", first_user)
def test_owner_can_list_workspace_members_with_full_user_fields(self):
self.client.force_authenticate(user=self.owner)
response = self.client.get(
f"/api/workspace-memberships/?workspace={self.workspace.id}"
)
self.assertEqual(response.status_code, 200)
payload = (
response.data.get("items", response.data)
if isinstance(response.data, dict)
else response.data
)
self.assertGreaterEqual(len(payload), 1)
first_user = payload[0]["user"]
self.assertIn("mobile", first_user)
def test_admin_cannot_change_owner_membership_but_canonical_owner_can(self):
extra_owner_membership = WorkspaceMembership.objects.create(
workspace=self.workspace,
user=self.extra_owner,
role=WorkspaceMembership.Role.OWNER,
is_active=True,
)
self.client.force_authenticate(user=self.admin)
admin_response = self.client.patch(
f"/api/workspace-memberships/{extra_owner_membership.id}/",
{"role": WorkspaceMembership.Role.ADMIN},
format="json",
)
self.client.force_authenticate(user=self.owner)
owner_response = self.client.patch(
f"/api/workspace-memberships/{extra_owner_membership.id}/",
{"role": WorkspaceMembership.Role.ADMIN},
format="json",
)
self.assertEqual(admin_response.status_code, 403)
self.assertEqual(owner_response.status_code, 200)
def test_admin_cannot_add_or_change_admin_memberships(self):
admin_membership = WorkspaceMembership.objects.get(
workspace=self.workspace,
user=self.admin,
is_deleted=False,
)
self.client.force_authenticate(user=self.admin)
create_response = self.client.post(
"/api/workspace-memberships/",
{
"workspace": str(self.workspace.id),
"user": str(self.member.id),
"role": WorkspaceMembership.Role.ADMIN,
},
format="json",
)
update_response = self.client.patch(
f"/api/workspace-memberships/{admin_membership.id}/",
{"role": WorkspaceMembership.Role.MEMBER},
format="json",
)
delete_response = self.client.delete(
f"/api/workspace-memberships/{admin_membership.id}/"
)
self.assertEqual(create_response.status_code, 403)
self.assertEqual(update_response.status_code, 403)
self.assertEqual(delete_response.status_code, 403)
def test_owner_can_validate_and_commit_member_import_with_rate(self):
PriceUnit.objects.create(code="IRT", name="Toman", local_name="Toman", symbol="Toman")
target = self._user(21)
self.client.force_authenticate(user=self.owner)
validate_response = self.client.post(
"/api/workspace-memberships/import/validate/",
{
"workspace": str(self.workspace.id),
"rows": [
{
"line": 2,
"mobile": target.mobile,
"role": "member",
"hourly_rate": "150000",
"currency": "IRT",
}
],
},
format="json",
)
self.assertEqual(validate_response.status_code, 200)
self.assertTrue(validate_response.data["can_commit"])
self.assertEqual(validate_response.data["summary"]["valid"], 1)
self.assertTrue(validate_response.data["import_token"])
commit_response = self.client.post(
"/api/workspace-memberships/import/commit/",
{
"workspace": str(self.workspace.id),
"import_token": validate_response.data["import_token"],
},
format="json",
)
self.assertEqual(commit_response.status_code, 201)
self.assertEqual(commit_response.data["created_memberships"], 1)
self.assertEqual(commit_response.data["created_or_updated_rates"], 1)
self.assertTrue(
WorkspaceMembership.objects.filter(
workspace=self.workspace,
user=target,
role=WorkspaceMembership.Role.MEMBER,
is_active=True,
).exists()
)
rate = WorkspaceUserRate.objects.get(workspace=self.workspace, user=target)
self.assertEqual(rate.hourly_rate, Decimal("150000.00"))
self.assertEqual(rate.currency, "IRT")
self.assertTrue(
HourlyRateHistory.objects.filter(
workspace=self.workspace,
user=target,
hourly_rate=Decimal("150000.00"),
currency="IRT",
).exists()
)
def test_member_import_rejects_invalid_rows(self):
PriceUnit.objects.create(code="USD", name="Dollar", local_name="Dollar", symbol="$")
target = self._user(22)
self.client.force_authenticate(user=self.owner)
response = self.client.post(
"/api/workspace-memberships/import/validate/",
{
"workspace": str(self.workspace.id),
"rows": [
{"line": 2, "mobile": "", "role": "member"},
{"line": 3, "mobile": "09120000000", "role": "member"},
{"line": 4, "mobile": target.mobile, "role": "owner"},
{"line": 5, "mobile": target.mobile, "role": "member"},
{"line": 6, "mobile": self.member.mobile, "role": "member"},
{"line": 7, "mobile": self.guest.mobile, "role": "guest", "hourly_rate": "10"},
{"line": 8, "mobile": self.admin.mobile, "role": "admin", "hourly_rate": "0", "currency": "USD"},
{"line": 9, "mobile": self.extra_owner.mobile, "role": "guest", "hourly_rate": "10", "currency": "XXX"},
],
},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertFalse(response.data["can_commit"])
self.assertIsNone(response.data["import_token"])
self.assertEqual(response.data["summary"]["invalid"], 8)
def test_admin_import_follows_role_assignment_rules(self):
target = self._user(23)
self.client.force_authenticate(user=self.admin)
response = self.client.post(
"/api/workspace-memberships/import/validate/",
{
"workspace": str(self.workspace.id),
"rows": [{"line": 2, "mobile": target.mobile, "role": "admin"}],
},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertFalse(response.data["can_commit"])
self.assertIn("permission", response.data["rows"][0]["messages"][0].lower())
def test_member_cannot_import_workspace_members(self):
target = self._user(24)
self.client.force_authenticate(user=self.member)
response = self.client.post(
"/api/workspace-memberships/import/validate/",
{
"workspace": str(self.workspace.id),
"rows": [{"line": 2, "mobile": target.mobile, "role": "member"}],
},
format="json",
)
self.assertEqual(response.status_code, 403)
def test_import_commit_rejects_expired_token(self):
self.client.force_authenticate(user=self.owner)
response = self.client.post(
"/api/workspace-memberships/import/commit/",
{
"workspace": str(self.workspace.id),
"import_token": "invalid-token",
},
format="json",
)
self.assertEqual(response.status_code, 400)
def test_admin_can_delete_only_owned_clients_tags_and_projects(self):
self.client.force_authenticate(user=self.owner)
owner_client_response = self.client.post(
"/api/clients/",
{
"workspace_id": str(self.workspace.id),
"name": "Owner Client",
"notes": "",
},
format="json",
)
owner_tag_response = self.client.post(
"/api/tags/",
{
"workspace_id": str(self.workspace.id),
"name": "Owner Tag",
"color": "#123456",
},
format="json",
)
owner_project_response = self.client.post(
"/api/projects/",
{
"workspace": str(self.workspace.id),
"name": "Owner Project",
"description": "",
"client": None,
},
format="json",
)
self.client.force_authenticate(user=self.admin)
admin_client_response = self.client.post(
"/api/clients/",
{
"workspace_id": str(self.workspace.id),
"name": "Admin Client",
"notes": "",
},
format="json",
)
admin_tag_response = self.client.post(
"/api/tags/",
{
"workspace_id": str(self.workspace.id),
"name": "Admin Tag",
"color": "#654321",
},
format="json",
)
admin_project_response = self.client.post(
"/api/projects/",
{
"workspace": str(self.workspace.id),
"name": "Admin Project",
"description": "",
"client": None,
},
format="json",
)
delete_owner_client = self.client.delete(
f"/api/clients/{owner_client_response.data['id']}/"
)
delete_owner_tag = self.client.delete(
f"/api/tags/{owner_tag_response.data['id']}/"
)
delete_owner_project = self.client.delete(
f"/api/projects/{owner_project_response.data['id']}/"
)
delete_admin_client = self.client.delete(
f"/api/clients/{admin_client_response.data['id']}/"
)
delete_admin_tag = self.client.delete(
f"/api/tags/{admin_tag_response.data['id']}/"
)
delete_admin_project = self.client.delete(
f"/api/projects/{admin_project_response.data['id']}/"
)
self.assertEqual(delete_owner_client.status_code, 403)
self.assertEqual(delete_owner_tag.status_code, 403)
self.assertIn(delete_owner_project.status_code, {403, 404})
self.assertEqual(delete_admin_client.status_code, 204)
self.assertEqual(delete_admin_tag.status_code, 204)
self.assertEqual(delete_admin_project.status_code, 204)