Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
380 lines
17 KiB
Python
380 lines
17 KiB
Python
import os
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
|
||
from _common import SETTINGS, agent, jira_cfg, no_sleep
|
||
from fake_hub import FakeHub
|
||
from fake_jira import FakeJira
|
||
|
||
|
||
class RunnerTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.fake = FakeJira()
|
||
self.hub = FakeHub(SETTINGS)
|
||
self.dir = tempfile.TemporaryDirectory()
|
||
self.journal = agent.Journal(os.path.join(self.dir.name, "journal.jsonl"))
|
||
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
|
||
self.runner = agent.Runner(cfg, agent.HubClient(cfg), agent.JiraClient(cfg), self.journal)
|
||
self.sleeps = []
|
||
self.runner.sleep = self.sleeps.append
|
||
self.runner.jira.sleep = no_sleep
|
||
self.runner.idle_limit = 1
|
||
self.fake.add("AURORA-1", duedate="2026-10-20")
|
||
|
||
def tearDown(self):
|
||
self.fake.stop()
|
||
self.hub.stop()
|
||
self.dir.cleanup()
|
||
|
||
def kinds(self):
|
||
return [(r["kind"], r["command_id"]) for r in self.journal.records()]
|
||
|
||
def claim_in_flight(self, ctype, params, pause=0):
|
||
cid = self.hub.enqueue(ctype, params, pause=pause)
|
||
self.hub.queue.remove(cid)
|
||
self.hub.commands[cid]["status"] = "claimed"
|
||
self.hub.in_flight = cid
|
||
return cid
|
||
|
||
def test_processes_queue(self):
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.hub.enqueue("refresh", {"keys": ["AURORA-1"]})
|
||
self.runner.loop()
|
||
self.assertEqual(self.hub.results[1]["status"], "done")
|
||
self.assertEqual(self.hub.results[1]["result"], {"changed": ["duedate"]})
|
||
self.assertEqual(self.hub.results[1]["jira_calls"], 2)
|
||
self.assertEqual(self.hub.results[2]["status"], "done")
|
||
self.assertEqual(len(self.hub.states), 1)
|
||
self.assertEqual(self.hub.states[0]["issues"][0]["duedate"], "2026-10-23")
|
||
self.assertEqual(self.kinds(), [("claimed", 1), ("started", 1), ("finished", 1), ("acked", 1), ("claimed", 2), ("started", 2), ("finished", 2), ("acked", 2)])
|
||
self.assertEqual(self.sleeps, []) # пауз не было: last_call_at пуст, потом pause 0
|
||
|
||
def test_pause_waits_from_last_jira_call(self):
|
||
self.runner.jira.last_call_at = time.monotonic()
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=3)
|
||
self.runner.loop()
|
||
self.assertEqual(len(self.sleeps), 3)
|
||
self.assertAlmostEqual(sum(self.sleeps), 3, delta=0.1)
|
||
self.assertEqual(self.hub.results[1]["status"], "done")
|
||
|
||
def test_result_retried_when_hub_down(self):
|
||
self.hub.fail_results = 2
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.runner.loop()
|
||
self.assertEqual(self.hub.results[1]["status"], "done")
|
||
self.assertEqual(self.sleeps, [10, 10])
|
||
self.assertEqual(self.journal.unacked_finished(), [])
|
||
|
||
def test_result_retried_on_connection_error(self):
|
||
calls = []
|
||
real = self.runner.hub.result
|
||
|
||
def flaky(cid, payload):
|
||
calls.append(1)
|
||
if len(calls) < 3:
|
||
raise ConnectionError("нет связи")
|
||
return real(cid, payload)
|
||
|
||
self.runner.hub.result = flaky
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.runner.loop()
|
||
self.assertEqual(self.sleeps, [10, 10])
|
||
self.assertEqual(self.hub.results[1]["status"], "done")
|
||
|
||
def test_result_rejected_is_acked(self):
|
||
for cid, code in ((1, 404), (2, 409), (3, 422)):
|
||
self.hub.reject_results = {cid: code}
|
||
self.journal.finished(cid, "done", {}, None, 0, 0)
|
||
self.assertTrue(self.runner.send_result(self.journal.unacked_finished()[0]))
|
||
self.assertEqual(self.journal.unacked_finished(), [])
|
||
|
||
def test_result_rejected_in_loop(self):
|
||
self.hub.reject_results = {1: 404}
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.runner.loop()
|
||
self.assertNotIn(1, self.hub.results)
|
||
self.assertEqual(self.journal.unacked_finished(), [])
|
||
self.assertEqual(self.kinds()[-1], ("acked", 1))
|
||
|
||
def test_recover_unfinished_started(self):
|
||
cid = self.claim_in_flight("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-20"}, pause=99)
|
||
self.journal.started(cid)
|
||
self.runner.loop()
|
||
self.assertEqual(self.hub.results[cid]["result"], {"changed": []}) # уже так в Jira — записи нет
|
||
self.assertEqual(self.journal.unfinished_started(), [])
|
||
self.assertEqual(self.sleeps, []) # пауза при восстановлении не применяется
|
||
self.assertEqual([r[0] for r in self.fake.writes()], [])
|
||
|
||
def test_recover_ensure_issue_finds_duplicate(self):
|
||
self.fake.add("AURORA-7", reporter="me", summary="Новая задача")
|
||
before = len(self.fake.issues)
|
||
cid = self.claim_in_flight("ensure_issue", {"temp_key": "new:1", "issue_type": "task", "summary": "Новая задача"})
|
||
self.journal.started(cid)
|
||
self.runner.loop()
|
||
self.assertEqual(self.hub.results[cid]["status"], "done")
|
||
self.assertEqual(self.hub.results[cid]["result"], {"key": "AURORA-7", "existed": True})
|
||
self.assertEqual(len(self.fake.issues), before)
|
||
self.assertEqual(self.fake.writes(), [])
|
||
|
||
def test_recover_skips_command_not_claimed_anymore(self):
|
||
cid = self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.hub.queue.remove(cid)
|
||
self.hub.commands[cid]["status"] = "cancelled"
|
||
self.journal.started(cid)
|
||
self.runner.loop()
|
||
self.assertNotIn(cid, self.hub.results)
|
||
self.assertEqual(self.kinds(), [("started", cid), ("acked", cid)])
|
||
|
||
def test_recover_unacked_finished(self):
|
||
cid = self.claim_in_flight("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.journal.started(cid)
|
||
self.journal.finished(cid, "done", {"changed": ["duedate"]}, None, 2, 0)
|
||
self.hub.enqueue("refresh", {"keys": ["AURORA-1"]})
|
||
self.runner.loop()
|
||
self.assertEqual(self.hub.results[cid]["result"], {"changed": ["duedate"]})
|
||
self.assertEqual(self.hub.results[2]["status"], "done")
|
||
self.assertEqual(self.kinds()[:3], [("started", 1), ("finished", 1), ("acked", 1)])
|
||
|
||
def test_stop_during_pause(self):
|
||
self.runner.jira.last_call_at = time.monotonic()
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=5)
|
||
self.runner.sleep = lambda s: setattr(self.runner, "stop", True)
|
||
self.runner.loop()
|
||
self.assertEqual([r["kind"] for r in self.journal.records()], ["claimed"]) # started нет
|
||
self.assertEqual(self.hub.results, {})
|
||
self.assertEqual(self.hub.commands[1]["status"], "claimed")
|
||
|
||
def test_stop_during_pause_then_restart_executes_once(self):
|
||
self.runner.jira.last_call_at = time.monotonic()
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=5)
|
||
self.runner.sleep = lambda s: setattr(self.runner, "stop", True)
|
||
self.runner.loop()
|
||
self.assertEqual(self.journal.unfinished_started(), [1])
|
||
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
|
||
second = agent.Runner(cfg, agent.HubClient(cfg), agent.JiraClient(cfg), agent.Journal(self.journal.path))
|
||
second.sleep = self.sleeps.append
|
||
second.idle_limit = 1
|
||
second.loop()
|
||
self.assertEqual(self.hub.results[1]["status"], "done")
|
||
self.assertEqual(len(self.fake.writes()), 1)
|
||
self.assertEqual(self.journal.unfinished_started(), [])
|
||
self.assertEqual(self.sleeps, []) # при восстановлении пауза не применяется
|
||
|
||
def test_stop_between_claim_and_execute_is_recorded(self):
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
real = self.runner.hub.claim
|
||
|
||
def claim_then_stop(*a):
|
||
resp = real(*a)
|
||
self.runner.stop = True
|
||
return resp
|
||
|
||
self.runner.hub.claim = claim_then_stop
|
||
self.runner.loop()
|
||
self.assertEqual([r["kind"] for r in self.journal.records()], ["claimed"])
|
||
self.assertEqual(self.fake.writes(), [])
|
||
self.assertEqual(self.journal.unfinished_started(), [1])
|
||
|
||
def test_stop_during_execution_finishes_and_reports(self):
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.hub.enqueue("refresh", {"keys": ["AURORA-1"]})
|
||
self.runner.load_settings()
|
||
run = self.runner.executor.run
|
||
|
||
def run_and_signal(cmd):
|
||
self.runner.stop = True # сигнал пришёл во время выполнения
|
||
return run(cmd)
|
||
|
||
self.runner.executor.run = run_and_signal
|
||
self.runner.loop()
|
||
self.assertEqual(self.hub.results[1]["status"], "done")
|
||
self.assertNotIn(2, self.hub.results) # вторая команда не начата
|
||
self.assertEqual(self.kinds(), [("claimed", 1), ("started", 1), ("finished", 1), ("acked", 1)])
|
||
|
||
def test_stop_during_execution_hub_down_single_attempt(self):
|
||
self.hub.fail_results = 5
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.runner.load_settings()
|
||
run = self.runner.executor.run
|
||
self.runner.executor.run = lambda cmd: (setattr(self.runner, "stop", True), run(cmd))[1]
|
||
self.runner.loop()
|
||
self.assertEqual(self.sleeps, [])
|
||
self.assertEqual([r["command_id"] for r in self.journal.unacked_finished()], [1])
|
||
|
||
def test_unexpected_exception_becomes_failed(self):
|
||
self.runner.load_settings()
|
||
|
||
def boom(cmd):
|
||
raise KeyError("summary")
|
||
|
||
self.runner.executor.run = boom
|
||
self.hub.enqueue("ensure_issue", {"temp_key": "new:1", "issue_type": "task"})
|
||
self.runner.loop()
|
||
self.assertEqual(self.hub.results[1]["status"], "failed")
|
||
self.assertIn("KeyError", self.hub.results[1]["error"])
|
||
self.assertEqual(self.journal.unfinished_started(), [])
|
||
self.assertEqual(self.journal.unacked_finished(), [])
|
||
|
||
def test_recover_waits_when_hub_down(self):
|
||
cid = self.claim_in_flight("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-20"})
|
||
self.journal.started(cid)
|
||
real = self.runner.hub.command
|
||
calls = []
|
||
|
||
def flaky(i):
|
||
calls.append(1)
|
||
if len(calls) == 1:
|
||
raise ConnectionError("нет связи")
|
||
return real(i)
|
||
|
||
self.runner.hub.command = flaky
|
||
self.runner.loop()
|
||
self.assertEqual(self.sleeps, [10])
|
||
self.assertEqual(self.hub.results[cid]["status"], "done")
|
||
|
||
def test_settings_reload_on_version_change(self):
|
||
self.runner.load_settings()
|
||
self.runner.settings["settings_version"] = 0
|
||
self.runner.loop()
|
||
self.assertEqual(self.runner.settings["settings_version"], 1)
|
||
|
||
def test_settings_reloaded_before_command(self):
|
||
self.runner.load_settings()
|
||
self.hub.settings["settings_version"] = 2
|
||
self.hub.settings["pacing"] = dict(self.hub.settings["pacing"], ensure_fields=0)
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
seen = []
|
||
real = self.runner.execute
|
||
self.runner.execute = lambda cmd: (seen.append(self.runner.settings["settings_version"]), real(cmd))
|
||
self.runner.loop()
|
||
self.assertEqual(seen, [2])
|
||
|
||
def test_hub_down_at_start_keeps_polling(self):
|
||
calls = []
|
||
|
||
def hub_settings():
|
||
calls.append(1)
|
||
if len(calls) < 3:
|
||
raise ConnectionError("нет связи")
|
||
return dict(SETTINGS)
|
||
|
||
self.runner.hub.settings = hub_settings
|
||
self.runner.loop()
|
||
self.assertEqual(len(calls), 3)
|
||
self.assertEqual(self.sleeps, [1, 1]) # пол паузы на ошибках хаба — 1 с
|
||
|
||
def test_retry_after_connection_drop_creates_one_issue(self):
|
||
self.fake.drop_next_post_issue = True
|
||
self.hub.enqueue("ensure_issue", {"temp_key": "new:1", "issue_type": "task", "summary": "Новая задача"})
|
||
self.runner.loop()
|
||
r = self.hub.results[1]
|
||
self.assertEqual(r["status"], "done")
|
||
self.assertEqual(r["result"]["existed"], True)
|
||
self.assertEqual(len([k for k in self.fake.issues if k != "AURORA-1"]), 1)
|
||
self.assertAlmostEqual(sum(self.sleeps), 30, delta=1.5) # ожидание 30 с перед повтором
|
||
self.assertEqual(r["jira_calls"], 2) # поиск + повторный поиск (оборванный ответ не считается)
|
||
self.assertEqual([k for k, _ in self.kinds()].count("started"), 1)
|
||
|
||
def test_retry_exhausted_gives_failed_unknown(self):
|
||
self.runner.load_settings()
|
||
runs = []
|
||
|
||
def down(cmd):
|
||
runs.append(1)
|
||
return {"status": "failed", "result": None, "error": "нет связи", "jira_calls": 1, "rate_limited": 0, "retryable": True}
|
||
|
||
self.runner.executor.run = down
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.runner.loop()
|
||
self.assertEqual(len(runs), 4) # первый запуск + 3 повтора
|
||
r = self.hub.results[1]
|
||
self.assertEqual(r["status"], "failed")
|
||
self.assertIn("результат неизвестен", r["error"])
|
||
self.assertEqual(r["jira_calls"], 4)
|
||
|
||
def test_stop_during_retry_wait_leaves_command_for_recovery(self):
|
||
self.runner.load_settings()
|
||
self.runner.executor.run = lambda cmd: {"status": "failed", "result": None, "error": "x", "jira_calls": 1,
|
||
"rate_limited": 0, "retryable": True}
|
||
self.runner.sleep = lambda s: setattr(self.runner, "stop", True)
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.runner.loop()
|
||
self.assertEqual(self.journal.unfinished_started(), [1])
|
||
self.assertEqual(self.hub.results, {})
|
||
|
||
def test_agent_version_in_result(self):
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
self.runner.loop()
|
||
self.assertEqual(self.hub.results[1]["agent_version"], agent.AGENT_VERSION)
|
||
|
||
def test_hub_auth_error_waits_60(self):
|
||
def denied(*a):
|
||
raise agent.HttpError(401, "unauthorized")
|
||
|
||
self.runner.hub.claim = denied
|
||
self.runner.sleep = lambda s: (self.sleeps.append(s), setattr(self.runner, "stop", True))
|
||
self.runner.loop()
|
||
self.assertEqual(self.sleeps, [60])
|
||
|
||
def test_wrong_hub_token_401_real_hub(self):
|
||
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token="bad")
|
||
r = agent.Runner(cfg, agent.HubClient(cfg), agent.JiraClient(cfg), self.journal)
|
||
r.sleep = lambda s: (self.sleeps.append(s), setattr(r, "stop", True))
|
||
r.loop()
|
||
self.assertEqual(self.sleeps, [60])
|
||
|
||
def test_settings_reload_failure_retried_before_command(self):
|
||
self.runner.load_settings()
|
||
self.hub.settings["settings_version"] = 2
|
||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||
real = self.runner.hub.settings
|
||
calls = []
|
||
|
||
def flaky():
|
||
calls.append(1)
|
||
if len(calls) < 3:
|
||
raise ConnectionError("нет связи")
|
||
return real()
|
||
|
||
self.runner.hub.settings = flaky
|
||
self.runner.loop()
|
||
self.assertEqual(len(calls), 3)
|
||
self.assertEqual(self.sleeps, [1, 1])
|
||
self.assertEqual(self.runner.settings["settings_version"], 2)
|
||
self.assertEqual(self.hub.results[1]["status"], "done")
|
||
|
||
def test_claim_timeout_at_least_45(self):
|
||
seen = []
|
||
real = agent.Http.call
|
||
agent.Http.call = lambda self_, m, p, body=None, params=None, timeout=None: (seen.append(timeout), {"command": None, "settings_version": 1})[1]
|
||
try:
|
||
agent.HubClient(jira_cfg(self.fake, hub_url="http://x", hub_token="t", timeout=5)).claim(1, 1, None)
|
||
finally:
|
||
agent.Http.call = real
|
||
self.assertEqual(seen, [45])
|
||
|
||
def test_signal_handler(self):
|
||
import signal
|
||
from unittest import mock
|
||
handlers = {}
|
||
seen = {}
|
||
|
||
def fake_loop(runner):
|
||
handlers[signal.SIGINT](signal.SIGINT, None)
|
||
seen["stop_after_first"] = runner.stop
|
||
with self.assertRaises(SystemExit) as cm:
|
||
handlers[signal.SIGINT](signal.SIGINT, None)
|
||
seen["code"] = cm.exception.code
|
||
|
||
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
|
||
with mock.patch.object(agent.signal, "signal", lambda n, h: handlers.__setitem__(n, h)), \
|
||
mock.patch.object(agent.Runner, "loop", fake_loop):
|
||
agent.cmd_run(cfg, self.journal.path)
|
||
self.assertTrue(seen["stop_after_first"])
|
||
self.assertEqual(seen["code"], 130)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|