112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
# ============================================================
|
||
# 项目分析：绑定登录用户 + 写入租户知识库
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
||
from server.knowledge.assets import KnowledgeStore, get_knowledge
|
||
from server.knowledge.ingest import bind_world_to_identity, ingest_project_analyze_report
|
||
from server.knowledge.retrieval import hybrid_search
|
||
from server.state.seed import empty_world
|
||
|
||
|
||
def test_bind_world_to_current_user():
|
||
world = empty_world()
|
||
identity = IdentityContext(
|
||
user_id=42,
|
||
username="ruiyang_admin",
|
||
fullname="锐扬管理员",
|
||
tenant_uuid="tenant-ruiyang-demo",
|
||
roles=("admin",),
|
||
)
|
||
token = bind_identity(identity)
|
||
try:
|
||
meta = bind_world_to_identity(world, sources=["湖南锐扬MOM主数据收集表.xlsx"])
|
||
assert world["meta"]["ownerUserId"] == 42
|
||
assert world["meta"]["ownerUsername"] == "ruiyang_admin"
|
||
assert world["meta"]["tenantUuid"] == "tenant-ruiyang-demo"
|
||
assert "湖南锐扬" in (meta.get("sources") or [])[0]
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_ingest_analyze_report_into_tenant_kb(tmp_path, monkeypatch):
|
||
identity = IdentityContext(
|
||
user_id=7,
|
||
username="planner_a",
|
||
fullname="计划员甲",
|
||
tenant_uuid="tenant-kb-demo",
|
||
roles=("planner",),
|
||
)
|
||
token = bind_identity(identity)
|
||
try:
|
||
# 隔离租户知识库到临时文件，避免污染本机 store
|
||
kb_path = tmp_path / "kb.json"
|
||
store = KnowledgeStore(path=str(kb_path))
|
||
store.assets = []
|
||
store._write()
|
||
monkeypatch.setattr("server.knowledge.assets.get_knowledge", lambda: store)
|
||
|
||
world = empty_world()
|
||
deep = {
|
||
"projectName": "湖南锐扬",
|
||
"sources": ["MOM主数据收集表.xlsx"],
|
||
"markdown": "# 湖南锐扬分析\n\n物料与设备已解析，可排产前需补工艺标准工时。",
|
||
"summary": {"materials": 12, "orders": 3, "equipment": 5, "bom": 8},
|
||
"materials": [
|
||
{"code": "P-001", "name": "成品阀", "type": "FINISHED_PRODUCT"},
|
||
{"code": "M-010", "name": "阀体毛坯", "type": "RAW"},
|
||
],
|
||
"equipment": [
|
||
{"code": "EQ-01", "name": "数控车", "capabilities": "车削"},
|
||
],
|
||
"orders": [
|
||
{"orderNo": "SO-1", "productCode": "P-001", "quantity": 10, "dueDate": "2026-08-01"},
|
||
],
|
||
"plan": ["补标准工时", "核对 BOM 用量"],
|
||
}
|
||
world["flexMaterials"] = deep["materials"] + [
|
||
{"code": "M-LAST-999", "name": "完整集末尾物料", "type": "RAW_MATERIAL"},
|
||
]
|
||
world["flexEquipment"] = deep["equipment"] + [
|
||
{"code": "EQ-LAST-999", "name": "完整集末尾设备", "status": "RUNNING"},
|
||
]
|
||
world["flexOrders"] = deep["orders"]
|
||
world["flexBom"] = [
|
||
{"productCode": "P-001", "materialCode": "M-LAST-999", "quantity": 2},
|
||
]
|
||
world["flexRoutings"] = [
|
||
{"productCode": "P-001", "seq": 99, "operationCode": "OP-LAST-999",
|
||
"operationName": "完整集末尾工序", "stdTimePerUnit": 12},
|
||
]
|
||
applied = ingest_project_analyze_report(deep, world=world)
|
||
assert applied.get("assetId")
|
||
assert applied.get("chunkCount", 0) >= 1
|
||
assert applied.get("ownerUserId") == 7
|
||
assert applied.get("tenantUuid") == "tenant-kb-demo"
|
||
assert world["meta"]["ownerUserId"] == 7
|
||
|
||
asset = store.get(applied["assetId"])
|
||
assert asset and "项目分析" in (asset.get("title") or "")
|
||
tags = asset.get("tags") or []
|
||
assert any(str(t).startswith("user:planner_a") for t in tags)
|
||
assert any(str(t).startswith("uid:7") for t in tags)
|
||
chunk_text = "\n".join(str(c.get("text") or "") for c in (asset.get("chunks") or []))
|
||
assert "M-LAST-999" in chunk_text
|
||
assert "EQ-LAST-999" in chunk_text
|
||
assert "OP-LAST-999" in chunk_text
|
||
assert "完整数据集" in tags
|
||
|
||
units = store.iter_search_units()
|
||
hits = hybrid_search(units, "阀体毛坯 数控车", top_k=5)
|
||
assert hits
|
||
assert any("锐扬" in (h.get("title") or "") or "阀" in (h.get("snippet") or h.get("text") or "") for h in hits)
|
||
finally:
|
||
reset_identity(token)
|
||
# 清掉可能缓存的租户 store（若测试间复用 get_knowledge）
|
||
try:
|
||
from server.knowledge import assets as assets_mod
|
||
assets_mod._stores.pop("tenant-kb-demo", None)
|
||
except Exception:
|
||
pass
|