"""Task 10 静态+路由回归网:客户端 key 租户化 / brand fetch 机制 / 会话站点标。 - 静态断言仿 test_theme_static 风格(Path(__file__) 锚定仓库静态文件); - 品牌值已自 widget.js/chat.js/chat.html 迁入 seed_profiles/*/tenant.json (断言语义等价迁移:锚 seed_profiles 值或锚 fetch 机制); - 服务端配套:WS init 的 site 与 REST ?site= 记入 client_info["site"]。 """ import asyncio import json from pathlib import Path from urllib.parse import urlparse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.pool import NullPool from app.engine.models import Conversation REPO = Path(__file__).resolve().parents[1] STATIC = REPO / "static" SEED = REPO / "seed_profiles" def _read(rel): return (STATIC / rel).read_text(encoding="utf-8") # ---- 客户端 key 租户化 --------------------------------------------------------- def test_widget_keys_namespaced_by_slug(): src = _read("widget.js") assert "cc-widget-host-" in src # host id 拼租户 slug（模板串形式） assert "widget_invite_dismissed_" in src # sessionStorage key 同 def test_chat_session_key_namespaced_by_slug(): src = _read("js/chat.js") assert "cc_session_id_" in src # 独立页 localStorage 会话 key 同 # ---- brand fetch 机制(迁移后品牌事实源 = seed_profiles)------------------------ def test_widget_fetches_brand_with_site_param(): src = _read("widget.js") assert "/api/public/brand" in src # 启动即取品牌 assert "?site=" in src # data-site 进 query(空域名站点接入) assert "no-store" in src # 品牌热更新即时生效 def test_widget_ws_init_carries_site(): src = _read("widget.js") assert "site: siteAttr" in src # WS init 消息带 site def test_brand_values_live_in_seed_profiles(): """等价迁移锚:原锚 widget.js 硬编码品牌文案的用例改锚 seed_profiles 值 (品牌事实源自 Task 10 起迁入 tenant.json)。""" jc = json.loads((SEED / "jiancai" / "tenant.json").read_text(encoding="utf-8")) b = jc["config"]["brand"] assert b["title"] == "HESUPIPE Support" assert b["whatsapp"] == "+8615937108317" assert b["email"] == "hesu@hesuu.com" assert b["privacy_url"] == "https://hesupipe.com/privacy-policy" assert b["topics"] == ["Get Quotation", "Pipe Specifications", "Delivery & Shipping"] es = next(s for s in jc["sites"] if s["slug"] == "es") assert es["brand"]["title"] == "HESUPIPE Soporte" auto = json.loads((SEED / "auto" / "tenant.json").read_text(encoding="utf-8")) ab = auto["config"]["brand"] assert ab["title"] == "HESU Support" assert ab["whatsapp"] == "+8619037179250" assert ab["topics"] == ["Get Quotation", "Distributor Cooperation"] def _brand_identity_values(): """复审动态化: 从全部 seed_profiles/*/tenant.json(租户默认门面 + 各站点 覆盖)收集品牌身份值作禁词表——第三租户落地即自动纳管,无需改本测试。 只收身份字段(title/whatsapp/email/privacy_url/welcome/topics); color 与 avatar 文件名是中性 UI 默认(widget.js 的 DEFAULT_COLOR= #066fc0 恰等于 jiancai 品牌色、DEFAULT_AVATAR_PATH 兜底文件同名), 收进来会误伤合法兜底。privacy_url 额外拆主机名(含去 www. 裸域名), whatsapp 额外收去 + 变体(客户端硬编码裸号也算残留)。""" values = set() for tenant_file in sorted(SEED.glob("*/tenant.json")): cfg = json.loads(tenant_file.read_text(encoding="utf-8")) brands = [cfg.get("config", {}).get("brand") or {}] brands += [s.get("brand") or {} for s in cfg.get("sites", [])] for b in brands: for key in ("title", "whatsapp", "email", "privacy_url", "welcome"): v = str(b.get(key) or "").strip() if v: values.add(v) values.add(v.lstrip("+")) for t in b.get("topics") or []: if str(t).strip(): values.add(str(t).strip()) host = urlparse(str(b.get("privacy_url") or "")).hostname or "" if host: values.add(host) if host.startswith("www."): values.add(host[4:]) assert values, "禁词表收集为空——seed_profiles 结构变了?" return values def test_no_brand_copy_left_in_client_files(): """等价迁移锚:四份客户端文件不再内嵌任何租户品牌值(联系号/邮箱/品牌名/ 隐私 URL/welcome/topics),品牌只经 /api/public/brand 下发。禁词表动态 收集自 seed_profiles(见 _brand_identity_values)。""" forbidden = _brand_identity_values() for rel in ("widget.js", "js/chat.js", "chat.html", "widget-test.html"): src = _read(rel) for v in forbidden: assert v not in src, f"{rel} 残留品牌值: {v!r}" # ---- 服务端配套:会话打站点标 --------------------------------------------------- def _read_client_info(db_path: Path, session_id: str) -> dict: """直读租户库验证 client_info——新开 NullPool engine(不复用 TestClient 循环里的 Registry engine,跨循环复用会死锁;同文件 WAL 多连接读安全)。""" async def q(): engine = create_async_engine( f"sqlite+aiosqlite:///{db_path.as_posix()}", poolclass=NullPool) try: async with AsyncSession(engine) as s: conv = (await s.execute( select(Conversation).where( Conversation.session_id == session_id))).scalar_one() return dict(conv.client_info or {}) finally: await engine.dispose() return asyncio.run(q()) def test_ws_init_site_recorded_in_client_info(app_client, tmp_path): with app_client.websocket_connect("/ws/chat") as ws: ws.send_text(json.dumps({"source": "widget", "site": "es"})) sid = ws.receive_json()["session_id"] assert _read_client_info(tmp_path / "jiancai.db", sid).get("site") == "es" def test_ws_init_without_site_no_tag(app_client, tmp_path): with app_client.websocket_connect("/ws/chat") as ws: ws.send_text(json.dumps({"source": "widget"})) sid = ws.receive_json()["session_id"] assert "site" not in _read_client_info(tmp_path / "jiancai.db", sid) def test_rest_session_site_query_recorded(app_client, tmp_path): sid = app_client.post("/api/chat/session?site=es", json={}).json()["session_id"] assert _read_client_info(tmp_path / "jiancai.db", sid).get("site") == "es" def test_rest_session_without_site_no_tag(app_client, tmp_path): sid = app_client.post("/api/chat/session", json={}).json()["session_id"] assert "site" not in _read_client_info(tmp_path / "jiancai.db", sid) # ---- 复审: 专用 site 通道优先级(WS 与 REST 对齐) ------------------------------- def test_ws_dedicated_site_wins_over_self_reported(app_client, tmp_path): """WS init 同时带专用 site 字段与 client_info 自报 site 时,专用字段赢 (与 REST ?site= 覆盖 req.client_info 的优先级对齐——专用通道是嵌入 代码显式声明,可信度高于任意自报 dict)。""" with app_client.websocket_connect("/ws/chat") as ws: ws.send_text(json.dumps({"source": "widget", "site": "es", "client_info": {"site": "main"}})) sid = ws.receive_json()["session_id"] assert _read_client_info(tmp_path / "jiancai.db", sid).get("site") == "es" def test_rest_site_query_wins_over_body_client_info(app_client, tmp_path): # 同上语义的 REST 钉子: ?site= 覆盖请求体 client_info 自报的 site sid = app_client.post( "/api/chat/session?site=es", json={"client_info": {"site": "main"}} ).json()["session_id"] assert _read_client_info(tmp_path / "jiancai.db", sid).get("site") == "es"