qiaoqichaoandClaude Code
d56c699ce6
按 docs/v2/05-implementation-plan.md §1 候选迁移表与 claude_use/T00 白名单， 从 55e2cdb 抽取 224 个文件（2.48 MB）到独立目录 F:/mini-OpenClaw-v2。 - 排除：凭据(.env/server-deploy.conf/旧DELIVERY.md)、运行数据(xingbang_data 等 4 个 SQLite)、依赖(venv/node_modules 符号链接)、归档(_backup*/test-evidence/ ori_file)、Office 临时文件 - 目标目录独立 git 仓库，初始恢复点；Gitea 远程与分支待 OQ-06 Q2 定案 - 未部署、未启动服务、未调用模型 Co-Authored-By: Claude Code <noreply@anthropic.com>
186 lines
10 KiB
Python
186 lines
10 KiB
Python
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from api import operations_center as api
|
|
from api import risk_agent_config as catalog
|
|
from services.project_assistant import ProjectAssistantService
|
|
from services.operations_demo import DemoService, PROJECT
|
|
from services.assistant_worker import DurableWorker
|
|
from services.assistant_store import get, listing
|
|
|
|
|
|
class OperationsCenterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
|
self.path = Path(self.temp.name)
|
|
self.current = ProjectAssistantService(self.path / "current.sqlite3")
|
|
self.lab = DemoService(self.path / "lab.sqlite3")
|
|
self.patches = [patch.object(api, "current_service", lambda: self.current), patch.object(api, "demo_service", lambda: self.lab),
|
|
patch.object(catalog, "DB_PATH", self.path / "catalog.sqlite3"),
|
|
patch.dict(os.environ, {"XINGBANG_AUTH_USERS":"", "XINGBANG_REQUIRE_AUTH":"0", "XINGBANG_LOCAL_MODEL_URL":"", "XINGBANG_ASSISTANT_ENABLED":"1", "XINGBANG_ASSISTANT_DEMO_ENABLED":"1"})]
|
|
for p in self.patches:
|
|
p.start()
|
|
app = FastAPI()
|
|
app.include_router(api.router)
|
|
self.client = TestClient(app)
|
|
self.worker = DurableWorker(self.lab)
|
|
self.root = "/api/xingbang/risk-agent/project-assistant/operations"
|
|
|
|
def tearDown(self):
|
|
self.worker.stop()
|
|
self.client.close()
|
|
for p in reversed(self.patches):
|
|
p.stop()
|
|
self.temp.cleanup()
|
|
|
|
def get(self, path):
|
|
return self.client.get(self.root + path)
|
|
|
|
def post(self, path, body=None, key="test-run"):
|
|
return self.client.post(self.root + path, json=body or {}, headers={"Idempotency-Key":key})
|
|
|
|
def submit(self, key="test-run", model=False):
|
|
result = self.post("/lab/runs", {"source_revision":self.lab.source()["project"]["revision"], "model":model}, key)
|
|
self.assertEqual(result.status_code, 202, result.text)
|
|
return result.json()
|
|
|
|
def edit(self, **overrides):
|
|
s = self.lab.source()
|
|
body = {"revision":s["project"]["revision"], "amount":s["project"]["amount"], "liabilities":s["finance"]["liabilities"], "dd_liabilities":s["finance"]["dd_liabilities"], "cashflow":s["finance"]["cashflow"], "duplicate":bool(s["assets"][2]["active"]), "enabled":True, **overrides}
|
|
result = self.post("/lab/source", body)
|
|
self.assertEqual(result.status_code, 200, result.text)
|
|
return body
|
|
|
|
def test_real_sql_worker_and_scoped_monitor(self):
|
|
self.assertEqual(len(self.lab.list_projects()), 1)
|
|
self.assertEqual(len(self.lab.source()["assets"]), 3)
|
|
initial = self.submit()
|
|
self.assertEqual(self.get("/overview?domain=lab").json()["counts"]["queued"], 1)
|
|
self.worker.process_next()
|
|
detail = self.get("/runs/" + initial["id"] + "?domain=lab").json()
|
|
self.assertEqual(detail["run"]["status"], "partial")
|
|
self.assertEqual(detail["run"]["summary"], {"coverage":"1.0800", "open":4, "resolved":0})
|
|
self.assertTrue(all("SQLite" in e["source"] for e in detail["run"]["evidence"] if e["field"] == "lease.total_value"))
|
|
self.assertEqual(detail["source_revision"], 1)
|
|
self.assertEqual(self.get("/overview?domain=current").json()["total_runs"], 0)
|
|
self.assertNotIn("model_config", detail["run"])
|
|
self.assertNotIn("identity", detail["run"])
|
|
|
|
def test_source_changes_require_sync_and_keep_snapshots(self):
|
|
first = self.submit()
|
|
self.worker.process_next()
|
|
original = self.lab.resource("input_snapshot", first["input_snapshot_id"])
|
|
self.edit(amount="4000", dd_liabilities="13200", cashflow="900", duplicate=False)
|
|
self.assertEqual(self.lab.resource("project", PROJECT)["values"]["facility.amount"], "5000")
|
|
second = self.submit("second")
|
|
self.worker.process_next()
|
|
result = self.lab.resource("run", second["id"])
|
|
self.assertEqual(result["summary"]["coverage"], "1.3500")
|
|
self.assertEqual(result["summary"]["open"], 0)
|
|
self.assertEqual(result["status"], "completed")
|
|
self.assertEqual(self.lab.resource("input_snapshot", first["input_snapshot_id"]), original)
|
|
self.assertEqual(self.current.get_project("DEMO-B")["facts"]["facility_amount"], "5000")
|
|
|
|
def test_frozen_configuration_and_idempotency(self):
|
|
first = self.submit()
|
|
self.assertEqual(self.submit()["id"], first["id"])
|
|
source = self.get("/lab/source").json()
|
|
change = self.post("/lab/rule", {"version":"2.0.0", "revision":source["release"]["revision"]})
|
|
self.assertEqual(change.status_code, 200)
|
|
self.worker.process_next()
|
|
detail = self.get("/runs/" + first["id"] + "?domain=lab").json()
|
|
rule = next(c for c in detail["components"] if c["id"] == "lab.coverage")
|
|
self.assertEqual(rule["version"], "1.0.0")
|
|
self.assertTrue(rule["changed"])
|
|
second = self.submit("new-version")
|
|
self.assertEqual(next(c for c in second["component_versions"] if c["id"] == "lab.coverage")["version"], "2.0.0")
|
|
|
|
def test_pause_conflict_cancel_and_invalid_amount(self):
|
|
body = self.edit(enabled=False)
|
|
self.assertEqual(self.post("/lab/source", body).status_code, 409)
|
|
self.assertEqual(self.post("/lab/runs", {"source_revision":2}).status_code, 422)
|
|
with self.lab.connect() as db:
|
|
self.assertEqual(len(listing(db, "run")), 0)
|
|
body.update(revision=2,amount="NaN")
|
|
self.assertEqual(self.post("/lab/source", body).status_code, 422)
|
|
self.edit(enabled=True)
|
|
run = self.submit()
|
|
self.assertEqual(self.post("/lab/runs/" + run["id"] + "/cancel").status_code, 200)
|
|
self.assertFalse(self.worker.process_next())
|
|
|
|
def test_no_fake_natural_language_or_model_success(self):
|
|
self.assertEqual(self.post("/lab/runs", {"source_revision":1,"requirement":"重点现金流"}).status_code, 422)
|
|
run = self.submit(model=True)
|
|
self.worker.process_next()
|
|
run = self.lab.resource("run", run["id"])
|
|
model = next(s for s in run["steps"] if s["step_id"] == "corporate.analysis")
|
|
self.assertEqual(model["status"], "failed")
|
|
self.assertIn("未配置", model["errors"][0]["message"])
|
|
self.assertEqual(run["status"], "partial")
|
|
|
|
def test_evaluation_is_real_and_isolated(self):
|
|
before = self.lab.source()
|
|
result = self.post("/lab/evaluate")
|
|
self.assertEqual(result.status_code, 200, result.text)
|
|
self.assertTrue(result.json()["passed"], result.text)
|
|
self.assertEqual([len(r["actual_codes"]) for r in result.json()["cases"]], [4,2,0,1])
|
|
self.assertEqual(self.lab.source(), before)
|
|
self.assertEqual(self.get("/overview?domain=lab").json()["total_runs"], 0)
|
|
self.assertEqual(len(self.get("/overview?domain=lab").json()["evaluations"]), 1)
|
|
|
|
def test_authorization_and_demo_switch(self):
|
|
run = self.submit()
|
|
accounts = [{"id":"reviewer","token":"r","role":"reviewer","projects":["DEMO-A"]}, {"id":"manager","token":"m","role":"manager","projects":[PROJECT]}]
|
|
with patch.dict(os.environ, {"XINGBANG_AUTH_USERS":json.dumps(accounts)}):
|
|
for token, code in (("r",403),("m",200)):
|
|
self.assertEqual(self.client.get(self.root+"/overview?domain=lab", headers={"Authorization":"Bearer "+token}).status_code, code)
|
|
self.assertEqual(self.client.get(self.root+"/runs/"+run["id"]+"?domain=lab", headers={"Authorization":"Bearer r"}).status_code,403)
|
|
self.assertEqual(self.client.post(self.root+"/lab/evaluate", headers={"Authorization":"Bearer m"},json={}).status_code,403)
|
|
self.assertEqual(self.client.get(self.root+"/health", headers={"Authorization":"Bearer m"}).status_code,403)
|
|
self.assertEqual(self.client.get(self.root+"/overview?domain=current", headers={"Authorization":"Bearer r"}).json()["projects"][0]["id"],"DEMO-A")
|
|
with patch.dict(os.environ, {"XINGBANG_ASSISTANT_DEMO_ENABLED":"0"}):
|
|
self.assertEqual(self.get("/lab/source").status_code,403)
|
|
|
|
def test_navigation_is_role_aware_and_keeps_project_context(self):
|
|
root = Path(__file__).resolve().parents[1]
|
|
nav = (root / "src/components/xingbang/AgentNav.tsx").read_text(encoding="utf-8")
|
|
selector = (root / "src/components/xingbang/IdentitySelector.tsx").read_text(encoding="utf-8")
|
|
self.assertNotIn("四 Agent 工作区",nav)
|
|
for label in ("业务助手","项目工作台","业务时效","配置治理"):
|
|
self.assertIn(label,nav)
|
|
self.assertIn("当前身份",selector)
|
|
self.assertIn("选择身份",selector)
|
|
for label in ("运行中心","检查依据","演示与验收","兼容入口"):
|
|
self.assertNotIn(label,nav)
|
|
self.assertIn("ROLE_LABEL",nav)
|
|
self.assertNotIn("typeof window",nav)
|
|
self.assertIn('href.includes("?")',nav)
|
|
|
|
def test_business_ui_keeps_simple_traceable_flows(self):
|
|
root = Path(__file__).resolve().parents[1]
|
|
workspace = (root / "src/app/xingbang-workbench/page.tsx").read_text(encoding="utf-8")
|
|
assistant = (root / "src/components/xingbang/BusinessAssistant.tsx").read_text(encoding="utf-8")
|
|
dashboard = (root / "src/app/operations-center/page.tsx").read_text(encoding="utf-8")
|
|
for label in ("待评审项目池", "客户经理提交", "领取并开始评审", "全链路评审", "批准并生成评审报告", "怎么解决"):
|
|
self.assertIn(label, workspace)
|
|
self.assertIn("detailText(item.detail)", workspace)
|
|
self.assertNotIn("<p>{item.detail}</p>", workspace)
|
|
self.assertNotIn('"方案比较"', workspace)
|
|
self.assertNotIn('tab === "scenario"', workspace)
|
|
self.assertNotIn("mini-openclaw", assistant)
|
|
self.assertNotIn("<ChatPanel/>", assistant)
|
|
self.assertIn('className="ba-workspace"', assistant)
|
|
self.assertIn("当前身份只会调用本项目授权的 Skills、规则和业务工具", assistant)
|
|
self.assertIn("IdentitySelector", assistant)
|
|
self.assertIn("IdentitySelector", dashboard)
|
|
self.assertIn("统计范围", dashboard)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|