浅念
128 lines
5.3 KiB
Python
128 lines
5.3 KiB
Python
"""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 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 test_no_brand_copy_left_in_client_files():
|
|
"""等价迁移锚:三份客户端文件不再内嵌任何租户品牌值(联系号/邮箱/品牌名/
|
|
隐私 URL),品牌只经 /api/public/brand 下发。"""
|
|
for rel in ("widget.js", "js/chat.js", "chat.html"):
|
|
src = _read(rel)
|
|
assert "HESUPIPE" not in src, f"{rel} 残留品牌名"
|
|
assert "8615937108317" not in src and "8619037179250" not in src, f"{rel} 残留联系号"
|
|
assert "hesuu.com" not in src and "borderlesscar.com" not in src, f"{rel} 残留邮箱域名"
|
|
assert "hesupipe.com" not in src, f"{rel} 残留隐私 URL 域名"
|
|
|
|
|
|
# ---- 服务端配套:会话打站点标 ---------------------------------------------------
|
|
|
|
|
|
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)
|