From c21e4ee3359f97ee645e6d34630ed660387ab37d Mon Sep 17 00:00:00 2001 From: zhi Date: Wed, 18 Mar 2026 04:02:29 +0000 Subject: [PATCH] =?UTF-8?q?test(P13.2):=20task=20state-machine=20tests=20?= =?UTF-8?q?=E2=80=94=2034=20tests=20covering=20transitions,=20assignee=20g?= =?UTF-8?q?uards,=20comments,=20permissions,=20edit=20restrictions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/conftest.py | 4 + tests/test_task_transitions.py | 564 +++++++++++++++++++++++++++++++++ 2 files changed, 568 insertions(+) create mode 100644 tests/test_task_transitions.py diff --git a/tests/conftest.py b/tests/conftest.py index 565f280..ee1e3da 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,6 +26,10 @@ try: import app.models.apikey # noqa: F401 except ImportError: pass +try: + import app.models.webhook # noqa: F401 +except ImportError: + pass TEST_DATABASE_URL = "sqlite://" # in-memory diff --git a/tests/test_task_transitions.py b/tests/test_task_transitions.py new file mode 100644 index 0000000..b4cdf6d --- /dev/null +++ b/tests/test_task_transitions.py @@ -0,0 +1,564 @@ +"""P13.2 — Task state-machine transition tests. + +Covers: +- pending → open: success, milestone not undergoing, deps not met +- open → undergoing: success, no assignee, non-assignee blocked +- undergoing → completed: success with comment, no comment fails, non-assignee blocked +- close from pending/open/undergoing: permission required +- reopen from completed/closed → open: distinct permissions +- invalid transitions: rejected by state machine +- edit restrictions: P5.7 body edit guards by status/assignee +""" +import json + +import pytest +from app.models.milestone import MilestoneStatus +from app.models.task import TaskStatus + + +# ----------------------------------------------------------------------- +# Helpers +# ----------------------------------------------------------------------- + +def _transition(client, task_id, new_status, headers, comment=None): + """POST /tasks/{id}/transition?new_status=...""" + body = {} + if comment is not None: + body["comment"] = comment + return client.post( + f"/tasks/{task_id}/transition?new_status={new_status}", + json=body, + headers=headers, + ) + + +# ----------------------------------------------------------------------- +# pending → open +# ----------------------------------------------------------------------- + +class TestPendingToOpen: + + def test_success( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """pending→open succeeds when milestone is undergoing and no deps.""" + admin_role, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=TaskStatus.PENDING) + + resp = _transition(client, task.id, "open", auth_header(user)) + assert resp.status_code == 200, resp.text + assert resp.json()["status"] == "open" + + def test_milestone_not_undergoing( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """pending→open rejected when milestone is still open.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.OPEN) + task = make_task(project.id, ms.id, user.id, status=TaskStatus.PENDING) + + resp = _transition(client, task.id, "open", auth_header(user)) + assert resp.status_code == 400 + assert "undergoing" in resp.json()["detail"].lower() + + def test_deps_not_satisfied( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """pending→open rejected when depend_on tasks are not completed.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + + dep_task = make_task(project.id, ms.id, user.id, status=TaskStatus.OPEN) + task = make_task( + project.id, ms.id, user.id, + status=TaskStatus.PENDING, + depend_on=json.dumps([dep_task.id]), + ) + + resp = _transition(client, task.id, "open", auth_header(user)) + assert resp.status_code == 400 + assert "depend" in resp.json()["detail"].lower() or "block" in resp.json()["detail"].lower() + + def test_deps_satisfied( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """pending→open succeeds when all depend_on tasks are completed.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + + dep_task = make_task(project.id, ms.id, user.id, status=TaskStatus.COMPLETED) + task = make_task( + project.id, ms.id, user.id, + status=TaskStatus.PENDING, + depend_on=json.dumps([dep_task.id]), + ) + + resp = _transition(client, task.id, "open", auth_header(user)) + assert resp.status_code == 200 + assert resp.json()["status"] == "open" + + +# ----------------------------------------------------------------------- +# open → undergoing +# ----------------------------------------------------------------------- + +class TestOpenToUndergoing: + + def test_success_assignee_starts( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Assignee can start their own task.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task( + project.id, ms.id, user.id, + status=TaskStatus.OPEN, + assignee_id=user.id, + ) + + resp = _transition(client, task.id, "undergoing", auth_header(user)) + assert resp.status_code == 200 + assert resp.json()["status"] == "undergoing" + db.refresh(task) + assert task.started_on is not None + + def test_no_assignee_fails( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Cannot start a task without an assignee.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=TaskStatus.OPEN) + + resp = _transition(client, task.id, "undergoing", auth_header(user)) + assert resp.status_code == 400 + assert "assignee" in resp.json()["detail"].lower() + + def test_non_assignee_blocked( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """A different user cannot start someone else's task.""" + _, mgr_role, _ = seed_roles_and_permissions + owner = make_user() + other = make_user() + project = make_project(owner_id=owner.id) + make_member(project.id, owner.id, mgr_role.id) + make_member(project.id, other.id, mgr_role.id) + ms = make_milestone(project.id, owner.id, status=MilestoneStatus.UNDERGOING) + task = make_task( + project.id, ms.id, owner.id, + status=TaskStatus.OPEN, + assignee_id=owner.id, + ) + + resp = _transition(client, task.id, "undergoing", auth_header(other)) + assert resp.status_code == 403 + assert "assigned" in resp.json()["detail"].lower() + + +# ----------------------------------------------------------------------- +# undergoing → completed +# ----------------------------------------------------------------------- + +class TestUndergoingToCompleted: + + def test_success_with_comment( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Assignee can complete a task with a completion comment.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task( + project.id, ms.id, user.id, + status=TaskStatus.UNDERGOING, + assignee_id=user.id, + ) + + resp = _transition(client, task.id, "completed", auth_header(user), comment="Done!") + assert resp.status_code == 200 + assert resp.json()["status"] == "completed" + db.refresh(task) + assert task.finished_on is not None + + def test_no_comment_fails( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Cannot complete without a comment.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task( + project.id, ms.id, user.id, + status=TaskStatus.UNDERGOING, + assignee_id=user.id, + ) + + resp = _transition(client, task.id, "completed", auth_header(user)) + assert resp.status_code == 400 + assert "comment" in resp.json()["detail"].lower() + + def test_empty_comment_fails( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Empty/whitespace comment is rejected.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task( + project.id, ms.id, user.id, + status=TaskStatus.UNDERGOING, + assignee_id=user.id, + ) + + resp = _transition(client, task.id, "completed", auth_header(user), comment=" ") + assert resp.status_code == 400 + assert "comment" in resp.json()["detail"].lower() + + def test_non_assignee_blocked( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Non-assignee cannot complete the task.""" + _, mgr_role, _ = seed_roles_and_permissions + owner = make_user() + other = make_user() + project = make_project(owner_id=owner.id) + make_member(project.id, owner.id, mgr_role.id) + make_member(project.id, other.id, mgr_role.id) + ms = make_milestone(project.id, owner.id, status=MilestoneStatus.UNDERGOING) + task = make_task( + project.id, ms.id, owner.id, + status=TaskStatus.UNDERGOING, + assignee_id=owner.id, + ) + + resp = _transition(client, task.id, "completed", auth_header(other), comment="I finished it") + assert resp.status_code == 403 + + +# ----------------------------------------------------------------------- +# Close task (from various states) +# ----------------------------------------------------------------------- + +class TestCloseTask: + + @pytest.mark.parametrize("initial_status", [ + TaskStatus.PENDING, + TaskStatus.OPEN, + TaskStatus.UNDERGOING, + ]) + def test_close_from_valid_states( + self, initial_status, + client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Close is allowed from pending/open/undergoing with permission.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=initial_status) + + resp = _transition(client, task.id, "closed", auth_header(user)) + assert resp.status_code == 200, resp.text + assert resp.json()["status"] == "closed" + + @pytest.mark.parametrize("initial_status", [ + TaskStatus.COMPLETED, + TaskStatus.CLOSED, + ]) + def test_close_from_terminal_states_fails( + self, initial_status, + client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Cannot close from completed or already closed.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=initial_status) + + resp = _transition(client, task.id, "closed", auth_header(user)) + assert resp.status_code == 400 + + def test_close_without_permission_fails( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """User without task.close permission cannot close.""" + from app.models.role_permission import Role + _, _, dev_role = seed_roles_and_permissions + + # Create a role with NO task.close permission + no_close_role = Role(name="viewer", is_global=False) + db.add(no_close_role) + db.commit() + + # Give viewer only basic perms (project.read, task.read) + from app.models.role_permission import Permission, RolePermission + for pname in ("project.read", "task.read"): + p = db.query(Permission).filter(Permission.name == pname).first() + if p: + db.add(RolePermission(role_id=no_close_role.id, permission_id=p.id)) + db.commit() + + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, no_close_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=TaskStatus.OPEN) + + resp = _transition(client, task.id, "closed", auth_header(user)) + assert resp.status_code == 403 + + +# ----------------------------------------------------------------------- +# Reopen (completed → open, closed → open) +# ----------------------------------------------------------------------- + +class TestReopen: + + def test_reopen_completed( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Reopen from completed → open with task.reopen_completed permission.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=TaskStatus.COMPLETED) + + resp = _transition(client, task.id, "open", auth_header(user)) + assert resp.status_code == 200 + assert resp.json()["status"] == "open" + # finished_on should be cleared + db.refresh(task) + assert task.finished_on is None + + def test_reopen_closed( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Reopen from closed → open with task.reopen_closed permission.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=TaskStatus.CLOSED) + + resp = _transition(client, task.id, "open", auth_header(user)) + assert resp.status_code == 200 + assert resp.json()["status"] == "open" + + def test_reopen_without_permission_fails( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """User without reopen permission cannot reopen.""" + from app.models.role_permission import Role, Permission, RolePermission + + # Create a role with task.close but NO reopen permissions + limited_role = Role(name="limited", is_global=False) + db.add(limited_role) + db.commit() + for pname in ("project.read", "task.read", "task.write", "task.close"): + p = db.query(Permission).filter(Permission.name == pname).first() + if p: + db.add(RolePermission(role_id=limited_role.id, permission_id=p.id)) + db.commit() + + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, limited_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=TaskStatus.COMPLETED) + + resp = _transition(client, task.id, "open", auth_header(user)) + assert resp.status_code == 403 + + +# ----------------------------------------------------------------------- +# Invalid transitions +# ----------------------------------------------------------------------- + +class TestInvalidTransitions: + + @pytest.mark.parametrize("from_status,to_status", [ + (TaskStatus.PENDING, "undergoing"), + (TaskStatus.PENDING, "completed"), + (TaskStatus.OPEN, "completed"), + (TaskStatus.OPEN, "pending"), + (TaskStatus.UNDERGOING, "open"), + (TaskStatus.UNDERGOING, "pending"), + (TaskStatus.COMPLETED, "undergoing"), + (TaskStatus.COMPLETED, "closed"), + (TaskStatus.CLOSED, "undergoing"), + (TaskStatus.CLOSED, "completed"), + ]) + def test_disallowed_transition( + self, from_status, to_status, + client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """State machine rejects transitions not in VALID_TRANSITIONS.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task( + project.id, ms.id, user.id, + status=from_status, + assignee_id=user.id, + ) + + resp = _transition(client, task.id, to_status, auth_header(user)) + assert resp.status_code == 400 + assert "cannot transition" in resp.json()["detail"].lower() + + +# ----------------------------------------------------------------------- +# Edit restrictions (PATCH) +# ----------------------------------------------------------------------- + +class TestEditRestrictions: + + def test_undergoing_body_edit_blocked( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Cannot PATCH body fields on an undergoing task.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=TaskStatus.UNDERGOING, assignee_id=user.id) + + resp = client.patch( + f"/tasks/{task.id}", + json={"title": "New Title"}, + headers=auth_header(user), + ) + assert resp.status_code == 400 + assert "undergoing" in resp.json()["detail"].lower() + + def test_completed_body_edit_blocked( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Cannot PATCH body fields on a completed task.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task(project.id, ms.id, user.id, status=TaskStatus.COMPLETED) + + resp = client.patch( + f"/tasks/{task.id}", + json={"title": "Changed"}, + headers=auth_header(user), + ) + assert resp.status_code == 400 + + def test_open_assignee_only_edit( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Open task with assignee: only assignee can edit body.""" + _, mgr_role, _ = seed_roles_and_permissions + owner = make_user() + other = make_user() + project = make_project(owner_id=owner.id) + make_member(project.id, owner.id, mgr_role.id) + make_member(project.id, other.id, mgr_role.id) + ms = make_milestone(project.id, owner.id, status=MilestoneStatus.UNDERGOING) + task = make_task( + project.id, ms.id, owner.id, + status=TaskStatus.OPEN, + assignee_id=owner.id, + ) + + # Other user cannot edit + resp = client.patch( + f"/tasks/{task.id}", + json={"title": "Hijack"}, + headers=auth_header(other), + ) + assert resp.status_code == 403 + + # Assignee can edit + resp = client.patch( + f"/tasks/{task.id}", + json={"title": "My Change"}, + headers=auth_header(owner), + ) + assert resp.status_code == 200 + assert resp.json()["title"] == "My Change" + + def test_open_no_assignee_anyone_edits( + self, client, db, make_user, make_project, make_milestone, + make_task, seed_roles_and_permissions, make_member, auth_header, + ): + """Open task without assignee: any project member can edit.""" + _, mgr_role, _ = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + make_member(project.id, user.id, mgr_role.id) + ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING) + task = make_task( + project.id, ms.id, user.id, + status=TaskStatus.OPEN, + assignee_id=None, + ) + + resp = client.patch( + f"/tasks/{task.id}", + json={"title": "Anyone's Change"}, + headers=auth_header(user), + ) + assert resp.status_code == 200 + assert resp.json()["title"] == "Anyone's Change"