from apps.audit.models import AuditLog from apps.modules import services as modules from apps.tasks.models import Task from tests.base import CoreTestCase URL = "/api/v1/tasks/" class TasksTests(CoreTestCase): def setUp(self): super().setUp() modules.enable_module(actor=self.root, organization=self.org_a, module_code="tasks") def create(self, user, **data): payload = {"title": "t", **data} return self.api(user, self.org_a).post(URL, payload, format="json") def test_employee_creates_own_task_and_sees_only_own(self): r = self.create(self.employee_a, title="mine") self.assertEqual(r.status_code, 201) self.assertEqual(r.data["assignee"]["id"], self.employee_a.pk) Task.objects.create(organization=self.org_a, title="other", created_by=self.manager_a, assignee=self.manager_a) titles = [t["title"] for t in self.api(self.employee_a, self.org_a).get(URL).data["results"]] self.assertEqual(titles, ["mine"]) def test_manager_sees_all_in_org(self): self.create(self.employee_a, title="a") self.create(self.manager_a, title="b") self.assertEqual(self.api(self.manager_a, self.org_a).get(URL).data["count"], 2) def test_employee_cannot_assign_to_others_but_manager_can(self): r = self.create(self.employee_a, assignee=self.manager_a.pk) self.assertEqual(r.status_code, 403) r = self.create(self.manager_a, assignee=self.employee_a.pk) self.assertEqual(r.status_code, 201) # the employee now sees it (assigned to them) and can update its status pk = r.data["id"] r = self.api(self.employee_a, self.org_a).patch(f"{URL}{pk}/", {"status": "in_progress"}, format="json") self.assertEqual(r.status_code, 200) self.assertEqual(r.data["status"], "in_progress") def test_assignee_must_be_active_member_of_same_org(self): r = self.create(self.manager_a, assignee=self.employee_b.pk) self.assertEqual(r.status_code, 400) membership = self.membership(self.employee_a, self.org_a) membership.status = "suspended" membership.save() self.assertEqual(self.create(self.manager_a, assignee=self.employee_a.pk).status_code, 400) def test_employee_cannot_touch_others_task_and_gets_404(self): task = Task.objects.create(organization=self.org_a, title="x", created_by=self.manager_a, assignee=self.manager_a) client = self.api(self.employee_a, self.org_a) self.assertEqual(client.get(f"{URL}{task.pk}/").status_code, 404) self.assertEqual(client.patch(f"{URL}{task.pk}/", {"status": "done"}, format="json").status_code, 404) def test_employee_cannot_reassign_or_delete(self): r = self.create(self.employee_a, title="mine") client = self.api(self.employee_a, self.org_a) pk = r.data["id"] self.assertEqual(client.patch(f"{URL}{pk}/", {"assignee": self.manager_a.pk}, format="json").status_code, 403) self.assertEqual(client.delete(f"{URL}{pk}/").status_code, 403) # no tasks.delete self.assertEqual(self.api(self.manager_a, self.org_a).delete(f"{URL}{pk}/").status_code, 204) def test_other_tenant_isolated(self): r = self.create(self.manager_a, title="secret") pk = r.data["id"] # manager_b has no access to org A at all, and header tricks do not help self.assertEqual(self.api(self.manager_b, self.org_a).get(URL).status_code, 403) modules.enable_module(actor=self.root, organization=self.org_b, module_code="tasks") self.assertEqual(self.api(self.manager_b, self.org_b).get(f"{URL}{pk}/").status_code, 404) def test_module_disabled_blocks_everything(self): modules.disable_module(actor=self.root, organization=self.org_b, module_code="tasks") self.assertEqual(self.api(self.manager_b, self.org_b).get(URL).status_code, 403) modules.disable_module(actor=self.root, organization=self.org_a, module_code="tasks") self.assertEqual(self.api(self.manager_a, self.org_a).get(URL).status_code, 403) self.assertEqual(Task.objects.count(), 0) # nothing deleted by disabling (none existed) def test_disabling_module_keeps_data(self): self.create(self.manager_a, title="keep") modules.disable_module(actor=self.root, organization=self.org_a, module_code="tasks") self.assertEqual(Task.objects.filter(organization=self.org_a).count(), 1) def test_validation_and_audit(self): self.assertEqual(self.create(self.manager_a, title=" ").status_code, 400) self.assertEqual(self.create(self.manager_a, status="bogus").status_code, 400) r = self.create(self.manager_a, title="v") self.assertEqual( self.api(self.manager_a, self.org_a).patch(f"{URL}{r.data['id']}/", {"status": "bogus"}, format="json").status_code, 400, ) self.assertTrue(AuditLog.objects.filter(action="task.create", organization=self.org_a).exists()) def test_assignees_endpoint_needs_assign_permission(self): self.assertEqual(self.api(self.employee_a, self.org_a).get(f"{URL}assignees/").status_code, 403) r = self.api(self.manager_a, self.org_a).get(f"{URL}assignees/") self.assertEqual(r.status_code, 200) self.assertNotIn(self.employee_b.pk, [u["id"] for u in r.data]) def test_navigation_shows_tasks_only_when_module_enabled(self): nav = self.ctx(self.employee_a, self.org_a) from apps.access.services import build_navigation self.assertIn("/tasks", [n["path"] for n in build_navigation(nav)]) modules.disable_module(actor=self.root, organization=self.org_a, module_code="tasks") nav = self.ctx(self.employee_a, self.org_a) self.assertNotIn("/tasks", [n["path"] for n in build_navigation(nav)]) # ---- assignment notice def test_assignee_sees_a_new_task_until_opened(self): pk = self.create(self.manager_a, title="for you", assignee=self.employee_a.pk).data["id"] emp = self.api(self.employee_a, self.org_a) self.assertTrue(emp.get(URL).data["results"][0]["is_new"]) self.assertEqual(emp.get(f"{URL}summary/").data["new_assigned"], 1) self.assertTrue(emp.get(f"{URL}{pk}/").data["is_new"]) # first open still shows it self.assertFalse(emp.get(f"{URL}{pk}/").data["is_new"]) self.assertEqual(emp.get(f"{URL}summary/").data["new_assigned"], 0) def test_own_and_unrelated_tasks_are_not_new(self): self.create(self.employee_a, title="mine") self.assertEqual(self.api(self.employee_a, self.org_a).get(f"{URL}summary/").data["new_assigned"], 0) pk = self.create(self.manager_a, assignee=self.employee_a.pk).data["id"] self.assertFalse(self.api(self.manager_a, self.org_a).get(f"{URL}{pk}/").data["is_new"]) def test_reassignment_is_new_again(self): pk = self.create(self.manager_a, assignee=self.employee_a.pk).data["id"] self.api(self.employee_a, self.org_a).get(f"{URL}{pk}/") self.api(self.manager_a, self.org_a).patch(f"{URL}{pk}/", {"assignee": self.manager_a2.pk}, format="json") self.assertEqual(self.api(self.manager_a2, self.org_a).get(f"{URL}summary/").data["new_assigned"], 1) # ---- review flow def submit(self, pk): return self.api(self.employee_a, self.org_a).patch(f"{URL}{pk}/", {"status": "review"}, format="json") def test_employee_cannot_finish_but_can_submit(self): pk = self.create(self.manager_a, assignee=self.employee_a.pk).data["id"] emp = self.api(self.employee_a, self.org_a) self.assertEqual(emp.patch(f"{URL}{pk}/", {"status": "done"}, format="json").status_code, 403) self.assertEqual(self.submit(pk).data["status"], "review") self.assertEqual(emp.post(f"{URL}{pk}/approve/").status_code, 403) self.assertEqual(emp.post(f"{URL}{pk}/reject/").status_code, 403) def test_approve_finishes_the_task(self): pk = self.create(self.manager_a, assignee=self.employee_a.pk).data["id"] self.submit(pk) r = self.api(self.manager_a, self.org_a).post(f"{URL}{pk}/approve/") self.assertEqual(r.status_code, 200) self.assertEqual((r.data["status"], r.data["reviewed_by"]["id"]), ("done", self.manager_a.pk)) # an approved task cannot be quietly reopened by the assignee self.assertEqual(self.api(self.employee_a, self.org_a).patch(f"{URL}{pk}/", {"status": "todo"}, format="json").status_code, 403) def test_reject_keeps_the_task_open_with_a_reason_and_notifies(self): pk = self.create(self.manager_a, assignee=self.employee_a.pk).data["id"] emp = self.api(self.employee_a, self.org_a) emp.get(f"{URL}{pk}/") self.submit(pk) r = self.api(self.manager_a, self.org_a).post(f"{URL}{pk}/reject/", {"note": "missing the report"}, format="json") self.assertEqual((r.data["status"], r.data["review_note"]), ("in_progress", "missing the report")) self.assertEqual(emp.get(f"{URL}summary/").data["new_assigned"], 1) self.assertEqual(emp.get(f"{URL}{pk}/").data["review_note"], "missing the report") def test_only_submitted_tasks_can_be_reviewed(self): pk = self.create(self.manager_a, assignee=self.employee_a.pk).data["id"] mgr = self.api(self.manager_a, self.org_a) self.assertEqual(mgr.post(f"{URL}{pk}/approve/").status_code, 400) self.assertEqual(mgr.post(f"{URL}{pk}/reject/").status_code, 400) self.assertEqual(self.create(self.manager_a, status="done").status_code, 400) def test_approver_can_mark_done_directly_and_summary_counts(self): pk = self.create(self.manager_a, assignee=self.employee_a.pk).data["id"] self.create(self.manager_a, title="open") self.assertEqual(self.api(self.manager_a, self.org_a).patch(f"{URL}{pk}/", {"status": "done"}, format="json").status_code, 200) s = self.api(self.manager_a, self.org_a).get(f"{URL}summary/").data self.assertEqual((s["todo"], s["done"], s["total"]), (1, 1, 2)) def test_review_endpoints_are_tenant_bound(self): pk = self.create(self.manager_a, assignee=self.employee_a.pk).data["id"] self.submit(pk) modules.enable_module(actor=self.root, organization=self.org_b, module_code="tasks") self.assertEqual(self.api(self.manager_b, self.org_b).post(f"{URL}{pk}/approve/").status_code, 404) def test_list_is_ordered_by_due_date_with_undated_last_and_big_pages(self): self.create(self.manager_a, title="late", due_date="2026-05-10") self.create(self.manager_a, title="none") self.create(self.manager_a, title="soon", due_date="2026-01-02") rows = self.api(self.manager_a, self.org_a).get(URL, {"page_size": 100}).data["results"] self.assertEqual([r["title"] for r in rows], ["soon", "late", "none"]) def test_open_filter_excludes_finished_and_done_history_is_newest_first(self): mgr = self.api(self.manager_a, self.org_a) first = self.create(self.manager_a, title="first").data["id"] second = self.create(self.manager_a, title="second").data["id"] self.create(self.manager_a, title="still open") for pk in (first, second): mgr.patch(f"{URL}{pk}/", {"status": "done"}, format="json") self.assertEqual([r["title"] for r in mgr.get(URL, {"status": "open"}).data["results"]], ["still open"]) done = [r["title"] for r in mgr.get(URL, {"status": "done"}).data["results"]] self.assertEqual(sorted(done), ["first", "second"]) self.assertEqual(done[0], "second") # finished last -> listed first # ---- quality rating & history def finished(self, assignee=None, **extra): pk = self.create(self.manager_a, assignee=(assignee or self.employee_a).pk, **extra).data["id"] self.api(assignee and self.manager_a or self.employee_a, self.org_a) self.api(self.employee_a, self.org_a).patch(f"{URL}{pk}/", {"status": "review"}, format="json") if not assignee else None self.api(self.manager_a, self.org_a).patch(f"{URL}{pk}/", {"status": "done"}, format="json") return pk def test_manager_rates_a_finished_task_with_five_levels(self): pk = self.finished() mgr = self.api(self.manager_a, self.org_a) for level in ("excellent", "good", "average", "poor", "terrible"): r = mgr.post(f"{URL}{pk}/rate/", {"quality": level}, format="json") self.assertEqual((r.status_code, r.data["quality"]), (200, level)) self.assertEqual(mgr.post(f"{URL}{pk}/rate/", {"quality": "amazing"}, format="json").status_code, 400) self.assertEqual(mgr.post(f"{URL}{pk}/rate/", {}, format="json").status_code, 400) def test_only_finished_tasks_can_be_rated_and_only_by_approvers(self): open_pk = self.create(self.manager_a, assignee=self.employee_a.pk).data["id"] mgr, emp = self.api(self.manager_a, self.org_a), self.api(self.employee_a, self.org_a) self.assertEqual(mgr.post(f"{URL}{open_pk}/rate/", {"quality": "good"}, format="json").status_code, 400) done_pk = self.finished() self.assertEqual(emp.post(f"{URL}{done_pk}/rate/", {"quality": "excellent"}, format="json").status_code, 403) mgr.post(f"{URL}{done_pk}/rate/", {"quality": "good"}, format="json") self.assertEqual(emp.get(f"{URL}{done_pk}/").data["quality"], "good") # the assignee can see their rating def test_reopening_clears_the_rating(self): pk = self.finished() mgr = self.api(self.manager_a, self.org_a) mgr.post(f"{URL}{pk}/rate/", {"quality": "excellent"}, format="json") r = mgr.patch(f"{URL}{pk}/", {"status": "in_progress"}, format="json") self.assertEqual(r.data["quality"], "") def test_rating_is_tenant_bound_and_audited(self): from apps.audit.models import AuditLog pk = self.finished() self.api(self.manager_a, self.org_a).post(f"{URL}{pk}/rate/", {"quality": "poor"}, format="json") self.assertTrue(AuditLog.objects.filter(action="task.rate", organization=self.org_a).exists()) modules.enable_module(actor=self.root, organization=self.org_b, module_code="tasks") self.assertEqual(self.api(self.manager_b, self.org_b).post(f"{URL}{pk}/rate/", {"quality": "good"}, format="json").status_code, 404) def test_staff_history_contains_only_their_own_tasks(self): mine = self.finished() # assigned to employee_a theirs = self.create(self.manager_a, title="not yours").data["id"] # assigned to the manager self.api(self.manager_a, self.org_a).patch(f"{URL}{theirs}/", {"status": "done"}, format="json") emp_rows = self.api(self.employee_a, self.org_a).get(URL, {"status": "done"}).data["results"] self.assertEqual([r["id"] for r in emp_rows], [mine]) all_rows = self.api(self.manager_a, self.org_a).get(URL, {"status": "done"}).data["results"] self.assertEqual({r["id"] for r in all_rows}, {mine, theirs}) def test_creator_also_sees_the_task_in_their_history(self): # an employee who handed work to a colleague is the "provider": it stays in their history from apps.access.models import Permission, RolePermission role = self.role(self.org_a, "employee") RolePermission.objects.update_or_create(role=role, permission=Permission.objects.get(code="tasks.assign"), defaults={"scope": "organization"}) helper = self.make_user("helper") self.join(helper, self.org_a, "employee") pk = self.create(self.employee_a, assignee=helper.pk).data["id"] self.api(self.manager_a, self.org_a).patch(f"{URL}{pk}/", {"status": "done"}, format="json") self.assertEqual([r["id"] for r in self.api(self.employee_a, self.org_a).get(URL, {"status": "done"}).data["results"]], [pk]) self.assertEqual([r["id"] for r in self.api(helper, self.org_a).get(URL, {"status": "done"}).data["results"]], [pk]) outsider = self.make_user("outsider") self.join(outsider, self.org_a, "employee") self.assertEqual(self.api(outsider, self.org_a).get(URL, {"status": "done"}).data["results"], []) def test_history_search_and_approval_order(self): a = self.finished(title="alpha report") b = self.finished(title="beta audit") mgr = self.api(self.manager_a, self.org_a) rows = mgr.get(URL, {"status": "done"}).data["results"] self.assertEqual([r["id"] for r in rows], [b, a]) # approved last -> first self.assertIsNotNone(rows[0]["reviewed_at"]) self.assertEqual([r["id"] for r in mgr.get(URL, {"status": "done", "search": "alpha"}).data["results"]], [a]) self.assertEqual([r["id"] for r in mgr.get(URL, {"status": "done", "search": "employee-a"}).data["count"] and mgr.get(URL, {"status": "done", "search": "employee-a"}).data["results"]], [b, a])