test: harden integration smoke coverage

scripts/smoke_fnos_storyforge_lan.sh

@@ -47,17 +47,17 @@ asr_url_file="$tmp_dir/asr-url.txt"
 asr_wav="$tmp_dir/asr.wav"
 asr_result="$tmp_dir/asr.json"
 
-echo "[1/7] check fnOS web"
+echo "[1/8] check fnOS web"
 curl_fetch "$WEB_URL/" >"$index_file"
 rg -Fq "StoryForge" "$index_file"
 echo "web ok"
 
-echo "[2/7] check runtime config"
+echo "[2/8] check runtime config"
 curl_fetch "$WEB_URL/assets/storyforge-runtime-config.js" >"$runtime_file"
 rg -Fq "$BACKEND_URL" "$runtime_file"
 echo "runtime config ok"
 
-echo "[3/7] check collector healthz"
+echo "[3/8] check collector healthz"
 curl_fetch "$BACKEND_URL/healthz" >"$health_file"
 python3 -c '
 import json, pathlib, sys
@@ -177,6 +177,25 @@ if not cutvideo.get("reachable"):
     raise SystemExit("cutvideo is not reachable")
 if not cutvideo.get("supports_uploads"):
     raise SystemExit("cutvideo uploads are not available")
+if cutvideo.get("deployment_label") != "NAS 隧道":
+    raise SystemExit(f"unexpected cutvideo deployment label: {cutvideo.get('deployment_label')!r}")
+n8n = payload.get("n8n") or {}
+if not n8n.get("reachable") or n8n.get("deployment_label") != "服务器":
+    raise SystemExit(f"unexpected n8n status: {n8n!r}")
+huobao = payload.get("huobao") or {}
+if not huobao.get("reachable") or huobao.get("deployment_label") != "服务器":
+    raise SystemExit(f"unexpected huobao status: {huobao!r}")
+asr = payload.get("asr") or {}
+if not asr.get("reachable") or asr.get("deployment_label") != "Windows":
+    raise SystemExit(f"unexpected asr status: {asr!r}")
+if not str(asr.get("active_device") or "").strip():
+    raise SystemExit("asr active_device missing")
+live_recorder = payload.get("live_recorder") or {}
+if not live_recorder.get("reachable") or live_recorder.get("deployment_label") != "NAS":
+    raise SystemExit(f"unexpected live recorder status: {live_recorder!r}")
+local_model = payload.get("local_model") or {}
+if str(local_model.get("error") or "") != "not_configured":
+    raise SystemExit(f"local_model should be not_configured, got {local_model.get('error')!r}")
 print("integrations ok")
 ' "$integrations_file" "$CUTVIDEO_URL"
 

scripts/smoke_public_storyforge.sh

@@ -26,11 +26,14 @@ health_file="$tmp_dir/health.json"
 html_file="$tmp_dir/index.html"
 js_file="$tmp_dir/app.js"
 openapi_file="$tmp_dir/openapi.json"
+session_file="$tmp_dir/session.json"
+token_file="$tmp_dir/token.txt"
+integrations_file="$tmp_dir/integrations.json"
 asr_url_file="$tmp_dir/asr-url.txt"
 asr_wav="$tmp_dir/asr.wav"
 asr_result="$tmp_dir/asr.json"
 
-echo "[1/4] check public healthz"
+echo "[1/7] check public healthz"
 curl_fetch "$BASE_URL/healthz" >"$health_file"
 python3 - "$health_file" "$asr_url_file" <<'PY'
 import json
@@ -46,7 +49,65 @@ pathlib.Path(sys.argv[2]).write_text(asr_url, encoding="utf-8")
 print("healthz ok")
 PY
 
