455 lines
18 KiB
Python
455 lines
18 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from fastapi import FastAPI
|
||
from httpx import ASGITransport, AsyncClient
|
||
from sqlalchemy import JSON, func, select
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test")
|
||
|
||
from app.api import pmflow_os_api # noqa: E402
|
||
from app.core.database import get_db # noqa: E402
|
||
from app.dependencies.auth import get_current_user # noqa: E402
|
||
from app.models import ( # noqa: E402
|
||
DevWorkspaceBinding,
|
||
FeedbackIngest,
|
||
FeishuItem,
|
||
GitHubWebhookDelivery,
|
||
ModelCatalogItem,
|
||
ModelEvalRun,
|
||
ModelRecommendationSnapshot,
|
||
PMFlowApproval,
|
||
PMFlowChangeItem,
|
||
PMFlowChangeSet,
|
||
PMFlowEvidenceLink,
|
||
PMFlowLaunchDraft,
|
||
PMFlowThread,
|
||
PMFlowThreadMessage,
|
||
ProgressEvent,
|
||
Project,
|
||
ProjectModelBinding,
|
||
PromptPackage,
|
||
ReleaseValidationRun,
|
||
RepoBinding,
|
||
RevenueEvent,
|
||
ROISnapshot,
|
||
SemanticFact,
|
||
Signal,
|
||
Task,
|
||
TaskCodeLink,
|
||
TruthNode,
|
||
User,
|
||
UserProject,
|
||
)
|
||
from app.models.base import Base # noqa: E402
|
||
|
||
|
||
TABLES = [
|
||
User.__table__,
|
||
FeishuItem.__table__,
|
||
Signal.__table__,
|
||
Project.__table__,
|
||
UserProject.__table__,
|
||
PMFlowLaunchDraft.__table__,
|
||
PMFlowThread.__table__,
|
||
PMFlowThreadMessage.__table__,
|
||
PMFlowChangeSet.__table__,
|
||
PMFlowChangeItem.__table__,
|
||
PMFlowApproval.__table__,
|
||
PMFlowEvidenceLink.__table__,
|
||
TruthNode.__table__,
|
||
Task.__table__,
|
||
SemanticFact.__table__,
|
||
ModelCatalogItem.__table__,
|
||
ProjectModelBinding.__table__,
|
||
ModelRecommendationSnapshot.__table__,
|
||
ModelEvalRun.__table__,
|
||
PromptPackage.__table__,
|
||
DevWorkspaceBinding.__table__,
|
||
RepoBinding.__table__,
|
||
TaskCodeLink.__table__,
|
||
ProgressEvent.__table__,
|
||
ReleaseValidationRun.__table__,
|
||
FeedbackIngest.__table__,
|
||
RevenueEvent.__table__,
|
||
ROISnapshot.__table__,
|
||
GitHubWebhookDelivery.__table__,
|
||
]
|
||
|
||
|
||
def _use_sqlite_json_types() -> None:
|
||
for table in TABLES:
|
||
for column in table.columns:
|
||
type_name = column.type.__class__.__name__.upper()
|
||
if type_name in {"JSONB", "ARRAY"}:
|
||
column.type = JSON()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def db_env():
|
||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||
_use_sqlite_json_types()
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(lambda sync_conn: Base.metadata.create_all(sync_conn, tables=TABLES))
|
||
maker = async_sessionmaker(engine, expire_on_commit=False)
|
||
async with maker() as session:
|
||
yield session
|
||
await engine.dispose()
|
||
|
||
|
||
async def _seed_user_project(session, *, tenant_id: int = 1, role: str = "MEMBER", member: bool = True):
|
||
user = User(
|
||
tenant_id=tenant_id,
|
||
username=f"user-{tenant_id}-{role}-{member}",
|
||
email=f"user-{tenant_id}-{role}-{member}@example.com",
|
||
password_hash="hashed",
|
||
name="Owner",
|
||
role=role,
|
||
assigned_domains=[],
|
||
workload=0,
|
||
status="active",
|
||
is_active=True,
|
||
)
|
||
session.add(user)
|
||
await session.flush()
|
||
project = Project(
|
||
tenant_id=tenant_id,
|
||
name="真实项目",
|
||
title="真实项目",
|
||
description="用于 Change Review 测试",
|
||
owner_id=user.id,
|
||
status="active",
|
||
assigned_role="BUILDER",
|
||
signal_count=0,
|
||
)
|
||
session.add(project)
|
||
await session.flush()
|
||
if member:
|
||
session.add(UserProject(tenant_id=tenant_id, user_id=user.id, project_id=project.id))
|
||
await session.commit()
|
||
return user, project
|
||
|
||
|
||
def _build_app(session, current_user) -> FastAPI:
|
||
app = FastAPI()
|
||
app.include_router(pmflow_os_api.router)
|
||
|
||
async def _fake_get_db():
|
||
yield session
|
||
|
||
app.dependency_overrides[get_db] = _fake_get_db
|
||
app.dependency_overrides[get_current_user] = lambda: current_user
|
||
return app
|
||
|
||
|
||
def _proposed_items():
|
||
return [
|
||
{
|
||
"group": "new_object",
|
||
"object_type": "Requirement",
|
||
"title": "新增需求：审查真实落库",
|
||
"summary": "把 AI 候选变更写入待审状态机。",
|
||
"diff": "新增真实 ChangeItem。",
|
||
"payload": {"summary": "候选需求"},
|
||
"confidence": 0.91,
|
||
"risk_level": "medium",
|
||
},
|
||
{
|
||
"group": "risk",
|
||
"object_type": "Risk",
|
||
"title": "新增风险：高风险需记录审批",
|
||
"summary": "高风险项本轮先记录 approval。",
|
||
"diff": "新增高风险候选。",
|
||
"payload": {"summary": "候选风险"},
|
||
"confidence": 0.78,
|
||
"risk_level": "high",
|
||
},
|
||
{
|
||
"group": "decision",
|
||
"object_type": "Decision",
|
||
"title": "待确认决策：先审查后 commit",
|
||
"summary": "C1 不直接写项目真相层。",
|
||
"diff": "新增决策候选。",
|
||
"payload": {"summary": "候选决策"},
|
||
"confidence": 0.83,
|
||
"risk_level": "medium",
|
||
},
|
||
]
|
||
|
||
|
||
def _commit_proposed_items():
|
||
return [
|
||
{
|
||
"group": "new_object",
|
||
"object_type": "Requirement",
|
||
"title": "新增需求：真实写入项目记忆",
|
||
"summary": "确认后的需求必须进入 truth_nodes。",
|
||
"diff": "写入 requirement truth node。",
|
||
"payload": {"summary": "确认后的需求", "stage": "planning"},
|
||
"confidence": 0.92,
|
||
"risk_level": "medium",
|
||
},
|
||
{
|
||
"group": "new_object",
|
||
"object_type": "Task",
|
||
"title": "新增任务：生成开发上下文包",
|
||
"summary": "确认后的任务必须进入 tasks。",
|
||
"diff": "写入 task。",
|
||
"payload": {
|
||
"summary": "确认后的任务",
|
||
"description": "生成开发上下文包并等待执行桥消费。",
|
||
"priority": "high",
|
||
"acceptanceCriteria": ["任务进入 tasks", "任务绑定 truth_node"],
|
||
},
|
||
"confidence": 0.88,
|
||
"risk_level": "medium",
|
||
},
|
||
{
|
||
"group": "risk",
|
||
"object_type": "Risk",
|
||
"title": "新增风险：commit 需要审批",
|
||
"summary": "高风险项必须有 approval 才能写入。",
|
||
"diff": "写入 high risk。",
|
||
"payload": {"summary": "确认后的高风险", "mitigation": "提交前检查 approval gate。"},
|
||
"confidence": 0.81,
|
||
"risk_level": "high",
|
||
},
|
||
]
|
||
|
||
|
||
async def _create_message_and_extraction(client, project_id: int):
|
||
message = await client.post(
|
||
f"/pmflow-os/projects/{project_id}/threads/thread-project-{project_id}/messages",
|
||
json={"content": "请把本轮对话提取成待审变更", "source": "web", "attachments": []},
|
||
)
|
||
assert message.status_code == 200
|
||
msg = message.json()["data"]
|
||
extraction = await client.post(
|
||
f"/pmflow-os/projects/{project_id}/extractions",
|
||
json={
|
||
"thread_id": msg["threadId"],
|
||
"message_id": msg["id"],
|
||
"scope": "latest_message",
|
||
"headline": "本轮待审变更",
|
||
"summary": "生成 3 个候选变更。",
|
||
"proposed_items": _proposed_items(),
|
||
},
|
||
)
|
||
assert extraction.status_code == 200
|
||
return msg, extraction.json()["data"]
|
||
|
||
|
||
async def _create_commit_extraction(client, project_id: int):
|
||
message = await client.post(
|
||
f"/pmflow-os/projects/{project_id}/threads/thread-project-{project_id}/messages",
|
||
json={"content": "请把本轮确认项写入项目真相层", "source": "web", "attachments": []},
|
||
)
|
||
assert message.status_code == 200
|
||
msg = message.json()["data"]
|
||
extraction = await client.post(
|
||
f"/pmflow-os/projects/{project_id}/extractions",
|
||
json={
|
||
"thread_id": msg["threadId"],
|
||
"message_id": msg["id"],
|
||
"scope": "latest_message",
|
||
"headline": "待提交变更",
|
||
"summary": "生成 3 个可提交候选。",
|
||
"proposed_items": _commit_proposed_items(),
|
||
},
|
||
)
|
||
assert extraction.status_code == 200
|
||
return msg, extraction.json()["data"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_change_review_message_extraction_detail_and_projection(db_env) -> None:
|
||
session = db_env
|
||
user, project = await _seed_user_project(session)
|
||
app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))
|
||
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
msg, detail = await _create_message_and_extraction(client, project.id)
|
||
truth = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")
|
||
|
||
assert msg["threadId"].isdigit()
|
||
assert await session.scalar(select(func.count(PMFlowThread.id))) == 1
|
||
assert await session.scalar(select(func.count(PMFlowThreadMessage.id))) == 1
|
||
assert await session.scalar(select(func.count(PMFlowChangeSet.id))) == 1
|
||
assert await session.scalar(select(func.count(PMFlowChangeItem.id))) == 3
|
||
assert await session.scalar(select(func.count(PMFlowEvidenceLink.id))) == 1
|
||
|
||
assert detail["changeSet"]["status"] == "open"
|
||
assert len(detail["items"]) == 3
|
||
assert detail["items"][0]["status"] == "pending_review"
|
||
assert detail["evidence"][0]["sourceType"] == "thread_message"
|
||
assert detail["evidence"][0]["sourceId"] == msg["id"]
|
||
|
||
truth_payload = truth.json()["data"]
|
||
assert truth_payload["loopStage"] == "review_open"
|
||
assert truth_payload["activeChangeSetId"] == detail["changeSet"]["id"]
|
||
assert len(truth_payload["changeSets"]) == 1
|
||
assert len(truth_payload["changeItems"]) == 3
|
||
assert len(truth_payload["evidence"]) == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_change_item_accept_edit_defer_write_approvals_and_do_not_write_memory(db_env) -> None:
|
||
session = db_env
|
||
user, project = await _seed_user_project(session)
|
||
app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))
|
||
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
_, detail = await _create_message_and_extraction(client, project.id)
|
||
item_ids = [item["id"] for item in detail["items"]]
|
||
accepted = await client.post(f"/pmflow-os/change-items/{item_ids[0]}/accept")
|
||
edited = await client.post(
|
||
f"/pmflow-os/change-items/{item_ids[1]}/edit",
|
||
json={"summary": "人工修订摘要", "diff": "人工修订 diff", "payload_patch": {"owner": "user"}, "reason": "收紧范围"},
|
||
)
|
||
deferred = await client.post(f"/pmflow-os/change-items/{item_ids[2]}/defer", json={"reason": "下一轮再看"})
|
||
truth = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")
|
||
|
||
assert accepted.status_code == 200
|
||
assert edited.status_code == 200
|
||
assert deferred.status_code == 200
|
||
assert await session.scalar(select(func.count(PMFlowApproval.id))) == 3
|
||
assert await session.scalar(select(func.count(TruthNode.id))) == 0
|
||
assert await session.scalar(select(func.count(Task.id))) == 0
|
||
assert await session.scalar(select(func.count(SemanticFact.id))) == 0
|
||
|
||
change_set = (await session.execute(select(PMFlowChangeSet))).scalar_one()
|
||
assert change_set.status == "reviewed"
|
||
edited_row = (await session.execute(select(PMFlowChangeItem).where(PMFlowChangeItem.id == int(item_ids[1])))).scalar_one()
|
||
assert edited_row.status == "edited"
|
||
assert edited_row.edited_payload["owner"] == "user"
|
||
assert truth.json()["data"]["projectMemory"]["sourceSummary"]["truth_nodes"] == 0
|
||
assert len(truth.json()["data"]["approvals"]) == 3
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_change_set_commit_writes_truth_nodes_tasks_and_semantic_facts(db_env) -> None:
|
||
session = db_env
|
||
user, project = await _seed_user_project(session)
|
||
app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))
|
||
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
_, detail = await _create_commit_extraction(client, project.id)
|
||
item_ids = [item["id"] for item in detail["items"]]
|
||
await client.post(f"/pmflow-os/change-items/{item_ids[0]}/accept")
|
||
await client.post(f"/pmflow-os/change-items/{item_ids[1]}/accept")
|
||
await client.post(
|
||
f"/pmflow-os/change-items/{item_ids[2]}/edit",
|
||
json={"summary": "人工确认后的高风险", "payload_patch": {"mitigation": "人工确认后再落库"}},
|
||
)
|
||
committed = await client.post(
|
||
f"/pmflow-os/change-sets/{detail['changeSet']['id']}/commit",
|
||
json={"mode": "atomic", "note": "确认写入项目记忆"},
|
||
)
|
||
truth = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")
|
||
|
||
assert committed.status_code == 200
|
||
payload = committed.json()["data"]
|
||
assert payload["changeSet"]["status"] == "committed"
|
||
assert len(payload["committedItems"]) == 3
|
||
assert payload["failedItems"] == []
|
||
assert await session.scalar(select(func.count(TruthNode.id))) == 3
|
||
assert await session.scalar(select(func.count(Task.id))) == 1
|
||
assert await session.scalar(select(func.count(SemanticFact.id))) == 4
|
||
|
||
task = (await session.execute(select(Task))).scalar_one()
|
||
assert task.priority == "high"
|
||
assert task.status == "todo"
|
||
assert task.manual_overrides["acceptance_criteria"] == ["任务进入 tasks", "任务绑定 truth_node"]
|
||
|
||
truth_payload = truth.json()["data"]
|
||
assert truth_payload["activeChangeSetId"] is None
|
||
assert truth_payload["projectMemory"]["sourceSummary"]["truth_nodes"] == 3
|
||
assert truth_payload["projectMemory"]["sourceSummary"]["semantic_facts"] == 4
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_change_set_commit_blocks_high_risk_without_approval_and_rolls_back(db_env) -> None:
|
||
session = db_env
|
||
user, project = await _seed_user_project(session)
|
||
app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))
|
||
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
_, detail = await _create_commit_extraction(client, project.id)
|
||
high_risk_id = detail["items"][2]["id"]
|
||
|
||
item = (await session.execute(select(PMFlowChangeItem).where(PMFlowChangeItem.id == int(high_risk_id)))).scalar_one()
|
||
item.status = "accepted"
|
||
change_set = (await session.execute(select(PMFlowChangeSet).where(PMFlowChangeSet.id == int(item.change_set_id)))).scalar_one()
|
||
change_set.status = "reviewed"
|
||
await session.commit()
|
||
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
blocked = await client.post(
|
||
f"/pmflow-os/change-sets/{detail['changeSet']['id']}/commit",
|
||
json={"item_ids": [high_risk_id], "mode": "atomic"},
|
||
)
|
||
|
||
assert blocked.status_code == 409
|
||
assert "high_risk_approval_required" in blocked.json()["message"]
|
||
assert await session.scalar(select(func.count(TruthNode.id))) == 0
|
||
assert await session.scalar(select(func.count(Task.id))) == 0
|
||
assert await session.scalar(select(func.count(SemanticFact.id))) == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_change_item_review_conflict_and_closed_changeset_guard(db_env) -> None:
|
||
session = db_env
|
||
user, project = await _seed_user_project(session)
|
||
app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))
|
||
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
_, detail = await _create_message_and_extraction(client, project.id)
|
||
item_id = detail["items"][0]["id"]
|
||
first = await client.post(f"/pmflow-os/change-items/{item_id}/accept")
|
||
second = await client.post(f"/pmflow-os/change-items/{item_id}/accept")
|
||
|
||
assert first.status_code == 200
|
||
assert second.status_code == 409
|
||
assert "item_already_reviewed" in second.json()["message"]
|
||
|
||
pending = (await session.execute(select(PMFlowChangeItem).where(PMFlowChangeItem.status == "pending_review"))).scalars().first()
|
||
change_set = (await session.execute(select(PMFlowChangeSet).where(PMFlowChangeSet.id == pending.change_set_id))).scalar_one()
|
||
change_set.status = "closed"
|
||
await session.commit()
|
||
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
blocked = await client.post(f"/pmflow-os/change-items/{pending.id}/accept")
|
||
|
||
assert blocked.status_code == 409
|
||
assert "changeset_not_open" in blocked.json()["message"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_change_review_project_permission_boundary(db_env) -> None:
|
||
session = db_env
|
||
owner, project = await _seed_user_project(session)
|
||
outsider, _ = await _seed_user_project(session, tenant_id=1, member=False)
|
||
owner_app = _build_app(session, SimpleNamespace(id=owner.id, tenant_id=1, role="MEMBER"))
|
||
|
||
async with AsyncClient(transport=ASGITransport(app=owner_app), base_url="http://test") as client:
|
||
_, detail = await _create_message_and_extraction(client, project.id)
|
||
|
||
outsider_app = _build_app(session, SimpleNamespace(id=outsider.id, tenant_id=1, role="MEMBER"))
|
||
async with AsyncClient(transport=ASGITransport(app=outsider_app), base_url="http://test") as client:
|
||
denied = await client.get(f"/pmflow-os/projects/{project.id}/change-sets/{detail['changeSet']['id']}")
|
||
denied_action = await client.post(f"/pmflow-os/change-items/{detail['items'][0]['id']}/accept")
|
||
|
||
assert denied.status_code == 403
|
||
assert denied_action.status_code == 403
|
||
|
||
cross_tenant_app = _build_app(session, SimpleNamespace(id=outsider.id, tenant_id=2, role="MEMBER"))
|
||
async with AsyncClient(transport=ASGITransport(app=cross_tenant_app), base_url="http://test") as client:
|
||
missing = await client.get(f"/pmflow-os/projects/{project.id}/change-sets/{detail['changeSet']['id']}")
|
||
|
||
assert missing.status_code == 404
|