107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
# ============================================================
|
||
# 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()
|