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).
243 lines
8.8 KiB
Python
243 lines
8.8 KiB
Python
"""Coverage batch v2 (E1): cloud/wait_ops.py pure helpers -- row
|
|
access, handoff validation, pr-check parsing, wait polling cores,
|
|
diagnostics assembly. All direct calls; no production code touched.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from cloud.wait_ops import ( # noqa: E402
|
|
_normalize_json_checks,
|
|
_row_dict,
|
|
build_pr_diagnostics,
|
|
create_handoff,
|
|
decode_wait_cursor,
|
|
detect_significant_change,
|
|
get_handoff,
|
|
handoff_to_dict,
|
|
interruptible_sleep,
|
|
parse_pr_checks,
|
|
poll_for_first_change,
|
|
poll_task_statuses,
|
|
report_handoff,
|
|
row_get,
|
|
significant_now,
|
|
worktree_status,
|
|
)
|
|
|
|
|
|
class RowHelpersTestCase(unittest.TestCase):
|
|
def test_row_get_fallbacks(self):
|
|
self.assertIsNone(row_get(None, "x"))
|
|
self.assertIsNone(row_get({}, "missing"))
|
|
self.assertEqual(row_get({"a": 1}, "a"), 1)
|
|
|
|
def test_row_dict_none_and_tuple(self):
|
|
self.assertIsNone(_row_dict(None, ["id", "x"]))
|
|
self.assertEqual(_row_dict(("a", "b"), ["id", "x"]),
|
|
{"id": "a", "x": "b"})
|
|
|
|
def test_handoff_progress_non_int_zero(self):
|
|
d = handoff_to_dict({"id": "h", "progress": "abc"})
|
|
self.assertEqual(d["progress"], 0)
|
|
|
|
def test_handoff_progress_clamped(self):
|
|
d = handoff_to_dict({"id": "h", "progress": 150})
|
|
self.assertEqual(d["progress"], 100)
|
|
|
|
|
|
class HandoffValidationTestCase(unittest.TestCase):
|
|
def test_create_target_too_long(self):
|
|
with self.assertRaises(ValueError):
|
|
create_handoff(None, "t", "p", "x" * 201, "u")
|
|
|
|
def test_create_detail_not_string(self):
|
|
with self.assertRaises(ValueError):
|
|
create_handoff(None, "t", "p", "host", "u", detail=123)
|
|
|
|
def test_create_detail_too_long(self):
|
|
with self.assertRaises(ValueError):
|
|
create_handoff(None, "t", "p", "host", "u",
|
|
detail="x" * 2001)
|
|
|
|
def test_create_valid(self):
|
|
conn = sqlite3.connect(":memory:")
|
|
d = create_handoff(conn, "t1", "p1", " host-b ", "u1")
|
|
self.assertEqual(d["status"], "pending")
|
|
self.assertEqual(d["target_host"], "host-b")
|
|
conn.close()
|
|
|
|
def test_get_missing_none(self):
|
|
conn = sqlite3.connect(":memory:")
|
|
self.assertIsNone(get_handoff(conn, "ghost"))
|
|
conn.close()
|
|
|
|
def test_report_bad_status(self):
|
|
ok, err = report_handoff(None, "h", "bogus")
|
|
self.assertFalse(ok)
|
|
self.assertIn("status must be one of", err)
|
|
|
|
def test_report_progress_not_int(self):
|
|
ok, err = report_handoff(None, "h", "running", progress=True)
|
|
self.assertFalse(ok)
|
|
self.assertIn("must be an integer", err)
|
|
|
|
def test_report_detail_not_string(self):
|
|
ok, err = report_handoff(None, "h", "running", detail=123)
|
|
self.assertFalse(ok)
|
|
self.assertIn("must be a string", err)
|
|
|
|
def test_report_detail_too_long(self):
|
|
ok, err = report_handoff(None, "h", "running",
|
|
detail="x" * 2001)
|
|
self.assertFalse(ok)
|
|
self.assertIn("too long", err)
|
|
|
|
|
|
class ParseChecksTestCase(unittest.TestCase):
|
|
def test_parse_header_single_part_nomatch_fallback(self):
|
|
checks, all_pass, failed, pending = parse_pr_checks(
|
|
"name\tstatus\nonly\t\nhello world\npass\nci\tpassing")
|
|
names = [(c["name"], c["state"]) for c in checks]
|
|
self.assertIn(("pass", "pass"), names)
|
|
self.assertIn(("ci", "pass"), names)
|
|
self.assertTrue(all_pass)
|
|
|
|
def test_worktree_status_unknown(self):
|
|
s = worktree_status("paused", "/tmp/x", False, False)
|
|
self.assertEqual(s["status"], "unknown")
|
|
|
|
|
|
class WaitCoresTestCase(unittest.TestCase):
|
|
def test_sleep_stop_event(self):
|
|
event = threading.Event()
|
|
event.set()
|
|
slept, interrupted = interruptible_sleep(5, event)
|
|
self.assertEqual(slept, 0.0)
|
|
self.assertTrue(interrupted)
|
|
|
|
def test_sleep_full(self):
|
|
slept, interrupted = interruptible_sleep(0.05)
|
|
self.assertAlmostEqual(slept, 0.05, places=2)
|
|
self.assertFalse(interrupted)
|
|
|
|
def test_poll_statuses_bad_timeout(self):
|
|
all_terminal, statuses, waited, timed_out = poll_task_statuses(
|
|
lambda: {}, timeout_seconds="bad")
|
|
self.assertTrue(all_terminal)
|
|
|
|
def test_cursor_empty(self):
|
|
with self.assertRaises(ValueError):
|
|
decode_wait_cursor("")
|
|
|
|
def test_cursor_not_dict(self):
|
|
cur = base64.urlsafe_b64encode(b"[1]").decode()
|
|
with self.assertRaises(ValueError):
|
|
decode_wait_cursor(cur)
|
|
|
|
def test_cursor_value_not_dict(self):
|
|
cur = base64.urlsafe_b64encode(
|
|
json.dumps({"a": 1}).encode()).decode()
|
|
with self.assertRaises(ValueError):
|
|
decode_wait_cursor(cur)
|
|
|
|
def test_significant_dirty_numbers(self):
|
|
self.assertIsNone(significant_now(
|
|
{"status": "running", "pending_approvals": "x",
|
|
"pending_input": "y", "pending_followups": "z"}))
|
|
|
|
def test_significant_needs_approval(self):
|
|
self.assertEqual(significant_now(
|
|
{"status": "running", "pending_approvals": 1,
|
|
"pending_input": 0, "pending_followups": 0}),
|
|
"needs_approval")
|
|
|
|
def test_significant_all_clear(self):
|
|
self.assertIsNone(significant_now(
|
|
{"status": "running", "pending_approvals": 0,
|
|
"pending_input": 0, "pending_followups": 0}))
|
|
|
|
def test_detect_non_dict_current(self):
|
|
self.assertIsNone(detect_significant_change(
|
|
None, {"t1": "notadict"}, ["t1"]))
|
|
|
|
def test_detect_needs_approval(self):
|
|
base = {"t1": {"status": "running", "pending_approvals": 0,
|
|
"pending_input": 0, "pending_followups": 0}}
|
|
cur = {"t1": {"status": "running", "pending_approvals": 1,
|
|
"pending_input": 0, "pending_followups": 0}}
|
|
task_id, kind, detail = detect_significant_change(
|
|
base, cur, ["t1"])
|
|
self.assertEqual((task_id, kind), ("t1", "needs_approval"))
|
|
|
|
def test_detect_needs_continue(self):
|
|
base = {"t1": {"status": "running", "pending_approvals": 0,
|
|
"pending_input": 0, "pending_followups": 0}}
|
|
cur = {"t1": {"status": "running", "pending_approvals": 0,
|
|
"pending_input": 0, "pending_followups": 2}}
|
|
task_id, kind, detail = detect_significant_change(
|
|
base, cur, ["t1"])
|
|
self.assertEqual((task_id, kind), ("t1", "needs_continue"))
|
|
|
|
def test_poll_first_bad_timeout_immediate(self):
|
|
def fetch():
|
|
return {"t1": {"status": "completed",
|
|
"pending_approvals": 0,
|
|
"pending_input": 0,
|
|
"pending_followups": 0}}
|
|
changed, summaries, waited, timed_out = poll_for_first_change(
|
|
fetch, ["t1"], timeout_seconds="bad")
|
|
self.assertEqual(changed["kind"], "terminal")
|
|
self.assertFalse(timed_out)
|
|
|
|
def test_poll_first_loop_finds_change(self):
|
|
seq = [{"t1": {"status": "running", "pending_approvals": 0,
|
|
"pending_input": 0, "pending_followups": 0}},
|
|
{"t1": {"status": "completed", "pending_approvals": 0,
|
|
"pending_input": 0, "pending_followups": 0}}]
|
|
it = iter(seq)
|
|
def fetch():
|
|
return next(it, seq[-1])
|
|
changed, summaries, waited, timed_out = poll_for_first_change(
|
|
fetch, ["t1"], timeout_seconds=5, poll_seconds=0.01)
|
|
self.assertEqual(changed["kind"], "terminal")
|
|
self.assertFalse(timed_out)
|
|
|
|
|
|
class DiagnosticsTestCase(unittest.TestCase):
|
|
def test_normalize_non_list(self):
|
|
self.assertEqual(_normalize_json_checks("x"), [])
|
|
|
|
def test_normalize_skips_nameless(self):
|
|
checks = _normalize_json_checks(
|
|
[{"state": "pass"}, {"name": "ci", "state": "pass"}])
|
|
self.assertEqual([c["name"] for c in checks], ["ci"])
|
|
|
|
def test_build_both_routes_fail(self):
|
|
def run(cmd):
|
|
return (1, "", "boom")
|
|
with self.assertRaises(RuntimeError):
|
|
build_pr_diagnostics(run, 7)
|
|
|
|
def test_build_check_runs_fetch_failed(self):
|
|
def run(cmd):
|
|
if "--json" in cmd:
|
|
return (0, json.dumps(
|
|
[{"name": "ci", "state": "fail"}]), "")
|
|
return (0, "", "")
|
|
with mock.patch("cloud.wait_ops._fetch_failed_check_runs",
|
|
side_effect=RuntimeError("api down")):
|
|
diag = build_pr_diagnostics(run, 7, owner_repo="o/r",
|
|
head_sha="deadbeef")
|
|
self.assertFalse(diag["check_runs"]["available"])
|
|
self.assertIn("api down", diag["check_runs"]["note"])
|