289 lines
11 KiB
Python
289 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from server.aps_domain.folder_pack import analyze_work_dir
|
|
from server.aps_domain.project_analyze import analyze_project_deep
|
|
from server.state.seed import empty_world
|
|
|
|
|
|
def _counter():
|
|
values: dict[str, int] = {}
|
|
|
|
def next_id(kind: str) -> int:
|
|
values[kind] = values.get(kind, 0) + 1
|
|
return values[kind]
|
|
|
|
return next_id
|
|
|
|
|
|
def test_analyze_project_deep_readonly_does_not_mutate_world(tmp_path: Path, monkeypatch):
|
|
def fail_write(*_args, **_kwargs):
|
|
raise AssertionError("readonly analysis must not call write paths")
|
|
|
|
def fail_ingest(*_args, **_kwargs):
|
|
raise AssertionError("readonly analysis must not write knowledge")
|
|
|
|
(tmp_path / "orders.csv").write_text(
|
|
"orderNo,productCode,quantity,deliveryDate,customerName\n"
|
|
"SO-1,PRODUCT-1,10,2026-08-20,CUSTOMER-1\n",
|
|
encoding="utf-8-sig",
|
|
)
|
|
(tmp_path / "materials.csv").write_text(
|
|
"code,name,type,unit,stock\n"
|
|
"PRODUCT-1,Product 1,FINISHED_PRODUCT,pc,100\n",
|
|
encoding="utf-8-sig",
|
|
)
|
|
(tmp_path / "routing.csv").write_text(
|
|
"productCode,operationCode,operationName,seq,stdTimePerUnit\n"
|
|
"PRODUCT-1,CUT,Cut,10,5\n",
|
|
encoding="utf-8-sig",
|
|
)
|
|
(tmp_path / "equipment.csv").write_text(
|
|
"code,name,capabilities\n"
|
|
"EQ-1,Cutter,CUT\n",
|
|
encoding="utf-8-sig",
|
|
)
|
|
|
|
read_world = empty_world()
|
|
before = json.dumps(read_world, sort_keys=True, default=str)
|
|
|
|
def fail_project_context(*_args, **_kwargs):
|
|
raise AssertionError("readonly analysis must not call ProjectStore.snapshot")
|
|
|
|
with monkeypatch.context() as patch:
|
|
patch.setattr("server.aps_domain.importers.apply_import_commit", fail_write)
|
|
patch.setattr("server.knowledge.ingest.ingest_project_analyze_report", fail_ingest)
|
|
patch.setattr(
|
|
"server.aps_domain.project_analyze._project_ctx",
|
|
fail_project_context,
|
|
)
|
|
readonly = analyze_project_deep(
|
|
read_world,
|
|
input_path=str(tmp_path),
|
|
next_id=_counter(),
|
|
apply=False,
|
|
)
|
|
after = json.dumps(read_world, sort_keys=True, default=str)
|
|
|
|
assert before == after
|
|
assert readonly["readOnly"] is True
|
|
assert readonly["applied"] == {}
|
|
assert readonly["pendingAdoption"]["mode"] == "folder"
|
|
assert readonly["pendingAdoption"]["batches"]
|
|
assert all(batch["okRows"] for batch in readonly["pendingAdoption"]["batches"])
|
|
assert readonly["pendingAdoption"]["filename"] == ""
|
|
# 订单行引用了尚未采用的产品，预览阶段就无法通过关联校验：这份文件不会进入
|
|
# 可提交批次，但必须原样出现在分析报告的 files 里，不能静默丢弃。
|
|
assert set(readonly["pendingAdoption"]["sourcePaths"]) == {
|
|
str(tmp_path / "materials.csv"),
|
|
str(tmp_path / "routing.csv"),
|
|
str(tmp_path / "equipment.csv"),
|
|
}
|
|
rejected = [
|
|
item for item in (readonly["folder"] or {}).get("files") or []
|
|
if item.get("name") == "orders.csv"
|
|
]
|
|
assert rejected and rejected[0]["errorCount"] >= 1
|
|
assert rejected[0]["errors"]
|
|
assert not any(
|
|
batch.get("kind") == "orders"
|
|
for batch in readonly["pendingAdoption"]["batches"]
|
|
)
|
|
|
|
applied = analyze_project_deep(
|
|
empty_world(),
|
|
input_path=str(tmp_path),
|
|
next_id=_counter(),
|
|
apply=True,
|
|
)
|
|
|
|
for key in ("summary", "canSchedule", "orders", "materials", "routings", "equipment"):
|
|
assert readonly[key] == applied[key]
|
|
|
|
|
|
def test_analyze_work_dir_sql_preview_without_apply(tmp_path: Path, monkeypatch):
|
|
sql = tmp_path / "mini.sql"
|
|
sql.write_text(
|
|
"""
|
|
CREATE TABLE `pl_order` (
|
|
`id` int, `code` varchar(64), `business_code` varchar(64),
|
|
`customer_business_code` varchar(64), `material_code` varchar(64),
|
|
`material_name` varchar(64), `parent_code` varchar(64), `craftl_code` varchar(64),
|
|
`machine_model` varchar(64), `dept` varchar(64), `people` varchar(64),
|
|
`drawing_code` varchar(64), `quantity` decimal(10,2), `man_quantity` decimal(10,2),
|
|
`all_quantity` decimal(10,2), `status` varchar(8), `stock_status` varchar(8),
|
|
`distribution_status` varchar(8), `type` varchar(8), `description` varchar(64),
|
|
`planned_end_time` datetime, `is_delete` varchar(8), `level` int
|
|
) ENGINE=InnoDB;
|
|
CREATE TABLE `r_production_craftl` (
|
|
`id` int, `craftl_code` varchar(64), `craftl_name` varchar(64),
|
|
`procedure_code` varchar(64), `procedure_name` varchar(64),
|
|
`output_material_code` varchar(64), `output_material_name` varchar(64),
|
|
`working_hours` decimal(10,2), `sort_no` int, `is_delete` varchar(8)
|
|
) ENGINE=InnoDB;
|
|
CREATE TABLE `md_equipment` (
|
|
`id` int, `code` varchar(64), `name` varchar(64), `is_delete` varchar(8)
|
|
) ENGINE=InnoDB;
|
|
CREATE TABLE `md_material` (
|
|
`id` int, `code` varchar(64), `name` varchar(64), `type` varchar(32),
|
|
`unit` varchar(8), `is_delete` varchar(8)
|
|
) ENGINE=InnoDB;
|
|
INSERT INTO `pl_order` VALUES
|
|
(1,'ORD1',NULL,'C1','FG001','Product 1',NULL,'CRAFT1',NULL,NULL,NULL,NULL,10,NULL,NULL,'0','0','0','S',NULL,'2026-08-01 00:00:00','0',5);
|
|
INSERT INTO `r_production_craftl` VALUES
|
|
(1,'CRAFT1','Product 1','OP10','Cut','WIP-X','Semi',0.5,10,'0'),
|
|
(2,'CRAFT1','Product 1','OP20','Finish','FG001','Product 1',0.8,20,'0');
|
|
INSERT INTO `md_equipment` VALUES (1,'EQ1','Cutter','0');
|
|
INSERT INTO `md_material` VALUES (1,'FG001','Product 1','Product','pc','0');
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
class FakePS:
|
|
def snapshot(self, include_messages=False):
|
|
return {
|
|
"projects": [{"id": "p1", "name": "Readonly", "workDir": str(tmp_path)}],
|
|
"sessions": [{"id": "s1", "projectId": "p1"}],
|
|
"files": [],
|
|
}
|
|
|
|
monkeypatch.setattr("server.state.projects.get_project_store", lambda: FakePS())
|
|
monkeypatch.setattr(
|
|
"server.aps_domain.project_analyze._project_ctx_readonly",
|
|
lambda _session_id: (
|
|
{"id": "p1", "name": "Readonly", "workDir": str(tmp_path)},
|
|
str(tmp_path),
|
|
),
|
|
)
|
|
world = empty_world()
|
|
for key in ("flexOrders", "flexMaterials", "flexRoutings", "flexEquipment", "flexBom"):
|
|
world[key] = []
|
|
before = json.dumps(world, sort_keys=True, default=str)
|
|
|
|
def fail_sql_write(*_args, **_kwargs):
|
|
raise AssertionError("SQL preview must not be written")
|
|
|
|
def fail_project_snapshot(*_args, **_kwargs):
|
|
raise AssertionError("SQL preview must not call ProjectStore.snapshot")
|
|
|
|
with monkeypatch.context() as patch:
|
|
patch.setattr(
|
|
"server.importers.sql_pack.apply_sql_pack_to_world",
|
|
fail_sql_write,
|
|
)
|
|
patch.setattr(
|
|
"server.aps_domain.folder_pack._project_work_dir",
|
|
fail_project_snapshot,
|
|
)
|
|
report = analyze_work_dir(world, "s1", apply_sql=False)
|
|
after = json.dumps(world, sort_keys=True, default=str)
|
|
|
|
assert before == after
|
|
assert report["sqlApplied"] is False
|
|
assert report["sqlStats"]["orders"] >= 1
|
|
assert report["sqlPreview"]["filename"] == "mini.sql"
|
|
assert report["sqlPreview"]["fieldMapText"]
|
|
assert report["canSchedule"] is True
|
|
assert world["flexOrders"] == []
|
|
assert world["flexRoutings"] == []
|
|
assert world["flexEquipment"] == []
|
|
|
|
applied = analyze_work_dir(empty_world(), "s1", apply_sql=True)
|
|
assert applied["sqlApplied"] is True
|
|
assert applied["sqlStats"] == report["sqlStats"]
|
|
|
|
deep_world = empty_world()
|
|
deep_before = json.dumps(deep_world, sort_keys=True, default=str)
|
|
deep = analyze_project_deep(
|
|
deep_world,
|
|
input_path=str(sql),
|
|
next_id=_counter(),
|
|
apply=False,
|
|
apply_sql=False,
|
|
)
|
|
deep_after = json.dumps(deep_world, sort_keys=True, default=str)
|
|
|
|
assert deep_before == deep_after
|
|
assert deep["readOnly"] is True
|
|
assert deep["pendingAdoption"]["mode"] == "sql"
|
|
assert deep["summary"]["orders"] >= 1
|
|
assert deep["summary"]["routings"] >= 1
|
|
assert deep["summary"]["equipment"] >= 1
|
|
assert deep["canSchedule"] is True
|
|
|
|
|
|
def test_readonly_knowledge_preview_never_seeds_or_applies_templates(monkeypatch):
|
|
from server.aps_domain.project_analyze import (
|
|
_fill_from_knowledge,
|
|
_recommend_templates_readonly,
|
|
)
|
|
from server.knowledge import routing_templates
|
|
|
|
def fail_seed():
|
|
raise AssertionError("readonly preview must not seed routing templates")
|
|
|
|
def fail_apply(*_args, **_kwargs):
|
|
raise AssertionError("readonly preview must not apply routing templates")
|
|
|
|
monkeypatch.setattr(routing_templates, "seed_templates", fail_seed)
|
|
monkeypatch.setattr(routing_templates, "apply_template_to_product", fail_apply)
|
|
|
|
recommendations = _recommend_templates_readonly("箱体 支架 铣削", top_k=1)
|
|
assert recommendations
|
|
assert recommendations[0]["code"] == "TPL-MILLING"
|
|
|
|
world = empty_world()
|
|
world["flexOrders"] = [{
|
|
"orderNo": "ORDER-1",
|
|
"productCode": "PRODUCT-1",
|
|
"productName": "箱体支架",
|
|
"status": "NEW",
|
|
}]
|
|
world["flexMaterials"] = [{"code": "PRODUCT-1", "name": "箱体支架"}]
|
|
world["flexRoutings"] = []
|
|
|
|
actions = _fill_from_knowledge(world, ["PRODUCT-1"], apply=False)
|
|
|
|
assert any(action.get("type") == "template_applied" for action in actions)
|
|
assert world["flexRoutings"]
|
|
|
|
|
|
def test_readonly_project_context_uses_ro_query_without_writes(tmp_path: Path, monkeypatch):
|
|
import sqlite3
|
|
|
|
from server.aps_domain.project_analyze import _project_ctx_readonly
|
|
|
|
db_path = tmp_path / "master.db"
|
|
connection = sqlite3.connect(db_path)
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE aps_chat_sessions (
|
|
id TEXT, project_id TEXT, scope TEXT, tenant_uuid TEXT, deleted INTEGER
|
|
);
|
|
CREATE TABLE aps_workspace_projects (
|
|
id TEXT, name TEXT, work_dir TEXT, tenant_uuid TEXT, deleted INTEGER
|
|
);
|
|
CREATE TABLE aps_project_members (
|
|
id INTEGER, project_id TEXT, tenant_uuid TEXT, user_id INTEGER,
|
|
status TEXT, deleted INTEGER
|
|
);
|
|
INSERT INTO aps_chat_sessions VALUES ('s1', 'p1', 'project', 'platform', 0);
|
|
INSERT INTO aps_workspace_projects VALUES ('p1', 'Readonly', 'C:/data', 'platform', 0);
|
|
INSERT INTO aps_project_members VALUES (1, 'p1', 'platform', 0, 'active', 0);
|
|
""",
|
|
)
|
|
connection.commit()
|
|
connection.close()
|
|
before = db_path.read_bytes()
|
|
|
|
monkeypatch.setenv("APS_DB_PATH", str(db_path))
|
|
monkeypatch.delenv("APS_PROJECTS_PATH", raising=False)
|
|
|
|
project, work = _project_ctx_readonly("s1")
|
|
|
|
assert project == {"id": "p1", "name": "Readonly", "workDir": "C:/data"}
|
|
assert work == "C:/data"
|
|
assert db_path.read_bytes() == before
|