test: code-to10 v2 coverage additions (14 files, production code unchanged)

(cherry picked from commit 473338c35d59bfd50929542e7f08c0a4a22c5987)

cloud/tests/test_v2cov_server_handlers4.py

@@ -0,0 +1,496 @@
+"""Coverage batch v2 (D4): cloud/server.py small handler clusters.
+
+Same stubbing discipline as D2/D3: the method under test is never
+stubbed; _send_json / _read_json are captured input, store classes
+are faked only where the test target is the close-swallow itself
+(the fake still performs the business call), and collaborator
+error injection covers only the handler's own error mapping.
+"""
+
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+from cloud.server import (
+    CloudControlPlaneHandler,
+    _auth_store,
+    init_db,
+)
+
+
+def _handler(db_path, upload_dir=None):
+    h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler)
+    h.server = SimpleNamespace(db_path=db_path, upload_dir=upload_dir)
+    sent = []
+    h._send_json = lambda status, obj, extra_headers=None: sent.append(
+        (status, obj))
+    h.sent = sent
+    return h
+
+
+def _conn(db_path):
+    c = sqlite3.connect(db_path)
+    c.row_factory = sqlite3.Row
+    return c
+
+
+class _Base(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "test.db")
+        init_db(self.db_path)
+        self.upload_dir = os.path.join(self.tmp.name, "uploads")
+        os.makedirs(self.upload_dir, exist_ok=True)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _h(self, body=None, err=None):
+        h = _handler(self.db_path, upload_dir=self.upload_dir)
+        h._read_json = lambda: (body, err)
+        return h
+
+    def _project(self, pid="p1", owner="u1", repo_url=None):
+        c = _conn(self.db_path)
+        c.execute(
+            "INSERT INTO projects(id, name, owner_id, created_at,"
+            " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)",
+            (pid, "P", owner, time.time(), repo_url))
+        c.commit()
+        c.close()
+        with _auth_store(self.db_path) as store:
+            store.create_project(pid, owner)
+
+    def _member(self, pid, uid, role):
+        c = _conn(self.db_path)
+        c.execute(
+            "INSERT OR REPLACE INTO project_members(project_id,"
+            " user_id, role) VALUES(?, ?, ?)", (pid, uid, role))
+        c.commit()
+        c.close()
+
+    def _task(self, tid="t1", pid="p1", status="completed",
+              created_by="u1", **extra):
+        cols = ["id", "project_id", "prompt", "status", "created_by",
+                "created_at", "updated_at"]
+        vals = [tid, pid, "do work", status, created_by, time.time(),
+                time.time()]
+        for key, val in extra.items():
+            cols.append(key)
+            vals.append(val)
+        c = _conn(self.db_path)
+        c.execute(
+            "INSERT INTO tasks(%s) VALUES(%s)"
+            % (", ".join(cols), ", ".join("?" * len(cols))), vals)
+        c.commit()
+        c.close()
+
+    def _env(self, eid="e1", pid="p1"):
+        c = _conn(self.db_path)
+        c.execute(
+            "INSERT INTO environments(id, project_id, name, branch,"
+            " executor, status, error_text, repo_url, created_by,"
+            " created_at, updated_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?,"
+            " ?, ?, ?)",
+            (eid, pid, "env", "main", "local", "ready", None,
+             "https://example.com/r.git", "u1", time.time(),
+             time.time()))
+        c.commit()
+        c.close()
+
+    def _last(self, h):
+        self.assertTrue(h.sent, "handler sent no response")
+        return h.sent[-1]
+
+
+class _ClosingConn:
+    def __init__(self):
+        self.conn = mock.Mock()
+        self.conn.close.side_effect = RuntimeError("close down")
+
+
+class _ClosingStore:
+    def __init__(self):
+        self.conn = _ClosingConn().conn
+
+    def close(self):
+        raise RuntimeError("close down")
+
+
+class EnvProjectMemberTestCase(_Base):
+    def test_get_environment_check_error(self):
+        self._project()
+        self._env()
+        h = self._h()
+        h._get_environment("u9", "e1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 403)
+
+    def test_archive_environment_missing_404(self):
+        h = self._h()
+        h._archive_environment("u1", "ghost")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_archive_environment_check_error(self):
+        self._project()
+        self._env()
+        h = self._h()
+        h._archive_environment("u9", "e1")
+        self.assertEqual(self._last(h)[0], 403)
+
+    def test_update_project_read_error(self):
+        self._project()
+        h = self._h(err="bad json")
+        h._update_project("u1", "p1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_resolve_invite_user_id_must_be_string(self):
+        h = self._h()
+        target, status, err = h._resolve_invite_target({"user_id": 42})
+        self.assertEqual(status, 400)
+        self.assertIsNone(target)
+
+    def test_resolve_invite_username_must_be_string(self):
+        h = self._h()
+        target, status, err = h._resolve_invite_target({"username": 42})
+        self.assertEqual(status, 400)
+        self.assertIsNone(target)
+
+    def test_member_invite_project_missing_404(self):
+        h = self._h(body={})
+        h._project_member_invite("u1", "ghost")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_member_invite_read_error(self):
+        self._project()
+        h = self._h(err="bad json")
+        h._project_member_invite("u1", "p1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_member_patch_project_missing_404(self):
+        h = self._h(body={})
+        h._project_member_patch("u1", "ghost", "u2")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_member_patch_read_error(self):
+        self._project()
+        h = self._h(err="bad json")
+        h._project_member_patch("u1", "p1", "u2")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_member_patch_unknown_role_400(self):
+        self._project()
+        h = self._h(body={"role": "superuser"})
+        h._project_member_patch("u1", "p1", "u2")
+        code, payload = self._last(h)
+        self.assertEqual(code, 400)
+        self.assertIn("unknown role", payload["error"])
+
+    def test_push_sub_delete_wrong_method_404(self):
+        h = self._h()
+        h._project_push_subscription_delete("u1", "GET", "p1",
+                                            ["p1", "push",
+                                             "subscriptions", "s1"])
+        self.assertEqual(self._last(h)[0], 404)
+
+    def _sub(self, sid="s1", pid="p1", uid="u2"):
+        from cloud.push import init_push_db
+        init_push_db(self.db_path)
+        c = _conn(self.db_path)
+        c.execute(
+            "INSERT INTO push_subscriptions(id, project_id, user_id,"
+            " channel, target, preference, created_at)"
+            " VALUES(?, ?, ?, ?, ?, ?, ?)",
+            (sid, pid, uid, "webhook", "https://example.com/hook",
+             "all", time.time()))
+        c.commit()
+        c.close()
+
+    def test_push_sub_delete_missing_404(self):
+        self._project()
+        h = self._h()
+        h._project_push_subscription_delete("u1", "DELETE", "p1",
+                                            ["p1", "push",
+                                             "subscriptions", "ghost"])
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_push_sub_delete_forbidden_403(self):
+        self._project(owner="owner1")
+        self._member("p1", "u1", "contributor")
+        self._sub()
+        h = self._h()
+        h._project_push_subscription_delete("u1", "DELETE", "p1",
+                                            ["p1", "push",
+                                             "subscriptions", "s1"])
+        self.assertEqual(self._last(h)[0], 403)
+
+
+class MemoryIdentityTestCase(_Base):
+    def test_memory_create_read_error(self):
+        h = self._h(err="bad json")
+        h._memory_create("u1", "p1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_memory_create_goal_required(self):
+        h = self._h(body={"conclusion": "done"})
+        h._memory_create("u1", "p1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 400)
+        self.assertIn("goal required", payload["error"])
+
+    def test_memory_create_close_failure_swallowed(self):
+        class _Store:
+            def __init__(self, db_path):
+                self.conn = _ClosingConn().conn
+
+            def add_summary(self, *a, **k):
+                return 99
+        h = self._h(body={"goal": "g", "conclusion": "c"})
+        with mock.patch("cloud.server.MemoryStore", _Store):
+            h._memory_create("u1", "p1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 201)
+        self.assertEqual(payload["id"], 99)
+
+    def test_memory_list_close_failure_swallowed(self):
+        class _Store:
+            def __init__(self, db_path):
+                self.conn = _ClosingConn().conn
+
+            def list_summaries(self, project_id):
+                return []
+
+            def build_prompt_context(self, project_id, top_k=5):
+                return ""
+        h = self._h()
+        with mock.patch("cloud.server.MemoryStore", _Store):
+            h._memory_list("p1", {})
+        self.assertEqual(self._last(h)[0], 200)
+
+    def test_memory_supersede_close_failure_swallowed(self):
+        from cloud.server import MemoryStore
+        store = MemoryStore(self.db_path)
+        try:
+            mid = store.add_summary("p1", "", "g", "c")
+        finally:
+            store.conn.close()
+
+        class _Store:
+            def __init__(self, db_path):
+                self.conn = _ClosingConn().conn
+
+            def deprecate(self, mid):
+                pass
+        h = self._h()
+        with mock.patch("cloud.server.MemoryStore", _Store):
+            h._memory_supersede("u1", "p1", str(mid))
+        code, payload = self._last(h)
+        self.assertEqual(code, 200)
+        self.assertEqual(payload["status"], "deprecated")
+
+    def test_identity_close_swallow(self):
+        h = self._h()
+        h._identity_close(_ClosingStore())
+        h._auto_memory_close(_ClosingStore())
+
+    def test_identity_profile_upsert_read_error(self):
+        h = self._h(err="bad json")
+        h._identity_profile_upsert("u1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_identity_profile_upsert_bad_preferences_400(self):
+        h = self._h(body={"preferences": "not-an-object"})
+        h._identity_profile_upsert("u1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 400)
+        self.assertIn("must be an object", payload["error"])
+
+    def test_identity_user_memory_create_read_error(self):
+        h = self._h(err="bad json")
+        h._identity_user_memory_create("u1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_identity_user_memory_create_bad_kind_400(self):
+        h = self._h(body={"kind": "bogus", "content": "x"})
+        h._identity_user_memory_create("u1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_identity_user_memory_supersede_read_error_body_404(self):
+        # Read error does not abort supersede: body defaults to {}
+        # (4294) and the missing-memory check still answers 404.
+        h = self._h(err="bad json")
+        h._identity_user_memory_supersede("u1", "ghost")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_identity_user_memory_supersede_key_error_404(self):
+        # Row check passes, then the supersede itself loses a race
+        # (KeyError) -- the handler maps it to the same 404.
+        class _Store:
+            def __init__(self, db_path):
+                self.conn = mock.Mock()
+
+            def get_memory(self, mem_id):
+                return {"scope": "user", "owner_user_id": "u1"}
+
+            def supersede_memory(self, *a, **k):
+                raise KeyError("gone")
+        h = self._h(body={})
+        with mock.patch("cloud.server.IdentityStore", _Store):
+            h._identity_user_memory_supersede("u1", "m1")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_identity_user_memory_delete_key_error_404(self):
+        class _Store:
+            def __init__(self, db_path):
+                self.conn = mock.Mock()
+
+            def get_memory(self, mem_id):
+                return {"scope": "user", "owner_user_id": "u1"}
+
+            def delete_memory(self, mem_id, user_id):
+                raise KeyError("gone")
+        h = self._h()
+        with mock.patch("cloud.server.IdentityStore", _Store):
+            h._identity_user_memory_delete("u1", "m1")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_identity_user_memory_supersede_bad_kind_400(self):
+        h = self._h(body={"kind": "fact", "content": "likes tea"})
+        h._identity_user_memory_create("u1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 201)
+        mem_id = payload["id"]
+        h2 = self._h(body={"replacement_kind": "bogus",
+                           "replacement_content": "x"})
+        h2._identity_user_memory_supersede("u1", mem_id)
+        self.assertEqual(self._last(h2)[0], 400)
+
+    def test_identity_user_memory_delete_missing_404(self):
+        h = self._h()
+        h._identity_user_memory_delete("u1", "ghost")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_identity_user_memory_delete_forbidden_404(self):
+        h = self._h(body={"kind": "fact", "content": "likes tea"})
+        h._identity_user_memory_create("u1")
+        mem_id = self._last(h)[1]["id"]
+        h2 = self._h()
+        h2._identity_user_memory_delete("u2", mem_id)
+        self.assertEqual(self._last(h2)[0], 404)
+
+    def test_identity_context_with_project(self):
+        self._project()
+        h = self._h()
+        h._identity_context("u1", {"project_id": ["p1"]})
+        code, payload = self._last(h)
+        self.assertEqual(code, 200)
+        self.assertEqual(payload["project_id"], "p1")
+
+    def test_identity_context_close_failure_swallowed(self):
+        self._project()
+
+        class _Store:
+            def __init__(self, db_path):
+                self.conn = _ClosingConn().conn
+
+            def build_prompt_context(self, project_id):
+                return ""
+        h = self._h()
+        with mock.patch("cloud.server.MemoryStore", _Store):
+            h._identity_context("u1", {"project_id": ["p1"]})
+        self.assertEqual(self._last(h)[0], 200)
+
+    def test_auto_memory_settings_read_error(self):
+        h = self._h(err="bad json")
+        h._auto_memory_settings_upsert("u1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_auto_memory_settings_body_must_be_object(self):
+        h = self._h(body=["not", "an", "object"])
+        h._auto_memory_settings_upsert("u1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 400)
+        self.assertIn("must be an object", payload["error"])
+
+    def test_auto_memory_log_bad_limit(self):
+        h = self._h()
+        h._auto_memory_log("u1", {"limit": ["abc"]})
+        code, payload = self._last(h)
+        self.assertEqual(code, 400)
+        self.assertIn("limit must be an integer", payload["error"])
+
+    def test_task_auto_memory_flags_read_error(self):
+        h = self._h(err="bad json")
+        h._task_auto_memory_flags("u1", "t1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_task_auto_memory_flags_body_must_be_object(self):
+        h = self._h(body=["x"])
+        h._task_auto_memory_flags("u1", "t1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_task_auto_memory_flags_check_error(self):
+        self._project()
+        self._task()
+        h = self._h(body={})
+        h._task_auto_memory_flags("u9", "t1")
+        self.assertEqual(self._last(h)[0], 403)
+
+    def test_project_identity_memory_create_read_error(self):
+        h = self._h(err="bad json")
+        h._project_identity_memory_create("u1", "p1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_project_identity_memories_list_bad_top_k(self):
+        h = self._h()
+        h._project_identity_memories_list("p1", {"top_k": ["abc"]})
+        code, payload = self._last(h)
+        self.assertEqual(code, 400)
+        self.assertIn("top_k must be an integer", payload["error"])
+
+    def test_project_identity_memory_supersede_none_body_404(self):
+        h = self._h(body=None)
+        h._project_identity_memory_supersede("u1", "p1", "ghost")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_project_identity_memory_supersede_read_error_body_404(self):
+        # Same read-error defaulting as the user-memory variant.
+        h = self._h(err="bad json")
+        h._project_identity_memory_supersede("u1", "p1", "ghost")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_project_identity_memory_supersede_key_error_404(self):
+        class _Store:
+            def __init__(self, db_path):
+                self.conn = mock.Mock()
+
+            def get_memory(self, mem_id):
+                return {"scope": "project", "project_id": "p1"}
+
+            def supersede_memory(self, *a, **k):
+                raise PermissionError("not yours")
+        h = self._h(body={})
+        with mock.patch("cloud.server.IdentityStore", _Store):
+            h._project_identity_memory_supersede("u1", "p1", "m1")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_project_identity_memory_supersede_bad_kind_400(self):
+        self._project()
+        h = self._h(body={"kind": "fact", "content": "likes tea"})
+        h._project_identity_memory_create("u1", "p1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 201)
+        mem_id = payload["id"]
+        h2 = self._h(body={"replacement_kind": "bogus",
+                           "replacement_content": "x"})
+        h2._project_identity_memory_supersede("u1", "p1", mem_id)
+        self.assertEqual(self._last(h2)[0], 400)