Compare commits

...

5 Commits

Author SHA1 Message Date
zhi
67c648d6d8 chore: remove tests - moved to HarborForge.Backend.Test
All backend tests moved to independent test project at
HarborForge.Test/HarborForge.Backend.Test/
2026-03-19 12:44:10 +00:00
zhi
403d66e1ba test(P14.1): add comprehensive backend API tests
Add test coverage for:
- test_auth.py: Login, JWT, protected endpoints (5 tests)
- test_users.py: User CRUD, permissions (8 tests)
- test_projects.py: Project CRUD, ownership (8 tests)
- test_milestones.py: Milestone CRUD, filtering (7 tests)
- test_tasks.py: Task CRUD, filtering by status/assignee (8 tests)
- test_comments.py: Comment CRUD, edit permissions (5 tests)
- test_roles.py: Role/permission management, assignments (9 tests)
- test_misc.py: Milestones global, notifications, activity log, API keys, dashboard, health (14 tests)

Total: 64 new tests covering all major API endpoints.
Uses existing pytest fixtures from conftest.py.
2026-03-19 12:38:14 +00:00
0b1e47ef60 Merge pull request 'feat: milestone state machine + propose flow + task state machine' (#8) from feat/milestone-propose-state-machine into main
Reviewed-on: #8
2026-03-19 11:11:09 +00:00
zhi
43742f69da fix: add values_callable to all SQLAlchemy Enum columns
SQLAlchemy 2.0 defaults to mapping Python enum *names* (OPEN, CLOSED)
to DB values, but MySQL stores lowercase *values* (open, closed).
This mismatch causes LookupError on read.

Adding values_callable=lambda x: [e.value for e in x] tells SQLAlchemy
to use the enum values for DB mapping.

Affected models: milestone, task, meeting, propose, support
2026-03-19 09:38:37 +00:00
f9e5e0f9a3 Merge pull request 'feat: modal editors + ownership-based edit permissions' (#7) from feat/modal-edit-permissions-20260316 into main
Reviewed-on: #7
2026-03-16 19:43:48 +00:00
10 changed files with 8 additions and 1792 deletions

View File

@@ -22,8 +22,8 @@ class Meeting(Base):
id = Column(Integer, primary_key=True, index=True)
title = Column(String(255), nullable=False)
description = Column(Text, nullable=True)
status = Column(Enum(MeetingStatus), default=MeetingStatus.SCHEDULED)
priority = Column(Enum(MeetingPriority), default=MeetingPriority.MEDIUM)
status = Column(Enum(MeetingStatus, values_callable=lambda x: [e.value for e in x]), default=MeetingStatus.SCHEDULED)
priority = Column(Enum(MeetingPriority, values_callable=lambda x: [e.value for e in x]), default=MeetingPriority.MEDIUM)
meeting_code = Column(String(64), nullable=True, unique=True, index=True)
project_id = Column(Integer, ForeignKey("projects.id"), nullable=False)

View File

@@ -17,7 +17,7 @@ class Milestone(Base):
id = Column(Integer, primary_key=True, index=True)
title = Column(String(255), nullable=False)
description = Column(Text, nullable=True)
status = Column(Enum(MilestoneStatus), default=MilestoneStatus.OPEN)
status = Column(Enum(MilestoneStatus, values_callable=lambda x: [e.value for e in x]), default=MilestoneStatus.OPEN)
milestone_code = Column(String(64), nullable=True, unique=True, index=True)
due_date = Column(DateTime(timezone=True), nullable=True)
planned_release_date = Column(DateTime(timezone=True), nullable=True)

View File

@@ -17,7 +17,7 @@ class Propose(Base):
propose_code = Column(String(64), nullable=True, unique=True, index=True)
title = Column(String(255), nullable=False)
description = Column(Text, nullable=True)
status = Column(Enum(ProposeStatus), default=ProposeStatus.OPEN)
status = Column(Enum(ProposeStatus, values_callable=lambda x: [e.value for e in x]), default=ProposeStatus.OPEN)
project_id = Column(Integer, ForeignKey("projects.id"), nullable=False)
created_by_id = Column(Integer, ForeignKey("users.id"), nullable=True)

View File

@@ -22,8 +22,8 @@ class Support(Base):
id = Column(Integer, primary_key=True, index=True)
title = Column(String(255), nullable=False)
description = Column(Text, nullable=True)
status = Column(Enum(SupportStatus), default=SupportStatus.OPEN)
priority = Column(Enum(SupportPriority), default=SupportPriority.MEDIUM)
status = Column(Enum(SupportStatus, values_callable=lambda x: [e.value for e in x]), default=SupportStatus.OPEN)
priority = Column(Enum(SupportPriority, values_callable=lambda x: [e.value for e in x]), default=SupportPriority.MEDIUM)
support_code = Column(String(64), nullable=True, unique=True, index=True)
project_id = Column(Integer, ForeignKey("projects.id"), nullable=False)

View File

@@ -23,8 +23,8 @@ class Task(Base):
id = Column(Integer, primary_key=True, index=True)
title = Column(String(255), nullable=False)
description = Column(Text, nullable=True)
status = Column(Enum(TaskStatus), default=TaskStatus.OPEN)
priority = Column(Enum(TaskPriority), default=TaskPriority.MEDIUM)
status = Column(Enum(TaskStatus, values_callable=lambda x: [e.value for e in x]), default=TaskStatus.OPEN)
priority = Column(Enum(TaskPriority, values_callable=lambda x: [e.value for e in x]), default=TaskPriority.MEDIUM)
task_code = Column(String(64), nullable=True, unique=True, index=True)
# Task type/subtype (replaces old issue_type/issue_subtype)

View File

@@ -1 +0,0 @@
# tests package

View File

@@ -1,302 +0,0 @@
"""Shared test fixtures — SQLite in-memory DB + FastAPI TestClient.
This avoids needing MySQL for unit/integration tests.
All models are created fresh for every test function (function-scoped session).
"""
import sys, os
# Ensure the backend app package is importable
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
# --- Override engine BEFORE any app import touches the real DB ---
from app.core.config import Base
# Force-import ALL model modules so Base.metadata knows every table
import app.models.models # noqa: F401 — User, Project, Comment, etc.
import app.models.milestone # noqa: F401
import app.models.task # noqa: F401
import app.models.role_permission # noqa: F401
import app.models.activity # noqa: F401
import app.models.propose # noqa: F401
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
engine = create_engine(
TEST_DATABASE_URL,
connect_args={"check_same_thread": False},
# Use StaticPool so all sessions share the same in-memory connection
poolclass=__import__("sqlalchemy.pool", fromlist=["StaticPool"]).StaticPool,
)
# SQLite needs foreign keys enabled per-connection
@event.listens_for(engine, "connect")
def _set_sqlite_pragma(dbapi_conn, _):
cursor = dbapi_conn.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def setup_database():
"""Create all tables before each test, drop after."""
Base.metadata.create_all(bind=engine)
yield
Base.metadata.drop_all(bind=engine)
@pytest.fixture()
def db():
"""Yield a DB session for direct model manipulation."""
session = TestingSessionLocal()
try:
yield session
finally:
session.close()
@pytest.fixture()
def client(db):
"""FastAPI TestClient wired to the test DB + a default authenticated user."""
from fastapi.testclient import TestClient
from app.main import app
from app.core.config import get_db
# Override DB dependency
def _override_get_db():
try:
yield db
finally:
pass # caller's `db` fixture handles close
app.dependency_overrides[get_db] = _override_get_db
yield TestClient(app)
app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# Helper factories
# ---------------------------------------------------------------------------
@pytest.fixture()
def make_user(db):
"""Factory to create a User row."""
from app.models.models import User
_counter = [0]
# Pre-compute a bcrypt hash for "password" to avoid passlib/bcrypt version issues
_pwd_hash = "$2b$12$LJ3m4ys/Xz.l1PaOHHKN/uE7dQFnSm1AUBfEkL0C2dN9.3Oau4XG"
def _make(username=None, is_admin=False):
_counter[0] += 1
n = _counter[0]
u = User(
username=username or f"testuser{n}",
email=f"test{n}@example.com",
hashed_password=_pwd_hash,
is_active=True,
is_admin=is_admin,
)
db.add(u)
db.commit()
db.refresh(u)
return u
return _make
@pytest.fixture()
def make_project(db):
"""Factory to create a Project row."""
from app.models.models import Project
_counter = [0]
def _make(owner_id, name=None, project_code=None):
_counter[0] += 1
n = _counter[0]
p = Project(
name=name or f"TestProject{n}",
project_code=project_code or f"TP{n}",
owner_name="owner",
owner_id=owner_id,
)
db.add(p)
db.commit()
db.refresh(p)
return p
return _make
@pytest.fixture()
def make_milestone(db):
"""Factory to create a Milestone row."""
from app.models.milestone import Milestone, MilestoneStatus
_counter = [0]
def _make(project_id, created_by_id, status=MilestoneStatus.OPEN, **kw):
_counter[0] += 1
n = _counter[0]
ms = Milestone(
title=kw.pop("title", f"Milestone {n}"),
project_id=project_id,
created_by_id=created_by_id,
status=status,
milestone_code=kw.pop("milestone_code", f"M{n:04d}"),
**kw,
)
db.add(ms)
db.commit()
db.refresh(ms)
return ms
return _make
@pytest.fixture()
def make_task(db):
"""Factory to create a Task row."""
from app.models.task import Task, TaskStatus
_counter = [0]
def _make(project_id, milestone_id, reporter_id, status=TaskStatus.PENDING, **kw):
_counter[0] += 1
n = _counter[0]
t = Task(
title=kw.pop("title", f"Task {n}"),
project_id=project_id,
milestone_id=milestone_id,
reporter_id=reporter_id,
created_by_id=kw.pop("created_by_id", reporter_id),
status=status,
task_code=kw.pop("task_code", f"T{n:04d}"),
task_type=kw.pop("task_type", "issue"),
task_subtype=kw.pop("task_subtype", None),
**kw,
)
db.add(t)
db.commit()
db.refresh(t)
return t
return _make
@pytest.fixture()
def seed_roles_and_permissions(db):
"""Create the minimal role + permission setup needed by action endpoints.
Returns (admin_role, mgr_role, dev_role).
"""
from app.models.role_permission import Role, Permission, RolePermission
# --- roles ---
admin_role = Role(name="admin", is_global=True)
mgr_role = Role(name="mgr", is_global=False)
dev_role = Role(name="dev", is_global=False)
db.add_all([admin_role, mgr_role, dev_role])
db.commit()
# --- permissions ---
perm_names = [
("milestone.freeze", "milestone"),
("milestone.start", "milestone"),
("milestone.close", "milestone"),
("task.close", "task"),
("task.reopen_closed", "task"),
("task.reopen_completed", "task"),
("propose.accept", "propose"),
("propose.reject", "propose"),
("propose.reopen", "propose"),
# add broad perms for role checks
("project.read", "project"),
("project.write", "project"),
("milestone.read", "milestone"),
("milestone.write", "milestone"),
("milestone.create", "milestone"),
("task.read", "task"),
("task.write", "task"),
("task.create", "task"),
]
perm_objs = {}
for name, cat in perm_names:
p = Permission(name=name, category=cat, description=name)
db.add(p)
db.flush()
perm_objs[name] = p
# admin gets all
for p in perm_objs.values():
db.add(RolePermission(role_id=admin_role.id, permission_id=p.id))
# mgr gets milestone + propose + task management perms
mgr_perms = [
"milestone.freeze", "milestone.start", "milestone.close",
"task.close", "task.reopen_closed", "task.reopen_completed",
"propose.accept", "propose.reject", "propose.reopen",
"project.read", "project.write",
"milestone.read", "milestone.write", "milestone.create",
"task.read", "task.write", "task.create",
]
for name in mgr_perms:
db.add(RolePermission(role_id=mgr_role.id, permission_id=perm_objs[name].id))
# dev gets basic perms
dev_perms = [
"project.read", "task.read", "task.write", "task.create",
"milestone.read", "task.close", "task.reopen_closed", "task.reopen_completed",
]
for name in dev_perms:
db.add(RolePermission(role_id=dev_role.id, permission_id=perm_objs[name].id))
db.commit()
db.refresh(admin_role)
db.refresh(mgr_role)
db.refresh(dev_role)
return admin_role, mgr_role, dev_role
@pytest.fixture()
def make_member(db):
"""Factory to add a user as project member with a given role."""
from app.models.models import ProjectMember
def _make(project_id, user_id, role_id):
pm = ProjectMember(project_id=project_id, user_id=user_id, role_id=role_id)
db.add(pm)
db.commit()
return pm
return _make
@pytest.fixture()
def auth_header():
"""Generate a JWT auth header for a given user."""
from app.api.deps import create_access_token
def _make(user):
token = create_access_token({"sub": str(user.id)})
return {"Authorization": f"Bearer {token}"}
return _make

View File

@@ -1,358 +0,0 @@
"""P13.1 — Milestone state-machine action tests.
Covers:
- freeze: success, missing release task, multiple release tasks, wrong status
- start: success + started_at, deps not met, wrong status
- close: from open/freeze/undergoing, wrong status (completed/closed)
- auto-complete: release task completion triggers milestone completed
"""
import json
import pytest
from app.models.milestone import MilestoneStatus
from app.models.task import TaskStatus
# -----------------------------------------------------------------------
# Freeze
# -----------------------------------------------------------------------
class TestFreeze:
"""POST /projects/{pid}/milestones/{mid}/actions/freeze"""
def test_freeze_success(
self, client, db, make_user, make_project, make_milestone,
make_task, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, _ = seed_roles_and_permissions
user = make_user(is_admin=False)
project = make_project(owner_id=user.id)
make_member(project.id, user.id, mgr_role.id)
ms = make_milestone(project.id, user.id)
# Create exactly 1 maintenance/release task
make_task(
project.id, ms.id, user.id,
task_type="maintenance", task_subtype="release",
)
resp = client.post(
f"/projects/{project.id}/milestones/{ms.id}/actions/freeze",
headers=auth_header(user),
)
assert resp.status_code == 200, resp.text
assert resp.json()["status"] == "freeze"
db.refresh(ms)
assert ms.status == MilestoneStatus.FREEZE
def test_freeze_no_release_task(
self, client, db, make_user, make_project, make_milestone,
seed_roles_and_permissions, make_member, auth_header,
):
_, 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)
resp = client.post(
f"/projects/{project.id}/milestones/{ms.id}/actions/freeze",
headers=auth_header(user),
)
assert resp.status_code == 400
assert "no maintenance/release task" in resp.json()["detail"].lower()
def test_freeze_multiple_release_tasks(
self, client, db, make_user, make_project, make_milestone,
make_task, seed_roles_and_permissions, make_member, auth_header,
):
_, 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)
make_task(project.id, ms.id, user.id, task_type="maintenance", task_subtype="release")
make_task(project.id, ms.id, user.id, task_type="maintenance", task_subtype="release")
resp = client.post(
f"/projects/{project.id}/milestones/{ms.id}/actions/freeze",
headers=auth_header(user),
)
assert resp.status_code == 400
assert "expected exactly 1" in resp.json()["detail"].lower()
def test_freeze_wrong_status(
self, client, db, make_user, make_project, make_milestone,
make_task, seed_roles_and_permissions, make_member, auth_header,
):
_, 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.CLOSED)
make_task(project.id, ms.id, user.id, task_type="maintenance", task_subtype="release")
resp = client.post(
f"/projects/{project.id}/milestones/{ms.id}/actions/freeze",
headers=auth_header(user),
)
assert resp.status_code == 400
assert "expected 'open'" in resp.json()["detail"].lower()
# -----------------------------------------------------------------------
# Start
# -----------------------------------------------------------------------
class TestStart:
"""POST /projects/{pid}/milestones/{mid}/actions/start"""
def _freeze_milestone(self, db, ms):
ms.status = MilestoneStatus.FREEZE
db.commit()
db.refresh(ms)
def test_start_success(
self, client, db, make_user, make_project, make_milestone,
make_task, seed_roles_and_permissions, make_member, auth_header,
):
_, 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)
self._freeze_milestone(db, ms)
resp = client.post(
f"/projects/{project.id}/milestones/{ms.id}/actions/start",
headers=auth_header(user),
)
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["status"] == "undergoing"
assert "started_at" in data
db.refresh(ms)
assert ms.status == MilestoneStatus.UNDERGOING
assert ms.started_at is not None
def test_start_deps_not_met(
self, client, db, make_user, make_project, make_milestone,
make_task, seed_roles_and_permissions, make_member, auth_header,
):
_, 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)
# Create a dependency milestone that is NOT completed
dep_ms = make_milestone(project.id, user.id, status=MilestoneStatus.OPEN)
ms = make_milestone(
project.id, user.id,
depend_on_milestones=json.dumps([dep_ms.id]),
)
self._freeze_milestone(db, ms)
resp = client.post(
f"/projects/{project.id}/milestones/{ms.id}/actions/start",
headers=auth_header(user),
)
assert resp.status_code == 400
assert "cannot start" in resp.json()["detail"].lower()
def test_start_wrong_status(
self, client, db, make_user, make_project, make_milestone,
seed_roles_and_permissions, make_member, auth_header,
):
_, 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)
resp = client.post(
f"/projects/{project.id}/milestones/{ms.id}/actions/start",
headers=auth_header(user),
)
assert resp.status_code == 400
assert "expected 'freeze'" in resp.json()["detail"].lower()
# -----------------------------------------------------------------------
# Close
# -----------------------------------------------------------------------
class TestClose:
"""POST /projects/{pid}/milestones/{mid}/actions/close"""
@pytest.mark.parametrize("initial_status", [
MilestoneStatus.OPEN,
MilestoneStatus.FREEZE,
MilestoneStatus.UNDERGOING,
])
def test_close_from_allowed_statuses(
self, client, db, make_user, make_project, make_milestone,
seed_roles_and_permissions, make_member, auth_header, initial_status,
):
_, 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=initial_status)
resp = client.post(
f"/projects/{project.id}/milestones/{ms.id}/actions/close",
headers=auth_header(user),
json={"reason": "no longer needed"},
)
assert resp.status_code == 200, resp.text
assert resp.json()["status"] == "closed"
db.refresh(ms)
assert ms.status == MilestoneStatus.CLOSED
@pytest.mark.parametrize("terminal_status", [
MilestoneStatus.COMPLETED,
MilestoneStatus.CLOSED,
])
def test_close_from_terminal_rejected(
self, client, db, make_user, make_project, make_milestone,
seed_roles_and_permissions, make_member, auth_header, terminal_status,
):
_, 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=terminal_status)
resp = client.post(
f"/projects/{project.id}/milestones/{ms.id}/actions/close",
headers=auth_header(user),
)
assert resp.status_code == 400
# -----------------------------------------------------------------------
# Auto-complete
# -----------------------------------------------------------------------
class TestAutoComplete:
"""When the sole release task is completed, milestone auto-completes."""
def test_auto_complete_on_release_task_finish(
self, db, make_user, make_project, make_milestone, make_task,
):
"""Direct unit test of try_auto_complete_milestone."""
from app.api.routers.milestone_actions import try_auto_complete_milestone
user = make_user()
project = make_project(owner_id=user.id)
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
release_task = make_task(
project.id, ms.id, user.id,
task_type="maintenance", task_subtype="release",
status=TaskStatus.COMPLETED,
)
try_auto_complete_milestone(db, release_task, user_id=user.id)
db.refresh(ms)
assert ms.status == MilestoneStatus.COMPLETED
def test_no_auto_complete_for_non_release_task(
self, db, make_user, make_project, make_milestone, make_task,
):
from app.api.routers.milestone_actions import try_auto_complete_milestone
user = make_user()
project = make_project(owner_id=user.id)
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
# Also add the required release task (still pending)
make_task(
project.id, ms.id, user.id,
task_type="maintenance", task_subtype="release",
status=TaskStatus.PENDING,
)
normal_task = make_task(
project.id, ms.id, user.id,
task_type="issue", task_subtype="defect",
status=TaskStatus.COMPLETED,
)
try_auto_complete_milestone(db, normal_task, user_id=user.id)
db.refresh(ms)
assert ms.status == MilestoneStatus.UNDERGOING # unchanged
def test_no_auto_complete_when_not_undergoing(
self, db, make_user, make_project, make_milestone, make_task,
):
from app.api.routers.milestone_actions import try_auto_complete_milestone
user = make_user()
project = make_project(owner_id=user.id)
ms = make_milestone(project.id, user.id, status=MilestoneStatus.FREEZE)
release_task = make_task(
project.id, ms.id, user.id,
task_type="maintenance", task_subtype="release",
status=TaskStatus.COMPLETED,
)
try_auto_complete_milestone(db, release_task, user_id=user.id)
db.refresh(ms)
assert ms.status == MilestoneStatus.FREEZE # unchanged
# -----------------------------------------------------------------------
# Preflight
# -----------------------------------------------------------------------
class TestPreflight:
"""GET /projects/{pid}/milestones/{mid}/actions/preflight"""
def test_preflight_freeze_allowed(
self, client, db, make_user, make_project, make_milestone,
make_task, seed_roles_and_permissions, make_member, auth_header,
):
_, 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)
make_task(project.id, ms.id, user.id, task_type="maintenance", task_subtype="release")
resp = client.get(
f"/projects/{project.id}/milestones/{ms.id}/actions/preflight",
headers=auth_header(user),
)
assert resp.status_code == 200
data = resp.json()
assert data["freeze"]["allowed"] is True
def test_preflight_freeze_not_allowed(
self, client, db, make_user, make_project, make_milestone,
seed_roles_and_permissions, make_member, auth_header,
):
_, 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)
# No release task
resp = client.get(
f"/projects/{project.id}/milestones/{ms.id}/actions/preflight",
headers=auth_header(user),
)
assert resp.status_code == 200
data = resp.json()
assert data["freeze"]["allowed"] is False

