diff --git a/README.md b/README.md index 67fb294..c028647 100644 --- a/README.md +++ b/README.md @@ -1 +1,52 @@ # HarborForge.Backend.Test + +Independent test suite for HarborForge.Backend. + +## Setup + +```bash +# Create virtual environment +python3 -m venv venv +source venv/bin/activate + +# Install dependencies +pip install -r requirements.txt +``` + +## Run Tests + +```bash +# Run all tests +pytest + +# Run specific test file +pytest tests/test_auth.py + +# Run with verbose output +pytest -v + +# Run with coverage (requires pytest-cov) +pytest --cov=../HarborForge.Backend/app --cov-report=html +``` + +## Test Structure + +| File | Tests | Coverage | +|------|-------|----------| +| `test_auth.py` | 5 | Login, JWT, protected endpoints | +| `test_users.py` | 8 | User CRUD, permissions | +| `test_projects.py` | 8 | Project CRUD, ownership | +| `test_milestones.py` | 7 | Milestone CRUD, filtering | +| `test_tasks.py` | 8 | Task CRUD, filtering | +| `test_comments.py` | 5 | Comment CRUD, permissions | +| `test_roles.py` | 9 | Role/permission management | +| `test_milestone_actions.py` | 17 | Milestone state machine actions | +| `test_task_transitions.py` | 34 | Task state machine transitions | +| `test_propose.py` | 19 | Propose CRUD, accept/reject/reopen | +| `test_misc.py` | 14 | Notifications, activity log, API keys, dashboard | + +**Total: 134 tests** + +## How It Works + +Tests import the backend code from `../HarborForge.Backend/` via path manipulation in `conftest.py`. Uses SQLite in-memory database for fast, isolated tests. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9ec7962 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,6 @@ +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +addopts = "-v --tb=short" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f68c3ee --- /dev/null +++ b/requirements.txt @@ -0,0 +1,22 @@ +# HarborForge.Backend.Test dependencies +# Tests the HarborForge.Backend as a separate project + +# Backend dependencies (must match Backend/requirements.txt) +fastapi==0.109.0 +uvicorn[standard]==0.27.0 +sqlalchemy==2.0.25 +pymysql==1.1.0 +pydantic==2.5.3 +pydantic-settings==2.1.0 +python-jose[cryptography]==3.3.0 +passlib[bcrypt]==1.7.4 +bcrypt==4.0.1 +python-multipart==0.0.6 +alembic==1.13.1 +python-dotenv==1.0.0 +httpx==0.27.0 +requests==2.31.0 + +# Test dependencies +pytest==8.0.0 +pytest-asyncio==0.23.0 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/__pycache__/__init__.cpython-312.pyc b/tests/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..c0e310b Binary files /dev/null and b/tests/__pycache__/__init__.cpython-312.pyc differ diff --git a/tests/__pycache__/conftest.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/conftest.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..4f2086f Binary files /dev/null and b/tests/__pycache__/conftest.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_auth.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_auth.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..a8e9b5f Binary files /dev/null and b/tests/__pycache__/test_auth.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_comments.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_comments.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..62437a2 Binary files /dev/null and b/tests/__pycache__/test_comments.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_milestone_actions.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_milestone_actions.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..5938629 Binary files /dev/null and b/tests/__pycache__/test_milestone_actions.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_milestones.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_milestones.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..2b79d1c Binary files /dev/null and b/tests/__pycache__/test_milestones.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_misc.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_misc.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..38131d1 Binary files /dev/null and b/tests/__pycache__/test_misc.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_projects.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_projects.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..54daa9c Binary files /dev/null and b/tests/__pycache__/test_projects.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_propose.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_propose.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..b39cff1 Binary files /dev/null and b/tests/__pycache__/test_propose.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_roles.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_roles.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..30b5d9c Binary files /dev/null and b/tests/__pycache__/test_roles.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_task_transitions.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_task_transitions.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..a14b7e4 Binary files /dev/null and b/tests/__pycache__/test_task_transitions.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_tasks.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_tasks.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..3ecf674 Binary files /dev/null and b/tests/__pycache__/test_tasks.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_users.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_users.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..b879827 Binary files /dev/null and b/tests/__pycache__/test_users.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..5e84575 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,304 @@ +"""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 +# Backend is at ../../../HarborForge.Backend/ relative to this file +_backend_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", "HarborForge.Backend")) +sys.path.insert(0, _backend_path) + +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 diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..37777cc --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,59 @@ +"""P14.1 — Auth API tests. + +Covers: +- Login with valid credentials +- Login with invalid credentials +- Token refresh +- Protected endpoint access with/without token +""" +import pytest + + +class TestAuth: + """Authentication endpoints.""" + + def test_login_success(self, client, db, make_user): + """Valid login returns JWT token.""" + user = make_user(username="testuser", password="testpass123") + + resp = client.post( + "/auth/token", + data={"username": "testuser", "password": "testpass123"} + ) + assert resp.status_code == 200 + data = resp.json() + assert "access_token" in data + assert data["token_type"] == "bearer" + + def test_login_invalid_password(self, client, db, make_user): + """Invalid password returns 401.""" + make_user(username="testuser", password="testpass123") + + resp = client.post( + "/auth/token", + data={"username": "testuser", "password": "wrongpass"} + ) + assert resp.status_code == 401 + + def test_login_nonexistent_user(self, client, db): + """Non-existent user returns 401.""" + resp = client.post( + "/auth/token", + data={"username": "nosuchuser", "password": "anypass"} + ) + assert resp.status_code == 401 + + def test_protected_endpoint_without_token(self, client): + """Accessing protected endpoint without token returns 401.""" + resp = client.get("/users/me") + assert resp.status_code == 401 + + def test_protected_endpoint_with_token(self, client, db, make_user, auth_header): + """Accessing protected endpoint with valid token succeeds.""" + user = make_user() + + resp = client.get("/users/me", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert data["id"] == user.id + assert data["username"] == user.username diff --git a/tests/test_comments.py b/tests/test_comments.py new file mode 100644 index 0000000..e67da4d --- /dev/null +++ b/tests/test_comments.py @@ -0,0 +1,180 @@ +"""P14.1 — Comments API tests. + +Covers: +- List comments for task +- Create comment +- Update comment +- Delete comment +- Comment permissions +""" +import pytest + + +class TestComments: + """Comment management endpoints.""" + + def test_list_comments(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """List comments for a task.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + from app.models.models import Comment + + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + task = Task( + title="Test Task", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add(task) + db.commit() + + # Add comments + comment1 = Comment(content="Comment 1", task_id=task.id, author_id=user.id) + comment2 = Comment(content="Comment 2", task_id=task.id, author_id=user.id) + db.add_all([comment1, comment2]) + db.commit() + + resp = client.get(f"/projects/{project.id}/tasks/{task.id}/comments", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert len(data) == 2 + + def test_create_comment(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Create comment on task.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + task = Task( + title="Test Task", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add(task) + db.commit() + + resp = client.post( + f"/projects/{project.id}/tasks/{task.id}/comments", + json={"content": "This is a test comment"}, + headers=auth_header(user) + ) + assert resp.status_code == 201 + data = resp.json() + assert data["content"] == "This is a test comment" + assert data["author_id"] == user.id + + def test_update_comment(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Update own comment.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + from app.models.models import Comment + + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + task = Task( + title="Test Task", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add(task) + db.commit() + + comment = Comment(content="Original", task_id=task.id, author_id=user.id) + db.add(comment) + db.commit() + + resp = client.patch( + f"/projects/{project.id}/tasks/{task.id}/comments/{comment.id}", + json={"content": "Updated content"}, + headers=auth_header(user) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["content"] == "Updated content" + + def test_delete_comment(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Delete comment.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + from app.models.models import Comment + + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + task = Task( + title="Test Task", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add(task) + db.commit() + + comment = Comment(content="To delete", task_id=task.id, author_id=user.id) + db.add(comment) + db.commit() + + resp = client.delete( + f"/projects/{project.id}/tasks/{task.id}/comments/{comment.id}", + headers=auth_header(user) + ) + assert resp.status_code == 204 + + def test_cannot_edit_others_comment(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Cannot edit another user's comment.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user1 = make_user(username="user1") + user2 = make_user(username="user2") + project = make_project() + make_member(project.id, user1.id, dev_role.id) + make_member(project.id, user2.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + from app.models.models import Comment + + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + task = Task( + title="Test Task", project_id=project.id, milestone_id=milestone.id, + reporter_id=user1.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add(task) + db.commit() + + comment = Comment(content="User1's comment", task_id=task.id, author_id=user1.id) + db.add(comment) + db.commit() + + resp = client.patch( + f"/projects/{project.id}/tasks/{task.id}/comments/{comment.id}", + json={"content": "Hacked!"}, + headers=auth_header(user2) + ) + assert resp.status_code == 403 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 diff --git a/tests/test_milestones.py b/tests/test_milestones.py new file mode 100644 index 0000000..4c42f20 --- /dev/null +++ b/tests/test_milestones.py @@ -0,0 +1,148 @@ +"""P14.1 — Milestones CRUD API tests. + +Covers: +- List milestones (project-scoped) +- Get milestone by ID +- Create milestone +- Update milestone +- Delete milestone +- Milestone filtering and sorting +""" +import pytest +from datetime import datetime, timedelta + + +class TestMilestonesCRUD: + """Milestone CRUD endpoints.""" + + def test_list_milestones(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """List milestones for a project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + # Create milestones + from app.models.milestone import Milestone, MilestoneStatus + milestone1 = Milestone(title="Milestone 1", project_id=project.id, status=MilestoneStatus.OPEN) + milestone2 = Milestone(title="Milestone 2", project_id=project.id, status=MilestoneStatus.OPEN) + db.add_all([milestone1, milestone2]) + db.commit() + + resp = client.get(f"/projects/{project.id}/milestones", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert len(data) == 2 + + def test_get_milestone_by_id(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Get specific milestone.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + milestone = Milestone( + title="Test Milestone", + description="Test desc", + project_id=project.id, + status=MilestoneStatus.OPEN + ) + db.add(milestone) + db.commit() + + resp = client.get( + f"/projects/{project.id}/milestones/{milestone.id}", + headers=auth_header(user) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["id"] == milestone.id + assert data["title"] == "Test Milestone" + + def test_create_milestone(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Create new milestone.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project(project_code="PROJ") + make_member(project.id, user.id, dev_role.id) + + due_date = (datetime.now() + timedelta(days=30)).isoformat() + resp = client.post( + f"/projects/{project.id}/milestones", + json={ + "title": "New Milestone", + "description": "Milestone description", + "due_date": due_date + }, + headers=auth_header(user) + ) + assert resp.status_code == 201 + data = resp.json() + assert data["title"] == "New Milestone" + assert data["status"] == "open" + assert data["milestone_code"].startswith("PROJ:") + + def test_update_milestone(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Update milestone (allowed in open status).""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + milestone = Milestone( + title="Old Title", + project_id=project.id, + status=MilestoneStatus.OPEN + ) + db.add(milestone) + db.commit() + + resp = client.patch( + f"/projects/{project.id}/milestones/{milestone.id}", + json={"title": "Updated Title"}, + headers=auth_header(user) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["title"] == "Updated Title" + + def test_delete_milestone(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Delete milestone.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + milestone = Milestone(title="To Delete", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + resp = client.delete( + f"/projects/{project.id}/milestones/{milestone.id}", + headers=auth_header(user) + ) + assert resp.status_code == 204 + + def test_milestone_status_filter(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Filter milestones by status.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + open_ms = Milestone(title="Open", project_id=project.id, status=MilestoneStatus.OPEN) + closed_ms = Milestone(title="Closed", project_id=project.id, status=MilestoneStatus.CLOSED) + db.add_all([open_ms, closed_ms]) + db.commit() + + resp = client.get( + f"/projects/{project.id}/milestones?status=open", + headers=auth_header(user) + ) + assert resp.status_code == 200 + data = resp.json() + assert all(m["status"] == "open" for m in data) diff --git a/tests/test_misc.py b/tests/test_misc.py new file mode 100644 index 0000000..fc6b57d --- /dev/null +++ b/tests/test_misc.py @@ -0,0 +1,264 @@ +"""P14.1 — Misc API tests. + +Covers: +- Milestones global list +- Notifications +- Activity log +- API Keys +- Webhooks +- Export +- Dashboard stats +- Health check +""" +import pytest + + +class TestMilestonesGlobal: + """Global milestones endpoints.""" + + def test_list_all_milestones(self, client, db, make_user, auth_header): + """List all milestones (global endpoint).""" + user = make_user() + + resp = client.get("/milestones", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + + def test_list_milestones_with_project_filter(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Filter milestones by project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + milestone = Milestone(title="Test", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + resp = client.get(f"/milestones?project_id={project.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert all(m["project_id"] == project.id for m in data) + + def test_get_milestone_detail(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Get milestone by ID (global endpoint).""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + milestone = Milestone(title="Test", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + resp = client.get(f"/milestones/{milestone.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert data["id"] == milestone.id + + def test_milestone_progress(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Get milestone progress.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + + milestone = Milestone(title="Test", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + # Add tasks + task1 = Task( + title="Done", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.CLOSED, priority=TaskPriority.MEDIUM + ) + task2 = Task( + title="Open", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add_all([task1, task2]) + db.commit() + + resp = client.get(f"/milestones/{milestone.id}/progress", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert "total_issues" in data + assert "completed" in data + assert "progress_pct" in data + + +class TestNotifications: + """Notifications endpoints.""" + + def test_list_notifications(self, client, db, make_user, auth_header): + """List user notifications.""" + user = make_user() + + resp = client.get(f"/notifications?user_id={user.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + + def test_notification_count(self, client, db, make_user, auth_header): + """Get unread notification count.""" + user = make_user() + + resp = client.get(f"/notifications/count?user_id={user.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert "unread" in data + assert data["user_id"] == user.id + + def test_mark_notification_read(self, client, db, make_user, auth_header): + """Mark notification as read.""" + user = make_user() + + from app.models.notification import Notification + notification = Notification( + user_id=user.id, + type="test", + title="Test", + message="Test message", + is_read=False + ) + db.add(notification) + db.commit() + + resp = client.post(f"/notifications/{notification.id}/read", headers=auth_header(user)) + assert resp.status_code == 200 + assert resp.json()["status"] == "read" + + +class TestActivityLog: + """Activity log endpoints.""" + + def test_list_activity(self, client, db, make_user, auth_header): + """List activity logs.""" + user = make_user() + + resp = client.get("/activity", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + + def test_list_activity_with_filters(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Filter activity by entity.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + + from app.models.activity import ActivityLog + activity = ActivityLog( + action="create", + entity_type="project", + entity_id=project.id, + user_id=user.id, + details="Created project" + ) + db.add(activity) + db.commit() + + resp = client.get( + f"/activity?entity_type=project&entity_id={project.id}", + headers=auth_header(user) + ) + assert resp.status_code == 200 + data = resp.json() + assert all(a["entity_type"] == "project" for a in data) + + +class TestAPIKeys: + """API Key management.""" + + def test_create_api_key(self, client, db, make_user, auth_header): + """Create API key.""" + user = make_user() + + resp = client.post( + "/api-keys", + json={"name": "Test Key", "user_id": user.id}, + headers=auth_header(user) + ) + assert resp.status_code == 201 + data = resp.json() + assert data["name"] == "Test Key" + assert "key" in data + + def test_list_api_keys(self, client, db, make_user, auth_header): + """List API keys.""" + user = make_user() + + resp = client.get("/api-keys", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + + def test_revoke_api_key(self, client, db, make_user, auth_header): + """Revoke API key.""" + user = make_user() + + resp = client.post( + "/api-keys", + json={"name": "To Revoke", "user_id": user.id}, + headers=auth_header(user) + ) + key_id = resp.json()["id"] + + resp = client.delete(f"/api-keys/{key_id}", headers=auth_header(user)) + assert resp.status_code == 204 + + +class TestDashboard: + """Dashboard stats.""" + + def test_dashboard_stats(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Get dashboard statistics.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + resp = client.get("/dashboard/stats", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert "total" in data + assert "by_status" in data + assert "by_type" in data + assert "by_priority" in data + + def test_dashboard_stats_with_project_filter(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Get dashboard stats for specific project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + resp = client.get(f"/dashboard/stats?project_id={project.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert "total" in data + + +class TestHealth: + """Health check.""" + + def test_health_check(self, client): + """Health endpoint returns ok.""" + resp = client.get("/health") + assert resp.status_code == 200 + data = resp.json() + assert data["status"] == "healthy" + + def test_version(self, client): + """Version endpoint.""" + resp = client.get("/version") + assert resp.status_code == 200 + data = resp.json() + assert "version" in data + assert "name" in data diff --git a/tests/test_projects.py b/tests/test_projects.py new file mode 100644 index 0000000..e3be968 --- /dev/null +++ b/tests/test_projects.py @@ -0,0 +1,108 @@ +"""P14.1 — Projects API tests. + +Covers: +- List projects +- Get project by ID +- Create project +- Update project +- Delete project +- Project ownership and permissions +""" +import pytest + + +class TestProjects: + """Project management endpoints.""" + + def test_list_projects(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions): + """User can list projects they have access to.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project1 = make_project(name="Project 1") + project2 = make_project(name="Project 2") + + resp = client.get("/projects", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + + def test_get_project_by_id(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions): + """Get specific project details.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project(name="Test Project", owner_id=user.id) + + resp = client.get(f"/projects/{project.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert data["id"] == project.id + assert data["name"] == "Test Project" + + def test_create_project(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """User can create project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + + resp = client.post( + "/projects", + json={ + "name": "New Project", + "description": "Test description", + "project_code": "TEST" + }, + headers=auth_header(user) + ) + assert resp.status_code == 201 + data = resp.json() + assert data["name"] == "New Project" + assert data["project_code"] == "TEST" + assert "id" in data + + def test_update_project(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions): + """Project owner can update project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project(name="Old Name", owner_id=user.id) + + resp = client.patch( + f"/projects/{project.id}", + json={"name": "Updated Name"}, + headers=auth_header(user) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["name"] == "Updated Name" + + def test_delete_project(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions): + """Project owner can delete project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project(owner_id=user.id) + + resp = client.delete(f"/projects/{project.id}", headers=auth_header(user)) + assert resp.status_code == 204 + + def test_non_owner_cannot_delete_project(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Non-owner cannot delete project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + owner = make_user(username="owner") + other = make_user(username="other") + project = make_project(owner_id=owner.id) + make_member(project.id, other.id, dev_role.id) + + resp = client.delete(f"/projects/{project.id}", headers=auth_header(other)) + assert resp.status_code == 403 + + def test_project_code_generation(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Project code is auto-generated if not provided.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + + resp = client.post( + "/projects", + json={"name": "Auto Code Project"}, + headers=auth_header(user) + ) + assert resp.status_code == 201 + data = resp.json() + assert data["project_code"].startswith("P") diff --git a/tests/test_propose.py b/tests/test_propose.py new file mode 100644 index 0000000..97469e2 --- /dev/null +++ b/tests/test_propose.py @@ -0,0 +1,559 @@ +"""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 diff --git a/tests/test_roles.py b/tests/test_roles.py new file mode 100644 index 0000000..45d171b --- /dev/null +++ b/tests/test_roles.py @@ -0,0 +1,182 @@ +"""P14.1 — Roles and Permissions API tests. + +Covers: +- List roles +- Get role by ID +- Create role +- Update role +- Delete role +- Assign role to user +- Check permissions +""" +import pytest + + +class TestRoles: + """Role management endpoints.""" + + def test_list_roles(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """List all roles.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + + resp = client.get("/roles", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert len(data) >= 3 # admin, mgr, dev at minimum + + def test_get_role_by_id(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Get specific role.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + + resp = client.get(f"/roles/{admin_role.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert data["id"] == admin_role.id + assert "name" in data + + def test_create_role(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Admin can create new role.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + + resp = client.post( + "/roles", + json={ + "name": "tester", + "description": "Test role", + "is_global": False + }, + headers=auth_header(admin) + ) + assert resp.status_code == 201 + data = resp.json() + assert data["name"] == "tester" + + def test_update_role(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Admin can update role.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + + resp = client.patch( + f"/roles/{dev_role.id}", + json={"description": "Updated description"}, + headers=auth_header(admin) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["description"] == "Updated description" + + def test_delete_role(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Admin can delete non-default role.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + + # Create a role to delete + resp = client.post( + "/roles", + json={"name": "temp-role", "description": "To delete"}, + headers=auth_header(admin) + ) + role_id = resp.json()["id"] + + resp = client.delete(f"/roles/{role_id}", headers=auth_header(admin)) + assert resp.status_code == 204 + + def test_cannot_delete_admin_role(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Cannot delete admin role.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + + resp = client.delete(f"/roles/{admin_role.id}", headers=auth_header(admin)) + assert resp.status_code == 400 + + +class TestPermissions: + """Permission checking endpoints.""" + + def test_check_permission_true(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Check permission returns true when granted.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + # Dev should have view permission + resp = client.get( + f"/projects/{project.id}/check-permission?permission=view", + headers=auth_header(user) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["has_permission"] is True + + def test_check_permission_false(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Check permission returns false when not granted.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + # Add as guest (viewer role) + from app.models.role_permission import Role + guest_role = db.query(Role).filter(Role.name == "guest").first() + if not guest_role: + guest_role = Role(name="guest", description="Guest", is_global=False) + db.add(guest_role) + db.commit() + make_member(project.id, user.id, guest_role.id) + + resp = client.get( + f"/projects/{project.id}/check-permission?permission=admin", + headers=auth_header(user) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["has_permission"] is False + + +class TestRoleAssignments: + """Role assignment endpoints.""" + + def test_assign_role_to_user(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Assign role to project member.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + user = make_user(username="member") + project = make_project() + + resp = client.post( + f"/projects/{project.id}/members", + json={"user_id": user.id, "role_id": dev_role.id}, + headers=auth_header(admin) + ) + assert resp.status_code == 201 + + def test_change_user_role(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Change user's role in project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + user = make_user(username="member") + project = make_project() + make_member(project.id, user.id, dev_role.id) + + resp = client.patch( + f"/projects/{project.id}/members/{user.id}", + json={"role_id": mgr_role.id}, + headers=auth_header(admin) + ) + assert resp.status_code == 200 + + def test_remove_user_from_project(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Remove user from project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + user = make_user(username="member") + project = make_project() + make_member(project.id, user.id, dev_role.id) + + resp = client.delete( + f"/projects/{project.id}/members/{user.id}", + headers=auth_header(admin) + ) + assert resp.status_code == 204 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" diff --git a/tests/test_tasks.py b/tests/test_tasks.py new file mode 100644 index 0000000..8263e26 --- /dev/null +++ b/tests/test_tasks.py @@ -0,0 +1,211 @@ +"""P14.1 — Tasks CRUD API tests. + +Covers: +- List tasks (project-scoped, milestone-scoped) +- Get task by ID +- Create task +- Update task +- Delete task +- Task filtering by status, assignee, etc. +""" +import pytest +from datetime import datetime, timedelta + + +class TestTasksCRUD: + """Task CRUD endpoints.""" + + def test_list_tasks(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """List tasks for a project.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + # Create milestone and tasks + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + task1 = Task( + title="Task 1", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + task2 = Task( + title="Task 2", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add_all([task1, task2]) + db.commit() + + resp = client.get(f"/projects/{project.id}/tasks", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert len(data) >= 2 + + def test_get_task_by_id(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Get specific task.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + task = Task( + title="Test Task", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.HIGH, + task_type="issue" + ) + db.add(task) + db.commit() + + resp = client.get(f"/projects/{project.id}/tasks/{task.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert data["id"] == task.id + assert data["title"] == "Test Task" + assert data["task_type"] == "issue" + + def test_create_task(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Create new task.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project(project_code="PROJ") + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + resp = client.post( + f"/projects/{project.id}/tasks", + json={ + "title": "New Task", + "description": "Task description", + "milestone_id": milestone.id, + "task_type": "issue", + "priority": "high" + }, + headers=auth_header(user) + ) + assert resp.status_code == 201 + data = resp.json() + assert data["title"] == "New Task" + assert data["status"] == "open" + assert data["task_code"].startswith("PROJ:") + + def test_update_task(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Update task.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + task = Task( + title="Old Title", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add(task) + db.commit() + + resp = client.patch( + f"/projects/{project.id}/tasks/{task.id}", + json={"title": "Updated Title"}, + headers=auth_header(user) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["title"] == "Updated Title" + + def test_delete_task(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Delete task.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + task = Task( + title="To Delete", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add(task) + db.commit() + + resp = client.delete(f"/projects/{project.id}/tasks/{task.id}", headers=auth_header(user)) + assert resp.status_code == 204 + + def test_task_filter_by_status(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Filter tasks by status.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + open_task = Task( + title="Open", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + closed_task = Task( + title="Closed", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, status=TaskStatus.CLOSED, priority=TaskPriority.MEDIUM + ) + db.add_all([open_task, closed_task]) + db.commit() + + resp = client.get(f"/projects/{project.id}/tasks?status=open", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert all(t["status"] == "open" for t in data) + + def test_task_filter_by_assignee(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member): + """Filter tasks by assignee.""" + admin_role, mgr_role, dev_role = seed_roles_and_permissions + user = make_user() + assignee = make_user(username="assignee") + project = make_project() + make_member(project.id, user.id, dev_role.id) + + from app.models.milestone import Milestone, MilestoneStatus + from app.models.task import Task, TaskStatus, TaskPriority + milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN) + db.add(milestone) + db.commit() + + assigned_task = Task( + title="Assigned", project_id=project.id, milestone_id=milestone.id, + reporter_id=user.id, assignee_id=assignee.id, + status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM + ) + db.add(assigned_task) + db.commit() + + resp = client.get(f"/projects/{project.id}/tasks?assignee_id={assignee.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert all(t["assignee_id"] == assignee.id for t in data) diff --git a/tests/test_users.py b/tests/test_users.py new file mode 100644 index 0000000..5f5fb91 --- /dev/null +++ b/tests/test_users.py @@ -0,0 +1,100 @@ +"""P14.1 — Users API tests. + +Covers: +- List users +- Get user by ID +- Create user +- Update user +- Delete user +- User self-service restrictions +""" +import pytest + + +class TestUsers: + """User management endpoints.""" + + def test_list_users(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Admin can list all users.""" + seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + make_user(username="user1") + make_user(username="user2") + + resp = client.get("/users", headers=auth_header(admin)) + assert resp.status_code == 200 + data = resp.json() + assert len(data) >= 2 + + def test_get_user_by_id(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Get specific user details.""" + seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + user = make_user(username="testuser") + + resp = client.get(f"/users/{user.id}", headers=auth_header(admin)) + assert resp.status_code == 200 + data = resp.json() + assert data["id"] == user.id + assert data["username"] == "testuser" + + def test_create_user(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Admin can create new user.""" + seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + + resp = client.post( + "/users", + json={ + "username": "newuser", + "password": "newpass123", + "is_admin": False + }, + headers=auth_header(admin) + ) + assert resp.status_code == 201 + data = resp.json() + assert data["username"] == "newuser" + assert "id" in data + + def test_update_user(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Admin can update user.""" + seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + user = make_user(username="testuser") + + resp = client.patch( + f"/users/{user.id}", + json={"username": "updateduser"}, + headers=auth_header(admin) + ) + assert resp.status_code == 200 + data = resp.json() + assert data["username"] == "updateduser" + + def test_delete_user(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Admin can delete user.""" + seed_roles_and_permissions + admin = make_user(username="admin", is_admin=True) + user = make_user(username="testuser") + + resp = client.delete(f"/users/{user.id}", headers=auth_header(admin)) + assert resp.status_code == 204 + + def test_regular_user_cannot_list_all_users(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """Non-admin cannot list all users.""" + seed_roles_and_permissions + user = make_user(username="regular") + + resp = client.get("/users", headers=auth_header(user)) + assert resp.status_code == 403 + + def test_user_can_view_self(self, client, db, make_user, auth_header, seed_roles_and_permissions): + """User can view their own profile.""" + seed_roles_and_permissions + user = make_user(username="testuser") + + resp = client.get(f"/users/{user.id}", headers=auth_header(user)) + assert resp.status_code == 200 + data = resp.json() + assert data["id"] == user.id