"""垂直切片测试：幂等原子事务、IDOR fail-closed、CSRF、SSE 与回滚。""" import json from httpx import AsyncClient from sqlalchemy import select from app import models from tests.conftest import login MESSAGE_BODY = {"content": [{"type": "text", "text": "你好，魂镜"}], "start_run": True} async def _create_conversation(client: AsyncClient, headers: dict) -> str: resp = await client.post( "/v1/conversations", json={"title": "切片"}, headers=headers ) assert resp.status_code == 201, resp.text return resp.json()["id"] async def _post_message(client: AsyncClient, headers: dict, cid: str, key: str) -> object: return await client.post( f"/v1/conversations/{cid}/messages", json=MESSAGE_BODY, headers={**headers, "Idempotency-Key": key}, ) class TestMessagePipeline: async def test_create_and_post_message(self, client: AsyncClient) -> None: headers = await login(client, "alice@example.com") cid = await _create_conversation(client, headers) resp = await _post_message(client, headers, cid, "idem-1") assert resp.status_code == 201, resp.text body = resp.json() assert body["message"]["id"].startswith("message_") assert body["message"]["content"][0]["id"].startswith("block_") assert body["run"]["status"] == "completed" # 空状态闭环 async def test_idempotent_replay_returns_original(self, client: AsyncClient) -> None: headers = await login(client, "bob@example.com") cid = await _create_conversation(client, headers) first = await _post_message(client, headers, cid, "idem-replay") assert first.status_code == 201 second = await _post_message(client, headers, cid, "idem-replay") assert second.status_code == 200 # 重放返回原资源 assert second.json()["message"]["id"] == first.json()["message"]["id"] async def test_same_key_different_body_conflicts(self, client: AsyncClient) -> None: headers = await login(client, "carol@example.com") cid = await _create_conversation(client, headers) await _post_message(client, headers, cid, "idem-conflict") resp = await client.post( f"/v1/conversations/{cid}/messages", json={"content": [{"type": "text", "text": "不同内容"}], "start_run": True}, headers={**headers, "Idempotency-Key": "idem-conflict"}, ) assert resp.status_code == 409 async def test_missing_idempotency_key_rejected(self, client: AsyncClient) -> None: headers = await login(client, "dave@example.com") cid = await _create_conversation(client, headers) resp = await client.post( f"/v1/conversations/{cid}/messages", json=MESSAGE_BODY, headers=headers ) assert resp.status_code == 422 async def test_unknown_block_type_rejected(self, client: AsyncClient) -> None: headers = await login(client, "erin@example.com") cid = await _create_conversation(client, headers) resp = await client.post( f"/v1/conversations/{cid}/messages", json={"content": [{"type": "image", "file_id": "x", "alt": "y"}]}, headers={**headers, "Idempotency-Key": "idem-bad"}, ) assert resp.status_code == 422 async def test_cross_domain_conversation_returns_404(self, client: AsyncClient) -> None: alice = await login(client, "ida@example.com") cid = await _create_conversation(client, alice) mallory = await login(client, "mallory@example.com") resp = await _post_message(client, mallory, cid, "idem-idor") assert resp.status_code == 404 # fail-closed，不泄露存在性 async def test_csrf_token_required_for_state_change(self, client: AsyncClient) -> None: await login(client, "frank@example.com") resp = await client.post("/v1/conversations", json={"title": "无 CSRF"}) assert resp.status_code == 403 async def test_unauthenticated_rejected(self, client: AsyncClient) -> None: resp = await client.post("/v1/conversations", json={"title": "匿名"}) assert resp.status_code == 401 class TestRunSse: async def test_sse_snapshot_and_terminal(self, client: AsyncClient) -> None: headers = await login(client, "grace@example.com") cid = await _create_conversation(client, headers) resp = await _post_message(client, headers, cid, "idem-sse") run_id = resp.json()["run"]["id"] sse = await client.get(f"/v1/runs/{run_id}/events") assert sse.status_code == 200 assert sse.headers["content-type"].startswith("text/event-stream") lines = [ln for ln in sse.text.splitlines() if ln.startswith("data: ")] events = [json.loads(ln.removeprefix("data: ")) for ln in lines] assert events[0]["type"] == "run.status.changed" assert events[0]["data"]["status"] == "completed" assert events[-1]["type"] == "message.completed" async def test_cross_domain_run_returns_404(self, client: AsyncClient) -> None: headers = await login(client, "heidi@example.com") cid = await _create_conversation(client, headers) resp = await _post_message(client, headers, cid, "idem-sse2") run_id = resp.json()["run"]["id"] await login(client, "eve@example.com") sse = await client.get(f"/v1/runs/{run_id}/events") assert sse.status_code == 404 class TestAtomicity: async def test_failure_leaves_no_orphans(self, client: AsyncClient, db) -> None: from app.conversations import service headers = await login(client, "ivan@example.com") cid = await _create_conversation(client, headers) class BoomRun: def __init__(self, *args, **kwargs): raise RuntimeError("注入失败：Run 写入阶段崩溃") original = service.Run service.Run = BoomRun # type: ignore[assignment] try: resp = await _post_message(client, headers, cid, "idem-atomic") assert resp.status_code == 500 finally: service.Run = original # type: ignore[assignment] async with db() as session: # 会话创建的合法审计/Outbox 保留；消息相关记录必须为零 assert (await session.scalars(select(models.Message))).all() == [] assert (await session.scalars(select(models.Run))).all() == [] audits = (await session.scalars(select(models.AuditEvent))).all() assert [a for a in audits if a.aggregate_type == "message"] == [] outbox = (await session.scalars(select(models.OutboxEvent))).all() assert [e for e in outbox if e.aggregate_type == "message"] == [] assert (await session.scalars(select(models.IdempotencyRecord))).all() == []