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.assertIn("mini-openclaw", assistant)
        self.assertIn("<ChatPanel/>", assistant)
        self.assertIn("IdentitySelector", assistant)
        self.assertIn("IdentitySelector", dashboard)
        self.assertIn("统计范围", dashboard)


if __name__ == "__main__":
    unittest.main()