""" test_integration_seams.py — 汇合联调接缝验证（无 Docker，不依赖 Redis/DB） 验证三条核心接缝： 1. AIE→BE 接缝：pipeline_steps / pipeline_io 数据流契约 2. BE→FE 接缝：SSE 事件字段与 generationStore 期望字段对齐 3. FE→BE 接缝：API 请求 DTO 与 BE 路由入参对齐 原则：纯 Python/结构断言，不联网，不起服务。 联调真实运行需 docker compose up + 真实 API Key。 """ import sys, os sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) import pytest import json # ═══════════════════════════════════════════════════════════════ # Seam 1 — AIE→BE: pipeline_steps 数据流 # ═══════════════════════════════════════════════════════════════ class TestAiePipelineStepsContract: """pipeline_steps.build_product_dict 产出符合 AI 引擎 dict 契约""" def _make_mock_product(self): """模拟 ORM Product 对象（duck typing）""" class P: name = "倍分子素颜霜" category = "美妆护肤" selling_points = '["轻薄不厚重","水润自然"]' style_tone = "素人分享风" text_angles = '["痛点切入","场景型"]' custom_prompt = "" image_path = None # 前端图片上传完成前为空 return P() def test_build_product_dict_keys(self): from app.workers.pipeline_steps import build_product_dict d = build_product_dict(self._make_mock_product()) for key in ("name", "category", "selling_points", "style_tone", "text_angles", "custom_prompt", "image_path"): assert key in d, f"build_product_dict 缺字段: {key}" def test_build_product_dict_lists_parsed(self): from app.workers.pipeline_steps import build_product_dict d = build_product_dict(self._make_mock_product()) assert isinstance(d["selling_points"], list) assert isinstance(d["text_angles"], list) def test_build_product_dict_no_key_field(self): """绝不含任何 key/token 相关字段（基石B）""" from app.workers.pipeline_steps import build_product_dict d = build_product_dict(self._make_mock_product()) for forbidden in ("api_key", "key", "token", "secret", "encrypted"): assert forbidden not in d, f"product_dict 不应含 {forbidden}" # ═══════════════════════════════════════════════════════════════ # Seam 2 — BE→FE: SSE 事件字段对齐 # ═══════════════════════════════════════════════════════════════ class TestSseEventFieldAlignment: """BE 推的 SSE 事件字段与 FE useSse handleEvent 期望的字段一一对应""" # FE useSse.ts 中 handleEvent 使用的字段（从 FE 代码提取） EXPECTED_FIELDS = { "task_started": {"total_text", "total_image"}, "text_progress": {"done", "total"}, "text_candidate": {"candidate_id", "angle_label", "content", "score"}, "image_progress": {"done", "total"}, "image_candidate": {"candidate_id", "strategy", "url", "role"}, "batch_failed": {"batch", "reason", "retryable"}, "flywheel_injected": {"recent_preference", "reject_reasons"}, } # BE pipeline_io 实际推的事件字段（从代码提取） ACTUAL_FIELDS = { "task_started": { "task_id": 1, "total_text": 5, "total_image": 3, }, "text_progress": { "done": 1, "total": 5, }, "text_candidate": { "candidate_id": 1, "angle_label": "场景型", "content": "测试正文", "score": 88, }, "image_progress": { "done": 1, "total": 3, }, "image_candidate": { "candidate_id": 1, "strategy": "A", "url": "/uploads/ws/1/1/01_hook.jpg", "role": "hook", }, "batch_failed": { "batch": "hook", "reason": "502", "retryable": True, }, "flywheel_injected": { "recent_preference": "最近偏好角度：场景型", "reject_reasons": [], }, } def test_all_fe_required_fields_present(self): """FE 期望的每个字段都在 BE 实际推的 data 中""" for event_type, required_fields in self.EXPECTED_FIELDS.items(): actual = self.ACTUAL_FIELDS.get(event_type, {}) for field in required_fields: assert field in actual, ( f"SSE 事件 [{event_type}] 缺字段 [{field}]，FE 会拿到 undefined" ) def test_no_extra_pii_in_sse_events(self): """SSE 事件不含 key/token/password 等敏感字段""" for event_type, data in self.ACTUAL_FIELDS.items(): for field in data: assert "key" not in field.lower() and "token" not in field.lower(), ( f"SSE 事件 [{event_type}] 含敏感字段 [{field}]" ) # ═══════════════════════════════════════════════════════════════ # Seam 3 — FE→BE: CreateTaskRequest DTO 对齐 # ═══════════════════════════════════════════════════════════════ class TestCreateTaskRequestContract: """FE tasks/new/page.tsx 发的请求体与 BE CreateTaskRequest 契约对齐（独立 DTO 验证，不依赖 motor）""" @pytest.fixture(autouse=True) def _stub_motor(self, monkeypatch): """database.py 顶层 import motor 在测试环境可能缺失，mock 掉避免 ModuleNotFoundError""" import sys import types if "motor" not in sys.modules: motor_mod = types.ModuleType("motor") motor_async = types.ModuleType("motor.motor_asyncio") motor_async.AsyncIOMotorClient = object sys.modules["motor"] = motor_mod sys.modules["motor.motor_asyncio"] = motor_async def _get_dto(self): from app.api.v1.tasks import CreateTaskRequest return CreateTaskRequest def test_create_task_required_fields(self): Req = self._get_dto() req = Req(product_id=1, theme="今日主题", text_count=8, image_count=5, track="ai") assert req.product_id == 1 assert req.theme == "今日主题" assert req.track == "ai" def test_count_validation(self): import pydantic Req = self._get_dto() with pytest.raises(pydantic.ValidationError): Req(product_id=1, text_count=0, track="ai") # 0 < 1 with pytest.raises(pydantic.ValidationError): Req(product_id=1, text_count=21, track="ai") # 21 > 20 def test_track_validation(self): import pydantic Req = self._get_dto() with pytest.raises(pydantic.ValidationError): Req(product_id=1, track="auto") # 非法 track def test_import_track_no_key_check(self): """轨B import 字段校验通过（业务层 key 检查在 service，不在 DTO）""" Req = self._get_dto() req = Req(product_id=1, track="import") assert req.track == "import" # ═══════════════════════════════════════════════════════════════ # Seam 4 — SSE ticket 流程 # ═══════════════════════════════════════════════════════════════ class TestSseTicket: """ticket 签发/消费/防重放逻辑（用 mock redis）""" def _make_mock_redis(self): store: dict = {} class MockRedis: def setex(self, key, ttl, value): store[key] = value def getdel(self, key): return store.pop(key, None) return MockRedis(), store def test_issue_and_consume(self): from app.core.sse_ticket import issue_ticket, consume_ticket r, _ = self._make_mock_redis() ticket = issue_ticket(r, task_id=42, workspace_id=7) result = consume_ticket(r, ticket) assert result == (42, 7) def test_consume_once_only(self): from app.core.sse_ticket import issue_ticket, consume_ticket r, _ = self._make_mock_redis() ticket = issue_ticket(r, task_id=1, workspace_id=1) consume_ticket(r, ticket) # 第一次消费 result = consume_ticket(r, ticket) # 第二次 → None assert result is None def test_invalid_ticket_returns_none(self): from app.core.sse_ticket import consume_ticket r, _ = self._make_mock_redis() assert consume_ticket(r, "fake-ticket") is None def test_empty_ticket_returns_none(self): from app.core.sse_ticket import consume_ticket r, _ = self._make_mock_redis() assert consume_ticket(r, "") is None # ═══════════════════════════════════════════════════════════════ # Seam 5 — SSE sse_utils 格式化 # ═══════════════════════════════════════════════════════════════ class TestSseFormat: """format_sse 输出符合 text/event-stream 规范，FE EventSource 能解析""" def test_format_contains_event_and_data(self): from app.utils.sse_utils import format_sse out = format_sse("text_candidate", {"candidate_id": 1, "score": 92}, seq=3) assert "event: text_candidate" in out assert "data:" in out assert "id: 3" in out def test_format_ends_with_double_newline(self): from app.utils.sse_utils import format_sse out = format_sse("task_done", {"status": "pending_selection"}) assert out.endswith("\n\n") def test_data_is_valid_json(self): from app.utils.sse_utils import format_sse out = format_sse("heartbeat", {"ts": 1234567.0}) data_line = [l for l in out.split("\n") if l.startswith("data:")][0] json.loads(data_line[len("data:"):].strip()) # 不应抛异常