tests/__init__.py

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

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}"}

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"]