aps-agent/tests/golden/test_web_demo_seed.py

107 lines
4.1 KiB
Python
Raw Normal View History

# ============================================================
# Web 端演示数据懒播种黄金测试
# 口径:仅 APS_MODE=web + APS_WEB_DEMO_SEED=1 + 租户 0 项目时播种;
# 桌面端/默认(未开开关)/已有项目的租户一律不播种。
# ============================================================
from __future__ import annotations
import json
import os
import pytest
from server.auth.context import IdentityContext
from server.db.database import reset_engine
def _identity(tenant: str, user_id: int = 1001) -> IdentityContext:
return IdentityContext(
user_id=user_id, username="demo-tester", fullname="演示测试",
tenant_uuid=tenant, roles=("planner",),
)
@pytest.fixture()
def web_env(tmp_path, monkeypatch):
"""隔离的 Web 数据目录 + 独立 SQLite。"""
data_dir = tmp_path / "server" / "data"
monkeypatch.setenv("APS_MODE", "web")
monkeypatch.setenv("APS_DATA_DIR", str(data_dir))
monkeypatch.setenv("APS_DB_PATH", str(data_dir / "master.db"))
monkeypatch.setenv("APS_WORLD_PATH", str(data_dir / "world.json"))
monkeypatch.delenv("APS_HOME", raising=False)
reset_engine()
yield data_dir
reset_engine()
def test_seed_runs_when_enabled_and_tenant_empty(web_env, monkeypatch):
monkeypatch.setenv("APS_WEB_DEMO_SEED", "1")
from server.state.projects import ProjectStore
snap = ProjectStore(_identity("tenant-demo-a")).snapshot()
demo = [p for p in snap["projects"] if p["id"].startswith("demo-")]
assert len(demo) == 1
assert demo[0]["name"].startswith("演示项目")
pid = demo[0]["id"]
# 世界数据落在数据目录下(tenants/<tenant>/projects/<pid>/world.json)
world_path = web_env / "tenants" / "tenant-demo-a" / "projects" / pid / "world.json"
assert world_path.exists()
world = json.loads(world_path.read_text(encoding="utf-8"))
assert len(world["salesOrders"]) == 7
assert len(world["flexOrders"]) == 10
assert world["lines"] and world["materials"]
# 演示对话已写入会话
sid = next(s["id"] for s in snap["sessions"] if s["projectId"] == pid)
messages = snap["messages"][sid]
assert any(m.get("role") == "assistant" for m in messages)
# marker 已落盘
assert (web_env / "tenants" / "tenant-demo-a" / "demo-seeded.ok").exists()
def test_seed_skipped_by_default(web_env):
from server.state.projects import ProjectStore
snap = ProjectStore(_identity("tenant-demo-b")).snapshot()
assert not [p for p in snap["projects"] if p["id"].startswith("demo-")]
def test_seed_skipped_on_desktop(web_env, monkeypatch):
monkeypatch.setenv("APS_WEB_DEMO_SEED", "1")
monkeypatch.setenv("APS_MODE", "desktop")
from server.state.projects import ProjectStore
snap = ProjectStore(_identity("tenant-demo-c")).snapshot()
assert not [p for p in snap["projects"] if p["id"].startswith("demo-")]
def test_seed_not_repeated_after_marker(web_env, monkeypatch):
monkeypatch.setenv("APS_WEB_DEMO_SEED", "1")
from server.state.projects import ProjectStore
store = ProjectStore(_identity("tenant-demo-d"))
snap = store.snapshot()
pid = next(p["id"] for p in snap["projects"] if p["id"].startswith("demo-"))
store.delete_project(pid)
# 用户删光演示项目后不回填(marker 已存在)
snap2 = ProjectStore(_identity("tenant-demo-d")).snapshot()
assert not [p for p in snap2["projects"] if p["id"].startswith("demo-")]
def test_seed_per_tenant_isolated(web_env, monkeypatch):
monkeypatch.setenv("APS_WEB_DEMO_SEED", "1")
from server.state.projects import ProjectStore
snap_a = ProjectStore(_identity("tenant-e-a")).snapshot()
snap_b = ProjectStore(_identity("tenant-e-b")).snapshot()
pa = next(p for p in snap_a["projects"] if p["id"].startswith("demo-"))
pb = next(p for p in snap_b["projects"] if p["id"].startswith("demo-"))
assert pa["id"] != pb["id"]
# 各自世界文件独立
assert (web_env / "tenants" / "tenant-e-a" / "projects" / pa["id"] / "world.json").exists()
assert (web_env / "tenants" / "tenant-e-b" / "projects" / pb["id"] / "world.json").exists()