"""新端点测试：记忆 / 知识库 / 成本 / 模型总览。 只用标准库 http.client，不跑真实模型，不联网。 每个测试用独立临时 SQLite 与临时端口。 """ import http.client import json import os import sys import tempfile import threading import unittest from pathlib import Path from urllib.parse import quote try: from cloud.auth import AuthStore from cloud.cost import CostStore from cloud.server import create_server except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.cost import CostStore from cloud.server import create_server class EndpointsTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") self.auth = AuthStore(self.db_path) self.owner_id, self.owner_key = self.auth.create_user("owner") self.viewer_id, self.viewer_key = self.auth.create_user("viewer") self.outsider_id, self.outsider_key = self.auth.create_user("outsider") self.server = create_server(self.db_path, "127.0.0.1", 0) self.port = self.server.server_address[1] self.thread = threading.Thread( target=self.server.serve_forever, daemon=True) self.thread.start() status, self.project = self._req( "POST", "/projects", {"name": "demo"}, self.owner_key) self.assertEqual(status, 201) self.auth.set_member( self.project["id"], self.owner_id, self.viewer_id, "viewer") # outsider 自己的项目，用于跨项目隔离测试。 status, self.other = self._req( "POST", "/projects", {"name": "other"}, self.outsider_key) self.assertEqual(status, 201) def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) try: self.auth.conn.close() except Exception: pass self.tmp.cleanup() def _req(self, method, path, body=None, key=None): conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) hdrs = {} payload = None if body is not None: payload = json.dumps(body).encode("utf-8") hdrs["Content-Type"] = "application/json" if key is not None: hdrs["X-API-Key"] = key try: conn.request(method, path, body=payload, headers=hdrs) resp = conn.getresponse() data = resp.read() finally: conn.close() try: parsed = json.loads(data.decode("utf-8")) if data else None except ValueError: parsed = None return resp.status, parsed def _pid(self): return self.project["id"] # -- 记忆 ---------------------------------------------------------- def test_memory_create_list_supersede(self): status, created = self._req( "POST", "/projects/%s/memory" % self._pid(), {"thread_id": "t1", "goal": "加法", "conclusion": "已加 add2", "files_changed": ["mod.py"]}, self.owner_key) self.assertEqual(status, 201, created) status, body = self._req( "GET", "/projects/%s/memory" % self._pid(), key=self.viewer_key) self.assertEqual(status, 200) self.assertEqual(len(body["summaries"]), 1) self.assertIn("add2", body["prompt_context"]) status, body = self._req( "POST", "/projects/%s/memory/%s/supersede" % (self._pid(), created["id"]), {}, self.owner_key) self.assertEqual(status, 200) self.assertEqual(body["status"], "deprecated") status, body = self._req( "GET", "/projects/%s/memory" % self._pid(), key=self.owner_key) self.assertEqual(body["summaries"], []) def test_memory_no_key_401_and_outsider_403(self): status, _ = self._req( "GET", "/projects/%s/memory" % self._pid()) self.assertEqual(status, 401) status, _ = self._req( "GET", "/projects/%s/memory" % self._pid(), key=self.outsider_key) self.assertEqual(status, 403) status, _ = self._req( "POST", "/projects/%s/memory" % self._pid(), {"goal": "g", "conclusion": "c"}, self.viewer_key) self.assertEqual(status, 403) def test_memory_cross_project_supersede_404(self): status, created = self._req( "POST", "/projects/%s/memory" % self._pid(), {"goal": "g", "conclusion": "c"}, self.owner_key) self.assertEqual(status, 201) # outsider 拿自己项目的路径去废弃别人的摘要：应 403/404，不成功。 status, _ = self._req( "POST", "/projects/%s/memory/%s/supersede" % (self.other["id"], created["id"]), {}, self.outsider_key) self.assertIn(status, (403, 404)) def test_memory_unknown_project_404(self): status, _ = self._req( "GET", "/projects/nope/memory", key=self.owner_key) self.assertEqual(status, 404) # -- 知识库 --------------------------------------------------------- def test_kb_index_search_and_isolation(self): status, _ = self._req( "POST", "/projects/%s/kb/index" % self._pid(), {"path": "README.md", "content": "用户登录流程说明", "kind": "readme"}, self.owner_key) self.assertEqual(status, 201) status, body = self._req( "GET", "/projects/%s/kb/search?q=%s" % (self._pid(), quote("登录")), key=self.viewer_key) self.assertEqual(status, 200) self.assertEqual(len(body["results"]), 1) # outsider 在自己项目里搜不到别人的文档。 status, body = self._req( "GET", "/projects/%s/kb/search?q=%s" % (self.other["id"], quote("登录")), key=self.outsider_key) self.assertEqual(status, 200) self.assertEqual(body["results"], []) # 无 key 401，越权 403。 status, _ = self._req( "GET", "/projects/%s/kb/search?q=%s" % (self._pid(), quote("登录"))) self.assertEqual(status, 401) status, _ = self._req( "GET", "/projects/%s/kb/search?q=%s" % (self._pid(), quote("登录")), key=self.outsider_key) self.assertEqual(status, 403) # -- 成本 ----------------------------------------------------------- def test_task_and_project_cost(self): status, task = self._req( "POST", "/tasks", {"project_id": self._pid(), "prompt": "做加法"}, self.owner_key) self.assertEqual(status, 201) store = CostStore(self.db_path) try: store.record(task["id"], self._pid(), "gpt-6.1-sol", input_tokens=100, output_tokens=50, usd_micros=1234, source="actual", fallback_count=1) store.record_estimated(task["id"], self._pid(), "gpt-6.1-sol", "prompt" * 10, "output" * 10) finally: store._conn.close() status, body = self._req( "GET", "/tasks/%s/cost" % task["id"], key=self.viewer_key) self.assertEqual(status, 200) self.assertEqual(body["actual_tokens"], 150) self.assertGreater(body["estimated_tokens"], 0) self.assertEqual(body["fallback_count"], 1) status, _ = self._req( "GET", "/tasks/%s/cost" % task["id"], key=self.outsider_key) self.assertEqual(status, 403) status, _ = self._req( "GET", "/tasks/nope/cost", key=self.owner_key) self.assertEqual(status, 404) status, body = self._req( "GET", "/projects/%s/cost" % self._pid(), key=self.owner_key) self.assertEqual(status, 200) self.assertEqual(body["overall"]["turns"], 2) # -- 模型总览 ------------------------------------------------------- def test_models_overview_isolated(self): status, task = self._req( "POST", "/tasks", {"project_id": self._pid(), "prompt": "x"}, self.owner_key) store = CostStore(self.db_path) try: store.record(task["id"], self._pid(), "gpt-6.1-sol", input_tokens=10, output_tokens=5) finally: store._conn.close() status, body = self._req("GET", "/models", key=self.owner_key) self.assertEqual(status, 200) self.assertIn("gpt-6.1-sol", body["priority_list"]) models = {m["model"] for m in body["model_stats"]} self.assertIn("gpt-6.1-sol", models) # outsider 只能看到自己项目的统计（目前为空），看不到别人的。 status, body = self._req("GET", "/models", key=self.outsider_key) self.assertEqual(status, 200) models = {m["model"] for m in body["model_stats"]} self.assertNotIn("gpt-6.1-sol", models) status, _ = self._req("GET", "/models") self.assertEqual(status, 401) if __name__ == "__main__": unittest.main()