diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..65140f2 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# tests package diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..565f280 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,298 @@ +"""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 + +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 diff --git a/tests/test_milestone_actions.py b/tests/test_milestone_actions.py new file mode 100644 index 0000000..7a3d61f --- /dev/null +++ b/tests/test_milestone_actions.py @@ -0,0 +1,358 @@ +"""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