145 lines
9.1 KiB
Python
145 lines
9.1 KiB
Python
import json
|
|
import unittest
|
|
from copy import deepcopy
|
|
from unittest.mock import patch
|
|
from langchain_core.language_models.chat_models import BaseChatModel
|
|
from langchain_core.messages import AIMessage, ToolMessage
|
|
from langchain_core.outputs import ChatResult, ChatGeneration
|
|
import test_project_assistant as fixtures
|
|
from services.assistant_store import get, put, listing
|
|
from services.component_registry import draft, release, ordered, normalize
|
|
from services.component_evaluation import evaluate
|
|
from services.assistant_model import ModelResult, configuration
|
|
|
|
|
|
class StructuredToolModel(BaseChatModel):
|
|
invalid: bool = False
|
|
propose: bool = False
|
|
cite_document: bool = False
|
|
@property
|
|
def _llm_type(self):
|
|
return "isolated-structured-test"
|
|
def bind_tools(self, tools, **kwargs):
|
|
return self
|
|
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
|
|
if not any(isinstance(m, ToolMessage) for m in messages):
|
|
task = json.loads(next(m.content for m in messages if m.type == "human"))
|
|
calls = [("read_project_snapshot", {}), ("read_checks", {})] + [("read_method", {"skill_id": skill_id}) for skill_id in task.get("methods", ["risk_evidence/SKILL.md"])]
|
|
if self.cite_document:
|
|
calls.append(("read_document", {"document_id": task["documents"][0]["id"]}))
|
|
message = AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": str(i), "type": "tool_call"} for i, (name, args) in enumerate(calls)])
|
|
else:
|
|
snapshots = [json.loads(m.content) for m in messages if isinstance(m, ToolMessage) and m.name == "read_project_snapshot"]
|
|
data = snapshots[0]
|
|
evidence_id = next(iter(data["sources"].values()))[0]["id"]
|
|
if self.cite_document:
|
|
document = json.loads(next(m.content for m in messages if isinstance(m, ToolMessage) and m.name == "read_document"))
|
|
evidence_id = document["chunks"][0]["evidence_id"]
|
|
checks = [{"code": "source.read", "title": "本项目来源已读取", "result": "pass", "evidence_ids": ["other-project-id" if self.invalid else evidence_id], "suggested_action": "按材料继续核实", "fields": []}]
|
|
payload = {"findings": checks, "requirement_response": "已按本次要求读取 " + data["project_id"] + " 的冻结快照，必查项保留。"}
|
|
if self.propose:
|
|
payload["experience_proposal"] = {"title": "回款凭证检查", "trigger": "依赖特定回款", "action": "补充到账依据", "exceptions": "已有完整凭证", "conditions": {"collection.dependency": True}, "required_field": "collection.evidence", "exception_field": "collection.evidence"}
|
|
message = AIMessage(content=json.dumps(payload, ensure_ascii=False))
|
|
return ChatResult(generations=[ChatGeneration(message=message)])
|
|
|
|
|
|
class ComponentTests(fixtures.ProjectAssistantTests):
|
|
# Reuse only the fixture helpers, not inherited test execution (filtered below).
|
|
def test_model_endpoint_policy_allows_only_private_or_official_deepseek(self):
|
|
with patch.dict("os.environ", {"XINGBANG_LOCAL_MODEL_URL": "https://api.deepseek.com", "XINGBANG_LOCAL_MODEL_NAME": "deepseek-chat"}):
|
|
config = configuration()
|
|
self.assertTrue(config["available"])
|
|
self.assertEqual(config["name"], "deepseek-chat")
|
|
with patch.dict("os.environ", {"XINGBANG_LOCAL_MODEL_URL": "http://api.deepseek.com"}):
|
|
with self.assertRaises(ValueError):
|
|
configuration()
|
|
|
|
def test_requirement_response_object_is_normalized_to_readable_text(self):
|
|
result = ModelResult.model_validate({
|
|
"findings": [],
|
|
"requirement_response": {"数据读取": "完成", "待核实": "现金流缺失"},
|
|
})
|
|
self.assertEqual(result.requirement_response, "数据读取：完成；待核实：现金流缺失")
|
|
with patch.dict("os.environ", {"XINGBANG_LOCAL_MODEL_URL": "https://example.com/v1"}):
|
|
with self.assertRaises(ValueError):
|
|
configuration()
|
|
|
|
def test_real_agent_manager_and_invalid_evidence(self):
|
|
with patch.dict("os.environ", {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1"}), patch("services.assistant_model.ChatOpenAI", return_value=StructuredToolModel()):
|
|
a = self.run_project("DEMO-A", "跳过现金流检查")
|
|
b = self.run_project("DEMO-B", "重点看现金流")
|
|
self.assertEqual(a["status"], "completed")
|
|
self.assertEqual(a["summary"]["open"], 0)
|
|
model = next(s for s in b["steps"] if s["step_id"] == "corporate.analysis")
|
|
self.assertEqual(model["trace"]["engine"], "mini-openclaw/AgentManager")
|
|
self.assertIn("DEMO-B", model["outputs"]["requirement_response"])
|
|
self.assertTrue(any(f["code"] == "financial.cashflow.missing" for f in b["findings"]))
|
|
self.assertFalse(set(e["id"] for e in a["evidence"]) & set(e["id"] for e in b["evidence"]))
|
|
with patch.dict("os.environ", {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1"}), patch("services.assistant_model.ChatOpenAI", return_value=StructuredToolModel(invalid=True)):
|
|
bad = self.run_project("DEMO-A")
|
|
self.assertEqual(bad["status"], "partial")
|
|
self.assertIn("未授权证据", next(s for s in bad["steps"] if s["step_id"] == "corporate.analysis")["errors"][0]["message"])
|
|
|
|
def test_feedback_proposal_uses_agent_and_stays_private_draft(self):
|
|
first = self.run_project()
|
|
feedback = self.post("/projects/DEMO-B/feedback", {"category": "rule_omission", "target_kind": "run", "target_id": first["id"], "note": "回款依赖项目需要凭证检查"}).json()
|
|
with patch.dict("os.environ", {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1"}), patch("services.assistant_model.ChatOpenAI", return_value=StructuredToolModel(propose=True)):
|
|
queued = self.post("/projects/DEMO-B/runs", {"feedback_id": feedback["id"], "requirement": "提炼经验草稿"}).json()
|
|
self.worker.process_next()
|
|
run = self.service.resource("run", queued["id"])
|
|
self.assertIn("proposed_experience_id", run)
|
|
draft = self.service.resource("experience", run["proposed_experience_id"])
|
|
self.assertEqual(draft["status"], "draft")
|
|
self.assertEqual(draft["scope"], "project")
|
|
self.assertEqual(draft["feedback_id"], feedback["id"])
|
|
self.assertEqual(self.project("DEMO-C")["experiences"], [])
|
|
|
|
def test_document_chunk_citation_is_persisted_with_true_locator(self):
|
|
with patch.dict("os.environ", {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1"}), patch("services.assistant_model.ChatOpenAI", return_value=StructuredToolModel(cite_document=True)):
|
|
run = self.run_project("DEMO-A")
|
|
self.assertEqual(run["status"], "completed")
|
|
refs = [e for e in run["evidence"] if e["field"] == "document.text"]
|
|
self.assertEqual(len(refs), 1)
|
|
self.assertEqual(refs[0]["locator"], "第1行")
|
|
self.assertTrue(refs[0]["source"].endswith(".csv"))
|
|
|
|
def test_component_publish_gate_and_frozen_queue(self):
|
|
queued = self.post("/projects/DEMO-B/runs").json()
|
|
with self.service.connect() as db:
|
|
original = get(db, "component_version", "corporate.coverage@1.0.0")
|
|
body = {k: v for k, v in original.items() if k not in ("revision", "component_id")}
|
|
body.update(id="corporate.coverage", version="1.0.1", name="新版覆盖测算")
|
|
candidate = draft(db, body, "author")
|
|
with self.assertRaises(ValueError):
|
|
release(db, candidate["id"], "reviewer")
|
|
evaluation = evaluate(db, candidate["id"], "author")
|
|
self.assertTrue(evaluation["passed"], evaluation)
|
|
release(db, candidate["id"], "reviewer")
|
|
self.worker.process_next()
|
|
old = self.service.resource("run", queued["id"])
|
|
self.assertEqual(next(s for s in old["components"]["skills"] if s["id"] == "corporate.coverage")["version"], "1.0.0")
|
|
newer = self.run_project()
|
|
self.assertEqual(next(s for s in newer["component_versions"] if s["id"] == "corporate.coverage")["version"], "1.0.1")
|
|
with self.service.connect() as db:
|
|
body.update(version="1.0.2", denominator="lease.total_value")
|
|
bad = draft(db, body, "author")
|
|
self.assertFalse(evaluate(db, bad["id"], "author")["passed"])
|
|
with self.assertRaises(ValueError):
|
|
release(db, bad["id"], "reviewer")
|
|
|
|
def test_cycles_missing_dependencies_and_output_conflicts(self):
|
|
for items in ([{"id": "a", "depends_on": ["b"]}], [{"id": "a", "depends_on": ["b"]}, {"id": "b", "depends_on": ["a"]}],
|
|
[{"id": "a", "engine": "ratio", "output": "same"}, {"id": "b", "engine": "ratio", "output": "same"}]):
|
|
with self.assertRaises(ValueError):
|
|
ordered(items)
|
|
|
|
|
|
# unittest's loader would otherwise repeat all integration tests through inheritance.
|
|
for name in list(fixtures.ProjectAssistantTests.__dict__):
|
|
if name.startswith("test_"):
|
|
setattr(ComponentTests, name, None)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|