"""Tests for cloud/identity.py (system-level identity + memory). Scope of this batch, stated so a future reader does not over-read it: cloud layer only. No test here touches host.py, CODEX_HOME or any Codex thread; the store and the HTTP endpoints only read/write the control-plane database. Run from the repo root: python3 -m unittest discover -s cloud/tests -p "test_identity_memory.py" -v """ import http.client import json import os import sys import tempfile import threading import unittest from pathlib import Path try: from cloud.auth import AuthStore from cloud.db import POSTGRES_SCHEMA from cloud.identity import IdentityStore from cloud.migrations import MIGRATION_TABLES 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.db import POSTGRES_SCHEMA from cloud.identity import IdentityStore from cloud.migrations import MIGRATION_TABLES from cloud.server import create_server def _tmp_db(): fd, path = tempfile.mkstemp(suffix=".db") os.close(fd) os.unlink(path) return path class IdentityStoreTest(unittest.TestCase): def setUp(self): self.db = _tmp_db() self.store = IdentityStore(self.db) def tearDown(self): try: self.store.conn.close() finally: if os.path.exists(self.db): os.unlink(self.db) def test_profile_partial_update_keeps_other_fields(self): self.store.upsert_profile( "u1", display_name="Alice", timezone="Asia/Shanghai", preferences={"lang": "zh"}) got = self.store.upsert_profile("u1", display_name="Alicia") self.assertEqual(got["display_name"], "Alicia") self.assertEqual(got["timezone"], "Asia/Shanghai") self.assertEqual(got["preferences"], {"lang": "zh"}) def test_profile_rejects_secret_preferences(self): with self.assertRaises(ValueError): self.store.upsert_profile("u1", preferences={"api_key": "x"}) with self.assertRaises(ValueError): self.store.upsert_profile("u1", preferences={"token": "x"}) def test_user_memory_visible_only_to_owner(self): self.store.add_memory("u1", "user", "preference", "喜欢简洁中文") self.assertEqual(len(self.store.list_user_memories("u1")), 1) self.assertEqual(self.store.list_user_memories("u2"), []) self.assertNotIn("喜欢简洁中文", self.store.build_identity_context("u2")) def test_scope_project_validation(self): with self.assertRaises(ValueError): self.store.add_memory("u1", "user", "fact", "x", project_id="p1") with self.assertRaises(ValueError): self.store.add_memory("u1", "project", "fact", "x") with self.assertRaises(ValueError): self.store.add_memory("u1", "project", "nope", "x", project_id="p1") with self.assertRaises(ValueError): self.store.add_memory("u1", "project", "fact", "", project_id="p1") def test_supersede_chain_hides_old_from_context(self): old = self.store.add_memory("u1", "user", "preference", "旧偏好") _old, new = self.store.supersede_memory( old, "u1", replacement_content="新偏好") self.assertIsNotNone(new) ctx = self.store.build_identity_context("u1") self.assertIn("新偏好", ctx) self.assertNotIn("旧偏好", ctx) row = self.store.get_memory(old) self.assertEqual(row["status"], "superseded") self.assertEqual(row["superseded_by"], new) def test_supersede_user_memory_by_non_owner_forbidden(self): mid = self.store.add_memory("u1", "user", "fact", "我的事实") with self.assertRaises(PermissionError): self.store.supersede_memory(mid, "u2") with self.assertRaises(PermissionError): self.store.delete_memory(mid, "u2") def test_context_caps_and_project_section(self): for i in range(7): self.store.add_memory("u1", "user", "fact", "用户事实%d" % i) self.store.add_memory("u1", "project", "convention", "项目约定", project_id="p1") ctx = self.store.build_identity_context("u1", "p1") self.assertIn("[用户记忆]", ctx) self.assertIn("[项目记忆]", ctx) self.assertIn("项目约定", ctx) # Only the newest 5 user facts may appear. self.assertNotIn("用户事实0", ctx) self.assertNotIn("用户事实1", ctx) self.assertIn("用户事实6", ctx) # Without a project, no project section leaks in. self.assertNotIn("[项目记忆]", self.store.build_identity_context("u1")) def test_pg_schema_and_migration_lists_include_new_tables(self): schema_text = "\n".join(POSTGRES_SCHEMA) self.assertIn("CREATE TABLE IF NOT EXISTS user_profiles", schema_text) self.assertIn("CREATE TABLE IF NOT EXISTS identity_memories", schema_text) self.assertIn("user_profiles", MIGRATION_TABLES) self.assertIn("identity_memories", MIGRATION_TABLES) class IdentityEndpointsTest(unittest.TestCase): """HTTP-level: routes, self-only user scope, project permission.""" 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") 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_profile_and_user_memory_roundtrip(self): status, body = self._req( "PUT", "/users/me/profile", {"display_name": "Alice", "timezone": "Asia/Shanghai", "preferences": {"lang": "zh"}}, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["display_name"], "Alice") status, body = self._req( "GET", "/users/me/profile", key=self.owner_key) self.assertEqual(status, 200) self.assertEqual(body["preferences"], {"lang": "zh"}) status, body = self._req( "POST", "/users/me/memories", {"kind": "preference", "content": "喜欢简洁中文"}, self.owner_key) self.assertEqual(status, 201, body) status, body = self._req( "GET", "/users/me/memories", key=self.viewer_key) self.assertEqual(status, 200) # Viewer asks for *their own* memories: owner's must not leak. self.assertEqual(body["memories"], []) status, body = self._req( "GET", "/users/me/identity-context", key=self.owner_key) self.assertEqual(status, 200) self.assertIn("喜欢简洁中文", body["identity_context"]) def test_project_memory_permission_and_outsider_403(self): status, body = self._req( "POST", "/projects/%s/identity-memories" % self._pid(), {"kind": "convention", "content": "提交前先跑测试"}, self.owner_key) self.assertEqual(status, 201, body) status, body = self._req( "GET", "/projects/%s/identity-memories" % self._pid(), key=self.viewer_key) self.assertEqual(status, 200) self.assertEqual(len(body["memories"]), 1) status, _ = self._req( "GET", "/projects/%s/identity-memories" % self._pid(), key=self.outsider_key) self.assertEqual(status, 403) status, _ = self._req( "GET", "/users/me/identity-context?project_id=%s" % self._pid(), key=self.outsider_key) self.assertEqual(status, 403) def test_user_memory_supersede_and_delete_via_http(self): status, created = self._req( "POST", "/users/me/memories", {"kind": "fact", "content": "旧事实"}, self.owner_key) self.assertEqual(status, 201) mid = created["id"] status, body = self._req( "POST", "/users/me/memories/%s/supersede" % mid, {"replacement_content": "新事实"}, self.viewer_key) # Another user's memory: 404, not a leak-shaped 403. self.assertEqual(status, 404) status, body = self._req( "POST", "/users/me/memories/%s/supersede" % mid, {"replacement_content": "新事实"}, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["status"], "superseded") status, body = self._req( "DELETE", "/users/me/memories/%s" % body["replacement_id"], key=self.owner_key) self.assertEqual(status, 200) if __name__ == "__main__": unittest.main()