119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
# ============================================================
|
||
# M-A 黄金测试：SQLite 主数据入库 → 投影回 world → PoolEngine 可排
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from server.engines import PoolEngine
|
||
from server.state.seed import seed_world
|
||
|
||
|
||
@pytest.fixture()
|
||
def db_env(tmp_path, monkeypatch):
|
||
"""隔离数据库：每个用例独立 master.db。"""
|
||
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "master.db"))
|
||
monkeypatch.delenv("APS_DB_DISABLED", raising=False)
|
||
from server.db.database import reset_engine
|
||
reset_engine()
|
||
yield
|
||
reset_engine()
|
||
|
||
|
||
def test_world_to_db_and_projection_roundtrip(db_env):
|
||
from server.db.sync import db_has_master, db_to_world, world_to_db
|
||
|
||
world = seed_world()
|
||
counts = world_to_db(world)
|
||
assert counts["materials"] == len(world["materials"])
|
||
assert counts["flexRoutings"] == len(world["flexRoutings"])
|
||
assert counts["flexParams"] == 1
|
||
assert db_has_master()
|
||
|
||
# 投影到空 world：主数据键零损耗还原
|
||
shadow: dict = {}
|
||
back = db_to_world(shadow)
|
||
assert back["materials"] == len(world["materials"])
|
||
assert shadow["materials"] == world["materials"]
|
||
assert shadow["flexRoutings"] == world["flexRoutings"]
|
||
assert shadow["salesOrders"] == world["salesOrders"]
|
||
assert shadow["flexParams"] == world["flexParams"]
|
||
|
||
|
||
def test_projection_world_schedulable_by_pool_engine(db_env):
|
||
from server.db.sync import db_to_world, world_to_db
|
||
|
||
world = seed_world()
|
||
world_to_db(world)
|
||
|
||
# 投影世界补运行态骨架后可直接排产
|
||
shadow = seed_world()
|
||
for k in ("flexScheduleVersions", "flexVirtualLines", "flexWorkOrders", "flexConflicts"):
|
||
shadow[k] = []
|
||
db_to_world(shadow)
|
||
|
||
counters: dict[str, int] = {}
|
||
|
||
def nid(kind: str) -> int:
|
||
counters[kind] = counters.get(kind, 0) + 1
|
||
return counters[kind]
|
||
|
||
result = PoolEngine().solve(shadow, nid, sort_mode="BOTTLENECK")
|
||
assert result["orderCount"] > 0
|
||
assert result["woCount"] > 0
|
||
|
||
|
||
def test_multi_project_isolation(db_env):
|
||
from server.db.sync import (
|
||
db_to_world, get_active_project, list_projects,
|
||
set_active_project, world_to_db,
|
||
)
|
||
|
||
world = seed_world()
|
||
world_to_db(world) # 写入默认 demo 项目
|
||
assert get_active_project()["code"] == "demo"
|
||
|
||
# 切到康尼项目：空主数据，互不污染
|
||
set_active_project("kangni", "康尼机电")
|
||
shadow: dict = {}
|
||
assert db_to_world(shadow) == {}
|
||
world_to_db({"materials": [{"id": 1, "code": "KN-001", "name": "康尼物料"}]})
|
||
|
||
kn: dict = {}
|
||
db_to_world(kn)
|
||
assert [m["code"] for m in kn["materials"]] == ["KN-001"]
|
||
|
||
# 切回 demo：原数据完好
|
||
set_active_project("demo")
|
||
demo: dict = {}
|
||
db_to_world(demo)
|
||
assert len(demo["materials"]) == len(world["materials"])
|
||
assert {p["code"] for p in list_projects()} >= {"demo", "kangni"}
|
||
|
||
|
||
def test_demo_pack_replay_and_date_rebase(tmp_path):
|
||
from server.state.packs import export_pack, load_pack
|
||
|
||
world = seed_world()
|
||
pack_path = str(tmp_path / "demo.json")
|
||
export_pack(world, pack_path, name="演示")
|
||
|
||
# 伪造 30 天前生成的包 → 加载后日期整体平移 30 天
|
||
import json
|
||
with open(pack_path, "r", encoding="utf-8") as f:
|
||
pack = json.load(f)
|
||
from datetime import datetime, timedelta
|
||
pack["baseDate"] = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
|
||
with open(pack_path, "w", encoding="utf-8") as f:
|
||
json.dump(pack, f, ensure_ascii=False)
|
||
|
||
loaded = load_pack(pack_path)
|
||
|
||
def shift(day: str, days: int) -> str:
|
||
return (datetime.strptime(day, "%Y-%m-%d") + timedelta(days=days)).strftime("%Y-%m-%d")
|
||
|
||
# 30 天前的包 → 全部日期前移 30 天对齐「今天」
|
||
assert loaded["salesOrders"][0]["deliveryDate"] == shift(world["salesOrders"][0]["deliveryDate"], 30)
|
||
assert loaded["flexOrders"][0]["dueDate"] == shift(world["flexOrders"][0]["dueDate"], 30)
|
||
assert len(loaded["materials"]) == len(world["materials"])
|