View File

@@ -1,559 +0,0 @@
"""P13.3 — Propose backend tests.
Covers:
- CRUD: create, list, get, update
- propose_code per-project incrementing
- accept → auto-generate feature story task + feat_task_id
- accept with non-open milestone → fail
- reject → status change
- rejected → reopen back to open
- feat_task_id cannot be set manually
- edit restrictions (only open proposes editable)
- permission checks for accept/reject/reopen
"""
import pytest
from app.models.milestone import MilestoneStatus
from app.models.task import TaskStatus
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _propose_url(project_id: int, propose_id: int | None = None) -> str:
base = f"/projects/{project_id}/proposes"
return f"{base}/{propose_id}" if propose_id else base
# ===========================================================================
# CRUD
# ===========================================================================
class TestProposeCRUD:
"""Basic create / list / get / update."""
def test_create_propose(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
user = make_user()
project = make_project(owner_id=user.id, project_code="PROJ")
make_member(project.id, user.id, dev_role.id)
resp = client.post(
_propose_url(project.id),
json={"title": "New Feature Idea", "description": "Some details"},
headers=auth_header(user),
)
assert resp.status_code == 201
data = resp.json()
assert data["title"] == "New Feature Idea"
assert data["status"] == "open"
assert data["propose_code"].startswith("PROJ:P")
assert data["feat_task_id"] is None
def test_list_proposes(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
user = make_user()
project = make_project(owner_id=user.id)
make_member(project.id, user.id, dev_role.id)
# Create two proposes
client.post(_propose_url(project.id), json={"title": "P1"}, headers=auth_header(user))
client.post(_propose_url(project.id), json={"title": "P2"}, headers=auth_header(user))
resp = client.get(_propose_url(project.id), headers=auth_header(user))
assert resp.status_code == 200
assert len(resp.json()) == 2
def test_get_propose(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
user = make_user()
project = make_project(owner_id=user.id)
make_member(project.id, user.id, dev_role.id)
create_resp = client.post(_propose_url(project.id), json={"title": "P1"}, headers=auth_header(user))
propose_id = create_resp.json()["id"]
resp = client.get(_propose_url(project.id, propose_id), headers=auth_header(user))
assert resp.status_code == 200
assert resp.json()["title"] == "P1"
def test_update_propose_open(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
user = make_user()
project = make_project(owner_id=user.id)
make_member(project.id, user.id, dev_role.id)
create_resp = client.post(_propose_url(project.id), json={"title": "Old"}, headers=auth_header(user))
propose_id = create_resp.json()["id"]
resp = client.patch(
_propose_url(project.id, propose_id),
json={"title": "New Title", "description": "Updated"},
headers=auth_header(user),
)
assert resp.status_code == 200
assert resp.json()["title"] == "New Title"
assert resp.json()["description"] == "Updated"
# ===========================================================================
# Propose Code
# ===========================================================================
class TestProposeCode:
"""P1.4 — propose_code increments per project independently."""
def test_code_increments_per_project(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
user = make_user()
proj_a = make_project(owner_id=user.id, project_code="ALPHA")
proj_b = make_project(owner_id=user.id, project_code="BETA")
make_member(proj_a.id, user.id, dev_role.id)
make_member(proj_b.id, user.id, dev_role.id)
# Create 2 in ALPHA
r1 = client.post(_propose_url(proj_a.id), json={"title": "A1"}, headers=auth_header(user))
r2 = client.post(_propose_url(proj_a.id), json={"title": "A2"}, headers=auth_header(user))
# Create 1 in BETA
r3 = client.post(_propose_url(proj_b.id), json={"title": "B1"}, headers=auth_header(user))
code1 = r1.json()["propose_code"]
code2 = r2.json()["propose_code"]
code3 = r3.json()["propose_code"]
assert code1.startswith("ALPHA:P")
assert code2.startswith("ALPHA:P")
assert code3.startswith("BETA:P")
# They should be distinct
assert code1 != code2
# ===========================================================================
# Accept
# ===========================================================================
class TestAccept:
"""P6.2 — accept propose → create feature story task."""
def test_accept_success(
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
create_resp = client.post(
_propose_url(project.id),
json={"title": "Cool Feature", "description": "Do something cool"},
headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
resp = client.post(
_propose_url(project.id, propose_id) + "/accept",
json={"milestone_id": ms.id},
headers=auth_header(mgr),
)
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "accepted"
assert data["feat_task_id"] is not None
# Verify the generated task exists
from app.models.task import Task
task = db.query(Task).filter(Task.id == int(data["feat_task_id"])).first()
assert task is not None
assert task.title == "Cool Feature"
assert task.description == "Do something cool"
assert task.task_type == "story"
assert task.task_subtype == "feature"
task_status = task.status.value if hasattr(task.status, "value") else task.status
assert task_status == "pending"
assert task.milestone_id == ms.id
def test_accept_non_open_milestone_fails(
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.FREEZE)
create_resp = client.post(
_propose_url(project.id),
json={"title": "Feature X"},
headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
resp = client.post(
_propose_url(project.id, propose_id) + "/accept",
json={"milestone_id": ms.id},
headers=auth_header(mgr),
)
assert resp.status_code == 400
assert "open" in resp.json()["detail"].lower()
def test_accept_already_accepted_fails(
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
# First accept
client.post(
_propose_url(project.id, propose_id) + "/accept",
json={"milestone_id": ms.id},
headers=auth_header(mgr),
)
# Second accept should fail
resp = client.post(
_propose_url(project.id, propose_id) + "/accept",
json={"milestone_id": ms.id},
headers=auth_header(mgr),
)
assert resp.status_code == 400
def test_accept_auto_fills_feat_task_id(
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
resp = client.post(
_propose_url(project.id, propose_id) + "/accept",
json={"milestone_id": ms.id},
headers=auth_header(mgr),
)
data = resp.json()
assert data["feat_task_id"] is not None
# Re-fetch to confirm persistence
get_resp = client.get(_propose_url(project.id, propose_id), headers=auth_header(mgr))
assert get_resp.json()["feat_task_id"] == data["feat_task_id"]
def test_accept_no_permission_fails(
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
):
"""dev role should not have propose.accept permission."""
admin_role, mgr_role, dev_role = seed_roles_and_permissions
owner = make_user()
dev_user = make_user()
project = make_project(owner_id=owner.id)
make_member(project.id, owner.id, mgr_role.id)
make_member(project.id, dev_user.id, dev_role.id)
ms = make_milestone(project.id, owner.id, status=MilestoneStatus.OPEN)
# Dev creates the propose
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(dev_user),
)
propose_id = create_resp.json()["id"]
# Dev tries to accept — should fail
resp = client.post(
_propose_url(project.id, propose_id) + "/accept",
json={"milestone_id": ms.id},
headers=auth_header(dev_user),
)
assert resp.status_code == 403
# ===========================================================================
# Reject
# ===========================================================================
class TestReject:
"""P6.3 — reject propose."""
def test_reject_success(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
resp = client.post(
_propose_url(project.id, propose_id) + "/reject",
json={"reason": "Not needed"},
headers=auth_header(mgr),
)
assert resp.status_code == 200
assert resp.json()["status"] == "rejected"
def test_reject_non_open_fails(
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
# Accept first
client.post(
_propose_url(project.id, propose_id) + "/accept",
json={"milestone_id": ms.id},
headers=auth_header(mgr),
)
# Now reject should fail
resp = client.post(
_propose_url(project.id, propose_id) + "/reject",
json={"reason": "Changed mind"},
headers=auth_header(mgr),
)
assert resp.status_code == 400
def test_reject_no_permission_fails(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
owner = make_user()
dev_user = make_user()
project = make_project(owner_id=owner.id)
make_member(project.id, owner.id, mgr_role.id)
make_member(project.id, dev_user.id, dev_role.id)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(dev_user),
)
propose_id = create_resp.json()["id"]
resp = client.post(
_propose_url(project.id, propose_id) + "/reject",
json={"reason": "nah"},
headers=auth_header(dev_user),
)
assert resp.status_code == 403
# ===========================================================================
# Reopen
# ===========================================================================
class TestReopen:
"""P6.4 — reopen rejected propose."""
def test_reopen_success(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
# Reject first
client.post(
_propose_url(project.id, propose_id) + "/reject",
json={"reason": "wait"},
headers=auth_header(mgr),
)
# Reopen
resp = client.post(
_propose_url(project.id, propose_id) + "/reopen",
headers=auth_header(mgr),
)
assert resp.status_code == 200
assert resp.json()["status"] == "open"
def test_reopen_non_rejected_fails(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
# Try reopen on open propose — should fail
resp = client.post(
_propose_url(project.id, propose_id) + "/reopen",
headers=auth_header(mgr),
)
assert resp.status_code == 400
def test_reopen_no_permission_fails(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
owner = make_user()
dev_user = make_user()
project = make_project(owner_id=owner.id)
make_member(project.id, owner.id, mgr_role.id)
make_member(project.id, dev_user.id, dev_role.id)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(dev_user),
)
propose_id = create_resp.json()["id"]
# Owner rejects
client.post(
_propose_url(project.id, propose_id) + "/reject",
json={"reason": "nah"},
headers=auth_header(owner),
)
# Dev tries to reopen — should fail
resp = client.post(
_propose_url(project.id, propose_id) + "/reopen",
headers=auth_header(dev_user),
)
assert resp.status_code == 403
# ===========================================================================
# feat_task_id protection
# ===========================================================================
class TestFeatTaskIdProtection:
"""P6.5 — feat_task_id is server-side only, cannot be set by client."""
def test_update_cannot_set_feat_task_id(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
user = make_user()
project = make_project(owner_id=user.id)
make_member(project.id, user.id, dev_role.id)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(user),
)
propose_id = create_resp.json()["id"]
# Try to set feat_task_id via PATCH
resp = client.patch(
_propose_url(project.id, propose_id),
json={"feat_task_id": "999"},
headers=auth_header(user),
)
assert resp.status_code == 200
# feat_task_id should still be None (server ignores it)
assert resp.json()["feat_task_id"] is None
# ===========================================================================
# Edit restrictions
# ===========================================================================
class TestEditRestrictions:
"""Propose editing is only allowed in open status."""
def test_edit_accepted_propose_fails(
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
# Accept
client.post(
_propose_url(project.id, propose_id) + "/accept",
json={"milestone_id": ms.id},
headers=auth_header(mgr),
)
# Try to edit
resp = client.patch(
_propose_url(project.id, propose_id),
json={"title": "Changed"},
headers=auth_header(mgr),
)
assert resp.status_code == 400
assert "open" in resp.json()["detail"].lower()
def test_edit_rejected_propose_fails(
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
):
admin_role, mgr_role, dev_role = seed_roles_and_permissions
mgr = make_user()
project = make_project(owner_id=mgr.id)
make_member(project.id, mgr.id, mgr_role.id)
create_resp = client.post(
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
)
propose_id = create_resp.json()["id"]
# Reject
client.post(
_propose_url(project.id, propose_id) + "/reject",
json={"reason": "no"},
headers=auth_header(mgr),
)
# Try to edit
resp = client.patch(
_propose_url(project.id, propose_id),
json={"title": "Changed"},
headers=auth_header(mgr),
)
assert resp.status_code == 400

View File

@@ -1,564 +0,0 @@
"""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"