import json import os import tempfile import threading import time import unittest from unittest import mock from datetime import datetime, timedelta, timezone from _common import SETTINGS, agent, eva_cfg, no_sleep from fake_hub import FakeHub FETCH = {"url": "https://eva.example.ru/files/a.png", "name": "a.png"} STUB_ERROR = "wiki_fetch: не реализовано" # Executor каркаса: обе команды — failed class RunnerTest(unittest.TestCase): def setUp(self): self.hub = FakeHub(SETTINGS) self.dir = tempfile.TemporaryDirectory() self.journal = agent.Journal(os.path.join(self.dir.name, "journal.jsonl")) cfg = eva_cfg(hub_url=self.hub.base_url, hub_token=self.hub.token) self.runner = agent.Runner(cfg, agent.HubClient(cfg), agent.EvaClient(cfg), self.journal) self.sleeps = [] self.runner.sleep = self.sleeps.append self.runner.eva.sleep = no_sleep self.runner.idle_limit = 1 def tearDown(self): 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("wiki_fetch", FETCH) self.hub.enqueue("wiki_fetch", FETCH) self.runner.loop() self.assertEqual(self.hub.results[1]["status"], "failed") self.assertEqual(self.hub.results[1]["error"], STUB_ERROR) self.assertEqual(self.hub.results[1]["jira_calls"], 0) # поле протокола хаба: число вызовов Eva self.assertEqual(self.hub.results[2]["status"], "failed") 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_claim_sends_update_state_and_launcher(self): self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1) agent.write_update_state(self.runner.updater.state_path, "trial", 4, agent.AGENT_VERSION, "s") self.runner.loop() body = self.hub.claim_bodies[0] self.assertEqual(body["launcher_version"], 1) self.assertEqual(body["update_state"]["state"], "trial") self.assertEqual(agent.read_update_state(self.runner.updater.state_path)["state"], "ok") # первый успешный claim self.assertEqual(self.hub.claim_bodies[0]["agent_version"], agent.AGENT_VERSION) def test_claim_without_updater_sends_nulls(self): self.runner.loop() self.assertIsNone(self.hub.claim_bodies[0]["update_state"]) self.assertIsNone(self.hub.claim_bodies[0]["launcher_version"]) def test_idle_with_release_exits_for_update(self): rel = {"version": agent.AGENT_VERSION + 1, "sha256": "s", "content": "x"} self.hub.release = rel self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1) self.runner.updater.check = lambda r: r == {"version": agent.AGENT_VERSION + 1, "sha256": "s"} self.runner.idle_limit = 5 self.runner.loop() self.assertEqual(self.runner.exit_code, agent.EXIT_UPDATE) self.assertEqual(self.hub.claims, 1) def test_stop_during_check_exits_0(self): self.hub.release = {"version": agent.AGENT_VERSION + 1, "sha256": "s", "content": "x"} self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1) def check(r): self.runner.stop = True # Ctrl-C пришёл, когда версия уже staged return True self.runner.updater.check = check self.runner.idle_limit = 5 self.runner.loop() self.assertEqual(self.runner.exit_code, 0) # остановка человеком; staged доделает run.py при старте def test_command_first_then_update(self): self.hub.release = {"version": agent.AGENT_VERSION + 1, "sha256": "s", "content": "x"} self.hub.enqueue("wiki_fetch", FETCH) self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1) seen = [] self.runner.updater.check = lambda r: (seen.append(r), True)[1] self.runner.loop() self.assertEqual(self.hub.results[1]["status"], "failed") # команда выполнена и подтверждена self.assertEqual(len(seen), 1) # проверка — только после claim без команды self.assertEqual(self.runner.exit_code, agent.EXIT_UPDATE) def test_restart_pace_from_old_finished(self): at = (datetime.now(timezone.utc) - timedelta(seconds=10)).replace(microsecond=0).isoformat() self.journal.append({"kind": "finished", "command_id": 99, "status": "done", "result": None, "error": None, "jira_calls": 1, "rate_limited": 0, "at": at}) self.journal.append({"kind": "acked", "command_id": 99, "at": at}) self.hub.enqueue("wiki_fetch", FETCH, pause=30) self.runner.loop() self.assertAlmostEqual(sum(self.sleeps), 20, delta=1.5) def test_future_journal_time_means_full_pause(self): at = (datetime.now(timezone.utc) + timedelta(hours=2)).replace(microsecond=0).isoformat() self.journal.append({"kind": "acked", "command_id": 99, "at": at}) self.hub.enqueue("wiki_fetch", FETCH, pause=30) self.runner.loop() self.assertAlmostEqual(sum(self.sleeps), 30, delta=1.5) # часы ушли назад — пауза полная (M-6 #12) self.assertEqual(self.hub.results[1]["status"], "failed") def test_claim_after_restart_sends_no_journal_estimate(self): at = (datetime.now(timezone.utc) - timedelta(seconds=10)).replace(microsecond=0).isoformat() self.journal.append({"kind": "acked", "command_id": 99, "at": at}) self.hub.enqueue("wiki_fetch", FETCH) eva = self.runner.eva with mock.patch.object(agent.Executor, "do_wiki_fetch", lambda ex, p, cmd: (eva._touch(), ex._out("done", {"file": "abc", "size": 3}))[1]): self.runner.loop() self.assertIsNone(self.hub.claim_bodies[0].get("last_call_at")) # оценка журнала — только для паузы (M-7 #12) self.assertIsNotNone(self.hub.claim_bodies[-1].get("last_call_at")) # после вызова Eva — настоящее время self.assertIsNotNone(self.runner.eva.last_call_at) def test_restore_pace_keeps_existing_last_call(self): at = (datetime.now(timezone.utc) - timedelta(seconds=100)).replace(microsecond=0).isoformat() self.journal.append({"kind": "acked", "command_id": 99, "at": at}) mark = time.monotonic() self.runner.eva.last_call_at = mark self.runner.restore_pace() self.assertEqual(self.runner.eva.last_call_at, mark) def test_pause_waits_from_last_eva_call(self): self.runner.eva.last_call_at = time.monotonic() self.hub.enqueue("wiki_fetch", FETCH, 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"], "failed") def test_result_retried_when_hub_down(self): self.hub.fail_results = 2 self.hub.enqueue("wiki_fetch", FETCH) self.runner.loop() self.assertEqual(self.hub.results[1]["status"], "failed") 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("wiki_fetch", FETCH) self.runner.loop() self.assertEqual(self.sleeps, [10, 10]) self.assertEqual(self.hub.results[1]["status"], "failed") 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("wiki_fetch", FETCH) 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("wiki_fetch", FETCH, pause=99) self.journal.started(cid) self.runner.loop() self.assertEqual(self.hub.results[cid]["error"], STUB_ERROR) # команда выполнена заново, как решено в design 9 self.assertEqual(self.journal.unfinished_started(), []) self.assertAlmostEqual(sum(self.sleeps), 99, delta=1) # пауза из хаба, от записи started (M-9) def test_recover_does_not_journal_claimed_twice(self): cid = self.claim_in_flight("wiki_fetch", FETCH, pause=0) self.journal.claimed(cid) self.runner.loop() self.assertEqual(self.kinds(), [("claimed", cid), ("started", cid), ("finished", cid), ("acked", cid)]) def test_recover_skips_command_not_claimed_anymore(self): cid = self.hub.enqueue("wiki_fetch", FETCH) 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("wiki_fetch", FETCH) self.journal.started(cid) self.journal.finished(cid, "done", {"file": "abc", "size": 3}, None, 2, 0) self.hub.enqueue("wiki_fetch", FETCH) self.runner.loop() self.assertEqual(self.hub.results[cid]["result"], {"file": "abc", "size": 3}) self.assertEqual(self.hub.results[2]["status"], "failed") self.assertEqual(self.kinds()[:3], [("started", 1), ("finished", 1), ("acked", 1)]) def test_stop_during_long_claim(self): self.hub.claim_wait = 1.0 # хаб держит claim при пустой очереди (как claim_wait = 30) self.runner.idle_limit = None threading.Timer(0.3, lambda: setattr(self.runner, "stop", True)).start() start = time.monotonic() self.runner.loop() self.assertGreaterEqual(time.monotonic() - start, 0.9) # дождался ответа claim self.assertEqual(self.hub.claims, 1) self.assertEqual(self.journal.records(), []) def test_command_arrives_during_long_claim(self): self.hub.claim_wait = 5.0 def arrive(): self.hub.enqueue("wiki_fetch", FETCH) self.hub.claim_wait = 0 # следующий claim (после отчёта) — без ожидания threading.Timer(0.3, arrive).start() start = time.monotonic() self.runner.loop() self.assertLess(time.monotonic() - start, 4) # ответ пришёл с командой, а не по таймауту self.assertEqual(self.hub.results[1]["status"], "failed") def test_stop_during_pause(self): self.runner.eva.last_call_at = time.monotonic() self.hub.enqueue("wiki_fetch", FETCH, 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.eva.last_call_at = time.monotonic() self.hub.enqueue("wiki_fetch", FETCH, pause=5) self.runner.sleep = lambda s: setattr(self.runner, "stop", True) self.runner.loop() self.assertEqual(self.journal.unfinished_started(), [1]) cfg = eva_cfg(hub_url=self.hub.base_url, hub_token=self.hub.token) second = agent.Runner(cfg, agent.HubClient(cfg), agent.EvaClient(cfg), agent.Journal(self.journal.path)) second.sleep = self.sleeps.append second.idle_limit = 1 second.loop() self.assertEqual(self.hub.results[1]["status"], "failed") self.assertEqual(len([r for r in self.journal.records() if r["kind"] == "finished"]), 1) self.assertEqual(self.journal.unfinished_started(), []) self.assertAlmostEqual(sum(self.sleeps), 5, delta=1) # пауза из хаба от записи claimed (M-9) def test_stop_between_claim_and_execute_is_recorded(self): self.hub.enqueue("wiki_fetch", FETCH) 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.journal.unfinished_started(), [1]) def test_stop_during_execution_finishes_and_reports(self): self.hub.enqueue("wiki_fetch", FETCH) self.hub.enqueue("wiki_fetch", FETCH) 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"], "failed") 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("wiki_fetch", FETCH) 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("wiki_fetch", FETCH) 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("wiki_fetch", FETCH) 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"], "failed") 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["wiki_pacing"] = 0.5 self.hub.enqueue("wiki_fetch", FETCH) 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_waits_from_failure_even_if_last_call_is_old(self): # F-1: Eva недоступна с первого вызова, last_call_at старый (обрыв не обновляет его) — # ожидание перед повтором считается от момента отказа, а не от последнего успешного вызова. self.runner.load_settings() self.runner.eva.last_call_at = time.monotonic() - 600 runs = [] def flaky(cmd): runs.append(1) if len(runs) == 1: return {"status": "failed", "result": None, "error": "нет связи", "jira_calls": 0, "rate_limited": 0, "retryable": True} return {"status": "done", "result": {"changed": []}, "jira_calls": 1, "rate_limited": 0} self.runner.executor.run = flaky self.hub.enqueue("wiki_fetch", FETCH) self.runner.loop() self.assertEqual(len(runs), 2) self.assertEqual(self.hub.results[1]["status"], "done") self.assertAlmostEqual(sum(self.sleeps), 30, delta=0.1) # полные 30 с от момента отказа 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("wiki_fetch", FETCH) self.runner.loop() self.assertEqual(len(runs), 4) # первый запуск + 3 повтора r = self.hub.results[1] self.assertEqual(r["status"], "failed") self.assertIn("результат неизвестен", r["error"]) self.assertIn("(обрыв связи с Eva)", 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("wiki_fetch", FETCH) 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("wiki_fetch", FETCH) 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 = eva_cfg(hub_url=self.hub.base_url, hub_token="bad") r = agent.Runner(cfg, agent.HubClient(cfg), agent.EvaClient(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("wiki_fetch", FETCH) 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"], "failed") 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(eva_cfg(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 = eva_cfg(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) def test_prune_refused_while_agent_runs(self): old = "2020-01-01T00:00:00+00:00" self.journal.append({"kind": "finished", "command_id": 7, "status": "done", "result": None, "error": None, "jira_calls": 1, "rate_limited": 0, "at": old}) self.journal.append({"kind": "acked", "command_id": 7, "at": old}) held = agent.acquire_lock(self.journal.path) try: self.assertEqual(agent.cmd_prune(self.journal.path), 1) self.assertEqual(len(self.journal.records()), 2) # журнал не тронут (T2.3 agent#4) finally: held.close() self.assertEqual(agent.cmd_prune(self.journal.path), 0) self.assertEqual(self.journal.records(), []) def test_second_process_refused(self): """I-1 финального ревью: второй run с тем же журналом не стартует — иначе довыдача хаба отдаст ему ту же команду.""" from unittest import mock held = agent.acquire_lock(self.journal.path) self.assertIsNotNone(held) try: self.assertIsNone(agent.acquire_lock(self.journal.path)) cfg = eva_cfg(hub_url=self.hub.base_url, hub_token=self.hub.token) with mock.patch.object(agent.Runner, "loop", lambda r: self.fail("цикл не должен стартовать")): self.assertEqual(agent.cmd_run(cfg, self.journal.path), agent.EXIT_LOCKED) self.assertEqual(agent.EXIT_LOCKED, 73) finally: held.close() again = agent.acquire_lock(self.journal.path) self.assertIsNotNone(again) # после выхода первого — снова можно again.close() def test_future_journal_time_single_log_line(self): at = (datetime.now(timezone.utc) + timedelta(hours=2)).replace(microsecond=0).isoformat() self.journal.append({"kind": "acked", "command_id": 99, "at": at}) logs = [] orig = agent.log agent.log = logs.append try: self.runner.restore_pace() finally: agent.log = orig self.assertEqual(len(logs), 1, logs) # M-2 #18: без «не позже <будущее>» self.assertIn("в будущем", logs[0]) self.assertIsNotNone(self.runner.eva.last_call_at) def test_future_journal_time_then_stop_in_pause(self): at = (datetime.now(timezone.utc) + timedelta(hours=2)).replace(microsecond=0).isoformat() self.journal.append({"kind": "acked", "command_id": 99, "at": at}) self.hub.enqueue("wiki_fetch", FETCH, pause=30) self.runner.sleep = lambda s: setattr(self.runner, "stop", True) # Ctrl-C в первую секунду паузы self.runner.loop() self.assertEqual([r["kind"] for r in self.journal.records() if r["command_id"] == 1], ["claimed"]) self.assertEqual(self.runner.eva.calls, 0) # к Eva ни одного вызова (M-3 #18) self.assertEqual(self.hub.commands[1]["status"], "claimed") def test_retry_exhausted_on_hub_names_hub(self): self.runner.load_settings() logs = [] orig = agent.log agent.log = logs.append self.runner.executor.run = lambda cmd: {"status": "failed", "result": None, "error": "хаб: не ответил", "jira_calls": 0, "rate_limited": 0, "retryable": True, "peer": "хабом"} self.hub.enqueue("wiki_fetch", FETCH) try: self.runner.loop() finally: agent.log = orig r = self.hub.results[1] self.assertEqual(r["error"], "результат неизвестен (обрыв связи с хабом), повтор безопасен: хаб: не ответил") self.assertTrue(any("обрыв связи с хабом, повтор 1/3" in l for l in logs), logs) self.assertNotIn("peer", r) if __name__ == "__main__": unittest.main()