host
488aedbcad
(cherry picked from commit 1db02545f74c96b3b92be4fc32954659cec1f814) Reconcile-conflict: A14b E402 imports referencing cloud.handler_core (does not exist here; core methods live in server.py) — kept HEAD imports with incoming's noqa: E402 markers, dropped the unused core_mod import (verified unused elsewhere in both files).
938 lines
35 KiB
Python
938 lines
35 KiB
Python
"""Coverage batch v2 (D2): cloud/server.py handler validation branches.
|
|
|
|
Covers baseline gaps in _create_project, _create_environment,
|
|
_create_upload, _automation_create, _automation_bind,
|
|
_automation_heartbeat, _merge_task, _loop_run, _audit_loop_confirm,
|
|
_full_auto_execute and _create_task (gateway / attachment /
|
|
idempotency).
|
|
|
|
Stubbing discipline: the method under test is never stubbed. Its
|
|
sinks and predicate collaborators are: _send_json (captured output),
|
|
_read_json (request body), _db (a proxy connection that delegates
|
|
every statement except the one under test, for the TOCTOU
|
|
IntegrityError simulations -- the same pattern as the worker retry
|
|
rowcount-0 test), _validate_sub2api_key / _pushback_block /
|
|
_loop_verdict / _loop_confirm (collaborator error injection for the
|
|
handler's own error mapping), and _loop_ledger_store /
|
|
_loop_create_subtask (close/subtask failure injection). All other
|
|
logic -- validation, permission checks via the real AuthStore,
|
|
quota gates, driver result shaping -- runs for real.
|
|
"""
|
|
|
|
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))
|
|
|
|
import cloud.server as server_mod # noqa: E402
|
|
from cloud.server import CloudControlPlaneHandler, _auth_store, init_db # noqa: E402
|
|
from cloud.sub2api import Sub2apiError as _Sub2apiError # noqa: E402
|
|
|
|
|
|
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 _ProxyConn:
|
|
"""Delegate everything except the statements a test targets."""
|
|
|
|
def __init__(self, real, fail_insert_prefix=None,
|
|
fail_insert_exc=None, first_select_none=None,
|
|
close_exc=None):
|
|
self._real = real
|
|
self._fail_insert_prefix = fail_insert_prefix
|
|
self._fail_insert_exc = fail_insert_exc
|
|
self._first_select_none = first_select_none
|
|
self._seen_select = 0
|
|
self._close_exc = close_exc
|
|
|
|
def execute(self, sql, params=()):
|
|
text = sql.strip()
|
|
if (self._fail_insert_prefix
|
|
and text.startswith(self._fail_insert_prefix)):
|
|
raise self._fail_insert_exc
|
|
if (self._first_select_none
|
|
and self._first_select_none in text
|
|
and text.startswith("SELECT")):
|
|
self._seen_select += 1
|
|
if self._seen_select == 1:
|
|
class _Empty:
|
|
def fetchone(self):
|
|
return None
|
|
|
|
def fetchall(self):
|
|
return []
|
|
return _Empty()
|
|
return self._real.execute(sql, params)
|
|
|
|
def close(self):
|
|
if self._close_exc is not None:
|
|
raise self._close_exc
|
|
return self._real.close()
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._real, name)
|
|
|
|
|
|
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 _last(self, h):
|
|
self.assertTrue(h.sent, "handler sent no response")
|
|
return h.sent[-1]
|
|
|
|
|
|
class ProjectEnvUploadTestCase(_Base):
|
|
def test_create_project_read_error(self):
|
|
h = self._h(err="bad json")
|
|
h._create_project("u1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_create_project_name_required(self):
|
|
h = self._h(body={})
|
|
h._create_project("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("name required", payload["error"])
|
|
|
|
def test_create_project_id_must_be_string(self):
|
|
h = self._h(body={"name": "P", "id": 42})
|
|
h._create_project("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("id must be a string", payload["error"])
|
|
|
|
def test_create_project_auto_pr_bool(self):
|
|
# Bool branch (True -> 1): the 3351 line.
|
|
h = self._h(body={"name": "P", "auto_pr": True})
|
|
h._create_project("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 201)
|
|
self.assertEqual(payload["auto_pr"], 1)
|
|
|
|
def test_create_project_existing_409(self):
|
|
self._project()
|
|
h = self._h(body={"name": "P2", "id": "p1"})
|
|
h._create_project("u1")
|
|
self.assertEqual(self._last(h)[0], 409)
|
|
|
|
def test_create_project_integrity_error_409(self):
|
|
# TOCTOU simulation: SELECT finds nothing, INSERT races and
|
|
# loses; the handler must map it to the same 409. Each _db()
|
|
# call gets a fresh proxy so closing one connection cannot
|
|
# poison the next.
|
|
h = self._h(body={"name": "P", "id": "p9"})
|
|
h._db = lambda: _ProxyConn(
|
|
_conn(self.db_path),
|
|
fail_insert_prefix="INSERT INTO projects",
|
|
fail_insert_exc=sqlite3.IntegrityError("race"))
|
|
h._create_project("u1")
|
|
self.assertEqual(self._last(h)[0], 409)
|
|
|
|
def test_create_environment_read_error(self):
|
|
h = self._h(err="bad json")
|
|
h._create_environment("u1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_create_environment_project_id_required(self):
|
|
h = self._h(body={})
|
|
h._create_environment("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("project_id required", payload["error"])
|
|
|
|
def test_create_environment_name_too_long(self):
|
|
self._project(repo_url="https://example.com/r.git")
|
|
h = self._h(body={"project_id": "p1", "name": "x" * 101})
|
|
h._create_environment("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("name too long", payload["error"])
|
|
|
|
def test_create_environment_integrity_error_409(self):
|
|
self._project(repo_url="https://example.com/r.git")
|
|
h = self._h(body={"project_id": "p1", "name": "env1"})
|
|
h._db = lambda: _ProxyConn(
|
|
_conn(self.db_path),
|
|
fail_insert_prefix="INSERT INTO environments",
|
|
fail_insert_exc=sqlite3.IntegrityError("race"))
|
|
h._create_environment("u1")
|
|
self.assertEqual(self._last(h)[0], 409)
|
|
|
|
def _multipart(self, parts):
|
|
boundary = "BOUNDARY"
|
|
out = b""
|
|
for headers, data in parts:
|
|
out += ("--%s\r\n" % boundary).encode()
|
|
for key, val in headers:
|
|
out += ("%s: %s\r\n" % (key, val)).encode()
|
|
out += b"\r\n" + data + b"\r\n"
|
|
out += ("--%s--\r\n" % boundary).encode()
|
|
return out, "multipart/form-data; boundary=" + boundary
|
|
|
|
def test_create_upload_content_length_value_error(self):
|
|
h = self._h()
|
|
h.headers = {"Content-Type": "multipart/form-data; boundary=X",
|
|
"Content-Length": "not-a-number"}
|
|
h._read_body = lambda: b""
|
|
h._create_upload("u1")
|
|
code, payload = self._last(h)
|
|
# Length falls back to 0, then the empty body fails parsing.
|
|
self.assertEqual(code, 400)
|
|
|
|
def test_create_upload_too_large(self):
|
|
from cloud.uploads import MAX_RAW_BYTES
|
|
h = self._h()
|
|
h.headers = {"Content-Type": "multipart/form-data; boundary=X",
|
|
"Content-Length": str(MAX_RAW_BYTES + 1)}
|
|
h._read_body = lambda: b"x"
|
|
h._create_upload("u1")
|
|
self.assertEqual(self._last(h)[0], 413)
|
|
|
|
def test_create_upload_parse_error(self):
|
|
raw, ctype = self._multipart([])
|
|
h = self._h()
|
|
h.headers = {"Content-Type": ctype,
|
|
"Content-Length": str(len(b"garbage"))}
|
|
h._read_body = lambda: b"garbage"
|
|
h._create_upload("u1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_create_upload_no_files(self):
|
|
# Zero parts: fields and files are both empty, so the paths
|
|
# check passes vacuously and the no-files branch fires.
|
|
raw, ctype = self._multipart([])
|
|
h = self._h()
|
|
h.headers = {"Content-Type": ctype,
|
|
"Content-Length": str(len(raw))}
|
|
h._read_body = lambda: raw
|
|
h._create_upload("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("no files", payload["error"])
|
|
|
|
def test_create_upload_paths_mismatch(self):
|
|
raw, ctype = self._multipart([
|
|
([("Content-Disposition",
|
|
'form-data; name="f1"; filename="a.txt"'),
|
|
("Content-Type", "text/plain")], b"hello"),
|
|
([("Content-Disposition",
|
|
'form-data; name="f2"; filename="b.txt"'),
|
|
("Content-Type", "text/plain")], b"world"),
|
|
([("Content-Disposition", 'form-data; name="paths"')],
|
|
b'["a.txt"]')])
|
|
h = self._h()
|
|
h.headers = {"Content-Type": ctype,
|
|
"Content-Length": str(len(raw))}
|
|
h._read_body = lambda: raw
|
|
h._create_upload("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("数量不一致", payload["error"])
|
|
|
|
|
|
class AutomationTestCase(_Base):
|
|
def _auto(self, name="auto1", project_id="p1", prompt="do it"):
|
|
from cloud.automations import create_automation
|
|
c = _conn(self.db_path)
|
|
auto = create_automation(c, project_id, name, prompt, "u1")
|
|
c.close()
|
|
return auto
|
|
|
|
def _bind(self, aid, tid, mode="heartbeat"):
|
|
from cloud.lifecycle import bind_automation
|
|
c = _conn(self.db_path)
|
|
binding = bind_automation(c, aid, "p1", tid, mode)
|
|
c.close()
|
|
return binding
|
|
|
|
def test_automation_create_read_error(self):
|
|
h = self._h(err="bad json")
|
|
h._automation_create("u1", "p1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_automation_create_interval_none_defaults(self):
|
|
self._project()
|
|
h = self._h(body={"name": "a1", "prompt": "p",
|
|
"interval_seconds": None})
|
|
h._automation_create("u1", "p1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 201)
|
|
self.assertEqual(payload["interval_seconds"], 86400)
|
|
|
|
def test_automation_create_bad_first_run(self):
|
|
self._project()
|
|
h = self._h(body={"name": "a1", "prompt": "p",
|
|
"first_run_at": "not-a-time"})
|
|
h._automation_create("u1", "p1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("first_run_at", payload["error"])
|
|
|
|
def test_automation_create_run_once_must_be_bool(self):
|
|
self._project()
|
|
h = self._h(body={"name": "a1", "prompt": "p",
|
|
"run_once": "yes"})
|
|
h._automation_create("u1", "p1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("run_once", payload["error"])
|
|
|
|
def test_automation_create_generic_error_502(self):
|
|
self._project()
|
|
h = self._h(body={"name": "a1", "prompt": "p"})
|
|
with mock.patch("cloud.handler_automation._auto_create",
|
|
side_effect=RuntimeError("db boom")):
|
|
h._automation_create("u1", "p1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 502)
|
|
self.assertIn("db boom", payload["error"])
|
|
|
|
def test_automation_bind_read_error(self):
|
|
h = self._h(err="bad json")
|
|
h._automation_bind("u1", "p1", "a1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_automation_bind_missing_404(self):
|
|
h = self._h(body={"task_id": "t1"})
|
|
h._automation_bind("u1", "p1", "nope")
|
|
self.assertEqual(self._last(h)[0], 404)
|
|
|
|
def test_automation_bind_cancelled_409(self):
|
|
auto = self._auto()
|
|
c = _conn(self.db_path)
|
|
c.execute("UPDATE automations SET status='cancelled' WHERE id=?",
|
|
(auto["id"],))
|
|
c.commit()
|
|
c.close()
|
|
self._task()
|
|
h = self._h(body={"task_id": "t1"})
|
|
h._automation_bind("u1", "p1", auto["id"])
|
|
self.assertEqual(self._last(h)[0], 409)
|
|
|
|
def test_automation_bind_target_missing_404(self):
|
|
auto = self._auto()
|
|
h = self._h(body={"task_id": "ghost"})
|
|
h._automation_bind("u1", "p1", auto["id"])
|
|
self.assertEqual(self._last(h)[0], 404)
|
|
|
|
def test_automation_bind_target_cross_project_404(self):
|
|
auto = self._auto()
|
|
self._project(pid="p2", owner="u2")
|
|
self._task(tid="t2", pid="p2", created_by="u2")
|
|
h = self._h(body={"task_id": "t2"})
|
|
h._automation_bind("u1", "p1", auto["id"])
|
|
self.assertEqual(self._last(h)[0], 404)
|
|
|
|
def test_automation_bind_bad_mode_400(self):
|
|
auto = self._auto()
|
|
self._task()
|
|
h = self._h(body={"task_id": "t1", "mode": "bogus"})
|
|
h._automation_bind("u1", "p1", auto["id"])
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("heartbeat or cron", payload["error"])
|
|
|
|
def test_automation_heartbeat_read_error(self):
|
|
h = self._h(err="bad json")
|
|
h._automation_heartbeat("u1", "p1", "a1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_automation_heartbeat_missing_404(self):
|
|
h = self._h(body={})
|
|
h._automation_heartbeat("u1", "p1", "nope")
|
|
self.assertEqual(self._last(h)[0], 404)
|
|
|
|
def test_automation_heartbeat_no_binding_409(self):
|
|
auto = self._auto()
|
|
h = self._h(body={})
|
|
h._automation_heartbeat("u1", "p1", auto["id"])
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 409)
|
|
self.assertIn("no heartbeat binding", payload["error"])
|
|
|
|
def test_automation_heartbeat_cron_binding_409(self):
|
|
auto = self._auto()
|
|
self._task()
|
|
self._bind(auto["id"], "t1", mode="cron")
|
|
h = self._h(body={})
|
|
h._automation_heartbeat("u1", "p1", auto["id"])
|
|
self.assertEqual(self._last(h)[0], 409)
|
|
|
|
def test_automation_heartbeat_bound_task_missing_404(self):
|
|
auto = self._auto()
|
|
self._task()
|
|
self._bind(auto["id"], "t1")
|
|
c = _conn(self.db_path)
|
|
c.execute("DELETE FROM tasks WHERE id='t1'")
|
|
c.commit()
|
|
c.close()
|
|
h = self._h(body={})
|
|
h._automation_heartbeat("u1", "p1", auto["id"])
|
|
self.assertEqual(self._last(h)[0], 404)
|
|
|
|
def test_automation_heartbeat_bound_task_cross_project_404(self):
|
|
auto = self._auto()
|
|
self._task()
|
|
self._bind(auto["id"], "t1")
|
|
c = _conn(self.db_path)
|
|
c.execute("UPDATE tasks SET project_id='p2' WHERE id='t1'")
|
|
c.commit()
|
|
c.close()
|
|
h = self._h(body={})
|
|
h._automation_heartbeat("u1", "p1", auto["id"])
|
|
self.assertEqual(self._last(h)[0], 404)
|
|
|
|
def test_automation_heartbeat_quota_reject_returns(self):
|
|
# The bare return after the quota gate: quota lookup failure
|
|
# makes the gate answer 503, and heartbeat must return
|
|
# without delivering anything.
|
|
auto = self._auto()
|
|
self._task()
|
|
self._bind(auto["id"], "t1")
|
|
h = self._h(body={})
|
|
h._quota_status = mock.Mock(side_effect=RuntimeError("down"))
|
|
h._automation_heartbeat("u1", "p1", auto["id"])
|
|
self.assertEqual(len(h.sent), 1)
|
|
self.assertEqual(h.sent[0][0], 503)
|
|
|
|
def test_automation_heartbeat_message_must_be_string(self):
|
|
auto = self._auto()
|
|
self._task()
|
|
self._bind(auto["id"], "t1")
|
|
h = self._h(body={"message": 123})
|
|
h._automation_heartbeat("u1", "p1", auto["id"])
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("message required", payload["error"])
|
|
|
|
def test_automation_heartbeat_message_too_long(self):
|
|
auto = self._auto()
|
|
self._task()
|
|
self._bind(auto["id"], "t1")
|
|
h = self._h(body={"message": "x" * 20001})
|
|
h._automation_heartbeat("u1", "p1", auto["id"])
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("message too long", payload["error"])
|
|
|
|
|
|
class MergeLoopTestCase(_Base):
|
|
def _state(self):
|
|
return SimpleNamespace(
|
|
module_id="m", contract_version="v1", pushback_round=0,
|
|
failure_count=0, audit_fail_count=0,
|
|
acceptance_fail_count=0, rebuild_no=0, targeted_used=False,
|
|
status="open", pending_confirmation=False, retained_tests=[],
|
|
retained_passed_items=[], previous_artifacts=[], history=[])
|
|
|
|
def _loop_result(self, subtask=None):
|
|
return {
|
|
"next_state": self._state(), "action_kind": "verdict",
|
|
"reason": "r", "merge_blocked": False,
|
|
"release_blocked": False, "requires_reaudit": False,
|
|
"escalation_grants_user_acceptance": False,
|
|
"notification": None, "subtask": subtask}
|
|
|
|
def _confirm_result(self, subtask=None):
|
|
result = self._loop_result(subtask)
|
|
result.pop("requires_reaudit")
|
|
result.pop("escalation_grants_user_acceptance")
|
|
return result
|
|
|
|
def _closing_db(self):
|
|
# First _db() call serves _fetch_task (real); later calls
|
|
# serve the loop connection whose close() raises, so the
|
|
# finally-block except-pass lines run.
|
|
calls = {"n": 0}
|
|
|
|
def _db():
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
return _conn(self.db_path)
|
|
return _ProxyConn(_conn(self.db_path),
|
|
close_exc=RuntimeError("close boom"))
|
|
return _db
|
|
|
|
def _closing_store(self):
|
|
class _Store:
|
|
def close(self):
|
|
raise RuntimeError("store close boom")
|
|
return _Store()
|
|
|
|
def test_merge_read_error(self):
|
|
self._project()
|
|
self._task(status="completed")
|
|
h = self._h(err="bad json")
|
|
h._merge_task("u1", "t1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_merge_files_changed_must_be_list(self):
|
|
self._project()
|
|
self._task(status="completed")
|
|
h = self._h(body={"files_changed": "x"})
|
|
h._merge_task("u1", "t1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("files_changed must be a list", payload["error"])
|
|
|
|
def test_merge_pushback_unavailable_409(self):
|
|
# A1 改写：旧用例以 CLOUD_MERGE_GATE_MODE=off + 空 body 走到
|
|
# 推回检查；服务端删模式后空 body 在硬门禁体校验前已 400，
|
|
# 故改用完整硬证据 body，仍钉“推回状态不可用 → 409”这一性质。
|
|
self._project()
|
|
self._task(status="completed")
|
|
h = self._h(body={
|
|
"files_changed": ["apps/web/a.py"],
|
|
"owned_dir": "apps/web",
|
|
"branch_protection": "protected",
|
|
"ci_status": "success",
|
|
"ci_checks": [{"name": "ci", "status": "success"}],
|
|
"required_checks": ["ci"],
|
|
"contract_version": "v2",
|
|
"contract_frozen": True,
|
|
"coordinator_approved": True,
|
|
"unresolved_review_comments": 0,
|
|
})
|
|
with mock.patch("cloud.server._pushback_block",
|
|
side_effect=RuntimeError("chain down")):
|
|
h._merge_task("u1", "t1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 409)
|
|
self.assertEqual(payload["error"],
|
|
"audit pushback state unavailable")
|
|
self.assertTrue(payload["audit_pushback"]["merge_blocked"])
|
|
|
|
def test_merge_pushback_unavailable_audit_fails_too(self):
|
|
# A1 改写：同上，off + 空 body 改为完整硬证据 body，仍钉
|
|
# “审计本身也失败时 handler 仍回同一 409”。
|
|
# The audit best-effort except-pass: even when the audit
|
|
# write itself raises, the handler still answers the same 409.
|
|
self._project()
|
|
self._task(status="completed")
|
|
h = self._h(body={
|
|
"files_changed": ["apps/web/a.py"],
|
|
"owned_dir": "apps/web",
|
|
"branch_protection": "protected",
|
|
"ci_status": "success",
|
|
"ci_checks": [{"name": "ci", "status": "success"}],
|
|
"required_checks": ["ci"],
|
|
"contract_version": "v2",
|
|
"contract_frozen": True,
|
|
"coordinator_approved": True,
|
|
"unresolved_review_comments": 0,
|
|
})
|
|
real_auth = server_mod._auth_store
|
|
calls = {"n": 0}
|
|
|
|
def _auth_store(db_path):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
return real_auth(db_path)
|
|
raise RuntimeError("audit down")
|
|
with mock.patch("cloud.server._pushback_block",
|
|
side_effect=RuntimeError("chain down")), \
|
|
mock.patch("cloud.server._auth_store",
|
|
side_effect=_auth_store):
|
|
h._merge_task("u1", "t1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 409)
|
|
self.assertEqual(payload["error"],
|
|
"audit pushback state unavailable")
|
|
|
|
def test_loop_run_task_missing_404(self):
|
|
h = self._h()
|
|
h._loop_run("u1", "ghost", {}, "verdict")
|
|
self.assertEqual(self._last(h)[0], 404)
|
|
|
|
def test_loop_run_forbidden_403(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h()
|
|
h._loop_run("u9", "t1", {}, "verdict")
|
|
self.assertEqual(self._last(h)[0], 403)
|
|
|
|
def test_loop_run_verdict_body_400(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h()
|
|
h._loop_run("u1", "t1", {}, "verdict")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("verdict dict", payload["error"])
|
|
|
|
def test_loop_run_fix_body_400(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h()
|
|
h._loop_run("u1", "t1", {}, "fix")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_loop_run_correction_body_400(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h()
|
|
h._loop_run("u1", "t1", {}, "correction")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_loop_run_unknown_kind_400(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h()
|
|
h._loop_run("u1", "t1", {}, "bogus")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("unknown loop kind", payload["error"])
|
|
|
|
def test_loop_run_driver_value_error_400(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h()
|
|
with mock.patch("cloud.server._loop_verdict",
|
|
side_effect=ValueError("driver nope")):
|
|
h._loop_run("u1", "t1", {"verdict": {"verdict": "fail"}},
|
|
"verdict")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_loop_run_driver_exception_503(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h()
|
|
with mock.patch("cloud.server._loop_verdict",
|
|
side_effect=RuntimeError("ledger down")):
|
|
h._loop_run("u1", "t1", {"verdict": {"verdict": "fail"}},
|
|
"verdict")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 503)
|
|
self.assertEqual(payload["error"], "audit_loop_unavailable")
|
|
|
|
def test_loop_run_close_failures_swallowed_200(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h()
|
|
h._db = self._closing_db()
|
|
h._loop_ledger_store = lambda: self._closing_store()
|
|
with mock.patch("cloud.server._loop_verdict",
|
|
return_value=self._loop_result()):
|
|
h._loop_run("u1", "t1", {"verdict": {"verdict": "fail"}},
|
|
"verdict")
|
|
self.assertEqual(self._last(h)[0], 200)
|
|
|
|
def test_loop_run_subtask_error_503(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h()
|
|
h._loop_create_subtask = mock.Mock(
|
|
side_effect=RuntimeError("insert boom"))
|
|
with mock.patch("cloud.server._loop_verdict",
|
|
return_value=self._loop_result(
|
|
subtask={"kind": "rebuild"})):
|
|
h._loop_run("u1", "t1", {"verdict": {"verdict": "fail"}},
|
|
"verdict")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 503)
|
|
self.assertIn("subtask_error", payload)
|
|
self.assertIn("subtask_note", payload)
|
|
|
|
def test_confirm_task_missing_404(self):
|
|
h = self._h(body={})
|
|
h._audit_loop_confirm("u1", "ghost")
|
|
self.assertEqual(self._last(h)[0], 404)
|
|
|
|
def test_confirm_forbidden_403(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h(body={})
|
|
h._audit_loop_confirm("u9", "t1")
|
|
self.assertEqual(self._last(h)[0], 403)
|
|
|
|
def test_confirm_read_error(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h(err="bad json")
|
|
h._audit_loop_confirm("u1", "t1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_confirm_driver_value_error_400(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h(body={"user_confirmed": True, "confirmed_by": "u1"})
|
|
with mock.patch("cloud.server._loop_confirm",
|
|
side_effect=ValueError("driver nope")):
|
|
h._audit_loop_confirm("u1", "t1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_confirm_driver_exception_503(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h(body={"user_confirmed": True, "confirmed_by": "u1"})
|
|
with mock.patch("cloud.server._loop_confirm",
|
|
side_effect=RuntimeError("ledger down")):
|
|
h._audit_loop_confirm("u1", "t1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 503)
|
|
self.assertEqual(payload["error"], "audit_loop_unavailable")
|
|
|
|
def test_confirm_close_failures_swallowed_200(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h(body={"user_confirmed": True, "confirmed_by": "u1"})
|
|
h._db = self._closing_db()
|
|
h._loop_ledger_store = lambda: self._closing_store()
|
|
with mock.patch("cloud.server._loop_confirm",
|
|
return_value=self._confirm_result()):
|
|
h._audit_loop_confirm("u1", "t1")
|
|
self.assertEqual(self._last(h)[0], 200)
|
|
|
|
def test_confirm_subtask_error_503(self):
|
|
self._project()
|
|
self._task()
|
|
h = self._h(body={"user_confirmed": True, "confirmed_by": "u1"})
|
|
h._loop_create_subtask = mock.Mock(
|
|
side_effect=RuntimeError("insert boom"))
|
|
with mock.patch("cloud.server._loop_confirm",
|
|
return_value=self._confirm_result(
|
|
subtask={"kind": "rebuild"})):
|
|
h._audit_loop_confirm("u1", "t1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 503)
|
|
self.assertIn("subtask_error", payload)
|
|
|
|
|
|
class FullAutoCreateTaskTestCase(_Base):
|
|
def _full_auto_h(self, body, err=None, admin=False):
|
|
h = self._h(body, err)
|
|
h._audit_admin = lambda *args: None
|
|
return h
|
|
|
|
def test_full_auto_read_error(self):
|
|
h = self._full_auto_h(None, err="bad json")
|
|
h._full_auto_execute("u1")
|
|
self.assertEqual(self._last(h)[0], 400)
|
|
|
|
def test_full_auto_plan_error(self):
|
|
h = self._full_auto_h({})
|
|
h._full_auto_execute("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("repo must be", payload["error"])
|
|
|
|
def test_full_auto_not_admin_denied(self):
|
|
h = self._full_auto_h({"repo": "o/r"})
|
|
with mock.patch.dict(os.environ, {}, clear=False):
|
|
os.environ.pop("CODEX_GLOBAL_ADMINS", None)
|
|
h._full_auto_execute("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 403)
|
|
self.assertEqual(payload["reason"], "not_global_admin")
|
|
|
|
def test_full_auto_action_must_be_string(self):
|
|
h = self._full_auto_h({"repo": "o/r", "action": 123})
|
|
with mock.patch.dict(os.environ, {"CODEX_GLOBAL_ADMINS": "u1"}):
|
|
h._full_auto_execute("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("action must be a string", payload["error"])
|
|
|
|
def test_full_auto_disabled_denied(self):
|
|
h = self._full_auto_h({"repo": "o/r"})
|
|
with mock.patch.dict(os.environ, {"CODEX_GLOBAL_ADMINS": "u1"}):
|
|
h._full_auto_execute("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 403)
|
|
self.assertEqual(payload["reason"], "full_auto_disabled")
|
|
|
|
def _install(self, repo="o/r"):
|
|
from cloud.github_app import get_app_config
|
|
c = _conn(self.db_path)
|
|
get_app_config(c)
|
|
c.execute(
|
|
"INSERT OR REPLACE INTO github_app_config(id, app_id, slug,"
|
|
" client_id, webhook_secret, pem, html_url, created_at)"
|
|
" VALUES(1, '1', 'app', 'cid', 'sec', NULL, NULL, ?)",
|
|
(time.time(),))
|
|
from cloud.github_app import _ensure_installations_table
|
|
_ensure_installations_table(c)
|
|
c.execute(
|
|
"INSERT INTO github_app_installations(installation_id,"
|
|
" repo_full_name, created_at) VALUES(?, ?, ?)",
|
|
(123, repo, time.time()))
|
|
c.commit()
|
|
c.close()
|
|
|
|
def test_full_auto_token_missing_denied(self):
|
|
self._install()
|
|
h = self._full_auto_h({"repo": "o/r"})
|
|
h.server.full_auto_token_resolver = lambda db_path, repo: None
|
|
with mock.patch.dict(os.environ, {"CODEX_GLOBAL_ADMINS": "u1",
|
|
"FULL_AUTO_ENABLED": "1"}):
|
|
h._full_auto_execute("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 403)
|
|
self.assertEqual(payload["reason"], "token_missing")
|
|
|
|
def test_full_auto_protection_request_error_denied(self):
|
|
self._install()
|
|
h = self._full_auto_h({"repo": "o/r"})
|
|
h.server.full_auto_token_resolver = lambda db_path, repo: "tok"
|
|
from cloud.full_auto import FullAutoError
|
|
h.server.full_auto_http_client = mock.Mock(
|
|
side_effect=FullAutoError("net down", reason="net"))
|
|
with mock.patch.dict(os.environ, {"CODEX_GLOBAL_ADMINS": "u1",
|
|
"FULL_AUTO_ENABLED": "1"}):
|
|
h._full_auto_execute("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 403)
|
|
self.assertEqual(payload["reason"],
|
|
"branch_protection_unverified")
|
|
|
|
def test_full_auto_protection_not_200_denied(self):
|
|
self._install()
|
|
h = self._full_auto_h({"repo": "o/r"})
|
|
h.server.full_auto_token_resolver = lambda db_path, repo: "tok"
|
|
h.server.full_auto_http_client = lambda m, u, hd, b: (404, {})
|
|
with mock.patch.dict(os.environ, {"CODEX_GLOBAL_ADMINS": "u1",
|
|
"FULL_AUTO_ENABLED": "1"}):
|
|
h._full_auto_execute("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 403)
|
|
self.assertEqual(payload["reason"],
|
|
"branch_protection_unverified")
|
|
|
|
def test_create_task_gateway_sub2api_error_502(self):
|
|
self._project()
|
|
h = self._h(body={"project_id": "p1", "prompt": "hello"})
|
|
with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api"}), \
|
|
mock.patch("cloud.server._validate_sub2api_key",
|
|
side_effect=_Sub2apiError("key dead")):
|
|
h._create_task("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 502)
|
|
self.assertIn("key dead", payload["error"])
|
|
|
|
def test_create_task_gateway_generic_error_502(self):
|
|
self._project()
|
|
h = self._h(body={"project_id": "p1", "prompt": "hello"})
|
|
with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api"}), \
|
|
mock.patch("cloud.server._validate_sub2api_key",
|
|
side_effect=RuntimeError("net down")):
|
|
h._create_task("u1")
|
|
self.assertEqual(self._last(h)[0], 502)
|
|
|
|
def _upload_row(self, uid="up1"):
|
|
c = _conn(self.db_path)
|
|
c.execute(
|
|
"INSERT INTO uploads(id, created_by, created_at, status)"
|
|
" VALUES(?, ?, ?, 'pending')",
|
|
(uid, "u1", time.time()))
|
|
c.commit()
|
|
c.close()
|
|
|
|
def test_create_task_attachment_dir_missing_400(self):
|
|
self._project()
|
|
self._upload_row()
|
|
h = self._h(body={"project_id": "p1", "prompt": "hello",
|
|
"upload_id": "up1"})
|
|
h._create_task("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("暂存目录缺失", payload["error"])
|
|
|
|
def test_create_task_attachment_empty_400(self):
|
|
self._project()
|
|
self._upload_row()
|
|
os.makedirs(os.path.join(self.upload_dir, "up1"), exist_ok=True)
|
|
h = self._h(body={"project_id": "p1", "prompt": "hello",
|
|
"upload_id": "up1"})
|
|
h._create_task("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 400)
|
|
self.assertIn("附件包为空", payload["error"])
|
|
|
|
def test_create_task_idempotency_race_returns_existing(self):
|
|
self._project()
|
|
self._task(tid="t-old", idempotency_key="k1")
|
|
h = self._h(body={"project_id": "p1", "prompt": "hello",
|
|
"idempotency_key": "k1"})
|
|
h._db = lambda: _ProxyConn(
|
|
_conn(self.db_path),
|
|
fail_insert_prefix="INSERT INTO tasks",
|
|
fail_insert_exc=sqlite3.IntegrityError("race"),
|
|
first_select_none="idempotency_key = ?")
|
|
h._create_task("u1")
|
|
code, payload = self._last(h)
|
|
self.assertEqual(code, 200)
|
|
self.assertEqual(payload["id"], "t-old")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|