tests/test_xkb_memory_service_http.py
Rama main · réplica de lectura del repositorio original.
from __future__ import annotations
import importlib.util
import json
import os
import tempfile
import threading
import unittest
from unittest import mock
from pathlib import Path
from urllib.error import HTTPError
from urllib.request import Request, urlopen
SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "xkb_memory_service.py"
spec = importlib.util.spec_from_file_location("xkb_memory_service_http", SCRIPT)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
import sys
sys.modules[spec.name] = module
spec.loader.exec_module(module)
_wiki_patch = None
def setUpModule() -> None:
"""同 test_xkb_memory_service：不要讓測試載入真實知識庫或打 embedding API。"""
global _wiki_patch
from unittest import mock
_wiki_patch = mock.patch.object(module.KnowledgeCatalog, "_wiki_search", return_value=[])
_wiki_patch.start()
def tearDownModule() -> None:
if _wiki_patch is not None:
_wiki_patch.stop()
class XKBMemoryServiceHTTPTests(unittest.TestCase):
def setUp(self) -> None:
# 這是 HTTP 契約測試，不是語意後端測試。機器上沒有 gbrain / bun 時，
# 每個請求會卡滿子行程的 timeout 才靜靜回空陣列——原本那是 30 秒，
# 於是整個檔案看起來像「HTTP 服務壞了」。
self._timeout_patch = mock.patch.dict(os.environ, {"XKB_EMBEDDING_TIMEOUT": "1", "XKB_XBRAIN_TIMEOUT": "1"})
self._timeout_patch.start()
self.addCleanup(self._timeout_patch.stop)
self.tmp = tempfile.TemporaryDirectory()
self.store = module.Store(Path(self.tmp.name) / "memory.sqlite")
self.server = module.ThreadingHTTPServer(("127.0.0.1", 0), module.Handler)
self.server.store = self.store
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
self.base = f"http://127.0.0.1:{self.server.server_address[1]}"
session = self.store.open_session({"source": "http-fixture", "session_key": "http-session"})
self.store.start_turn({"session_id": session["session_id"], "turn_id": "http-turn", "query": "http query"})
self.completed = self.store.complete_turn(
"http-turn",
{"query": "http query", "answer": "user evidence", "content": {"answer": "user evidence"}},
)
def tearDown(self) -> None:
self.server.shutdown()
self.server.server_close()
# join 不設上限。原本 timeout=2，時間到就往下刪暫存目錄——如果那條
# 執行緒還在處理請求，它用的 sqlite 檔會在腳下被移走，同一個類別的
# 下一個測試就對著一個還在關閉的伺服器跑。這就是它在 discover 底下
# 偶爾出錯、單獨跑卻永遠通過的原因。shutdown() 已經呼叫過，join 會回來。
self.thread.join()
self.tmp.cleanup()
def post(self, path: str, payload: dict) -> tuple[int, dict]:
request = Request(
self.base + path,
data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urlopen(request, timeout=3) as response:
return response.status, json.loads(response.read())
except HTTPError as exc:
return exc.code, json.loads(exc.read())
def test_http_rejects_malformed_json_and_invalid_numeric_types(self) -> None:
request = Request(self.base + "/v1/recall", data=b"{not-json", headers={"Content-Type": "application/json"}, method="POST")
with self.assertRaises(HTTPError) as raised:
urlopen(request, timeout=3)
self.assertEqual(raised.exception.code, 400)
self.assertIn("malformed JSON", json.loads(raised.exception.read())["error"])
status, body = self.post("/v1/recall", {"query": "http query", "limit": "10"})
self.assertEqual(status, 400)
self.assertIn("limit must be an integer", body["error"])
def test_http_turn_start_returns_400_for_identity_conflicts(self) -> None:
session = self.store.open_session({"source": "http-fixture", "session_key": "identity-http", "namespace": "private"})
payload = {"session_id": session["session_id"], "turn_id": "identity-http-turn", "query": "same query"}
status, body = self.post("/v1/turns/start", payload)
self.assertEqual(status, 200)
self.assertFalse(body["resumed"])
status, body = self.post("/v1/turns/start", {**payload, "query": "different query"})
self.assertEqual(status, 400)
self.assertIn("query", body["error"])
status, body = self.post("/v1/turns/start", {**payload, "namespace": "shared"})
self.assertEqual(status, 400)
self.assertIn("namespace", body["error"])
def test_http_namespace_must_match_session_and_completion(self) -> None:
session = self.store.open_session({"source": "http-fixture", "session_key": "namespace-http", "namespace": "private"})
status, body = self.post("/v1/turns/start", {"session_id": session["session_id"], "turn_id": "namespace-http-turn", "query": "q", "namespace": "shared"})
self.assertEqual(status, 400)
self.assertIn("namespace", body["error"])
status, body = self.post("/v1/turns/start", {"session_id": session["session_id"], "turn_id": "namespace-http-turn", "query": "q"})
self.assertEqual(status, 200)
status, body = self.post("/v1/turns/namespace-http-turn/complete", {"query": "q", "answer": "a", "namespace": "shared"})
self.assertEqual(status, 400)
self.assertIn("namespace", body["error"])
def test_http_recall_exposes_context_governance_packet(self) -> None:
status, body = self.post("/v1/context", {"query": "http query", "namespace": "private"})
self.assertEqual(status, 200)
self.assertEqual(body["request_namespace"], "private")
self.assertIn("acl_policy", body)
self.assertIn("filtered_counts", body)
self.assertIn("semantic_backend", body)
self.assertIn("warnings", body)
class XKBMemoryServiceAuthTests(unittest.TestCase):
"""Once a token exists, identity must come from it and not from the body."""
TOKEN = "test-token-0123456789abcdef"
def setUp(self) -> None:
# 這是 HTTP 契約測試，不是語意後端測試。機器上沒有 gbrain / bun 時，
# 每個請求會卡滿子行程的 timeout 才靜靜回空陣列——原本那是 30 秒，
# 於是整個檔案看起來像「HTTP 服務壞了」。
self._timeout_patch = mock.patch.dict(os.environ, {"XKB_EMBEDDING_TIMEOUT": "1", "XKB_XBRAIN_TIMEOUT": "1"})
self._timeout_patch.start()
self.addCleanup(self._timeout_patch.stop)
self.tmp = tempfile.TemporaryDirectory()
self.store = module.Store(Path(self.tmp.name) / "memory.sqlite")
self.server = module.ThreadingHTTPServer(("127.0.0.1", 0), module.Handler)
self.server.store = self.store
self.server.auth = module.AuthPolicy({
"tokens": {
self.TOKEN: {"namespace": "team-a", "scopes": ["memory:read", "memory:write"]},
"readonly-token-0123456789ab": {"namespace": "team-a", "scopes": ["memory:read"]},
}
})
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
self.base = f"http://127.0.0.1:{self.server.server_address[1]}"
def tearDown(self) -> None:
self.server.shutdown()
self.server.server_close()
# join 不設上限。原本 timeout=2，時間到就往下刪暫存目錄——如果那條
# 執行緒還在處理請求，它用的 sqlite 檔會在腳下被移走，同一個類別的
# 下一個測試就對著一個還在關閉的伺服器跑。這就是它在 discover 底下
# 偶爾出錯、單獨跑卻永遠通過的原因。shutdown() 已經呼叫過，join 會回來。
self.thread.join()
self.tmp.cleanup()
def post(self, path: str, payload: dict, token: str | None = None) -> tuple[int, dict]:
headers = {"Content-Type": "application/json"}
if token:
headers["Authorization"] = f"Bearer {token}"
request = Request(self.base + path, data=json.dumps(payload).encode(), headers=headers, method="POST")
try:
with urlopen(request, timeout=3) as response:
return response.status, json.loads(response.read())
except HTTPError as exc:
return exc.code, json.loads(exc.read())
def test_health_stays_open_and_advertises_that_auth_is_on(self) -> None:
with urlopen(self.base + "/v1/health", timeout=3) as response:
body = json.loads(response.read())
self.assertTrue(body["ok"])
self.assertTrue(body["auth_required"])
def test_anonymous_is_refused_once_a_token_is_configured(self) -> None:
status, error = self.post("/v1/recall", {"query": "anything"})
self.assertEqual(status, 401)
self.assertIn("token", error["error"])
def test_invalid_token_is_refused(self) -> None:
status, _ = self.post("/v1/recall", {"query": "anything"}, token="wrong-token-0123456789abcd")
self.assertEqual(status, 401)
def test_token_namespace_cannot_be_overridden_by_the_body(self) -> None:
"""The whole point: claiming another namespace must fail, not silently work."""
status, error = self.post("/v1/recall", {"query": "anything", "namespace": "team-b"}, token=self.TOKEN)
self.assertEqual(status, 403)
status, body = self.post("/v1/recall", {"query": "anything"}, token=self.TOKEN)
self.assertEqual(status, 200)
self.assertEqual(body["request_namespace"], "team-a")
def test_scopes_separate_read_from_write(self) -> None:
status, _ = self.post("/v1/sessions/open", {"source": "x", "session_key": "k"}, token="readonly-token-0123456789ab")
self.assertEqual(status, 403)
status, _ = self.post("/v1/recall", {"query": "anything"}, token="readonly-token-0123456789ab")
self.assertEqual(status, 200)
def test_written_session_is_pinned_to_the_token_namespace(self) -> None:
status, session = self.post("/v1/sessions/open", {"source": "x", "session_key": "k", "namespace": "team-b"}, token=self.TOKEN)
self.assertEqual(status, 403)
status, session = self.post("/v1/sessions/open", {"source": "x", "session_key": "k"}, token=self.TOKEN)
self.assertEqual(status, 200)
self.assertEqual(session["namespace"], "team-a")
class AuthPolicyTests(unittest.TestCase):
def test_anonymous_only_while_no_token_exists(self) -> None:
self.assertTrue(module.AuthPolicy({}).allow_anonymous)
policy = module.AuthPolicy({"tokens": {"a-long-enough-token-value": {"namespace": "n"}}})
self.assertFalse(policy.allow_anonymous)
self.assertTrue(policy.enabled)
def test_config_errors_fail_closed(self) -> None:
with self.assertRaises(ValueError):
module.AuthPolicy({"tokens": {"short": {"namespace": "n"}}})
with self.assertRaises(ValueError):
module.AuthPolicy({"tokens": {"a-long-enough-token-value": {}}})
def test_unreadable_policy_file_is_an_error_not_open_access(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
bad = Path(tmp) / "auth.json"
bad.write_text("{not json", encoding="utf-8")
with self.assertRaises(ValueError):
module.AuthPolicy.load(bad)
# A missing file is the documented "no auth configured" default.
self.assertFalse(module.AuthPolicy.load(Path(tmp) / "absent.json").enabled)
if __name__ == "__main__":
unittest.main()