from __future__ import annotations import asyncio import os from datetime import timedelta 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_launch_api, pmflow_os_api, projects_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, IterationRecommendation, ModelCatalogItem, ModelEvalRun, ModelRecommendationSnapshot, PMFlowApproval, PMFlowChangeItem, PMFlowChangeSet, PMFlowEvidenceLink, OperationLog, PMFlowLaunchDraft, PMFlowLaunchDraftJob, PMFlowThread, PMFlowThreadMessage, ComplianceCheckRun, ProgressEvent, Project, ProjectModelBinding, PromptRelease, ProjectSignalLink, PromptPackage, ReleaseValidationRun, RepoBinding, RevenueEvent, ROISnapshot, SemanticFact, Signal, Task, TaskCodeLink, TruthNode, User, UserProject, ) from app.schemas.pmflow_launch import PMFlowLaunchDraftJobRequest # noqa: E402 from app.models.base import Base # noqa: E402 from app.services import pmflow_launch_service # noqa: E402 from app.utils.timezone import now_cn_naive # noqa: E402 TABLES = [ User.__table__, FeishuItem.__table__, Signal.__table__, Project.__table__, ProjectSignalLink.__table__, UserProject.__table__, PMFlowLaunchDraft.__table__, PMFlowLaunchDraftJob.__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__, PromptRelease.__table__, PromptPackage.__table__, DevWorkspaceBinding.__table__, RepoBinding.__table__, TaskCodeLink.__table__, ProgressEvent.__table__, ReleaseValidationRun.__table__, ComplianceCheckRun.__table__, FeedbackIngest.__table__, IterationRecommendation.__table__, RevenueEvent.__table__, ROISnapshot.__table__, GitHubWebhookDelivery.__table__, OperationLog.__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(session, *, tenant_id: int = 1, username: str = "owner"): user = User( tenant_id=tenant_id, username=f"{username}-{tenant_id}", email=f"{username}-{tenant_id}@example.com", password_hash="hashed", name="Owner", role="MEMBER", assigned_domains=[], workload=0, status="active", is_active=True, ) session.add(user) await session.commit() return user def _build_app(session, current_user) -> FastAPI: app = FastAPI() app.include_router(pmflow_launch_api.router) app.include_router(pmflow_os_api.router) app.include_router(projects_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 _draft_payload(**overrides): payload = { "title": "AI 会议纪要助手", "one_liner": "自动整理会议纪要并生成待办。", "target_user": "个人开发者、小团队负责人", "problem": "会议信息散落，行动项难追踪。", "solution": "解析会议文本并转成项目任务和风险。", "requirements": [ {"title": "导入会议文本", "summary": "支持粘贴会议纪要", "acceptance_criteria": ["能保存输入"]}, {"title": "提取行动项", "summary": "识别待办和负责人", "acceptance_criteria": ["输出任务列表"]}, {"title": "风险提醒", "summary": "识别不确定项", "acceptance_criteria": ["输出风险"]}, ], "tasks": [ {"title": "设计输入流程", "summary": "完成启动台输入", "group": "product", "priority": "medium", "requirement_index": 0}, {"title": "实现文本解析", "summary": "调用结构化服务", "group": "engineering", "priority": "high", "requirement_index": 1}, {"title": "补充失败测试", "summary": "覆盖失败态", "group": "qa", "priority": "medium", "requirement_index": 2}, ], "risks": [ {"title": "结构化质量风险", "summary": "会议内容过短会影响提取", "level": "medium", "mitigation": "提示补充上下文"}, {"title": "隐私风险", "summary": "会议内容可能包含敏感信息", "level": "high", "mitigation": "明确权限边界"}, ], "decisions": [ {"title": "先做文本输入", "reason": "首版不接录音和日历", "owner": "项目负责人"}, ], "evidence": [{"source": "user_input", "excerpt": "我想做一个 AI 会议纪要工具"}], "confidence": 0.82, "warnings": [], } payload.update(overrides) return payload def _patch_llm(monkeypatch, payload): async def _fake_generate_structured_object(*args, **kwargs): return SimpleNamespace(payload=payload, raw_text="{}", provider="test", model="test") monkeypatch.setattr(pmflow_launch_service.llm_service, "generate_structured_object", _fake_generate_structured_object) def _runtime_profile(**overrides): data = { "function_name": "pmflow_launch_draft_v1", "scene_key": "project_launch", "prompt_key": "PROMPT_PMFLOW_LAUNCH_DRAFT", "required_slot": "structured", "response_contract": "structured", "max_tokens": 7200, "repair_max_tokens": 3600, "temperature": 0.0, "foreground_wait_seconds": 10, "provider_timeout_seconds": 90, "job_timeout_seconds": 180, "max_retries": 1, "large_input_policy": "preprocess_then_generate", "response_mode": "fast", "response_format": "json_object", "profile_version": "test", "profile_source": "active", } data.update(overrides) return pmflow_launch_service.AIFunctionRuntimeProfile(**data) @pytest.mark.asyncio async def test_create_launch_draft_success_persists_draft(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) _patch_llm(monkeypatch, _draft_payload()) 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: response = await client.post("/pmflow-launch/drafts", json={"input": "我想做一个 AI 会议纪要工具"}) assert response.status_code == 200 payload = response.json()["data"] assert payload["draft_id"].startswith("draft_") rows = (await session.execute(select(PMFlowLaunchDraft))).scalars().all() assert len(rows) == 1 assert rows[0].status == "draft" assert rows[0].draft_payload["title"] == "AI 会议纪要助手" @pytest.mark.asyncio async def test_create_launch_draft_uses_platform_prompt_release(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session, tenant_id=12) release = PromptRelease( tenant_id=1, prompt_key="PROMPT_PMFLOW_LAUNCH_DRAFT", version=1, body_snapshot=( "ACTIVE_RELEASE_MARKER\n" "用户输入：{input_text}\n" "附件摘要：{attachments_json}\n" "输出 JSON：{{\"title\":\"string\",\"one_liner\":\"string\",\"target_user\":\"string\"," "\"problem\":\"string\",\"solution\":\"string\",\"requirements\":[],\"tasks\":[]," "\"risks\":[],\"decisions\":[],\"evidence\":[],\"confidence\":0,\"warnings\":[]}}" ), status="active", created_by=1, activated_by=1, activated_at=now_cn_naive().isoformat(), ) session.add(release) await session.commit() await session.refresh(release) captured: dict[str, object] = {} async def _fake_generate_structured_object(prompt, *args, **kwargs): captured["prompt"] = prompt captured["routing_meta"] = kwargs.get("routing_meta") return SimpleNamespace(payload=_draft_payload(), raw_text="{}", provider="test", model="test") monkeypatch.setattr(pmflow_launch_service.llm_service, "generate_structured_object", _fake_generate_structured_object) app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=user.tenant_id, role="MEMBER")) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.post("/pmflow-launch/drafts", json={"input": "我想做一个 AI 会议纪要工具"}) assert response.status_code == 200 assert "ACTIVE_RELEASE_MARKER" in str(captured["prompt"]) routing_meta = captured["routing_meta"] assert isinstance(routing_meta, dict) assert routing_meta["prompt_key"] == "PROMPT_PMFLOW_LAUNCH_DRAFT" assert routing_meta["prompt_source"] == "active_release" assert routing_meta["prompt_release_id"] == release.id assert routing_meta["prompt_runtime_tenant_id"] == 1 assert routing_meta["provider_timeout_seconds"] == 90 assert routing_meta["job_timeout_seconds"] == 180 assert routing_meta["large_input_policy"] == "preprocess_then_generate" @pytest.mark.asyncio async def test_draft_job_response_exposes_runtime_policy(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) async def _fake_resolve(self, *, stage: str = "active"): return _runtime_profile( foreground_wait_seconds=12, provider_timeout_seconds=120, job_timeout_seconds=240, max_retries=2, large_input_policy="direct_generate", ) monkeypatch.setattr(pmflow_launch_service.PMFlowLaunchService, "_resolve_draft_runtime_profile", _fake_resolve) service = pmflow_launch_service.PMFlowLaunchService(session, tenant_id=1) job = await service.create_draft_job( body=PMFlowLaunchDraftJobRequest(input="我想做一个 AI 会议纪要工具，面向小团队整理行动项。"), user_id=int(user.id), ) assert job.runtime_policy is not None assert job.runtime_policy.foreground_wait_seconds == 12 assert job.runtime_policy.provider_timeout_seconds == 120 assert job.runtime_policy.job_timeout_seconds == 240 assert job.runtime_policy.max_retries == 2 assert job.runtime_policy.large_input_policy == "direct_generate" @pytest.mark.asyncio async def test_draft_job_marks_failed_when_job_timeout_expires(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) async def _fake_resolve(self, *, stage: str = "active"): return _runtime_profile(job_timeout_seconds=10) async def _slow_generate_structured_object(*args, **kwargs): await asyncio.sleep(2) return SimpleNamespace(payload=_draft_payload(), raw_text="{}", provider="test", model="test") monkeypatch.setattr(pmflow_launch_service.PMFlowLaunchService, "_resolve_draft_runtime_profile", _fake_resolve) monkeypatch.setattr(pmflow_launch_service.PMFlowLaunchService, "_bounded_seconds", staticmethod(lambda *args, **kwargs: 1)) monkeypatch.setattr(pmflow_launch_service.llm_service, "generate_structured_object", _slow_generate_structured_object) service = pmflow_launch_service.PMFlowLaunchService(session, tenant_id=1) job = await service.create_draft_job( body=PMFlowLaunchDraftJobRequest(input="我想做一个 AI 会议纪要工具，面向小团队整理行动项。"), user_id=int(user.id), ) await service.run_draft_job(job_id=job.job_id) row = ( await session.execute(select(PMFlowLaunchDraftJob).where(PMFlowLaunchDraftJob.job_id == job.job_id)) ).scalar_one() assert row.status == "failed" assert row.error_code == "draft_job_timeout" @pytest.mark.asyncio async def test_large_input_policy_compacts_material_before_prompt(db_env) -> None: service = pmflow_launch_service.PMFlowLaunchService(db_env, tenant_id=1) body = PMFlowLaunchDraftJobRequest( input="产品目标：" + ("很长的项目材料 " * 300), attachments=[{"type": "text", "name": "research.md", "summary": "用户访谈：" + ("访谈内容 " * 1000)}], ) next_body, meta = service._apply_large_input_policy( body=body, routing_meta={ "input_bucket": "large", "large_input_policy": "preprocess_then_generate", }, ) assert meta["large_input_preprocessed"] is True assert len(next_body.input) <= 3900 assert next_body.attachments[0].summary is not None assert len(next_body.attachments[0].summary or "") <= 560 @pytest.mark.asyncio async def test_create_launch_draft_uses_product_sized_token_budget(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session, tenant_id=12) captured: dict[str, object] = {} async def _fake_generate_structured_object(*args, **kwargs): captured["max_tokens"] = kwargs.get("max_tokens") return SimpleNamespace(payload=_draft_payload(), raw_text="{}", provider="test", model="test") monkeypatch.setattr(pmflow_launch_service.llm_service, "generate_structured_object", _fake_generate_structured_object) app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=user.tenant_id, role="MEMBER")) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.post("/pmflow-launch/drafts", json={"input": "我想做一个 AI 会议纪要工具，帮助团队提取行动项"}) assert response.status_code == 200 assert captured["max_tokens"] == pmflow_launch_service.PMFlowLaunchService.DRAFT_GENERATION_MAX_TOKENS @pytest.mark.asyncio async def test_create_launch_draft_missing_core_fields_returns_422(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) _patch_llm(monkeypatch, _draft_payload(title="")) 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: response = await client.post("/pmflow-launch/drafts", json={"input": "我想做一个 AI 会议纪要工具，帮助团队提取行动项"}) assert response.status_code == 422 assert response.json()["code"] == 422 assert "draft_structured_failed" in response.json()["message"] assert response.json()["data"] is None @pytest.mark.asyncio async def test_create_launch_draft_rejects_low_information_input_before_llm(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) called = False async def _fake_generate_structured_object(*args, **kwargs): nonlocal called called = True return SimpleNamespace(payload=_draft_payload(), raw_text="{}", provider="test", model="test") monkeypatch.setattr(pmflow_launch_service.llm_service, "generate_structured_object", _fake_generate_structured_object) 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: response = await client.post("/pmflow-launch/drafts", json={"input": "你好"}) assert response.status_code == 422 assert "input_needs_more_info" in response.json()["message"] assert response.json()["data"] is None assert called is False @pytest.mark.asyncio async def test_create_launch_draft_count_warnings_are_explicit(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) _patch_llm(monkeypatch, _draft_payload(risks=[{"title": "单一风险", "summary": "不足", "level": "medium"}])) 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: response = await client.post("/pmflow-launch/drafts", json={"input": "我想做一个 AI 会议纪要工具，帮助团队提取行动项"}) assert response.status_code == 200 assert "risks_count_out_of_range" in " ".join(response.json()["data"]["warnings"]) @pytest.mark.asyncio async def test_launch_draft_job_runs_and_persists_completed_draft(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) _patch_llm(monkeypatch, _draft_payload()) service = pmflow_launch_service.PMFlowLaunchService(session, tenant_id=1) job = await service.create_draft_job( body=PMFlowLaunchDraftJobRequest(input="我想做一个 AI 会议纪要工具，帮助团队提取行动项", source="web"), user_id=int(user.id), ) assert job.status == "queued" await service.run_draft_job(job_id=job.job_id) completed = await service.get_draft_job(job_id=job.job_id, user_id=int(user.id)) assert completed.status == "completed" assert completed.draft_id is not None assert completed.draft is not None assert completed.draft.title == "AI 会议纪要助手" assert completed.changed_fields assert (await session.scalar(select(func.count(PMFlowLaunchDraftJob.id)))) == 1 assert (await session.scalar(select(func.count(PMFlowLaunchDraft.id)))) == 1 @pytest.mark.asyncio async def test_launch_draft_job_supersedes_active_job_when_user_supplements_input(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) _patch_llm(monkeypatch, _draft_payload()) service = pmflow_launch_service.PMFlowLaunchService(session, tenant_id=1) first = await service.create_draft_job( body=PMFlowLaunchDraftJobRequest(input="我想做一个 AI 会议纪要工具，帮助团队提取行动项", source="web"), user_id=int(user.id), ) second = await service.create_draft_job( body=PMFlowLaunchDraftJobRequest( input="我想做一个 AI 会议纪要工具，帮助团队提取行动项。补充：首版优先飞书接入。", source="web", mode="merge_supplement", revision_note="首版优先飞书接入", ), user_id=int(user.id), ) superseded = await service.get_draft_job(job_id=first.job_id, user_id=int(user.id)) assert superseded.status == "superseded" assert superseded.superseded_by_job_id == second.job_id assert second.status == "queued" async def _create_draft_via_api(session, user, monkeypatch): _patch_llm(monkeypatch, _draft_payload()) app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=user.tenant_id, role="MEMBER")) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.post("/pmflow-launch/drafts", json={"input": "我想做一个 AI 会议纪要工具"}) assert response.status_code == 200 return app, response.json()["data"]["draft_id"] @pytest.mark.asyncio async def test_confirm_launch_draft_creates_project_thread_message_truth_nodes_tasks_and_truth_state(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) app, draft_id = await _create_draft_via_api(session, user, monkeypatch) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: confirm = await client.post(f"/pmflow-launch/drafts/{draft_id}/confirm", json={"edited_draft": {}, "confirm": True}) truth = await client.get(f"/pmflow-os/projects/{confirm.json()['data']['project_id']}/truth-state") assert confirm.status_code == 200 data = confirm.json()["data"] assert data["project_id"] > 0 assert data["thread_id"] is not None assert data["thread_message_id"] is not None assert len(data["created_truth_node_ids"]) == 6 assert len(data["created_task_ids"]) == 3 assert (await session.scalar(select(func.count(Project.id)))) == 1 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(TruthNode.id)))) == 6 assert (await session.scalar(select(func.count(Task.id)))) == 3 truth_node = (await session.execute(select(TruthNode).where(TruthNode.node_type == "requirement"))).scalars().first() assert truth_node.content["metadata"]["launch_draft_id"] == draft_id assert truth_node.content["metadata"]["thread_message_id"] == data["thread_message_id"] task = (await session.execute(select(Task))).scalars().first() assert task.ai_breakdown_payload["launch_metadata"]["launch_draft_id"] == draft_id assert task.manual_overrides["launch_metadata"]["thread_message_id"] == data["thread_message_id"] truth_payload = truth.json()["data"] assert truth_payload["project"]["id"] == str(data["project_id"]) assert truth_payload["thread"]["id"] == str(data["thread_id"]) assert truth_payload["messages"][0]["content"] == "我想做一个 AI 会议纪要工具" assert truth_payload["requirements"] assert truth_payload["tasks"] @pytest.mark.asyncio async def test_confirm_launch_draft_is_idempotent(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) app, draft_id = await _create_draft_via_api(session, user, monkeypatch) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: first = await client.post(f"/pmflow-launch/drafts/{draft_id}/confirm", json={"edited_draft": {}, "confirm": True}) second = await client.post(f"/pmflow-launch/drafts/{draft_id}/confirm", json={"edited_draft": {}, "confirm": True}) assert first.status_code == 200 assert second.status_code == 200 assert first.json()["data"]["project_id"] == second.json()["data"]["project_id"] assert (await session.scalar(select(func.count(Project.id)))) == 1 assert (await session.scalar(select(func.count(TruthNode.id)))) == 6 assert (await session.scalar(select(func.count(Task.id)))) == 3 @pytest.mark.asyncio async def test_confirm_launch_draft_tenant_user_and_expiry_boundaries(db_env, monkeypatch) -> None: session = db_env owner = await _seed_user(session, tenant_id=1, username="owner") other = await _seed_user(session, tenant_id=1, username="other") app, draft_id = await _create_draft_via_api(session, owner, monkeypatch) other_app = _build_app(session, SimpleNamespace(id=other.id, tenant_id=1, role="MEMBER")) async with AsyncClient(transport=ASGITransport(app=other_app), base_url="http://test") as client: forbidden = await client.post(f"/pmflow-launch/drafts/{draft_id}/confirm", json={"edited_draft": {}, "confirm": True}) assert forbidden.status_code == 404 row = (await session.execute(select(PMFlowLaunchDraft).where(PMFlowLaunchDraft.draft_id == draft_id))).scalar_one() row.expires_at = now_cn_naive() - timedelta(minutes=1) await session.commit() 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: expired = await client.post(f"/pmflow-launch/drafts/{draft_id}/confirm", json={"edited_draft": {}, "confirm": True}) assert expired.status_code == 409 assert expired.json()["code"] == 409 assert "draft_expired" in expired.json()["message"] assert expired.json()["data"] is None @pytest.mark.asyncio async def test_confirm_launch_draft_rolls_back_on_critical_write_failure(db_env, monkeypatch) -> None: session = db_env user = await _seed_user(session) app, draft_id = await _create_draft_via_api(session, user, monkeypatch) async def _boom(*args, **kwargs): raise RuntimeError("task write failed") monkeypatch.setattr(pmflow_launch_service.PMFlowLaunchService, "_create_tasks", _boom) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: with pytest.raises(RuntimeError): await client.post(f"/pmflow-launch/drafts/{draft_id}/confirm", json={"edited_draft": {}, "confirm": True}) assert (await session.scalar(select(func.count(Project.id)))) == 0 assert (await session.scalar(select(func.count(TruthNode.id)))) == 0 @pytest.mark.asyncio async def test_project_create_duplicate_title_defaults_to_conflict(db_env) -> None: session = db_env user = await _seed_user(session) app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="OWNER")) payload = {"title": "同名项目", "description": "普通创建默认不吞掉重复标题。", "status": "active"} async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: first = await client.post("/projects/", json=payload) second = await client.post("/projects/", json=payload) assert first.status_code == 200 assert second.status_code == 409 assert second.json()["detail"] == "project title already exists" assert (await session.scalar(select(func.count(Project.id)))) == 1 @pytest.mark.asyncio async def test_projects_list_and_detail_are_scoped_by_role(db_env) -> None: session = db_env manager = await _seed_user(session, username="manager") builder = await _seed_user(session, username="builder") outsider = await _seed_user(session, username="outsider") owned_project = Project( tenant_id=1, name="Builder Owned", title="Builder Owned", description="Owned by builder", owner_id=builder.id, status="active", assigned_role="BUILDER", signal_count=0, ) member_project = Project( tenant_id=1, name="Builder Member", title="Builder Member", description="Builder is assigned", owner_id=outsider.id, status="active", assigned_role="BUILDER", signal_count=0, ) hidden_project = Project( tenant_id=1, name="Hidden Project", title="Hidden Project", description="No builder access", owner_id=outsider.id, status="active", assigned_role="BUILDER", signal_count=0, ) session.add_all([owned_project, member_project, hidden_project]) await session.flush() session.add(UserProject(tenant_id=1, user_id=builder.id, project_id=member_project.id)) await session.commit() manager_app = _build_app(session, SimpleNamespace(id=manager.id, tenant_id=1, role="OWNER")) builder_app = _build_app(session, SimpleNamespace(id=builder.id, tenant_id=1, role="BUILDER")) async with AsyncClient(transport=ASGITransport(app=manager_app), base_url="http://test") as client: manager_response = await client.get("/projects/?summary=1") assert manager_response.status_code == 200 assert {item["title"] for item in manager_response.json()["projects"]} == { "Builder Owned", "Builder Member", "Hidden Project", } async with AsyncClient(transport=ASGITransport(app=builder_app), base_url="http://test") as client: builder_response = await client.get("/projects/?summary=1") allowed_detail = await client.get(f"/projects/{member_project.id}") hidden_detail = await client.get(f"/projects/{hidden_project.id}") assert builder_response.status_code == 200 assert {item["title"] for item in builder_response.json()["projects"]} == { "Builder Owned", "Builder Member", } assert allowed_detail.status_code == 200 assert allowed_detail.json()["title"] == "Builder Member" assert hidden_detail.status_code == 404