diff --git a/src/agent.py b/src/agent.py index bd2fd91..7c45cc8 100644 --- a/src/agent.py +++ b/src/agent.py @@ -11,6 +11,7 @@ Python 3.9. Спека: jira-hub/openspec/changes/bridge-core/specs/agent/spec.m """ import argparse import base64 +import http.client import json import os import signal @@ -70,7 +71,7 @@ class Http: def __repr__(self): return "<%s %s>" % (self.__class__.__name__, self.base) - def call(self, method: str, path: str, body=None, params=None) -> Any: + def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any: url = self.base + path if params: url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None}) @@ -82,7 +83,7 @@ class Http: if data is not None: req.add_header("Content-Type", "application/json; charset=utf-8") try: - with urllib.request.urlopen(req, timeout=self.timeout, context=self.ctx) as resp: + with urllib.request.urlopen(req, timeout=timeout or self.timeout, context=self.ctx) as resp: raw = resp.read() status = resp.status except urllib.error.HTTPError as e: @@ -90,12 +91,12 @@ class Http: try: headers = {k.lower(): v for k, v in e.headers.items()} text = e.read().decode("utf-8", "replace") - except (OSError, ConnectionError, TimeoutError) as err: + except (OSError, ConnectionError, TimeoutError, http.client.HTTPException) as err: raise ConnectionError("%s: обрыв при чтении ответа: %s" % (self.base, err)) from None finally: e.close() raise HttpError(e.code, text, headers) from None - except (urllib.error.URLError, ConnectionError, TimeoutError, OSError) as e: + except (urllib.error.URLError, ConnectionError, TimeoutError, OSError, http.client.HTTPException) as e: raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None self._touch() try: @@ -181,7 +182,7 @@ class HubClient(Http): def claim(self, agent_version, settings_version, last_jira_call_at) -> dict: return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version, - "last_jira_call_at": last_jira_call_at}) + "last_jira_call_at": last_jira_call_at}, timeout=max(self.timeout, 45)) def settings(self) -> dict: return self.call("GET", "/agent/settings") @@ -255,6 +256,10 @@ class Journal: log("журнал: строка %d не читается, пропускаю" % n) return good + def claimed(self, command_id) -> None: + """Команда получена из claim (до паузы): с этого момента хаб считает её выданной агенту.""" + self.append({"kind": "claimed", "command_id": int(command_id)}) + def started(self, command_id) -> None: self.append({"kind": "started", "command_id": int(command_id)}) @@ -276,11 +281,11 @@ class Journal: return [r for cid, r in last.items() if cid not in acked] def unfinished_started(self) -> List[int]: - """Порядковый учёт: started открывает id, finished/acked закрывает; порядок — первое открытие.""" + """Порядковый учёт: claimed/started открывают id, finished/acked закрывает; порядок — первое открытие.""" opened = {} # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия for r in self.records(): kind, cid = r.get("kind"), r["command_id"] - if kind == "started": + if kind in ("claimed", "started"): opened[cid] = True elif kind in ("finished", "acked") and cid in opened: opened[cid] = False @@ -299,7 +304,7 @@ class Journal: cid = r["command_id"] if r.get("kind") == "acked": last_ack[cid] = i - elif r.get("kind") in ("started", "finished"): + elif r.get("kind") in ("claimed", "started", "finished"): last_open[cid] = i drop_upto = {} # command_id -> индекс последнего acked: всё до него включительно удаляется for cid, i in last_ack.items(): @@ -344,6 +349,10 @@ def _date(value) -> Optional[str]: return (value or "")[:10] or None +class ConfigGap(Exception): + """В настройках хаба нет того, что нужно команде (R-11): команда падает с понятным текстом, а не молча теряет поле.""" + + class Executor: def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None): self.jira, self.s, self.hub = jira, settings, hub @@ -359,10 +368,16 @@ class Executor: return self._out("failed", error="неизвестный тип команды %s" % cmd.get("type")) try: return handler(cmd.get("params") or {}, cmd) - except HttpError as e: - return self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500])) - except ConnectionError as e: + except ConfigGap as e: return self._out("failed", error=str(e)) + except HttpError as e: + out = self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500])) + out["retryable"] = e.code >= 500 # исход неизвестен: Runner повторит через проверку (R-10) + return out + except ConnectionError as e: + out = self._out("failed", error=str(e)) + out["retryable"] = True + return out def _out(self, status, result=None, error=None) -> dict: return {"status": status, "result": result, "error": error, @@ -389,19 +404,36 @@ class Executor: it = (self.s.get("issue_types") or {}).get(kind) or {} return {"id": str(it["id"])} if it.get("id") else {"name": it.get("name")} - def epic_fields(self, epic_key) -> dict: + def epic_field_id(self) -> str: ep = self.s.get("epic_link") or {} - if epic_key and ep.get("mode", "field") == "field" and ep.get("field_id"): - return {ep["field_id"]: epic_key} - return {} + if ep.get("mode", "field") != "field": + raise ConfigGap("epic_key передан, но epic_link.mode = %r (поддержан только «field»)" % ep.get("mode")) + if not ep.get("field_id"): + raise ConfigGap("epic_key передан, но не настроено epic_link.field_id") + return ep["field_id"] + + def epic_fields(self, epic_key) -> dict: + return {self.epic_field_id(): epic_key} if epic_key else {} + + def need_field(self, logical: str) -> str: + if not self.has_field(logical): + raise ConfigGap("поле передано, но не настроено field_ids.%s" % logical) + return self.fid(logical) # --- ensure_issue ---------------------------------------------------------- def do_ensure_issue(self, p: dict, cmd: dict) -> dict: - jql = "project = %s AND reporter = currentUser() AND created >= -7d" % self.s["project_key"] - found = self.jira.get(API + "/search", jql=jql, fields="summary", maxResults=200) - for it in found.get("issues", []): - if (it.get("fields", {}).get("summary") or "") == p["summary"]: - return self._out("done", {"key": it["key"], "existed": True}) + summary = (p["summary"] or "").strip() + jql = "project = %s AND reporter = currentUser() AND created >= -7d ORDER BY created DESC" % self.s["project_key"] + start = 0 + while start < 1000: + found = self.jira.get(API + "/search", jql=jql, fields="summary", maxResults=200, startAt=start) + issues = found.get("issues", []) + for it in issues: + if (it.get("fields", {}).get("summary") or "").strip() == summary: + return self._out("done", {"key": it["key"], "existed": True}) + start += len(issues) + if not issues or start >= int(found.get("total") or 0): + break fields = {"project": {"key": self.s["project_key"]}, "issuetype": self.issue_type_ref(p["issue_type"]), "summary": p["summary"]} if p.get("description") is not None: @@ -411,8 +443,8 @@ class Executor: if p.get("duedate"): fields[self.fid("duedate")] = p["duedate"] for logical in ("planned_start", "planned_end"): - if p.get(logical) and self.has_field(logical): - fields[self.fid(logical)] = p[logical] + if p.get(logical): + fields[self.need_field(logical)] = p[logical] if p.get("labels") is not None: fields[self.fid("labels")] = list(p["labels"]) if p.get("assignee"): @@ -437,17 +469,22 @@ class Executor: if "priority" in p: want = self.priority_value(p["priority"]) vkey = list(want)[0] - if ((cur.get(self.fid("priority")) or {}).get(vkey)) != want[vkey]: + have = (cur.get(self.fid("priority")) or {}).get(vkey) + if (None if have is None else str(have)) != str(want[vkey]): fields[self.fid("priority")] = want changed.append("priority") for logical in ("duedate", "planned_start", "planned_end"): - if logical in p and self.has_field(logical) and _date(cur.get(self.fid(logical))) != _date(p[logical]): + if logical not in p: + continue + if logical != "duedate" and p[logical] and not self.has_field(logical): + self.need_field(logical) + if self.has_field(logical) and _date(cur.get(self.fid(logical))) != _date(p[logical]): fields[self.fid(logical)] = p[logical] changed.append(logical) - if "epic_key" in p: - ep = self.s.get("epic_link") or {} - if ep.get("field_id") and (cur.get(ep["field_id"]) or None) != (p["epic_key"] or None): - fields[ep["field_id"]] = p["epic_key"] + if "epic_key" in p and (p["epic_key"] or (self.s.get("epic_link") or {}).get("field_id")): + fid = self.epic_field_id() + if (cur.get(fid) or None) != (p["epic_key"] or None): + fields[fid] = p["epic_key"] changed.append("epic_key") if "assignee" in p and ((cur.get("assignee") or {}).get("name") or None) != (p["assignee"] or None): fields["assignee"] = {"name": p["assignee"]} if p["assignee"] else None @@ -467,12 +504,19 @@ class Executor: # --- ensure_comment -------------------------------------------------------- def do_ensure_comment(self, p: dict, cmd: dict) -> dict: key, marker, body = p["key"], p["marker"], p["body"] - if marker not in body[:80]: + heads = (marker, "*%s.*" % marker) # маркер — только в начале тела (формат агента: «*маркер.* текст») + if not body.startswith(heads): body = "*%s.* %s" % (marker, body) - existing = None - for c in self.jira.get(API + "/issue/%s/comment" % key).get("comments", []): - if marker in (c.get("body") or "")[:80]: - existing = c + existing, start = None, 0 + while existing is None: + page = self.jira.get(API + "/issue/%s/comment" % key, startAt=start, maxResults=100) + comments = page.get("comments", []) + for c in comments: + if (c.get("body") or "").startswith(heads): + existing = c + break + start += len(comments) + if not comments or page.get("total") is None or start >= int(page["total"]): break if existing and (existing.get("body") or "") == body: return self._out("done", {"comment_id": existing["id"], "changed": False}) @@ -740,7 +784,7 @@ class Executor: chunk = issues[i:i + 100] chunk_keys = {x["key"] for x in chunk} pages += 1 - self.hub.state({"command_id": cmd.get("id"), "issues": chunk, + self.hub.state({"command_id": cmd.get("id"), "agent_version": AGENT_VERSION, "issues": chunk, "links": [l for l in links if l["source_key"] in chunk_keys or l["target_key"] in chunk_keys], "neighbor_keys": [k for k in neighbor_keys if k in chunk_keys], "page": pages, "last_page": i + 100 >= len(issues)}) @@ -754,6 +798,9 @@ class Executor: # Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте # --------------------------------------------------------------------------- class Runner: + RETRY_ATTEMPTS = 3 # повторов команды после обрыва связи с Jira / 5xx (R-10) + RETRY_MIN_WAIT = 30 # секунд до повтора + def __init__(self, cfg: dict, hub: HubClient, jira: JiraClient, journal: Journal): self.cfg, self.hub, self.jira, self.journal = cfg, hub, jira, journal self.settings = None # type: Optional[dict] @@ -766,6 +813,14 @@ class Runner: def poll(self) -> float: return max(float(self.cfg.get("poll_seconds", 5) or 0), 0) + def hub_wait(self, e: Exception, base: float) -> float: + """Залогировать сбой хаба и вернуть, сколько ждать. 401/403 — токен неверен: повторять часто бессмысленно.""" + if isinstance(e, HttpError) and e.code in (401, 403): + log("хаб: HTTP %d — неверный hub_token в agent.json, повтор через 60 с" % e.code) + return 60 + log("хаб: %s, повтор через %d с" % (e, base)) + return base + def load_settings(self) -> None: self.settings = self.hub.settings() self.executor = Executor(self.jira, self.settings, self.hub) @@ -777,7 +832,8 @@ class Runner: или удалена): считаем закрытой, чтобы не зациклиться. При остановке по сигналу делается одна попытка.""" cid = rec["command_id"] payload = {"status": rec["status"], "result": rec.get("result"), "error": rec.get("error"), - "jira_calls": rec.get("jira_calls", 0), "rate_limited": rec.get("rate_limited", 0)} + "jira_calls": rec.get("jira_calls", 0), "rate_limited": rec.get("rate_limited", 0), + "agent_version": AGENT_VERSION} while True: try: self.hub.result(cid, payload) @@ -788,43 +844,65 @@ class Runner: log("хаб отверг отчёт по #%s: %s — считаю закрытым" % (cid, e)) self.journal.acked(cid) return True - log("хаб: %s, повтор через 10 с" % e) + wait = self.hub_wait(e, 10) except ConnectionError as e: - log("хаб недоступен: %s, повтор через 10 с" % e) + wait = self.hub_wait(e, 10) if self.stop: log("остановка: отчёт по #%s не подтверждён, будет дослан при следующем старте" % cid) return False - self.sleep(10) + self.sleep(wait) - def wait_pause(self, pause_seconds) -> bool: - """Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы.""" + def wait_pause(self, pause_seconds, force=False) -> bool: + """Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы. + force: если вызовов Jira ещё не было, считать от текущего момента.""" pause = float(pause_seconds or 0) - if pause > 0 and self.jira.last_call_at is not None: - remaining = pause - (time.monotonic() - self.jira.last_call_at) + if pause > 0 and (self.jira.last_call_at is not None or force): + since = time.monotonic() - self.jira.last_call_at if self.jira.last_call_at is not None else 0 + remaining = pause - since if remaining > 0: - log(" пауза %.0f с до команды" % remaining) + log(" пауза %.0f с" % remaining) while remaining > 0 and not self.stop: step = min(remaining, 1.0) self.sleep(step) remaining -= step return not self.stop - def execute(self, cmd: dict) -> None: - log("#%s %s %s" % (cmd["id"], cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200])) - if not self.wait_pause(cmd.get("pause_seconds")): - return - self.journal.started(cmd["id"]) # started — только после паузы + def run_once(self, cmd: dict) -> dict: try: - out = self.executor.run(cmd) + return self.executor.run(cmd) except Exception as e: # баг в обработчике не должен ронять цикл и зацикливать восстановление log(" ! необработанная ошибка: %r" % e) - out = {"status": "failed", "result": None, + return {"status": "failed", "result": None, "error": "внутренняя ошибка агента: %s: %s" % (type(e).__name__, str(e)[:300]), "jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited} - self.journal.finished(cmd["id"], out["status"], out.get("result"), out.get("error"), out["jira_calls"], out["rate_limited"]) + + def execute(self, cmd: dict) -> None: + cid = cmd["id"] + log("#%s %s %s" % (cid, cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200])) + self.journal.claimed(cid) # сразу после claim, до паузы (R-9): остановка не теряет команду + if not self.wait_pause(cmd.get("pause_seconds")): + return + self.journal.started(cid) # started — только после паузы + calls = limited = attempt = 0 + while True: + out = self.run_once(cmd) + calls += out["jira_calls"] + limited += out["rate_limited"] + if not out.pop("retryable", False): + break + if attempt >= self.RETRY_ATTEMPTS: + out = {"status": "failed", "result": None, + "error": "результат неизвестен (обрыв связи с Jira), повтор безопасен: %s" % out.get("error")} + break + attempt += 1 + log(" ! обрыв связи с Jira, повтор %d/%d через проверку: %s" % (attempt, self.RETRY_ATTEMPTS, out.get("error"))) + if not self.wait_pause(max(float(cmd.get("pause_seconds") or 0), self.RETRY_MIN_WAIT), force=True): + return # остановка: finished не пишем, команду подхватит recover + out["jira_calls"], out["rate_limited"] = calls, limited + self.journal.finished(cid, out["status"], out.get("result"), out.get("error"), calls, limited) log(" → %s %s" % (out["status"], out.get("error") or json.dumps(out.get("result"), ensure_ascii=False))) rec = dict(out) - rec["command_id"] = cmd["id"] + rec["command_id"] = cid self.send_result(rec) def fetch_command(self, cid): @@ -836,12 +914,12 @@ class Runner: if e.code in (404, 409, 422): log("#%s: хаб ответил %s" % (cid, e)) return None - log("хаб: %s, повтор через 10 с" % e) + wait = self.hub_wait(e, 10) except ConnectionError as e: - log("хаб недоступен: %s, повтор через 10 с" % e) + wait = self.hub_wait(e, 10) if self.stop: raise InterruptedError(cid) - self.sleep(10) + self.sleep(wait) def recover(self) -> None: for rec in self.journal.unacked_finished(): @@ -861,10 +939,28 @@ class Runner: log("#%s уже %s в хабе — пропускаю" % (cid, cmd.get("status"))) self.journal.acked(cid) continue - log("#%s начата до перезапуска — выполняю заново через проверку" % cid) + log("#%s выдана до перезапуска — выполняю через проверку" % cid) cmd["pause_seconds"] = 0 self.execute(cmd) + def reload_settings_before(self, cmd) -> bool: + """Настройки сменились: перечитать ДО выполнения. False — остановка, команда не должна выполняться + (если она уже выдана хабом, фиксируем claimed — подхватит recover).""" + while True: + try: + self.load_settings() + return True + except (ConnectionError, HttpError) as e: + wait = self.hub_wait(e, max(self.poll, 1)) + if self.stop: + if cmd: + self.journal.claimed(cmd["id"]) + return False + if not cmd: + self.sleep(wait) + return False + self.sleep(wait) + def loop(self) -> None: idle, recovered = 0, False while not self.stop: @@ -872,8 +968,7 @@ class Runner: try: self.load_settings() except (ConnectionError, HttpError) as e: - log("хаб: %s" % e) - self.sleep(self.poll) + self.sleep(self.hub_wait(e, max(self.poll, 1))) continue if not recovered: self.recover() @@ -882,15 +977,12 @@ class Runner: try: resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.jira.last_call_iso) except (ConnectionError, HttpError) as e: - log("хаб: %s" % e) - self.sleep(self.poll) + self.sleep(self.hub_wait(e, max(self.poll, 1))) continue - if resp.get("settings_version") != self.settings.get("settings_version"): - try: - self.load_settings() # настройки — до выполнения команды - except (ConnectionError, HttpError) as e: - log("хаб: не перечитать настройки: %s" % e) cmd = resp.get("command") + if resp.get("settings_version") != self.settings.get("settings_version"): + if not self.reload_settings_before(cmd): + continue if cmd: idle = 0 self.execute(cmd) @@ -905,11 +997,16 @@ def cmd_run(cfg: dict, journal_path: str) -> int: runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path)) def on_signal(signum, frame): + try: + if runner.stop: + log("повторный сигнал — выхожу немедленно") + else: + log("сигнал %s: доделываю текущую команду и выхожу" % signum) + except RuntimeError: # print внутри обработчика, прервавшего другой print (reentrant call) + pass if runner.stop: - log("повторный сигнал — выхожу немедленно") sys.exit(130) runner.stop = True - log("сигнал %s: доделываю текущую команду и выхожу" % signum) signal.signal(signal.SIGINT, on_signal) signal.signal(signal.SIGTERM, on_signal) diff --git a/tests/fake_jira.py b/tests/fake_jira.py index b4efb55..05add91 100644 --- a/tests/fake_jira.py +++ b/tests/fake_jira.py @@ -2,6 +2,7 @@ только маршруты, которые использует агент. Семантика связей — как у настоящего Jira Server.""" import json import re +import socket import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse @@ -31,6 +32,8 @@ class FakeJira: self.id_seq = 10000 self.fail_next = [] # [(код, {заголовки})] — следующие ответы с ошибкой self.requests = [] # [(метод, путь)] + self.queries = [] # параметры запросов (q) — для проверки JQL + self.drop_next_post_issue = False # выполнить создание задачи и оборвать соединение без ответа self.lock = threading.Lock() fake = self @@ -104,7 +107,8 @@ class FakeJira: keys = [k.strip() for k in m.group(1).split(",")] return [k for k in keys if k in self.issues] if "reporter = currentUser()" in (jql or ""): - return [k for k, f in self.issues.items() if f["reporter"] == "me"] + keys = [k for k, f in self.issues.items() if f["reporter"] == "me"] + return keys[::-1] if "ORDER BY created DESC" in jql else keys return [k for k, f in self.issues.items() if "AWPCA" in f["labels"]] # --- маршрутизация ---------------------------------------------------------- @@ -115,6 +119,7 @@ class FakeJira: body = json.loads(h.rfile.read(length).decode("utf-8")) if length else {} with self.lock: self.requests.append((method, url.path)) + self.queries.append(q) if self.fail_next: code, headers = self.fail_next.pop(0) return self._send(h, code, {"errorMessages": ["имитация ошибки"]}, headers) @@ -124,6 +129,11 @@ class FakeJira: code, payload = 404, {"errorMessages": ["Issue Does Not Exist: %s" % e]} except Exception as e: # некорректный запрос к фейку code, payload = 400, {"errorMessages": ["фейк: %s: %s" % (type(e).__name__, e)]} + if self.drop_next_post_issue and method == "POST" and url.path == "/rest/api/2/issue": + self.drop_next_post_issue = False + h.close_connection = True + h.connection.shutdown(socket.SHUT_RDWR) # задача создана, ответа клиент не получит + return self._send(h, code, payload) def _send(self, h, code, payload, headers=None): @@ -193,7 +203,8 @@ class FakeJira: self.issues[key] lst = self.comments.setdefault(key, []) if method == "GET": - return 200, {"comments": lst} + start, maxr = int(q.get("startAt", 0)), int(q.get("maxResults", 1000)) + return 200, {"startAt": start, "maxResults": maxr, "total": len(lst), "comments": lst[start:start + maxr]} self.comment_seq += 1 lst.append({"id": str(self.comment_seq), "body": body["body"]}) return 201, lst[-1] diff --git a/tests/test_py39.py b/tests/test_py39.py index 8235b6c..170aa5c 100644 --- a/tests/test_py39.py +++ b/tests/test_py39.py @@ -17,7 +17,7 @@ class Py39Test(unittest.TestCase): def test_no_third_party_imports(self): with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f: tree = ast.parse(f.read()) - allowed = {"argparse", "base64", "json", "os", "signal", "ssl", "sys", "time", "urllib", "datetime", "typing"} + allowed = {"argparse", "base64", "json", "os", "signal", "ssl", "http", "sys", "time", "urllib", "datetime", "typing"} for node in ast.walk(tree): if isinstance(node, ast.Import): for a in node.names: diff --git a/tests/test_runner.py b/tests/test_runner.py index f62d905..6e62b24 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -47,7 +47,7 @@ class RunnerTest(unittest.TestCase): 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(), [("started", 1), ("finished", 1), ("acked", 1), ("started", 2), ("finished", 2), ("acked", 2)]) + 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): @@ -141,10 +141,41 @@ class RunnerTest(unittest.TestCase): 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.records(), []) + 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"]}) @@ -159,7 +190,7 @@ class RunnerTest(unittest.TestCase): self.runner.loop() self.assertEqual(self.hub.results[1]["status"], "done") self.assertNotIn(2, self.hub.results) # вторая команда не начата - self.assertEqual(self.kinds(), [("started", 1), ("finished", 1), ("acked", 1)]) + 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 @@ -231,7 +262,117 @@ class RunnerTest(unittest.TestCase): self.runner.hub.settings = hub_settings self.runner.loop() self.assertEqual(len(calls), 3) - self.assertEqual(self.sleeps, [0, 0]) + 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__":