Add live stage-code golden tests

pytest.ini

@@ -2,3 +2,4 @@
 pythonpath = .
 markers =
     integration: marks tests that need access to external services
+    golden_live: runs golden conversation cases against a live API and FastGPT

test/golden/README.md

@@ -0,0 +1,32 @@
+# Stage-code golden tests
+
+These tests call the running FastAPI service and its configured FastGPT
+workflow. Each conversation step asserts only the `nextStageCode` carried by
+the SSE `stage_code` event.
+
+Edit `cases.json` to add or update conversation cases:
+
+```json
+{
+  "name": "example",
+  "clientMode": "browser_addon",
+  "steps": [
+    {
+      "text": "继续",
+      "expect": {
+        "stageCode": "browser_addon.1002"
+      }
+    }
+  ]
+}
+```
+
+Start the API, then run:
+
+```bash
+RUN_GOLDEN=1 \
+GOLDEN_BASE_URL=http://127.0.0.1:8000 \
+pytest -m golden_live test/golden -v
+```
+
+Without `RUN_GOLDEN=1`, the cases are skipped and do not make network calls.

test/golden/cases.json

@@ -0,0 +1,26 @@
+[
+  {
+    "name": "direct_initial_stage",
+    "clientMode": "direct",
+    "steps": [
+      {
+        "text": "hi",
+        "expect": {
+          "stageCode": "1001"
+        }
+      }
+    ]
+  },
+  {
+    "name": "browser_addon_initial_stage",
+    "clientMode": "browser_addon",
+    "steps": [
+      {
+        "text": "hi",
+        "expect": {
+          "stageCode": "browser_addon.1001"
+        }
+      }
+    ]
+  }
+]

test/golden/test_stage_code_golden.py

@@ -0,0 +1,112 @@
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import uuid4
+
+import httpx
+import pytest
+
+
+CASES_PATH = Path(__file__).with_name("cases.json")
+RUN_GOLDEN = os.getenv("RUN_GOLDEN") == "1"
+BASE_URL = os.getenv("GOLDEN_BASE_URL", "http://127.0.0.1:8000").rstrip("/")
+
+
+def load_cases() -> list[dict]:
+    return json.loads(CASES_PATH.read_text(encoding="utf-8"))
+
+
+def parse_stage_codes(response: httpx.Response) -> list[str]:
+    stage_codes: list[str] = []
+    event_name = ""
+    data_lines: list[str] = []
+
+    def consume_event() -> None:
+        nonlocal event_name, data_lines
+        if event_name == "stage_code" and data_lines:
+            payload = json.loads("\n".join(data_lines))
+            stage_codes.append(payload["nextStageCode"])
+        event_name = ""
+        data_lines = []
+
+    for line in response.iter_lines():
+        if not line:
+            consume_event()
+        elif line.startswith("event:"):
+            event_name = line.removeprefix("event:").strip()
+        elif line.startswith("data:"):
+            data_lines.append(line.removeprefix("data:").strip())
+
+    consume_event()
+    return stage_codes
+
+
+def run_chat_step(
+    client: httpx.Client,
+    *,
+    session_id: str,
+    client_mode: str,
+    text: str,
+) -> list[str]:
+    payload = {
+        "sessionId": session_id,
+        "timeStamp": datetime.now(timezone.utc).isoformat(),
+        "text": text,
+        "clientMode": client_mode,
+        "needFormUpdate": False,
+        "useTextChunk": False,
+    }
+
+    with client.stream(
+        "POST",
+        "/chat",
+        params={"stream": "true"},
+        json=payload,
+    ) as response:
+        response.raise_for_status()
+        return parse_stage_codes(response)
+
+
+def cleanup_session(client: httpx.Client, session_id: str) -> None:
+    try:
+        client.request(
+            "DELETE",
+            "/delete_session",
+            json={
+                "sessionId": session_id,
+                "timeStamp": datetime.now(timezone.utc).isoformat(),
+            },
+        )
+    except httpx.HTTPError:
+        # A unique session ID is used for every run, so cleanup failure must not
+        # hide the stage-code assertion that the golden test is responsible for.
+        pass
+
+
+@pytest.mark.golden_live
+@pytest.mark.skipif(
+    not RUN_GOLDEN,
+    reason="Set RUN_GOLDEN=1 to run live golden cases",
+)
+@pytest.mark.parametrize("case", load_cases(), ids=lambda case: case["name"])
+def test_expected_stage_codes(case):
+    session_id = f"golden-{case['name']}-{uuid4().hex}"
+
+    with httpx.Client(base_url=BASE_URL, timeout=90.0) as client:
+        try:
+            for step_number, step in enumerate(case["steps"], start=1):
+                actual_codes = run_chat_step(
+                    client,
+                    session_id=session_id,
+                    client_mode=case["clientMode"],
+                    text=step["text"],
+                )
+                expected_code = step["expect"]["stageCode"]
+
+                assert actual_codes == [expected_code], (
+                    f"case={case['name']} step={step_number} "
+                    f"expected={[expected_code]} actual={actual_codes}"
+                )
+        finally:
+            cleanup_session(client, session_id)