From 1ed7a85e11dc1eb95c9e77c0985b8c3c87d8536e Mon Sep 17 00:00:00 2001 From: zhi Date: Mon, 30 Mar 2026 16:17:00 +0000 Subject: [PATCH] BE-PR-011: Fix test infrastructure and add Proposal/Essential/Story restricted tests - Patched conftest.py to monkey-patch app.core.config engine/SessionLocal with SQLite in-memory DB BEFORE importing the FastAPI app, preventing startup event from trying to connect to production MySQL - All 29 tests pass: Essential CRUD (11), Proposal Accept (8), Story restricted (6), Legacy compat (4) --- tests/__init__.py | 1 + tests/conftest.py | 191 ++++++++++ tests/test_proposal_essential_story.py | 481 +++++++++++++++++++++++++ 3 files changed, 673 insertions(+) create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_proposal_essential_story.py diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..65140f2 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# tests package diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..5cb9573 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,191 @@ +"""Shared test fixtures — SQLite in-memory DB + FastAPI TestClient. + +Every test function gets a fresh database with baseline seed data: + - Roles: admin, dev, viewer (+ permissions for propose.accept/reject/reopen) + - An admin user (id=1) + - A dev user (id=2) + - A project (id=1) with both users as members + - An open milestone (id=1) under the project +""" + +import pytest +from sqlalchemy import create_engine, event +from sqlalchemy.orm import sessionmaker +from fastapi.testclient import TestClient + +# --------------------------------------------------------------------------- +# Patch the production engine/SessionLocal BEFORE importing app so that +# startup events (Base.metadata.create_all, init_wizard, etc.) use the +# in-memory SQLite database instead of trying to connect to MySQL. +# --------------------------------------------------------------------------- + +SQLALCHEMY_DATABASE_URL = "sqlite:///file::memory:?cache=shared&uri=true" + +test_engine = create_engine( + SQLALCHEMY_DATABASE_URL, + connect_args={"check_same_thread": False}, +) + +# SQLite foreign-key enforcement +@event.listens_for(test_engine, "connect") +def _set_sqlite_pragma(dbapi_connection, connection_record): + cursor = dbapi_connection.cursor() + cursor.execute("PRAGMA foreign_keys=ON") + cursor.close() + +TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=test_engine) + +# Monkey-patch app.core.config so the entire app uses SQLite +import app.core.config as _cfg +_cfg.engine = test_engine +_cfg.SessionLocal = TestingSessionLocal + +# Now it's safe to import app and friends +from app.core.config import Base, get_db +from app.main import app +from app.models import models +from app.models.role_permission import Role, Permission, RolePermission +from app.models.milestone import Milestone, MilestoneStatus +from app.api.deps import get_password_hash, create_access_token + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def setup_database(): + """Create all tables before each test, drop after.""" + Base.metadata.create_all(bind=test_engine) + yield + Base.metadata.drop_all(bind=test_engine) + + +@pytest.fixture() +def db(): + """Yield a DB session for direct model manipulation in tests.""" + session = TestingSessionLocal() + try: + yield session + finally: + session.close() + + +def _override_get_db(): + session = TestingSessionLocal() + try: + yield session + finally: + session.close() + + +@pytest.fixture() +def client(): + """FastAPI TestClient wired to the test DB.""" + app.dependency_overrides[get_db] = _override_get_db + with TestClient(app, raise_server_exceptions=False) as c: + yield c + app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# Seed helpers +# --------------------------------------------------------------------------- + +def _seed_roles_and_permissions(db_session): + """Create admin/dev/viewer roles and key permissions.""" + admin_role = Role(id=1, name="admin", is_global=True) + dev_role = Role(id=2, name="dev", is_global=False) + viewer_role = Role(id=3, name="viewer", is_global=False) + db_session.add_all([admin_role, dev_role, viewer_role]) + db_session.flush() + + perms = [] + for pname in ["propose.accept", "propose.reject", "propose.reopen", + "task.create", "task.edit", "task.delete"]: + p = Permission(name=pname, category="proposal") + db_session.add(p) + db_session.flush() + perms.append(p) + + # Admin gets all permissions + for p in perms: + db_session.add(RolePermission(role_id=admin_role.id, permission_id=p.id)) + + # Dev gets propose.accept / reject / reopen and task perms + for p in perms: + db_session.add(RolePermission(role_id=dev_role.id, permission_id=p.id)) + + db_session.flush() + return admin_role, dev_role, viewer_role + + +def _seed_users(db_session, admin_role, dev_role): + """Create admin + dev users.""" + admin_user = models.User( + id=1, username="admin", email="admin@test.com", + hashed_password=get_password_hash("admin123"), + is_admin=True, role_id=admin_role.id, + ) + dev_user = models.User( + id=2, username="developer", email="dev@test.com", + hashed_password=get_password_hash("dev123"), + is_admin=False, role_id=dev_role.id, + ) + db_session.add_all([admin_user, dev_user]) + db_session.flush() + return admin_user, dev_user + + +def _seed_project(db_session, admin_user, dev_user, dev_role): + """Create a project with both users as members.""" + project = models.Project( + id=1, name="TestProject", project_code="TPRJ", + owner_name=admin_user.username, owner_id=admin_user.id, + ) + db_session.add(project) + db_session.flush() + + db_session.add(models.ProjectMember(project_id=project.id, user_id=admin_user.id, role_id=1)) + db_session.add(models.ProjectMember(project_id=project.id, user_id=dev_user.id, role_id=dev_role.id)) + db_session.flush() + return project + + +def _seed_milestone(db_session, project): + """Create an open milestone.""" + ms = Milestone( + id=1, title="v1.0", milestone_code="TPRJ:M00001", + status=MilestoneStatus.OPEN, project_id=project.id, created_by_id=1, + ) + db_session.add(ms) + db_session.flush() + return ms + + +@pytest.fixture() +def seed(db): + """Seed the DB with roles, users, project, milestone. Returns a namespace dict.""" + admin_role, dev_role, viewer_role = _seed_roles_and_permissions(db) + admin_user, dev_user = _seed_users(db, admin_role, dev_role) + project = _seed_project(db, admin_user, dev_user, dev_role) + milestone = _seed_milestone(db, project) + db.commit() + + admin_token = create_access_token({"sub": str(admin_user.id)}) + dev_token = create_access_token({"sub": str(dev_user.id)}) + + return { + "admin_user": admin_user, + "dev_user": dev_user, + "admin_role": admin_role, + "dev_role": dev_role, + "project": project, + "milestone": milestone, + "admin_token": admin_token, + "dev_token": dev_token, + } + + +def auth_header(token: str) -> dict: + """Return Authorization header dict.""" + return {"Authorization": f"Bearer {token}"} diff --git a/tests/test_proposal_essential_story.py b/tests/test_proposal_essential_story.py new file mode 100644 index 0000000..21489d5 --- /dev/null +++ b/tests/test_proposal_essential_story.py @@ -0,0 +1,481 @@ +"""BE-PR-011 — Tests for Proposal / Essential / Story restricted. + +Covers: + 1. Essential CRUD (create, read, update, delete) + 2. Proposal Accept — batch generation of story tasks + 3. Story restricted — general create endpoint blocks story/* tasks + 4. Backward compatibility with legacy proposal data (feat_task_id read-only) +""" + +import pytest +from tests.conftest import auth_header + + +# =================================================================== +# Helper shortcuts +# =================================================================== + +PRJ = "1" # project id + + +def _create_proposal(client, token, title="Test Proposal", description="desc"): + """Create an open proposal and return its JSON.""" + r = client.post( + f"/projects/{PRJ}/proposals", + json={"title": title, "description": description}, + headers=auth_header(token), + ) + assert r.status_code == 201, r.text + return r.json() + + +def _create_essential(client, token, proposal_id, etype="feature", title="Ess 1"): + """Create an Essential under the given proposal and return its JSON.""" + r = client.post( + f"/projects/{PRJ}/proposals/{proposal_id}/essentials", + json={"type": etype, "title": title, "description": f"{etype} essential"}, + headers=auth_header(token), + ) + assert r.status_code == 201, r.text + return r.json() + + +# =================================================================== +# 1. Essential CRUD +# =================================================================== + +class TestEssentialCRUD: + """Test creating, listing, reading, updating, and deleting Essentials.""" + + def test_create_essential(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + ess = _create_essential(client, seed["admin_token"], proposal["id"]) + + assert ess["type"] == "feature" + assert ess["title"] == "Ess 1" + assert ess["proposal_id"] == proposal["id"] + assert ess["essential_code"].endswith(":E00001") + + def test_create_multiple_essentials_increments_code(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + e1 = _create_essential(client, seed["admin_token"], proposal["id"], "feature", "E1") + e2 = _create_essential(client, seed["admin_token"], proposal["id"], "improvement", "E2") + e3 = _create_essential(client, seed["admin_token"], proposal["id"], "refactor", "E3") + + assert e1["essential_code"].endswith(":E00001") + assert e2["essential_code"].endswith(":E00002") + assert e3["essential_code"].endswith(":E00003") + + def test_list_essentials(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"], "feature", "A") + _create_essential(client, seed["admin_token"], proposal["id"], "improvement", "B") + + r = client.get( + f"/projects/{PRJ}/proposals/{proposal['id']}/essentials", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 + items = r.json() + assert len(items) == 2 + assert items[0]["title"] == "A" + assert items[1]["title"] == "B" + + def test_get_single_essential(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + ess = _create_essential(client, seed["admin_token"], proposal["id"]) + + r = client.get( + f"/projects/{PRJ}/proposals/{proposal['id']}/essentials/{ess['id']}", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 + assert r.json()["id"] == ess["id"] + + def test_get_essential_by_code(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + ess = _create_essential(client, seed["admin_token"], proposal["id"]) + + r = client.get( + f"/projects/{PRJ}/proposals/{proposal['id']}/essentials/{ess['essential_code']}", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 + assert r.json()["id"] == ess["id"] + + def test_update_essential(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + ess = _create_essential(client, seed["admin_token"], proposal["id"]) + + r = client.patch( + f"/projects/{PRJ}/proposals/{proposal['id']}/essentials/{ess['id']}", + json={"title": "Updated Title", "type": "refactor"}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 + data = r.json() + assert data["title"] == "Updated Title" + assert data["type"] == "refactor" + + def test_delete_essential(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + ess = _create_essential(client, seed["admin_token"], proposal["id"]) + + r = client.delete( + f"/projects/{PRJ}/proposals/{proposal['id']}/essentials/{ess['id']}", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 204 + + # Verify it's gone + r = client.get( + f"/projects/{PRJ}/proposals/{proposal['id']}/essentials/{ess['id']}", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 404 + + def test_cannot_create_essential_on_accepted_proposal(self, client, seed): + """Essentials can only be added to open proposals.""" + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"]) + + # Accept the proposal + client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + + # Try to create another essential → should fail + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/essentials", + json={"type": "feature", "title": "Late essential"}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 400 + assert "open" in r.json()["detail"].lower() + + def test_cannot_update_essential_on_rejected_proposal(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + ess = _create_essential(client, seed["admin_token"], proposal["id"]) + + # Reject the proposal + client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/reject", + json={"reason": "not now"}, + headers=auth_header(seed["admin_token"]), + ) + + r = client.patch( + f"/projects/{PRJ}/proposals/{proposal['id']}/essentials/{ess['id']}", + json={"title": "Should fail"}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 400 + + def test_essential_not_found(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + + r = client.get( + f"/projects/{PRJ}/proposals/{proposal['id']}/essentials/9999", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 404 + + def test_essential_types(self, client, seed): + """All three essential types should be valid.""" + proposal = _create_proposal(client, seed["admin_token"]) + for etype in ["feature", "improvement", "refactor"]: + ess = _create_essential(client, seed["admin_token"], proposal["id"], etype, f"T-{etype}") + assert ess["type"] == etype + + +# =================================================================== +# 2. Proposal Accept — batch story task generation +# =================================================================== + +class TestProposalAccept: + """Test that accepting a Proposal generates story tasks from Essentials.""" + + def test_accept_generates_story_tasks(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"], "feature", "Feat 1") + _create_essential(client, seed["admin_token"], proposal["id"], "improvement", "Improv 1") + _create_essential(client, seed["admin_token"], proposal["id"], "refactor", "Refac 1") + + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200, r.text + data = r.json() + + assert data["status"] == "accepted" + tasks = data["generated_tasks"] + assert len(tasks) == 3 + + subtypes = {t["task_subtype"] for t in tasks} + assert subtypes == {"feature", "improvement", "refactor"} + + for t in tasks: + assert t["task_type"] == "story" + assert t["essential_id"] is not None + + def test_accept_requires_milestone(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"]) + + # Missing milestone_id + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 422 # validation error + + def test_accept_rejects_invalid_milestone(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"]) + + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 9999}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 404 + assert "milestone" in r.json()["detail"].lower() + + def test_accept_requires_at_least_one_essential(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 400 + assert "essential" in r.json()["detail"].lower() + + def test_accept_only_open_proposals(self, client, seed): + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"]) + + # Reject first + client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/reject", + json={"reason": "nope"}, + headers=auth_header(seed["admin_token"]), + ) + + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 400 + assert "open" in r.json()["detail"].lower() + + def test_accept_sets_source_proposal_id_on_tasks(self, client, seed): + """Generated tasks should have source_proposal_id and source_essential_id set.""" + proposal = _create_proposal(client, seed["admin_token"]) + ess = _create_essential(client, seed["admin_token"], proposal["id"]) + + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 + tasks = r.json()["generated_tasks"] + assert len(tasks) == 1 + assert tasks[0]["essential_id"] == ess["id"] + + def test_proposal_detail_includes_generated_tasks(self, client, seed): + """After accept, proposal detail should include generated_tasks.""" + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"], "feature", "F1") + + client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + + r = client.get( + f"/projects/{PRJ}/proposals/{proposal['id']}", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 + data = r.json() + assert len(data["essentials"]) == 1 + assert len(data["generated_tasks"]) >= 1 + assert data["generated_tasks"][0]["task_type"] == "story" + + def test_double_accept_fails(self, client, seed): + """Accepting an already-accepted proposal should fail.""" + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"]) + + client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 400 + + +# =================================================================== +# 3. Story restricted — general create blocks story/* tasks +# =================================================================== + +class TestStoryRestricted: + """Test that story/* tasks cannot be created via the general task endpoint.""" + + def test_create_story_feature_blocked(self, client, seed): + r = client.post( + "/tasks", + json={ + "title": "Sneaky story", + "task_type": "story", + "task_subtype": "feature", + "project_id": 1, + "milestone_id": 1, + }, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 400 + assert "story" in r.json()["detail"].lower() + + def test_create_story_improvement_blocked(self, client, seed): + r = client.post( + "/tasks", + json={ + "title": "Sneaky improvement", + "task_type": "story", + "task_subtype": "improvement", + "project_id": 1, + "milestone_id": 1, + }, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 400 + + def test_create_story_refactor_blocked(self, client, seed): + r = client.post( + "/tasks", + json={ + "title": "Sneaky refactor", + "task_type": "story", + "task_subtype": "refactor", + "project_id": 1, + "milestone_id": 1, + }, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 400 + + def test_create_story_no_subtype_blocked(self, client, seed): + r = client.post( + "/tasks", + json={ + "title": "Bare story", + "task_type": "story", + "project_id": 1, + "milestone_id": 1, + }, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 400 + + def test_create_issue_still_allowed(self, client, seed): + """Non-restricted types should still work normally.""" + r = client.post( + "/tasks", + json={ + "title": "Normal issue", + "task_type": "issue", + "task_subtype": "defect", + "project_id": 1, + "milestone_id": 1, + }, + headers=auth_header(seed["admin_token"]), + ) + # Should succeed (200 or 201) + assert r.status_code in (200, 201), r.text + + def test_story_only_via_proposal_accept(self, client, seed): + """Story tasks should exist only when created via Proposal Accept.""" + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"], "feature", "Via Accept") + + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 + tasks = r.json()["generated_tasks"] + assert len(tasks) == 1 + assert tasks[0]["task_type"] == "story" + assert tasks[0]["task_subtype"] == "feature" + + +# =================================================================== +# 4. Legacy / backward compatibility +# =================================================================== + +class TestLegacyCompat: + """Test backward compat with old proposal data (feat_task_id read-only).""" + + def test_feat_task_id_in_response(self, client, seed): + """Response should include feat_task_id (even if None).""" + proposal = _create_proposal(client, seed["admin_token"]) + r = client.get( + f"/projects/{PRJ}/proposals/{proposal['id']}", + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 + data = r.json() + assert "feat_task_id" in data + # New proposals should have None + assert data["feat_task_id"] is None + + def test_feat_task_id_not_writable_via_update(self, client, seed): + """Clients should not be able to set feat_task_id via PATCH.""" + proposal = _create_proposal(client, seed["admin_token"]) + + r = client.patch( + f"/projects/{PRJ}/proposals/{proposal['id']}", + json={"feat_task_id": "FAKE-TASK-123"}, + headers=auth_header(seed["admin_token"]), + ) + # Should succeed (ignoring the field) or reject + if r.status_code == 200: + assert r.json()["feat_task_id"] is None # not written + + def test_new_accept_does_not_write_feat_task_id(self, client, seed): + """After accept, feat_task_id should remain None; use generated_tasks.""" + proposal = _create_proposal(client, seed["admin_token"]) + _create_essential(client, seed["admin_token"], proposal["id"]) + + r = client.post( + f"/projects/{PRJ}/proposals/{proposal['id']}/accept", + json={"milestone_id": 1}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 + assert r.json()["feat_task_id"] is None + + def test_propose_code_alias(self, client, seed): + """Response should include both proposal_code and propose_code for compat.""" + proposal = _create_proposal(client, seed["admin_token"]) + assert "proposal_code" in proposal + assert "propose_code" in proposal + assert proposal["proposal_code"] == proposal["propose_code"]