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

(cherry picked from commit 473338c35d59bfd50929542e7f08c0a4a22c5987)

cloud/tests/test_v2cov_server_handlers7.py

@@ -0,0 +1,615 @@
+"""Coverage batch v2 (D7): cloud/server.py module-level helpers and
+last small handler branches -- rate limiter, heartbeat, quota helpers,
+plan-params parsing, kill requests, gate validation, thread guards,
+oauth/github-app leftovers, todos/automation error mappings.
+
+Discipline as before: the function under test is never stubbed.
+"""
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from collections import deque
+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,
+    RateLimiter,
+    _apply_default_quota,
+    _auth_store,
+    _backfill_default_quotas,
+    _classify_goal_followup,
+    _deployed_commit,
+    _default_executor,
+    _full_auto_installation_verified,
+    _full_auto_parse_plan_params,
+    _make_rate_limiter,
+    _read_agent_entries,
+    _read_agent_entries_multi,
+    _resolve_shared_home_mcp,
+    _run_host_cli,
+    _task_to_dict,
+    _upload_dir_for,
+    _validate_hard_gate_body,
+    _worker_heartbeat_status,
+    init_db,
+    request_kill,
+    start_rate_limit_sweeper,
+    sweep_rate_limit_hits,
+    DbRateLimiter,
+)
+from cloud.server import _auth_store as _auth_store_cm
+
+
+def _handler(db_path, upload_dir=None, tenants_root=None):
+    h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler)
+    h.server = SimpleNamespace(db_path=db_path, upload_dir=upload_dir,
+                               tenants_root=tenants_root)
+    sent = []
+    h._send_json = lambda status, obj, extra_headers=None: sent.append(
+        (status, obj))
+    html = []
+    h._send_html = lambda status, body: html.append((status, body))
+    h.sent = sent
+    h.html = html
+    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,
+                     tenants_root=self.tmp.name)
+        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 _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 _last(self, h):
+        self.assertTrue(h.sent, "handler sent no response")
+        return h.sent[-1]
+
+
+class HelpersTestCase(_Base):
+    def test_run_host_cli_timeout(self):
+        fake = os.path.join(self.tmp.name, "host.py")
+        with open(fake, "w") as f:
+            f.write("# fake\n")
+        import subprocess as _sp
+        with mock.patch("cloud.server._host_py_path",
+                        return_value=fake), \
+                mock.patch("cloud.server.subprocess.run",
+                           side_effect=_sp.TimeoutExpired("host.py", 5)):
+            ok, err = _run_host_cli(["x"], "/tmp")
+        self.assertFalse(ok)
+        self.assertEqual(err, "host.py timed out")
+
+    def test_run_host_cli_os_error(self):
+        fake = os.path.join(self.tmp.name, "host.py")
+        with open(fake, "w") as f:
+            f.write("# fake\n")
+        with mock.patch("cloud.server._host_py_path",
+                        return_value=fake), \
+                mock.patch("cloud.server.subprocess.run",
+                           side_effect=OSError("no exec")):
+            ok, err = _run_host_cli(["x"], "/tmp")
+        self.assertFalse(ok)
+        self.assertIn("cannot run host.py", err)
+
+    def test_shared_home_tenants_root_raises(self):
+        with mock.patch("cloud.server._local_worker_codex_home",
+                        return_value=None), \
+                mock.patch("cloud.server._tenants_root_for",
+                           side_effect=RuntimeError("no root")):
+            self.assertIsNone(_resolve_shared_home_mcp(None))
+
+    def test_shared_home_tenants_root_empty(self):
+        with mock.patch("cloud.server._local_worker_codex_home",
+                        return_value=None), \
+                mock.patch("cloud.server._tenants_root_for",
+                           return_value=""):
+            self.assertIsNone(_resolve_shared_home_mcp(None))
+
+    def test_read_agent_entries_mixed_lines(self):
+        path = os.path.join(self.tmp.name, "agents.jsonl")
+        with open(path, "w") as f:
+            f.write("\n")
+            f.write("not json\n")
+            f.write("[1, 2]\n")
+            f.write('{"thread": "t", "nope": 1}\n')
+            f.write('{"thread": "t", "id": "a", "state": "start"}\n')
+            f.write('{"thread": "t", "id": "a", "state": "done"}\n')
+        entries = _read_agent_entries(path)
+        self.assertEqual(len(entries), 1)
+        self.assertEqual(entries[0]["state"], "done")
+        self.assertEqual(_read_agent_entries(
+            os.path.join(self.tmp.name, "missing.jsonl")), [])
+
+    def test_read_agent_entries_multi_overlap(self):
+        p1 = os.path.join(self.tmp.name, "a1.jsonl")
+        p2 = os.path.join(self.tmp.name, "a2.jsonl")
+        with open(p1, "w") as f:
+            f.write('{"thread": "t", "id": "a", "v": 1}\n')
+        with open(p2, "w") as f:
+            f.write('{"thread": "t", "id": "a", "v": 2}\n')
+        entries = _read_agent_entries_multi([p1, p2])
+        self.assertEqual(len(entries), 1)
+        self.assertEqual(entries[0]["v"], 2)
+
+    def test_rate_limiter_key_cleanup(self):
+        limiter = RateLimiter(per_minute=100, window_seconds=60,
+                              time_fn=lambda: 1000.0)
+        limiter._hits = {("k%d" % i): deque() for i in range(10001)}
+        self.assertTrue(limiter.allow("new"))
+        self.assertEqual(len(limiter._hits), 10001)
+
+    def test_db_limiter_ensure_twice(self):
+        limiter = DbRateLimiter(self.db_path)
+        c = _conn(self.db_path)
+        limiter._ensure_table(c)
+        limiter._ensure_table(c)
+        c.close()
+        self.assertTrue(limiter._ensured)
+
+    def test_db_limiter_ensure_inner_return(self):
+        # Inner early return needs a thread that passed the outer
+        # check, then blocks on the lock while the main thread flips
+        # _ensured: deterministic, no timing luck beyond the block.
+        limiter = DbRateLimiter(self.db_path)
+        limiter._ensured = False
+        c = _conn(self.db_path)
+        limiter._ensure_lock.acquire()
+        done = []
+        thread = threading.Thread(
+            target=lambda: done.append(limiter._ensure_table(c)))
+        thread.start()
+        time.sleep(0.2)
+        limiter._ensured = True
+        limiter._ensure_lock.release()
+        thread.join(timeout=5)
+        self.assertFalse(thread.is_alive())
+        c.close()
+
+    def test_make_rate_limiter_postgres(self):
+        with mock.patch.dict(os.environ, {"CLOUD_DB_BACKEND": "postgres"}):
+            limiter = _make_rate_limiter(self.db_path)
+        self.assertIsInstance(limiter, DbRateLimiter)
+
+    def test_sweep_postgres_no_lock(self):
+        conn = mock.Mock()
+        conn.execute.return_value.fetchone.return_value = (0,)
+        with mock.patch.dict(os.environ, {"CLOUD_DB_BACKEND": "postgres"}), \
+                mock.patch("cloud.server._db_connect",
+                           return_value=conn):
+            self.assertFalse(sweep_rate_limit_hits(self.db_path, 60))
+
+    def test_sweeper_loop_runs_and_dies(self):
+        limiter = DbRateLimiter(self.db_path)
+        with mock.patch("time.sleep",
+                        side_effect=[None, RuntimeError("stop")]), \
+                mock.patch("cloud.server.sweep_rate_limit_hits",
+                           side_effect=RuntimeError("sweep down")):
+            thread = start_rate_limit_sweeper(limiter, self.db_path)
+            self.assertIsNotNone(thread)
+            thread.join(timeout=5)
+            self.assertFalse(thread.is_alive())
+
+    def test_deployed_commit_missing_file(self):
+        self.assertIsNone(_deployed_commit())
+
+    def test_deployed_commit_success_path(self):
+        # Success path credits 1088/1091 in coverage's line map;
+        # the file lives in the repo root only for this test.
+        root = Path(__file__).resolve().parents[2]
+        target = root / "DEPLOYED_COMMIT"
+        self.assertFalse(target.exists(),
+                         "DEPLOYED_COMMIT must not pre-exist")
+        try:
+            target.write_text("  abc123\n", encoding="utf-8")
+            self.assertEqual(_deployed_commit(), "abc123")
+        finally:
+            target.unlink(missing_ok=True)
+
+    def test_worker_heartbeat_bad_override(self):
+        hb = os.path.join(self.tmp.name, "hb.json")
+        with open(hb, "w") as f:
+            json.dump({"last_seen": time.time()}, f)
+        server = SimpleNamespace(
+            db_path=None, worker_heartbeat_max_age_seconds="bogus",
+            worker_heartbeat_file=hb)
+        age, ok = _worker_heartbeat_status(server)
+        self.assertTrue(ok)
+
+    def test_worker_heartbeat_bad_env(self):
+        hb = os.path.join(self.tmp.name, "hb.json")
+        with open(hb, "w") as f:
+            json.dump({"last_seen": time.time()}, f)
+        server = SimpleNamespace(db_path=None, worker_heartbeat_file=hb)
+        with mock.patch.dict(
+                os.environ,
+                {"CLOUD_WORKER_HEARTBEAT_MAX_AGE_SECONDS": "bogus"}):
+            age, ok = _worker_heartbeat_status(server)
+        self.assertTrue(ok)
+
+    def test_upload_dir_fallback(self):
+        d = _upload_dir_for(SimpleNamespace())
+        self.assertIsInstance(d, str)
+        self.assertTrue(d)
+
+    def test_default_executor_local(self):
+        with mock.patch.dict(os.environ,
+                             {"CLOUD_DEFAULT_EXECUTOR": "local"}):
+            self.assertEqual(_default_executor(), "local")
+
+    def test_apply_default_quota_commit_failure(self):
+        c = _conn(self.db_path)
+
+        class _CommitFail:
+            def __init__(self, real):
+                self._real = real
+
+            def execute(self, *a, **k):
+                return self._real.execute(*a, **k)
+
+            def commit(self):
+                raise RuntimeError("commit down")
+
+            def __getattr__(self, name):
+                return getattr(self._real, name)
+        self.assertTrue(_apply_default_quota(_CommitFail(c), "u1"))
+        c.close()
+
+    def test_installation_verified_config_none(self):
+        c = _conn(self.db_path)
+        with mock.patch("cloud.server.get_app_config",
+                        return_value=None):
+            self.assertFalse(_full_auto_installation_verified(c))
+        c.close()
+
+    def test_installation_verified_config_error(self):
+        c = _conn(self.db_path)
+        with mock.patch("cloud.server.get_app_config",
+                        side_effect=RuntimeError("config down")):
+            self.assertFalse(_full_auto_installation_verified(c))
+        c.close()
+
+    def test_installation_verified_repo(self):
+        c = _conn(self.db_path)
+        with mock.patch("cloud.server.get_app_config",
+                        return_value={"id": 1}), \
+                mock.patch("cloud.server.find_installation_for_repo",
+                           return_value={"id": 9}):
+            self.assertTrue(
+                _full_auto_installation_verified(c, "o/r"))
+        c.close()
+
+    def test_installation_verified_any(self):
+        c = _conn(self.db_path)
+        with mock.patch("cloud.server.get_app_config",
+                        return_value={"id": 1}), \
+                mock.patch("cloud.server.list_installations",
+                           return_value=[{"id": 9}]):
+            self.assertTrue(_full_auto_installation_verified(c))
+        c.close()
+
+    def test_plan_params_checks_string(self):
+        params, err = _full_auto_parse_plan_params(
+            {}, {"repo": ["o/r"], "required_checks": ["a, b"]})
+        self.assertIsNone(err)
+        self.assertEqual(params["required_checks"], ("a", "b"))
+
+    def test_plan_params_checks_list(self):
+        params, err = _full_auto_parse_plan_params(
+            {"repo": "o/r", "required_checks": ["x", "y"]}, {})
+        self.assertIsNone(err)
+        self.assertEqual(params["required_checks"], ("x", "y"))
+
+    def test_plan_params_checks_bad_type(self):
+        params, err = _full_auto_parse_plan_params(
+            {"repo": "o/r", "required_checks": 5}, {})
+        self.assertIsNone(params)
+        self.assertIn("must be a list or string", err)
+
+    def test_plan_params_org_bad_type(self):
+        params, err = _full_auto_parse_plan_params(
+            {"repo": "o/r", "org": 123}, {})
+        self.assertIsNone(params)
+        self.assertIn("org must be a string", err)
+
+    def test_backfill_rollback_failure_zero(self):
+        c = _conn(self.db_path)
+
+        class _Failing:
+            def __init__(self, real):
+                self._real = real
+
+            def execute(self, sql, params=()):
+                if "INSERT OR IGNORE INTO user_quotas" in sql:
+                    raise RuntimeError("insert down")
+                return self._real.execute(sql, params)
+
+            def rollback(self):
+                raise RuntimeError("rollback down")
+
+            def __getattr__(self, name):
+                return getattr(self._real, name)
+        self.assertEqual(_backfill_default_quotas(_Failing(c)), 0)
+        c.close()
+
+    def test_auth_store_close_failure_swallowed(self):
+        class _FakeStore:
+            def __init__(self, db_path):
+                self.conn = mock.Mock()
+                self.conn.close.side_effect = RuntimeError("down")
+        with mock.patch("cloud.server.AuthStore", _FakeStore):
+            with _auth_store_cm(self.db_path) as store:
+                self.assertIsNotNone(store)
+
+    def _row(self, depends_on):
+        return {
+            "id": "t1", "project_id": "p1", "prompt": "p",
+            "status": "completed", "idempotency_key": None,
+            "depends_on": depends_on, "priority": 0,
+            "created_by": "u1", "created_at": 1.0, "updated_at": 2.0,
+            "last_heartbeat": None,
+        }
+
+    def test_task_to_dict_bad_depends_on(self):
+        d = _task_to_dict(self._row("not json"))
+        self.assertEqual(d["depends_on"], [])
+
+    def test_task_to_dict_non_list_depends_on(self):
+        d = _task_to_dict(self._row('{"a": 1}'))
+        self.assertEqual(d["depends_on"], [{"a": 1}])
+
+    def test_request_kill_connect_failure(self):
+        with mock.patch("cloud.server._connect",
+                        side_effect=RuntimeError("db down")):
+            self.assertFalse(request_kill(self.db_path, "t1", "u1"))
+
+    def test_request_kill_insert_failure(self):
+        c = _conn(self.db_path)
+
+        class _KillConn:
+            def __init__(self, real):
+                self._real = real
+
+            def execute(self, sql, params=()):
+                if "INSERT INTO kill_requests" in sql:
+                    raise RuntimeError("insert down")
+                return self._real.execute(sql, params)
+
+            def rollback(self):
+                raise RuntimeError("rollback down")
+
+            def __getattr__(self, name):
+                return getattr(self._real, name)
+        with mock.patch("cloud.server._connect",
+                        return_value=_KillConn(c)):
+            self.assertFalse(request_kill(self.db_path, "t1", "u1"))
+        c.close()
+
+    def test_gate_body_bad_entry(self):
+        self.assertIn("ci_checks entries must be objects",
+                      _validate_hard_gate_body(
+                          {"ci_checks": [{"name": 1}]}))
+
+    def test_gate_body_bad_bool(self):
+        self.assertIn("must be a boolean",
+                      _validate_hard_gate_body({"contract_frozen": "x"}))
+
+    def test_gate_body_bad_version(self):
+        self.assertIn("must be a string",
+                      _validate_hard_gate_body({"contract_version": 3}))
+
+    def test_gate_body_bad_expected(self):
+        self.assertIn("must be a string",
+                      _validate_hard_gate_body(
+                          {"contract_expected_version": 4}))
+
+    def test_classify_goal_followup_paused(self):
+        action, code, payload = _classify_goal_followup(
+            {"status": "paused"}, "做目标")
+        self.assertEqual(action, "error")
+        self.assertEqual(code, 409)
+
+
+class HandlerLeftoversTestCase(_Base):
+    def test_kb_index_close_failure_swallowed(self):
+        h = self._h(body={"path": "a.md", "content": "hi"})
+
+        class _Store:
+            def __init__(self, db_path):
+                self.conn = mock.Mock()
+                self.conn.close.side_effect = RuntimeError("down")
+
+            def index_file(self, project_id, path, content, kind="file"):
+                return "d1"
+        with mock.patch("cloud.server.KBStore", _Store):
+            h._kb_index("u1", "p1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 201)
+        self.assertEqual(payload["id"], "d1")
+
+    def test_create_followup_read_error(self):
+        self._project()
+        self._task()
+        h = self._h(err="bad json")
+        h._create_followup("u1", "t1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_thread_guard_cloud(self):
+        h = self._h()
+        code, msg = h._thread_task_guard({"executor": "cloud"})
+        self.assertEqual(code, 409)
+
+    def test_thread_mutate_resolve_error(self):
+        self._project()
+        self._task()
+        h = self._h()
+        h._resolve_task_thread = lambda task, **k: (
+            None, None, None, (404, "no thread"), None)
+        h._task_thread_mutate("u1", "t1", ["archive"],
+                              "thread_archived", "archived %s")
+        self.assertEqual(self._last(h)[0], 404)
+
+    def test_mcp_oauth_login_home_none(self):
+        h = self._h(body={})
+        h._mcp_server_oauth_login("u1", None, "srv")
+        code, payload = self._last(h)
+        self.assertEqual(code, 502)
+        self.assertIn("cannot resolve tenant codex-home",
+                      payload["error"])
+
+    def test_thread_read_no_thread_note(self):
+        self._project()
+        self._task()
+        h = self._h()
+        h._resolve_task_thread = lambda task, **k: (
+            {"id": "th1"}, "/tmp", "/tmp", None, None)
+        with mock.patch("cloud.server._run_host_cli",
+                        return_value=(True, "garbage")):
+            h._task_thread_read("u1", "t1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 200)
+        self.assertIn("no thread", str(payload))
+
+    def test_release_hard_gate_error(self):
+        self._project()
+        self._task(status="merged")
+        h = self._h(body={"files_changed": [], "ci_checks": "x"})
+        with mock.patch.dict(os.environ, {"CODEX_GLOBAL_ADMINS": "u1"}):
+            h._release_task("u1", "t1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 400)
+        self.assertIn("must be a list", payload["error"])
+
+    def test_coordinator_reserve_single_part(self):
+        h = self._h()
+        self.assertTrue(h._check_coordinator_reserve("u1", 1))
+
+    def test_oauth_login_close_failure_swallowed(self):
+        from cloud.server import OAuthStateStore as _RealStore
+
+        class _ClosingProxy:
+            def __init__(self, real):
+                self._real = real
+
+            def close(self):
+                raise RuntimeError("close down")
+
+            def __getattr__(self, name):
+                return getattr(self._real, name)
+
+        class _Store(_RealStore):
+            def __init__(self, db_path):
+                super().__init__(db_path)
+                self.conn = _ClosingProxy(self.conn)
+        h = self._h()
+        h.path = ("/oauth/github/login?client_id=c"
+                  "&redirect_uri=https://x.test/cb")
+        with mock.patch("cloud.server.OAuthStateStore", _Store):
+            h._oauth_login("/oauth/github/login")
+        # Either 200 (login started) or 400 (no client config): both
+        # pass through the close-failure swallow in the finally.
+        self.assertIn(self._last(h)[0], (200, 400))
+
+    def test_oauth_bind_initiate_read_error(self):
+        h = self._h(err="bad json")
+        h._authenticate = lambda: "u1"
+        h._oauth_bind_initiate("github")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_github_app_setup_page_org_action(self):
+        from cloud.github_app import SetupStateStore
+        store = SetupStateStore(self.db_path)
+        try:
+            state = store.create_state(issued_by="u1")
+        finally:
+            store.conn.close()
+        h = self._h()
+        h.path = ("/github-app/setup?state=%s&org=acme" % state)
+        with mock.patch.dict(
+                os.environ,
+                {"GITHUB_APP_PUBLIC_BASE": "https://api.example.test"}):
+            h._github_app_setup_page()
+        self.assertEqual(h.html[0][0], 200)
+        self.assertIn("organizations/acme", h.html[0][1])
+
+    def test_todos_create_read_error(self):
+        h = self._h(err="bad json")
+        h._todos_create("u1", "p1")
+        self.assertEqual(self._last(h)[0], 400)
+
+    def test_todos_delete_ghost_404(self):
+        h = self._h()
+        h._todos_delete("u1", "p1", "ghost")
+        code, payload = self._last(h)
+        self.assertEqual(code, 404)
+        self.assertIn("todo not found", payload["error"])
+
+    def test_automation_update_generic_failure_502(self):
+        h = self._h(body={"name": "new"})
+
+        class _FailAll:
+            def __init__(self, real):
+                self._real = real
+
+            def execute(self, *a, **k):
+                raise RuntimeError("db down")
+
+            def __getattr__(self, name):
+                return getattr(self._real, name)
+        h._db = lambda: _FailAll(_conn(self.db_path))
+        h._automation_update("u1", "p1", "a1")
+        self.assertEqual(self._last(h)[0], 502)