-echo "[2/5] check public ASR transcribe"
+echo "[2/7] check public auto-session"
+curl_fetch -X POST "$BASE_URL/v2/auth/auto-session" \
+  -H 'content-type: application/json' \
+  -d '{}' >"$session_file"
+python3 - "$session_file" "$token_file" <<'PY'
+import json
+import pathlib
+import sys
+
+payload = json.loads(pathlib.Path(sys.argv[1]).read_text())
+token = str(payload.get("token") or "")
+mode = str(payload.get("mode") or "")
+username = str((payload.get("account") or {}).get("username") or "")
+if not token:
+    raise SystemExit("auto-session did not return token")
+if mode != "auto":
+    raise SystemExit(f"unexpected mode: {mode!r}")
+if not username:
+    raise SystemExit("auto-session returned empty username")
+pathlib.Path(sys.argv[2]).write_text(token, encoding="utf-8")
+print(f"auto-session ok: {username}")
+PY
+token="$(cat "$token_file")"
+
+echo "[3/7] check public integrations health"
+curl_fetch "$BASE_URL/v2/integrations/health" \
+  -H "Authorization: Bearer $token" >"$integrations_file"
+python3 - "$integrations_file" <<'PY'
+import json
+import pathlib
+import sys
+
+payload = json.loads(pathlib.Path(sys.argv[1]).read_text())
+required = {
+    "n8n": "服务器",
+    "huobao": "服务器",
+    "asr": "Windows",
+    "live_recorder": "NAS",
+}
+for key, label in required.items():
+    detail = payload.get(key) or {}
+    if not detail.get("reachable"):
+        raise SystemExit(f"{key} is not reachable")
+    if detail.get("deployment_label") != label:
+        raise SystemExit(f"{key} deployment label mismatch: {detail.get('deployment_label')!r}")
+cutvideo = payload.get("cutvideo") or {}
+if not cutvideo.get("reachable"):
+    raise SystemExit("cutvideo is not reachable")
+if str(cutvideo.get("deployment_label") or "") not in {"Windows", "NAS 隧道"}:
+    raise SystemExit(f"cutvideo deployment label mismatch: {cutvideo.get('deployment_label')!r}")
+local_model = payload.get("local_model") or {}
+if str(local_model.get("error") or "") != "not_configured":
+    raise SystemExit(f"local_model should be not_configured, got {local_model.get('error')!r}")
+if str((payload.get("asr") or {}).get("active_device") or "").strip() == "":
+    raise SystemExit("asr active_device missing")
+print("integrations ok")
+PY
+
+echo "[4/7] check public ASR transcribe"
 python3 - "$asr_wav" <<'PY'
 import math
 import struct
@@ -88,19 +149,19 @@ PY
     ;;
 esac
 
-echo "[3/5] check public index"
+echo "[5/7] check public index"
 curl_fetch "$BASE_URL/" >"$html_file"
 rg -q "StoryForge" "$html_file"
 echo "index ok"
 
-echo "[4/5] check deployed web bundle"
+echo "[6/7] check deployed web bundle"
 curl_fetch "$BASE_URL/assets/app.js" >"$js_file"
 rg -q "select-platform" "$js_file"
 rg -q "trackingCursorMap" "$js_file"
 rg -q "renderPlatformSwitchChips" "$js_file"
 echo "bundle ok"
 
-echo "[5/5] check public openapi routes"
+echo "[7/7] check public openapi routes"
 curl_fetch "$BASE_URL/openapi.json" >"$openapi_file"
 for route in \
   '"/v2/xiaohongshu/accounts"' \

tests/test_production_baseline.py

@@ -247,6 +247,22 @@ class ProductionBaselineTests(unittest.TestCase):
             "/transcribe",
             "/api/bootstrap",
             "19181",
+            "deployment_label",
+            "not_configured",
+        ]:
+            self.assertIn(expected, content)
+
+    def test_repo_contains_public_smoke_script(self) -> None:
+        script_path = ROOT / "scripts" / "smoke_public_storyforge.sh"
+        self.assertTrue(script_path.exists(), str(script_path))
+        content = script_path.read_text(encoding="utf-8")
+        for expected in [
+            "/healthz",
+            "/v2/auth/auto-session",
+            "/v2/integrations/health",
+            "deployment_label",
+            "not_configured",
+            "active_device",
         ]:
             self.assertIn(expected, content)