Merge pull request 'Агент v9 и run.py v2: устойчивость самообновления, диагностика, фейки (#23)' (#24) from feat/23-agent-v9 into main
This commit is contained in:
commit
f28df32f66
@ -23,6 +23,10 @@ Jira и хаба. Спека: `../jira-hub/openspec/changes/bridge-core/`, пл
|
||||
падение в `trial` или 10 минут без `ok` — `rollback`; иначе перезапуск с паузой 5…300 с. Блокировка `run.py.lock`;
|
||||
SIGINT при живом агенте игнорирует (Ctrl-C получает агент), SIGTERM пересылает. Спека — «Запускатель run.py».
|
||||
|
||||
`reconcile` при старте и `in_trial` перед каждым запуском сверяют `update.json` с sha256 `agent.py` (таблица — design 1 change'а
|
||||
agent-v9-hardening); коды агента: 75 — подмена, 73 — журнал занят (не сбой версии); `OSError` при подмене — `rejected`,
|
||||
запускатель не падает.
|
||||
|
||||
## Границы
|
||||
- Код новой версии приходит только через API хаба (`GET /agent/release`) и пишется в каталог агента; общая папка для
|
||||
кода не используется.
|
||||
|
||||
@ -48,7 +48,7 @@
|
||||
## Первая подсистема
|
||||
Исполнитель команд и цикл агента (спека `agent/spec.md`, `commands/spec.md`) — реализованы целиком: все типы команд,
|
||||
журнал, восстановление после перезапуска, CLI `run|check|selftest|prune`, самообновление через `run.py`.
|
||||
Последний change — `../jira-hub/openspec/changes/archive/2026-10-14-agent-self-update/` (решения агента — `openspec/changes/archive/2026-10-14-agent-self-update/design.md`), план — `docs/superpowers/plans/2026-10-14-agent-self-update-agent.md`.
|
||||
Последний change — `../jira-hub/openspec/changes/agent-v9-hardening/` (решения агента — `openspec/changes/agent-v9-hardening/design.md`), план — `docs/superpowers/plans/2026-10-14-agent-v9-hardening-agent.md`.
|
||||
План и спека живут в `jira-hub`: `openspec/changes/bridge-core/` и `docs/superpowers/plans/2026-10-11-bridge-core-hub.md`; план агента лежит здесь: `docs/superpowers/plans/2026-10-11-bridge-core-agent.md`.
|
||||
|
||||
## Выкатка новой версии
|
||||
|
||||
18
README.md
18
README.md
@ -82,6 +82,7 @@ nohup python3 -u run.py > agent.log 2>&1 &
|
||||
считается упавшим и перезапускается). Первый сигнал — агент доделывает текущую команду, отправляет отчёт и выходит.
|
||||
Повторный сигнал — немедленный выход (код 130). Остановка во время паузы перед командой безопасна: команда уже
|
||||
записана в журнал как `claimed`, при следующем запуске агент запрашивает её у хаба и выполняет через проверку, выдержав паузу темпа от последней записи журнала.
|
||||
Если журнал занят другим процессом (осталась прежняя копия агента), `agent.py run` выходит с кодом 73, а `run.py` повторяет запуск с паузой и не считает это сбоем версии.
|
||||
|
||||
## Самообновление
|
||||
|
||||
@ -93,6 +94,13 @@ nohup python3 -u run.py > agent.log 2>&1 &
|
||||
на связь за 10 минут — `run.py` возвращает `agent.py.prev` и пишет `rolled_back`; эту версию агент больше не ставит,
|
||||
хаб показывает откат в `agent_status`. Код в общую папку не попадает: файлы пишутся только в каталог `run.py`.
|
||||
|
||||
Запускатель v2 при старте и перед каждым запуском сверяет `update.json` с sha256 файла `agent.py`: если `agent.py` уже
|
||||
новая версия, а `update.json` = `staged` (запускатель упал посреди подмены), пробный период продолжается; если
|
||||
`update.json` = `trial`, а `agent.py` другая версия (прерван откат или файл заменён вручную), пишется `rolled_back`
|
||||
«пробный период прерван…» и эта версия больше сама не ставится — вернуть её можно публикацией с номером выше. Если
|
||||
подмена не удалась (`OSError`, например диск полон), текущая версия остаётся, а новая получает `rejected` «подмена не
|
||||
удалась»: освободить место и опубликовать следующую версию. Временные файлы `*.tmp` удаляются при старте.
|
||||
|
||||
Файлы рядом с `agent.py`: `update.json` (состояние: `staged`, `trial`, `ok`, `rolled_back`, `rejected`),
|
||||
`agent.py.prev` (прежняя версия), `run.py.lock`. Не удалять во время работы.
|
||||
|
||||
@ -181,9 +189,11 @@ python3 agent.py prune
|
||||
`finished`, ждёт не меньше 30 с и повторяет команду через проверку («проверь, потом сделай») до 3 раз. Счётчики
|
||||
суммируются. После исчерпания — `failed` «результат неизвестен (обрыв связи с Jira), повтор безопасен». При остановке
|
||||
по сигналу команду подхватит восстановление.
|
||||
- **Версия.** Константа `AGENT_VERSION` в `agent.py` (сейчас 5) отправляется в каждом `claim`, в каждом отчёте
|
||||
- **Версия.** Константа `AGENT_VERSION` в `agent.py` (сейчас 9) отправляется в каждом `claim`, в каждом отчёте
|
||||
(`agent_version`) и в `/agent/state`. Версия 4 — сравнение текстов без различия `\r\n`/`\n`, полная пауза при времени
|
||||
журнала в будущем, `claim` без оценки из журнала, `prune` под блокировкой. Версия 5 — самообновление через `run.py`.
|
||||
Версия 9 — код 73 при занятом журнале, `OSError` при подготовке — временный сбой, проверка формата публикации, источник ошибки хаба в
|
||||
`refresh` (номера 6–8 заняты живой проверкой самообновления).
|
||||
Если хаб назначил `min_agent_version` выше, он пишет в `agent_status.warning`: «обновите агента»; команды
|
||||
при этом выдаются по-прежнему.
|
||||
|
||||
@ -203,9 +213,13 @@ python3 agent.py prune
|
||||
старый процесс (Ctrl-C или `kill <pid>`, дождаться «остановлен»), скопировать `agent.py` и `run.py` в каталог агента,
|
||||
`python3 agent.py check`, запустить `python3 run.py` (или `launchctl unload` + `load`). Два процесса с одним журналом работать не могут: хаб
|
||||
довыдаёт команду без отчёта, и второй процесс выполнял бы её одновременно с первым. Начиная с версии 3 второй процесс
|
||||
не стартует («агент уже запущен…», код 1) — блокировка `journal.jsonl.lock` рядом с журналом.
|
||||
не стартует («агент уже запущен…», код 73) — блокировка `journal.jsonl.lock` рядом с журналом.
|
||||
`agent.json`, `journal.jsonl` и `update.json` не трогать: журнал переживает обновление, а неподтверждённые отчёты досылаются при старте.
|
||||
|
||||
Переход на `run.py` v2: сначала агент должен обновиться до v9 (MCP `agent_status` — «обновление: ok v8 → v9»); затем
|
||||
`launchctl unload ~/Library/LaunchAgents/ru.aepif.jira-agent.plist`, скопировать `run.py` в `~/jira-agent-bundle`,
|
||||
`launchctl load …`; `agent_status` покажет «запускатель: v2».
|
||||
|
||||
## Известные ограничения / эксплуатация
|
||||
|
||||
В порядке важности:
|
||||
|
||||
1055
docs/superpowers/plans/2026-10-14-agent-v9-hardening-agent.md
Normal file
1055
docs/superpowers/plans/2026-10-14-agent-v9-hardening-agent.md
Normal file
File diff suppressed because it is too large
Load Diff
22
openspec/changes/agent-v9-hardening/design.md
Normal file
22
openspec/changes/agent-v9-hardening/design.md
Normal file
@ -0,0 +1,22 @@
|
||||
# agent-v9-hardening — решения агента
|
||||
|
||||
Спека и дизайн изменения — в `jira-hub/openspec/changes/agent-v9-hardening/` (issue aep/jira-hub#26, aep/jira-agent#23).
|
||||
Агентская часть — решения 1–7 дизайна хаба (запускатель `run.py` v2), 8–14 (агент v9), 15–17 (фейки и тесты),
|
||||
18 (выкатка).
|
||||
|
||||
Решения владельца 14.10.2026: один пакет — #22 (M-3…M-7, M-10), #18 (M-2, M-3, M-5, M-6), #4 (M-10, точность фейков);
|
||||
устаревший `trial` → `rolled_back`; `OSError` при подмене в запускателе → `rejected` без повтора; `run.py` v2 владелец
|
||||
ставит вручную после самообновления до v9.
|
||||
|
||||
## Decisions made without discussion
|
||||
|
||||
Рулинги R-1…R-5 записаны в `jira-hub/openspec/changes/agent-v9-hardening/design.md`
|
||||
(раздел «Decisions made without discussion»); здесь — только решения, принятые при исполнении агентской части.
|
||||
|
||||
| Маркер | Дата | Решение | Обоснование | Цена ошибки |
|
||||
|---|---|---|---|---|
|
||||
| R-6 | 2026-10-15 | В test_hub_4xx_in_refresh_not_retryable токен «неверный» заменён на «wrong-token» (ASCII) | Кириллица в заголовке Authorization даёт UnicodeEncodeError (latin-1) ещё до запроса, тест не проверял бы 401; проверка (текст «хаб: HTTP 401», retryable=False) не изменилась | Нет: только данные теста; поведение при не-ASCII токене в agent.json — наблюдение O-2 |
|
||||
| R-7 | 2026-10-15 | В run.py пустой/None sha256 в update.json (state=trial) считается устаревшим пробным периодом: условие `not sha or sha256_file(agent) != sha` в reconcile и in_trial (в плане — только `!= sha`) | Если agent.py не читается, sha256_file даёт None и совпал бы с отсутствующим sha — пробный период признан бы живым; пункты 1–2 Review Focus плана требуют обратного | Нет: усиление проверки; худший случай — лишний rolled_back при повреждённом update.json |
|
||||
| R-8 | 2026-10-15 | В README фраза про код 73 добавлена в конец абзаца «Остановка…» (якорь плана «и перезапускается.» в тексте — «перезапускается).») | Якорь плана не совпал с текстом README; смысл и текст предложения сохранены | Нет: размещение предложения в документации |
|
||||
| R-9 | 2026-10-15 | Таймеры подсказок в `Updater` раздельные по виду (`hints[kind]`), а не один общий, как сказано в решении 10 | Review Focus 4 плана: подсказка «запускайте через run.py» не должна глушить подсказку о неполной публикации и наоборот | Нет: лишняя строка в логе раз в час |
|
||||
| R-10 | 2026-10-15 | Финальное ревью I-1: «версия подтверждена», если при живом агенте в update.json увидено любое состояние кроме (`trial`, тот же sha); после выхода запасное правило — состояние не `trial` и sha `agent.py` = sha пробной версии. Подтверждённая версия не откатывается | Агент после `ok` сам пишет `rejected`/`staged` для следующей версии, запускатель опрашивает раз в секунду и мог не увидеть `ok`; спека откатывает только при действующем пробном периоде | Ложное «подтверждено» оставит больную версию без отката (худший случай — перезапуск по обычной паузе) |
|
||||
101
src/agent.py
101
src/agent.py
@ -33,7 +33,8 @@ import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
AGENT_VERSION = 5
|
||||
AGENT_VERSION = 9
|
||||
EXIT_LOCKED = 73 # «журнал занят другим процессом»: для run.py это не сбой версии (design 8)
|
||||
EXIT_UPDATE = 75 # «обновление готово»: run.py подменит agent.py на agent.py.new
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
API = "/rest/api/2"
|
||||
@ -283,7 +284,7 @@ class FolderTransport:
|
||||
os.remove(path)
|
||||
except OSError:
|
||||
pass
|
||||
status, text = int(resp.get("status") or 0), resp.get("body") or ""
|
||||
text = body or "" # status и body проверены выше (M-5 #18)
|
||||
if not 200 <= status < 300:
|
||||
raise HttpError(status, text)
|
||||
try:
|
||||
@ -502,6 +503,15 @@ class ConfigGap(Exception):
|
||||
"""В настройках хаба нет того, что нужно команде (R-11): команда падает с понятным текстом, а не молча теряет поле."""
|
||||
|
||||
|
||||
class HubError(Exception):
|
||||
"""Ошибка хаба внутри команды (refresh: known_keys, state): источник — хаб, не Jira (M-6 #18, design 12).
|
||||
Повтор — только при недоступности хаба и 5xx; refresh в Jira не пишет, повтор безопасен."""
|
||||
|
||||
def __init__(self, cause: Exception):
|
||||
super().__init__("хаб: %s" % cause)
|
||||
self.retryable = isinstance(cause, ConnectionError) or (isinstance(cause, HttpError) and cause.code >= 500)
|
||||
|
||||
|
||||
class Executor:
|
||||
def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None):
|
||||
self.jira, self.s, self.hub = jira, settings, hub
|
||||
@ -519,6 +529,11 @@ class Executor:
|
||||
return handler(cmd.get("params") or {}, cmd)
|
||||
except ConfigGap as e:
|
||||
return self._out("failed", error=str(e))
|
||||
except HubError as e:
|
||||
out = self._out("failed", error=str(e))
|
||||
out["retryable"] = e.retryable
|
||||
out["peer"] = "хабом"
|
||||
return out
|
||||
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)
|
||||
@ -532,6 +547,13 @@ class Executor:
|
||||
return {"status": status, "result": result, "error": error,
|
||||
"jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited}
|
||||
|
||||
def _hub(self, fn, *args):
|
||||
"""Вызов хаба из команды: его сбой — HubError, чтобы не выдать его за обрыв связи с Jira."""
|
||||
try:
|
||||
return fn(*args)
|
||||
except (ConnectionError, HttpError) as e:
|
||||
raise HubError(e) from None
|
||||
|
||||
def gap(self, cmd_type: str) -> None:
|
||||
"""Зазор между записями внутри одной команды — по политике хаба для этого типа."""
|
||||
seconds = int((self.s.get("pacing") or {}).get(cmd_type, 0) or 0)
|
||||
@ -907,7 +929,7 @@ class Executor:
|
||||
if not raws and errors: # ни один ключ не найден: это конфигурация/запрос, а не удалённые задачи
|
||||
raise errors[0]
|
||||
fetched = {r["key"] for r in raws}
|
||||
known = set(self.hub.known_keys()) if self.hub else set()
|
||||
known = set(self._hub(self.hub.known_keys)) if self.hub else set()
|
||||
linked = set()
|
||||
for r in raws:
|
||||
for l in self.links_from(r):
|
||||
@ -935,7 +957,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"), "agent_version": AGENT_VERSION, "issues": chunk,
|
||||
self._hub(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)})
|
||||
@ -954,6 +976,7 @@ class Updater:
|
||||
HINT_SECONDS = 3600 # подсказка «запускайте через run.py» — не чаще (R-10)
|
||||
SELFTEST_TIMEOUT = 120
|
||||
VERSION_RE = re.compile(r"^AGENT_VERSION = (\d+)[ \t]*\r?$", re.M)
|
||||
SHA_RE = re.compile(r"[0-9a-f]{64}")
|
||||
|
||||
def __init__(self, hub, base_dir: str, launcher_version: Optional[int], enabled: bool = True,
|
||||
config_path: Optional[str] = None):
|
||||
@ -963,7 +986,7 @@ class Updater:
|
||||
self.new_path = os.path.join(base_dir, "agent.py.new")
|
||||
self.agent_path = os.path.join(base_dir, "agent.py")
|
||||
self.retry_after = {} # (version, sha256) -> clock(), раньше которого не повторять
|
||||
self.hint_at = None # type: Optional[float]
|
||||
self.hints = {} # вид подсказки -> clock() последней записи в лог (не чаще HINT_SECONDS)
|
||||
self.clock = time.monotonic
|
||||
self.should_stop = lambda: False
|
||||
self.run_selftest = self._run_selftest
|
||||
@ -975,16 +998,27 @@ class Updater:
|
||||
"""Первый успешный claim новой версии: trial этой версии → ok (запускатель перестаёт ждать отката)."""
|
||||
st = self.state()
|
||||
if st and st.get("state") == "trial" and st.get("to_version") == AGENT_VERSION:
|
||||
write_update_state(self.state_path, "ok", st.get("from_version"), AGENT_VERSION, st.get("sha256"))
|
||||
try:
|
||||
write_update_state(self.state_path, "ok", st.get("from_version"), AGENT_VERSION, st.get("sha256"))
|
||||
except OSError as e: # диск полон — не падать; trial остался, повтор на следующем claim (M-3)
|
||||
log("не удалось записать update.json: %s" % e)
|
||||
return
|
||||
log("обновление до v%d подтверждено" % AGENT_VERSION)
|
||||
|
||||
def _hint(self, kind: str, msg: str) -> None:
|
||||
now = self.clock()
|
||||
at = self.hints.get(kind)
|
||||
if at is None or now - at >= self.HINT_SECONDS:
|
||||
self.hints[kind] = now
|
||||
log(msg)
|
||||
|
||||
def check(self, release) -> bool:
|
||||
"""Вызывается после claim без команды. True — agent.py.new проверен, update.json = staged: run выходит с 75."""
|
||||
if not self.enabled or not isinstance(release, dict):
|
||||
return False
|
||||
try:
|
||||
version, sha = int(release.get("version")), str(release.get("sha256"))
|
||||
except (TypeError, ValueError):
|
||||
version, sha = release.get("version"), release.get("sha256")
|
||||
if type(version) is not int or not isinstance(sha, str) or not self.SHA_RE.fullmatch(sha):
|
||||
self._hint("format", "хаб прислал неполную публикацию: version=%.80r, sha256=%.80r" % (version, sha)) # design 10
|
||||
return False
|
||||
if version <= AGENT_VERSION:
|
||||
return False
|
||||
@ -993,10 +1027,7 @@ class Updater:
|
||||
and st.get("sha256") == sha:
|
||||
return False
|
||||
if self.launcher_version is None:
|
||||
now = self.clock()
|
||||
if self.hint_at is None or now - self.hint_at >= self.HINT_SECONDS:
|
||||
self.hint_at = now
|
||||
log("доступна v%d; для самообновления запускайте через run.py" % version)
|
||||
self._hint("launcher", "доступна v%d; для самообновления запускайте через run.py" % version)
|
||||
return False
|
||||
until = self.retry_after.get((version, sha))
|
||||
if until is not None and self.clock() < until:
|
||||
@ -1022,22 +1053,31 @@ class Updater:
|
||||
if self.VERSION_RE.findall(content) != [str(version)]:
|
||||
return self._reject(version, sha, "AGENT_VERSION в файле не %d" % version)
|
||||
tmp = self.new_path + ".tmp"
|
||||
with open(tmp, "wb") as f:
|
||||
f.write(data)
|
||||
try:
|
||||
os.chmod(tmp, os.stat(self.agent_path).st_mode & 0o777)
|
||||
except OSError:
|
||||
pass
|
||||
os.replace(tmp, self.new_path)
|
||||
try: # диск полон, нет прав — временный сбой, не трейсбек (design 9)
|
||||
with open(tmp, "wb") as f:
|
||||
f.write(data)
|
||||
try:
|
||||
os.chmod(tmp, os.stat(self.agent_path).st_mode & 0o777)
|
||||
except OSError:
|
||||
pass
|
||||
os.replace(tmp, self.new_path)
|
||||
except OSError as e:
|
||||
return self._transient(version, sha, "файлы: %s" % e)
|
||||
if self.should_stop():
|
||||
self._drop_new()
|
||||
return False
|
||||
code, out = self.run_selftest(self.new_path)
|
||||
try:
|
||||
code, out = self.run_selftest(self.new_path)
|
||||
except OSError as e:
|
||||
return self._transient(version, sha, "selftest: %s" % e)
|
||||
if self.should_stop(): # Ctrl-C уронил и selftest — это не отказ версии
|
||||
self._drop_new()
|
||||
return False
|
||||
if code == 0:
|
||||
write_update_state(self.state_path, "staged", AGENT_VERSION, version, sha)
|
||||
try:
|
||||
write_update_state(self.state_path, "staged", AGENT_VERSION, version, sha)
|
||||
except OSError as e:
|
||||
return self._transient(version, sha, "update.json: %s" % e)
|
||||
log("v%d готова, перезапуск через run.py" % version)
|
||||
return True
|
||||
if code is None:
|
||||
@ -1058,7 +1098,7 @@ class Updater:
|
||||
for path in (self.new_path, self.new_path + ".tmp"):
|
||||
try:
|
||||
os.remove(path)
|
||||
except FileNotFoundError:
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def _transient(self, version, sha, reason) -> bool:
|
||||
@ -1069,7 +1109,10 @@ class Updater:
|
||||
|
||||
def _reject(self, version, sha, reason) -> bool:
|
||||
self._drop_new()
|
||||
write_update_state(self.state_path, "rejected", AGENT_VERSION, version, sha, reason)
|
||||
try:
|
||||
write_update_state(self.state_path, "rejected", AGENT_VERSION, version, sha, reason)
|
||||
except OSError as e:
|
||||
log("не удалось записать update.json: %s" % e)
|
||||
log("v%d отвергнута: %s" % (version, reason))
|
||||
return False
|
||||
|
||||
@ -1119,7 +1162,8 @@ class Runner:
|
||||
ago = (datetime.now(timezone.utc) - at).total_seconds()
|
||||
if ago < 0:
|
||||
log("журнал: время последней записи в будущем (%s) — пауза перед первой командой полная" % at.isoformat())
|
||||
ago = 0
|
||||
self.jira.last_call_at = time.monotonic()
|
||||
return # вторая строка «не позже <будущее>» противоречила бы первой (M-2 #18)
|
||||
self.jira.last_call_at = time.monotonic() - ago
|
||||
log("темп: последний вызов Jira не позже %s (по журналу)"
|
||||
% at.astimezone(timezone.utc).replace(microsecond=0).isoformat())
|
||||
@ -1190,14 +1234,15 @@ class Runner:
|
||||
out = self.run_once(cmd)
|
||||
calls += out["jira_calls"]
|
||||
limited += out["rate_limited"]
|
||||
peer = out.pop("peer", "Jira") # с кем оборвалась связь: Jira или хаб (refresh)
|
||||
if not out.pop("retryable", False):
|
||||
break
|
||||
if attempt >= self.RETRY_ATTEMPTS:
|
||||
out = {"status": "failed", "result": None,
|
||||
"error": "результат неизвестен (обрыв связи с Jira), повтор безопасен: %s" % out.get("error")}
|
||||
"error": "результат неизвестен (обрыв связи с %s), повтор безопасен: %s" % (peer, out.get("error"))}
|
||||
break
|
||||
attempt += 1
|
||||
log(" ! обрыв связи с Jira, повтор %d/%d через проверку: %s" % (attempt, self.RETRY_ATTEMPTS, out.get("error")))
|
||||
log(" ! обрыв связи с %s, повтор %d/%d через проверку: %s" % (peer, attempt, self.RETRY_ATTEMPTS, out.get("error")))
|
||||
# F-1: обрыв не обновляет last_call_at, поэтому ждём фиксированно от момента отказа, а не от последнего вызова
|
||||
if not self.wait_seconds(max(float(cmd.get("pause_seconds") or 0), self.RETRY_MIN_WAIT)):
|
||||
return # остановка: finished не пишем, команду подхватит recover
|
||||
@ -1320,7 +1365,7 @@ def cmd_run(cfg: dict, journal_path: str, config_path: Optional[str] = None) ->
|
||||
lock = acquire_lock(journal_path)
|
||||
if lock is None:
|
||||
log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path)
|
||||
return 1
|
||||
return EXIT_LOCKED
|
||||
try:
|
||||
return _run(cfg, journal_path, config_path)
|
||||
finally:
|
||||
|
||||
145
src/run.py
145
src/run.py
@ -19,8 +19,9 @@ import time
|
||||
from datetime import datetime, timezone
|
||||
from typing import List, Optional
|
||||
|
||||
LAUNCHER_VERSION = 1
|
||||
LAUNCHER_VERSION = 2
|
||||
EXIT_UPDATE = 75 # агент: «agent.py.new проверен, update.json = staged»
|
||||
EXIT_LOCKED = 73 # агент: «журнал занят другим процессом» — не сбой версии (design 6)
|
||||
TRIAL_SECONDS = 600 # новая версия должна записать ok за это время
|
||||
KILL_GRACE = 60 # после SIGTERM до SIGKILL
|
||||
BACKOFF_START, BACKOFF_MAX = 5, 300
|
||||
@ -41,20 +42,36 @@ def read_state(path: str) -> Optional[dict]:
|
||||
return data if isinstance(data, dict) else None
|
||||
|
||||
|
||||
def write_state(path: str, state: str, base: Optional[dict], error=None) -> None:
|
||||
def write_state(path: str, state: str, base: Optional[dict], error=None) -> bool:
|
||||
"""Атомарная запись update.json. False — не записано (диск полон): запускатель не падает (design 5)."""
|
||||
base = base or {}
|
||||
data = {"state": state, "from_version": base.get("from_version"), "to_version": base.get("to_version"),
|
||||
"sha256": base.get("sha256"), "error": error,
|
||||
"at": datetime.now(timezone.utc).replace(microsecond=0).isoformat()}
|
||||
tmp = path + ".tmp"
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, ensure_ascii=False)
|
||||
os.replace(tmp, path)
|
||||
try:
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, ensure_ascii=False)
|
||||
os.replace(tmp, path)
|
||||
except OSError as e:
|
||||
log("не удалось записать update.json: %s" % e)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def sha256_file(path: str) -> str:
|
||||
with open(path, "rb") as f:
|
||||
return hashlib.sha256(f.read()).hexdigest()
|
||||
def sha256_file(path: str) -> Optional[str]:
|
||||
try:
|
||||
with open(path, "rb") as f:
|
||||
return hashlib.sha256(f.read()).hexdigest()
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def remove(path: str) -> None:
|
||||
try:
|
||||
os.remove(path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def acquire_lock(base_dir: str):
|
||||
@ -96,34 +113,80 @@ class Launcher:
|
||||
child.send_signal(signal.SIGTERM)
|
||||
|
||||
# --- файлы версии ----------------------------------------------------------
|
||||
def recover_files(self) -> None:
|
||||
def reconcile(self) -> None:
|
||||
"""Сверка update.json с файлами при старте (design 1): запускатель мог упасть посреди подмены или отката."""
|
||||
for tmp in (self.new + ".tmp", self.prev + ".tmp", self.state_path + ".tmp"):
|
||||
remove(tmp)
|
||||
if not os.path.exists(self.agent) and os.path.exists(self.prev):
|
||||
os.replace(self.prev, self.agent)
|
||||
log("agent.py не найден — восстановлен из agent.py.prev")
|
||||
if os.path.exists(self.new):
|
||||
st = read_state(self.state_path)
|
||||
if st and st.get("state") == "staged":
|
||||
st = read_state(self.state_path)
|
||||
state, sha = (st or {}).get("state"), (st or {}).get("sha256")
|
||||
if state == "staged":
|
||||
if sha and sha256_file(self.agent) == sha: # подмена сделана, trial не записан (M-5)
|
||||
remove(self.new)
|
||||
write_state(self.state_path, "trial", st)
|
||||
log("v%s уже на месте — пробный запуск" % st.get("to_version"))
|
||||
elif sha and os.path.exists(self.new) and sha256_file(self.new) == sha:
|
||||
log("найдена подготовленная версия — ставлю")
|
||||
self.apply_update()
|
||||
else:
|
||||
os.remove(self.new)
|
||||
remove(self.new)
|
||||
write_state(self.state_path, "rejected", st, "подготовленный файл потерян")
|
||||
log("подготовленная v%s потеряна — отвергнута" % st.get("to_version"))
|
||||
return
|
||||
remove(self.new)
|
||||
if state == "trial" and (not sha or sha256_file(self.agent) != sha):
|
||||
self.drop_stale_trial(st)
|
||||
|
||||
def in_trial(self) -> bool:
|
||||
"""Пробный период для этого запуска: trial и agent.py — именно эта версия (design 2)."""
|
||||
st = read_state(self.state_path)
|
||||
if not (st and st.get("state") == "trial"):
|
||||
return False
|
||||
if not st.get("sha256") or sha256_file(self.agent) != st.get("sha256"):
|
||||
self.drop_stale_trial(st)
|
||||
return False
|
||||
return True
|
||||
|
||||
def drop_stale_trial(self, st: dict) -> None:
|
||||
"""agent.py не та версия, что в trial (прерван откат, ручная подмена): пробного периода нет, версию не ставить (M-3)."""
|
||||
reason = "пробный период прерван: agent.py не v%s" % st.get("to_version")
|
||||
write_state(self.state_path, "rolled_back", st, reason)
|
||||
log(reason)
|
||||
|
||||
def apply_update(self) -> None:
|
||||
st = read_state(self.state_path)
|
||||
if not (st and st.get("state") == "staged" and os.path.exists(self.new) and sha256_file(self.new) == st.get("sha256")):
|
||||
if os.path.exists(self.new):
|
||||
os.remove(self.new)
|
||||
remove(self.new)
|
||||
write_state(self.state_path, "rejected", st, "файл изменился после проверки")
|
||||
log("обновление отвергнуто: файл изменился после проверки")
|
||||
return
|
||||
tmp = self.prev + ".tmp"
|
||||
with open(self.agent, "rb") as src, open(tmp, "wb") as dst:
|
||||
dst.write(src.read())
|
||||
os.replace(tmp, self.prev)
|
||||
os.replace(self.new, self.agent) # agent.py существует на каждом шаге
|
||||
write_state(self.state_path, "trial", st)
|
||||
try:
|
||||
with open(self.agent, "rb") as src, open(tmp, "wb") as dst:
|
||||
dst.write(src.read())
|
||||
os.replace(tmp, self.prev)
|
||||
os.replace(self.new, self.agent) # agent.py существует на каждом шаге
|
||||
except OSError as e:
|
||||
self.fail_update(st, e)
|
||||
return
|
||||
if not write_state(self.state_path, "trial", st):
|
||||
try:
|
||||
os.replace(self.prev, self.agent) # без trial новая версия осталась бы без пробного периода
|
||||
except OSError as e:
|
||||
log("не удалось вернуть agent.py.prev: %s" % e)
|
||||
self.fail_update(st, "update.json не записан")
|
||||
return
|
||||
log("v%s → v%s: пробный запуск" % (st.get("from_version"), st.get("to_version")))
|
||||
|
||||
def fail_update(self, st: dict, err) -> None:
|
||||
"""Подмена не удалась (диск полон, права): текущая версия остаётся, эта — rejected навсегда (design 4)."""
|
||||
remove(self.prev + ".tmp")
|
||||
remove(self.new)
|
||||
write_state(self.state_path, "rejected", st, "подмена не удалась: %s" % err)
|
||||
log("подмена v%s не удалась: %s" % (st.get("to_version"), err))
|
||||
|
||||
def rollback(self, st: Optional[dict], reason: str) -> bool:
|
||||
if not os.path.exists(self.prev):
|
||||
log("откат невозможен: нет agent.py.prev")
|
||||
@ -147,20 +210,26 @@ class Launcher:
|
||||
child.wait()
|
||||
|
||||
def wait_child(self, child, trial: bool):
|
||||
"""Код выхода и признак «пробный период истёк». В пробный период ждём ok не дольше trial_seconds."""
|
||||
"""(код выхода, «пробный период истёк», «версия подтверждена»). В пробный период ждём ok не дольше trial_seconds.
|
||||
Подтверждение — любое состояние update.json, кроме trial этой же версии, пока агент жив: запускатель сам
|
||||
update.json при живом агенте не пишет, значит это сделал подтвердивший агент (ok, затем rejected/staged для
|
||||
следующей версии — I-1)."""
|
||||
start = time.monotonic()
|
||||
st0 = read_state(self.state_path) if trial else None
|
||||
trial_sha = (st0 or {}).get("sha256")
|
||||
confirmed = False
|
||||
while True:
|
||||
try:
|
||||
return child.wait(1), False
|
||||
return child.wait(1), False, confirmed
|
||||
except subprocess.TimeoutExpired:
|
||||
pass
|
||||
if trial:
|
||||
if trial and not confirmed:
|
||||
st = read_state(self.state_path)
|
||||
if st and st.get("state") == "ok":
|
||||
trial = False
|
||||
if st and not (st.get("state") == "trial" and st.get("sha256") == trial_sha):
|
||||
confirmed = True
|
||||
elif time.monotonic() - start >= self.trial_seconds:
|
||||
self.terminate(child)
|
||||
return child.returncode, True
|
||||
return child.returncode, True, False
|
||||
|
||||
def pause(self, seconds: float) -> bool:
|
||||
"""Пауза перед перезапуском; False — остановлены во время паузы."""
|
||||
@ -180,22 +249,32 @@ class Launcher:
|
||||
return wait
|
||||
|
||||
def run(self) -> int:
|
||||
self.recover_files()
|
||||
self.reconcile()
|
||||
code = 0
|
||||
while True:
|
||||
if self.stopping: # сигнал пришёл между запусками — нового агента не запускаем (M-2)
|
||||
return code
|
||||
st = read_state(self.state_path)
|
||||
trial = bool(st and st.get("state") == "trial")
|
||||
trial = self.in_trial()
|
||||
trial_sha = (read_state(self.state_path) or {}).get("sha256") if trial else None
|
||||
started = time.monotonic()
|
||||
self.interrupted = False # Ctrl-C относится к этому запуску агента
|
||||
self.child = self.spawn()
|
||||
if self.stopping: # SIGTERM между Popen и присваиванием child до агента не дошёл
|
||||
self.child.send_signal(signal.SIGTERM)
|
||||
code, timed_out = self.wait_child(self.child, trial)
|
||||
code, timed_out, confirmed = self.wait_child(self.child, trial)
|
||||
self.child = None
|
||||
uptime = time.monotonic() - started
|
||||
st = read_state(self.state_path)
|
||||
ok = bool(st and st.get("state") == "ok")
|
||||
if trial and not (ok or confirmed) and st and not (st.get("state") == "trial" and st.get("sha256") == trial_sha) \
|
||||
and sha256_file(self.agent) == trial_sha:
|
||||
confirmed = True # ok и следующая запись агента случились между опросами (I-1)
|
||||
ok = ok or confirmed
|
||||
reason = "агент упал (код %s)" % code
|
||||
if timed_out and ok: # ответ claim пришёл в последнюю секунду (M-4, design 3)
|
||||
log("подтверждение пришло в последний момент — отката нет")
|
||||
self.backoff = BACKOFF_START
|
||||
continue
|
||||
if timed_out:
|
||||
if self.rollback(st, "не вышла на связь за 10 мин"):
|
||||
continue
|
||||
@ -207,13 +286,15 @@ class Launcher:
|
||||
if self.interrupted: # Ctrl-C во время подготовки: подмена сделана, trial — при следующем старте
|
||||
return 0
|
||||
continue
|
||||
elif trial and not (st and st.get("state") == "ok"):
|
||||
elif code == EXIT_LOCKED: # журнал держит другой процесс — версия ни при чём (M-7, design 6)
|
||||
reason = "журнал занят другим процессом (агент уже запущен)"
|
||||
elif trial and not ok:
|
||||
if self.rollback(st, "код %s через %d с" % (code, uptime)):
|
||||
continue
|
||||
if self.stopping:
|
||||
return code
|
||||
wait = self.next_backoff(uptime)
|
||||
log("агент упал (код %s), перезапуск через %d с" % (code, wait))
|
||||
log("%s, перезапуск через %d с" % (reason, wait))
|
||||
if not self.pause(wait):
|
||||
return code
|
||||
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
import json
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
||||
|
||||
@ -22,6 +23,7 @@ class FakeHub:
|
||||
self.in_flight = None
|
||||
self.release = None # {"version", "sha256", "content"} — опубликованная версия агента
|
||||
self.release_requests = 0
|
||||
self.claim_wait = 0 # с: при пустой очереди держать claim, как хаб
|
||||
self.lock = threading.Lock()
|
||||
hub = self
|
||||
|
||||
@ -61,8 +63,16 @@ class FakeHub:
|
||||
body = json.loads(h.rfile.read(length).decode("utf-8")) if length else {}
|
||||
if h.headers.get("Authorization") != "Bearer " + self.token:
|
||||
return self._send(h, 401, {"error": "unauthorized"})
|
||||
path = h.path.split("?")[0]
|
||||
if path == "/agent/claim" and self.claim_wait:
|
||||
end = time.monotonic() + self.claim_wait
|
||||
while time.monotonic() < end: # ждать без блокировки: другие запросы (enqueue) проходят
|
||||
with self.lock:
|
||||
if self.queue:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
with self.lock:
|
||||
code, payload = self.route(method, h.path.split("?")[0], body)
|
||||
code, payload = self.route(method, path, body)
|
||||
self._send(h, code, payload)
|
||||
|
||||
def _send(self, h, code, payload):
|
||||
|
||||
@ -4,9 +4,12 @@ import json
|
||||
import re
|
||||
import socket
|
||||
import threading
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
MSK = timezone(timedelta(hours=3))
|
||||
|
||||
WORKFLOW = { # статус → [(имя перехода, целевой статус)]
|
||||
"Ожидание разработки": [("Взять в разработку", "Разработка"), ("Ожидает анализа", "Ожидает анализа"), ("Снято", "Снято")],
|
||||
"Разработка": [("Малое исправление", "Готово к тестированию"), ("Готово", "Готово"), ("Снято", "Снято")],
|
||||
@ -24,6 +27,7 @@ CUSTOM = ("customfield_10801", "customfield_10802", "customfield_15302", "custom
|
||||
class FakeJira:
|
||||
def __init__(self, project="AURORA"):
|
||||
self.project = project
|
||||
self.now = datetime.now(MSK) # часы фейка: окно created >= -Nd считается от них
|
||||
self.issues = {}
|
||||
self.comments = {}
|
||||
self.links = []
|
||||
@ -64,10 +68,14 @@ class FakeJira:
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
|
||||
@staticmethod
|
||||
def ts(dt):
|
||||
return dt.astimezone(MSK).strftime("%Y-%m-%dT%H:%M:%S.000+0300")
|
||||
|
||||
def add(self, key, in_base=True, **fields):
|
||||
base = {"summary": "Тема " + key, "description": None, "status": "Ожидание разработки", "priority": "Средний",
|
||||
"assignee": None, "reporter": "someone", "duedate": None, "labels": ["AWPCA"] if in_base else [],
|
||||
"issuetype": "Задача", "created": "2026-10-01T10:00:00.000+0300", "updated": "2026-10-01T10:00:00.000+0300",
|
||||
"issuetype": "Задача", "created": self.ts(self.now - timedelta(hours=1)), "updated": "2026-10-01T10:00:00.000+0300",
|
||||
"sprint": None}
|
||||
self.id_seq += 1
|
||||
base["id"] = str(self.id_seq)
|
||||
@ -108,6 +116,14 @@ class FakeJira:
|
||||
return [k for k in keys if k in self.issues]
|
||||
if "reporter = currentUser()" in (jql or ""):
|
||||
keys = [k for k, f in self.issues.items() if f["reporter"] == "me"]
|
||||
m = re.search(r"project = (\w+)", jql)
|
||||
if m: # как Jira: только задачи проекта
|
||||
keys = [k for k in keys if k.startswith(m.group(1) + "-")]
|
||||
m = re.search(r"created >= -(\d+)d", jql)
|
||||
if m: # как Jira: только созданные за последние N дней
|
||||
since = self.now - timedelta(days=int(m.group(1)))
|
||||
keys = [k for k in keys
|
||||
if datetime.strptime(self.issues[k]["created"], "%Y-%m-%dT%H:%M:%S.%f%z") >= since]
|
||||
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"]]
|
||||
|
||||
@ -165,7 +181,7 @@ class FakeJira:
|
||||
self.add(key, summary=f.get("summary"), description=f.get("description"),
|
||||
priority=(f.get("priority") or {}).get("name", "Средний"), duedate=f.get("duedate"),
|
||||
labels=list(f.get("labels") or []), issuetype=self._type_name(f.get("issuetype")),
|
||||
reporter="me", created="2026-10-11T12:00:00.000+0300",
|
||||
reporter="me", created=self.ts(self.now),
|
||||
**{c: f.get(c) for c in CUSTOM})
|
||||
return 201, {"id": "1", "key": key}
|
||||
if p == "/rest/api/2/issueLinkType":
|
||||
|
||||
@ -214,7 +214,7 @@ class ClientsTest(unittest.TestCase):
|
||||
with contextlib.redirect_stdout(out):
|
||||
# хаб недоступен (порт 1), Jira доступен — selftest печатает версию и возвращает 3, а не 1
|
||||
self.assertEqual(agent.main(["--config", path, "selftest"]), 3)
|
||||
self.assertIn("jira-agent v5", out.getvalue())
|
||||
self.assertIn("jira-agent v%d" % agent.AGENT_VERSION, out.getvalue())
|
||||
self.assertIn("Jira: доступен", out.getvalue())
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
import unittest
|
||||
from datetime import timedelta
|
||||
|
||||
from _common import SETTINGS, agent, jira_cfg, no_sleep
|
||||
from fake_jira import FakeJira
|
||||
@ -146,6 +147,20 @@ class ExecutorFieldsTest(unittest.TestCase):
|
||||
out = self.run_cmd("ensure_issue", temp_key="n:2", issue_type="task", summary="Точная тема")
|
||||
self.assertEqual(out["result"], {"key": "AURORA-501", "existed": True})
|
||||
|
||||
def test_ensure_issue_old_or_foreign_is_not_duplicate(self):
|
||||
self.fake.add("AURORA-500", reporter="me", summary="Нужная задача",
|
||||
created=self.fake.ts(self.fake.now - timedelta(days=8))) # за окном created >= -7d
|
||||
self.fake.add("OTHER-1", reporter="me", summary="Нужная задача") # чужой проект
|
||||
out = self.run_cmd("ensure_issue", temp_key="n:1", issue_type="task", summary="Нужная задача")
|
||||
self.assertTrue(out["result"]["created"])
|
||||
self.assertNotIn(out["result"]["key"], ("AURORA-500", "OTHER-1"))
|
||||
|
||||
def test_ensure_issue_within_window_is_duplicate(self):
|
||||
self.fake.add("AURORA-500", reporter="me", summary="Нужная задача",
|
||||
created=self.fake.ts(self.fake.now - timedelta(days=6)))
|
||||
out = self.run_cmd("ensure_issue", temp_key="n:1", issue_type="task", summary="Нужная задача")
|
||||
self.assertEqual(out["result"], {"key": "AURORA-500", "existed": True})
|
||||
|
||||
def test_http_error_becomes_failed(self):
|
||||
self.fake.fail_next.append((400, {}))
|
||||
out = self.run_cmd("ensure_issue", temp_key="n:1", issue_type="task", summary="x")
|
||||
|
||||
@ -157,8 +157,8 @@ class FolderIntegrationTest(unittest.TestCase):
|
||||
def cfg(self):
|
||||
return jira_cfg(self.fake, hub_url="folder:" + self.root, hub_token=self.hub.token)
|
||||
|
||||
def test_version_is_5(self):
|
||||
self.assertEqual(agent.AGENT_VERSION, 5)
|
||||
def test_version_is_9(self):
|
||||
self.assertEqual(agent.AGENT_VERSION, 9)
|
||||
|
||||
def test_runner_through_folder(self):
|
||||
self.fake.add("AURORA-1", duedate="2026-10-20")
|
||||
|
||||
@ -6,6 +6,7 @@ import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import textwrap
|
||||
import time
|
||||
import unittest
|
||||
|
||||
from _common import SRC
|
||||
@ -67,6 +68,35 @@ V6_OK_THEN_STOP = fake("v6", '''
|
||||
''')
|
||||
V6_CRASH = fake("v6", 'sys.exit(1)\n')
|
||||
V6_HANG = fake("v6", 'import time\ntime.sleep(60)\n')
|
||||
V6_SEES_STATE = fake("v6", '''
|
||||
mark("seen:" + (state() or {}).get("state", "-"))
|
||||
write(**dict(state(), state="ok"))
|
||||
human_stop()
|
||||
''')
|
||||
V6_OK_ON_TERM = fake("v6", '''
|
||||
import signal, time
|
||||
if len(open(os.path.join(HERE, "runs.log")).read().split()) > 2:
|
||||
human_stop()
|
||||
def term(*a):
|
||||
write(**dict(state(), state="ok")) # ответ claim пришёл в последнюю секунду пробного периода
|
||||
sys.exit(0)
|
||||
signal.signal(signal.SIGTERM, term)
|
||||
time.sleep(60)
|
||||
''')
|
||||
|
||||
|
||||
# v6 подтвердился (ok), через delay с записал rejected для v10 с другим sha (как Updater._reject) и упал с кодом 1.
|
||||
V6_OK_THEN_REJECT_NEXT = fake("v6", '''
|
||||
import time
|
||||
n = len(open(os.path.join(HERE, "runs.log")).read().split())
|
||||
if n > 2:
|
||||
human_stop()
|
||||
st = state()
|
||||
write(**dict(st, state="ok"))
|
||||
time.sleep(DELAY)
|
||||
write(state="rejected", from_version=6, to_version=10, sha256="чужой", error="отвергнута")
|
||||
sys.exit(1)
|
||||
''')
|
||||
|
||||
|
||||
class LauncherTest(unittest.TestCase):
|
||||
@ -106,6 +136,12 @@ class LauncherTest(unittest.TestCase):
|
||||
self.put("update.json", json.dumps({"state": "staged", "from_version": 5, "to_version": to_version, "sha256": sha}))
|
||||
return sha
|
||||
|
||||
def sha(self, text):
|
||||
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
||||
|
||||
def path(self, name):
|
||||
return os.path.join(self.dir, name)
|
||||
|
||||
def test_update_applied_and_confirmed(self):
|
||||
self.put("agent.py", V5_UPDATES)
|
||||
sha = self.next_version(V6_OK_THEN_STOP)
|
||||
@ -189,7 +225,7 @@ class LauncherTest(unittest.TestCase):
|
||||
human_stop(130)
|
||||
'''))
|
||||
self.assertEqual(self.l.run(), 130)
|
||||
self.assertEqual(self.read("env.txt"), "1 run --journal j.jsonl")
|
||||
self.assertEqual(self.read("env.txt"), "2 run --journal j.jsonl")
|
||||
|
||||
def test_sigint_without_child_stops(self):
|
||||
self.l.on_sigint(signal.SIGINT, None)
|
||||
@ -287,6 +323,166 @@ class LauncherTest(unittest.TestCase):
|
||||
self.assertIn("запускатель уже работает", p.stdout.decode("utf-8"))
|
||||
self.assertFalse(os.path.exists(os.path.join(self.dir, "runs.log")))
|
||||
|
||||
def test_staged_already_replaced_resumes_trial(self):
|
||||
"""M-5: запускатель упал между os.replace(new, agent) и записью trial."""
|
||||
self.stage(V6_SEES_STATE)
|
||||
os.replace(self.path("agent.py.new"), self.path("agent.py"))
|
||||
self.put("agent.py.prev", fake("v5", "human_stop()\n"))
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
self.assertEqual(self.runs(), ["v6", "seen:trial"]) # пробный период, а не staged навсегда
|
||||
self.assertEqual(self.state()["state"], "ok")
|
||||
|
||||
def test_staged_without_file_rejected(self):
|
||||
self.put("agent.py", fake("v5", "human_stop()\n"))
|
||||
self.put("update.json", json.dumps({"state": "staged", "from_version": 5, "to_version": 6, "sha256": "a" * 64}))
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
st = self.state()
|
||||
self.assertEqual((st["state"], st["error"]), ("rejected", "подготовленный файл потерян"))
|
||||
self.assertEqual(self.runs(), ["v5"])
|
||||
|
||||
def test_stale_trial_marked_rolled_back(self):
|
||||
"""M-3: откат прерван — agent.py уже прежняя версия, update.json всё ещё trial."""
|
||||
self.l.trial_seconds = 1
|
||||
self.put("agent.py", fake("v8", "import time\ntime.sleep(2)\nhuman_stop()\n")) # дольше пробного периода
|
||||
self.put("update.json", json.dumps({"state": "trial", "from_version": 8, "to_version": 9, "sha256": "a" * 64}))
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
st = self.state()
|
||||
self.assertEqual((st["state"], st["to_version"], st["error"]),
|
||||
("rolled_back", 9, "пробный период прерван: agent.py не v9"))
|
||||
self.assertEqual(self.runs(), ["v8"]) # не остановлен через пробный период
|
||||
|
||||
def test_trial_without_sha_is_stale(self):
|
||||
self.put("agent.py", fake("v8", "human_stop()\n"))
|
||||
self.put("update.json", json.dumps({"state": "trial", "from_version": 8, "to_version": 9}))
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
self.assertEqual(self.state()["state"], "rolled_back")
|
||||
|
||||
def test_trial_survives_launcher_restart(self):
|
||||
"""M-10 #22: trial при перезапуске run.py — пробный период продолжается, падение даёт откат."""
|
||||
self.put("agent.py", V6_CRASH)
|
||||
self.put("agent.py.prev", fake("v5", "human_stop()\n"))
|
||||
self.put("update.json", json.dumps({"state": "trial", "from_version": 5, "to_version": 6, "sha256": self.sha(V6_CRASH)}))
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
self.assertEqual(self.runs(), ["v6", "v5"])
|
||||
self.assertEqual(self.state()["state"], "rolled_back")
|
||||
|
||||
def test_ok_at_last_moment_no_rollback(self):
|
||||
"""M-4: пробный период истёк, но агент записал ok, получив SIGTERM."""
|
||||
self.l.trial_seconds = 1
|
||||
self.l.kill_grace = 5
|
||||
self.put("agent.py", V5_UPDATES)
|
||||
self.next_version(V6_OK_ON_TERM)
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
self.assertEqual(self.read("agent.py"), V6_OK_ON_TERM)
|
||||
self.assertEqual(self.state()["state"], "ok")
|
||||
self.assertEqual(self.runs(), ["v5", "v6", "v6"])
|
||||
self.assertEqual(self.pauses, []) # перезапуск без паузы
|
||||
|
||||
def _confirmed_then_rejected_next(self, delay):
|
||||
self.put("agent.py", V5_UPDATES)
|
||||
self.next_version(V6_OK_THEN_REJECT_NEXT.replace("DELAY", repr(delay)))
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
self.assertEqual(self.read("agent.py"), V6_OK_THEN_REJECT_NEXT.replace("DELAY", repr(delay)))
|
||||
self.assertEqual(self.runs(), ["v5", "v6", "v6"])
|
||||
st = self.state()
|
||||
self.assertEqual((st["state"], st["to_version"]), ("rejected", 10)) # не rolled_back: v10 не заблокирована
|
||||
|
||||
def test_confirmed_then_next_rejected_no_rollback(self):
|
||||
"""I-1: агент записал ok, потом сам перезаписал update.json (rejected для v10) и упал — v6 здоровая, откат не нужен."""
|
||||
self._confirmed_then_rejected_next(2.5)
|
||||
|
||||
def test_confirmed_and_rejected_within_poll_no_rollback(self):
|
||||
"""I-1: ok и rejected записаны быстрее, чем запускатель опрашивает файл раз в секунду."""
|
||||
self._confirmed_then_rejected_next(0)
|
||||
|
||||
def test_locked_in_trial_no_rollback(self):
|
||||
"""M-7: журнал держит осиротевший агент — код 73 не откатывает новую версию."""
|
||||
v6 = fake("v6", '''
|
||||
n = len(open(os.path.join(HERE, "runs.log")).read().split())
|
||||
human_stop() if n > 1 else sys.exit(73)
|
||||
''')
|
||||
self.put("agent.py", v6)
|
||||
self.put("agent.py.prev", fake("v5", "human_stop()\n"))
|
||||
self.put("update.json", json.dumps({"state": "trial", "from_version": 5, "to_version": 6, "sha256": self.sha(v6)}))
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
self.assertEqual(self.runs(), ["v6", "v6"])
|
||||
self.assertEqual(self.pauses, [5])
|
||||
self.assertEqual(self.read("agent.py"), v6)
|
||||
self.assertEqual(self.state()["state"], "trial")
|
||||
|
||||
def test_locked_outside_trial_backs_off(self):
|
||||
self.put("agent.py", fake("v6", '''
|
||||
n = len(open(os.path.join(HERE, "runs.log")).read().split())
|
||||
human_stop() if n > 3 else sys.exit(73)
|
||||
'''))
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
self.assertEqual(self.pauses, [5, 10, 20])
|
||||
|
||||
def test_os_error_before_replace_rejected(self):
|
||||
"""M-6: копия в agent.py.prev.tmp не удалась (здесь — на месте файла каталог)."""
|
||||
self.put("agent.py", V5_UPDATES)
|
||||
self.next_version(V6_OK_THEN_STOP)
|
||||
os.mkdir(self.path("agent.py.prev.tmp"))
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
self.assertEqual(self.read("agent.py"), V5_UPDATES)
|
||||
st = self.state()
|
||||
self.assertEqual(st["state"], "rejected")
|
||||
self.assertTrue(st["error"].startswith("подмена не удалась: "), st["error"])
|
||||
self.assertFalse(os.path.exists(self.path("agent.py.new")))
|
||||
self.assertEqual(self.runs(), ["v5", "v5"])
|
||||
|
||||
def test_trial_not_written_reverts(self):
|
||||
orig = run.write_state
|
||||
def flaky(path, state, base, error=None):
|
||||
return False if state == "trial" else orig(path, state, base, error)
|
||||
run.write_state = flaky
|
||||
self.addCleanup(setattr, run, "write_state", orig)
|
||||
self.put("agent.py", V5_UPDATES)
|
||||
self.next_version(V6_OK_THEN_STOP)
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
self.assertEqual(self.read("agent.py"), V5_UPDATES) # agent.py.prev вернулся на место
|
||||
st = self.state()
|
||||
self.assertEqual((st["state"], st["error"]), ("rejected", "подмена не удалась: update.json не записан"))
|
||||
self.assertEqual(self.runs(), ["v5", "v5"])
|
||||
|
||||
def test_write_state_error_does_not_raise(self):
|
||||
self.assertFalse(run.write_state(self.path("нет/update.json"), "ok", {}))
|
||||
|
||||
def test_sha256_file_unreadable_is_none(self):
|
||||
self.assertIsNone(run.sha256_file(self.path("нет-файла")))
|
||||
|
||||
def test_tmp_files_removed_at_start(self):
|
||||
self.put("agent.py", fake("v5", "human_stop()\n"))
|
||||
for name in ("agent.py.new.tmp", "agent.py.prev.tmp", "update.json.tmp"):
|
||||
self.put(name, "остаток")
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
for name in ("agent.py.new.tmp", "agent.py.prev.tmp", "update.json.tmp"):
|
||||
self.assertFalse(os.path.exists(self.path(name)), name)
|
||||
|
||||
def test_sigterm_forwarded_and_no_restart(self):
|
||||
"""M-10 #22: on_sigterm — пересылка агенту, нового запуска нет."""
|
||||
import threading
|
||||
self.put("agent.py", fake("v5", '''
|
||||
import signal, time
|
||||
def term(*a):
|
||||
mark("term")
|
||||
sys.exit(0)
|
||||
signal.signal(signal.SIGTERM, term)
|
||||
open(os.path.join(HERE, "ready"), "w").close()
|
||||
time.sleep(30)
|
||||
'''))
|
||||
def send():
|
||||
for _ in range(300):
|
||||
if os.path.exists(self.path("ready")):
|
||||
break
|
||||
time.sleep(0.05)
|
||||
self.l.on_sigterm(signal.SIGTERM, None)
|
||||
t = threading.Thread(target=send)
|
||||
t.start()
|
||||
self.assertEqual(self.l.run(), 0)
|
||||
t.join()
|
||||
self.assertEqual(self.runs(), ["v5", "term"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@ -184,6 +184,29 @@ class RefreshTest(unittest.TestCase):
|
||||
self.assertEqual(self.hub.commands[cid]["status"], "done")
|
||||
self.assertEqual(self.hub.results[cid]["status"], "done")
|
||||
|
||||
def test_hub_unreachable_in_refresh_names_hub(self):
|
||||
self.fake.add("AURORA-1")
|
||||
self.ex.hub = agent.HubClient({"hub_url": "http://127.0.0.1:1", "hub_token": "t", "timeout": 2})
|
||||
out = self.ex.run({"id": 1, "type": "refresh", "params": {"keys": ["AURORA-1"]}})
|
||||
self.assertEqual(out["status"], "failed")
|
||||
self.assertTrue(out["error"].startswith("хаб: "), out["error"])
|
||||
self.assertTrue(out["retryable"])
|
||||
self.assertEqual(out["peer"], "хабом")
|
||||
|
||||
def test_hub_4xx_in_refresh_not_retryable(self):
|
||||
self.fake.add("AURORA-1")
|
||||
self.ex.hub = agent.HubClient({"hub_url": self.hub.base_url, "hub_token": "wrong-token", "timeout": 5})
|
||||
out = self.ex.run({"id": 1, "type": "refresh", "params": {"keys": ["AURORA-1"]}})
|
||||
self.assertEqual(out["status"], "failed")
|
||||
self.assertTrue(out["error"].startswith("хаб: HTTP 401"), out["error"])
|
||||
self.assertFalse(out["retryable"])
|
||||
|
||||
def test_jira_error_has_no_peer(self):
|
||||
self.fake.fail_next.append((500, {}))
|
||||
out = self.ex.run({"id": 1, "type": "refresh", "params": {"jql": "project = AURORA"}})
|
||||
self.assertTrue(out["retryable"])
|
||||
self.assertNotIn("peer", out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
import os
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
@ -59,7 +60,7 @@ class RunnerTest(unittest.TestCase):
|
||||
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"], 5)
|
||||
self.assertEqual(self.hub.claim_bodies[0]["agent_version"], agent.AGENT_VERSION)
|
||||
|
||||
def test_claim_without_updater_sends_nulls(self):
|
||||
self.runner.loop()
|
||||
@ -67,17 +68,17 @@ class RunnerTest(unittest.TestCase):
|
||||
self.assertIsNone(self.hub.claim_bodies[0]["launcher_version"])
|
||||
|
||||
def test_idle_with_release_exits_for_update(self):
|
||||
rel = {"version": 6, "sha256": "s", "content": "x"}
|
||||
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": 6, "sha256": "s"}
|
||||
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": 6, "sha256": "s", "content": "x"}
|
||||
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
|
||||
@ -88,7 +89,7 @@ class RunnerTest(unittest.TestCase):
|
||||
self.assertEqual(self.runner.exit_code, 0) # остановка человеком; staged доделает run.py при старте
|
||||
|
||||
def test_command_first_then_update(self):
|
||||
self.hub.release = {"version": 6, "sha256": "s", "content": "x"}
|
||||
self.hub.release = {"version": agent.AGENT_VERSION + 1, "sha256": "s", "content": "x"}
|
||||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||||
self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1)
|
||||
seen = []
|
||||
@ -224,6 +225,28 @@ class RunnerTest(unittest.TestCase):
|
||||
self.assertEqual(self.hub.results[2]["status"], "done")
|
||||
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("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||||
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"], "done")
|
||||
|
||||
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)
|
||||
@ -400,6 +423,7 @@ class RunnerTest(unittest.TestCase):
|
||||
r = self.hub.results[1]
|
||||
self.assertEqual(r["status"], "failed")
|
||||
self.assertIn("результат неизвестен", r["error"])
|
||||
self.assertIn("(обрыв связи с Jira)", r["error"])
|
||||
self.assertEqual(r["jira_calls"], 4)
|
||||
|
||||
def test_stop_during_retry_wait_leaves_command_for_recovery(self):
|
||||
@ -507,12 +531,55 @@ class RunnerTest(unittest.TestCase):
|
||||
self.assertIsNone(agent.acquire_lock(self.journal.path))
|
||||
cfg = jira_cfg(self.fake, 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), 1)
|
||||
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.jira.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("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, 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.fake.requests, []) # в Jira ни одного вызова (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("refresh", {"keys": ["AURORA-1"]})
|
||||
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()
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import signal
|
||||
import subprocess
|
||||
@ -24,7 +25,7 @@ def wait_for(pred, timeout=30):
|
||||
|
||||
|
||||
class SelfUpdateE2E(unittest.TestCase):
|
||||
"""run.py + настоящий agent.py: v5 видит публикацию v6, ставит её, v6 подтверждает; Ctrl-C группе — выход 0."""
|
||||
"""run.py + настоящий agent.py: v<N> видит публикацию v<N+1>, ставит её, v<N+1> подтверждает; Ctrl-C группе — выход 0."""
|
||||
|
||||
def setUp(self):
|
||||
self.jira = FakeJira()
|
||||
@ -39,9 +40,12 @@ class SelfUpdateE2E(unittest.TestCase):
|
||||
"auth": {"type": "basic", "user": "me", "password": "secret"}}}, f)
|
||||
with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f:
|
||||
src = f.read()
|
||||
content = src.replace("AGENT_VERSION = 5\n", "AGENT_VERSION = 6\n", 1)
|
||||
self.cur = int(re.search(r"^AGENT_VERSION = (\d+)", src, re.M).group(1))
|
||||
cur = "AGENT_VERSION = %d\n" % self.cur
|
||||
content = src.replace(cur, "AGENT_VERSION = %d\n" % (self.cur + 1), 1)
|
||||
self.assertNotEqual(content, src)
|
||||
self.hub.release = {"version": 6, "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), "content": content}
|
||||
self.hub.release = {"version": self.cur + 1, "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(),
|
||||
"content": content}
|
||||
self.content = content
|
||||
self.log = open(os.path.join(d, "out.log"), "wb")
|
||||
self.proc = subprocess.Popen([sys.executable, os.path.join(d, "run.py")], stdout=self.log,
|
||||
@ -63,7 +67,7 @@ class SelfUpdateE2E(unittest.TestCase):
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
|
||||
def test_update_to_v6_and_stop_by_ctrl_c(self):
|
||||
def test_update_to_next_and_stop_by_ctrl_c(self):
|
||||
ok = wait_for(lambda: (self.state() or {}).get("state") == "ok", 60)
|
||||
with open(os.path.join(self.tmp.name, "out.log"), encoding="utf-8", errors="replace") as f:
|
||||
out = f.read()
|
||||
@ -71,8 +75,8 @@ class SelfUpdateE2E(unittest.TestCase):
|
||||
with open(os.path.join(self.tmp.name, "agent.py"), encoding="utf-8") as f:
|
||||
self.assertEqual(f.read(), self.content)
|
||||
self.assertTrue(os.path.exists(os.path.join(self.tmp.name, "agent.py.prev")))
|
||||
self.assertTrue(wait_for(lambda: any(b.get("agent_version") == 6 for b in self.hub.claim_bodies)))
|
||||
last = [b for b in self.hub.claim_bodies if b.get("agent_version") == 6][-1]
|
||||
self.assertEqual(last["launcher_version"], 1)
|
||||
self.assertTrue(wait_for(lambda: any(b.get("agent_version") == self.cur + 1 for b in self.hub.claim_bodies)))
|
||||
last = [b for b in self.hub.claim_bodies if b.get("agent_version") == self.cur + 1][-1]
|
||||
self.assertEqual(last["launcher_version"], 2)
|
||||
os.killpg(self.proc.pid, signal.SIGINT) # Ctrl-C в терминале — всей группе
|
||||
self.assertEqual(self.proc.wait(30), 0)
|
||||
|
||||
@ -7,6 +7,9 @@ import unittest
|
||||
from _common import SETTINGS, agent
|
||||
from fake_hub import FakeHub
|
||||
|
||||
CUR = agent.AGENT_VERSION
|
||||
NEXT = CUR + 1
|
||||
|
||||
|
||||
class UpdateStateTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
@ -39,6 +42,25 @@ class UpdateStateTest(unittest.TestCase):
|
||||
st = agent.read_update_state(u.state_path)
|
||||
self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("ok", 4, agent.AGENT_VERSION, "s"))
|
||||
|
||||
def test_confirm_write_error_only_logged(self):
|
||||
"""M-3 финального ревью: диск полон в момент подтверждения — агент не падает, подтвердит на следующем claim."""
|
||||
u = agent.Updater(None, self.dir.name, 1)
|
||||
agent.write_update_state(u.state_path, "trial", 4, agent.AGENT_VERSION, "s")
|
||||
logs = []
|
||||
|
||||
def full(*a, **kw):
|
||||
raise OSError(28, "No space left on device")
|
||||
orig_log, orig_write = agent.log, agent.write_update_state
|
||||
agent.log, agent.write_update_state = logs.append, full
|
||||
try:
|
||||
u.confirm()
|
||||
finally:
|
||||
agent.log, agent.write_update_state = orig_log, orig_write
|
||||
self.assertEqual(u.state()["state"], "trial")
|
||||
self.assertTrue(any(l.startswith("не удалось записать update.json") for l in logs), logs)
|
||||
u.confirm()
|
||||
self.assertEqual(u.state()["state"], "ok")
|
||||
|
||||
def test_confirm_ignores_other_states(self):
|
||||
u = agent.Updater(None, self.dir.name, 1)
|
||||
u.confirm()
|
||||
@ -75,51 +97,121 @@ class UpdaterCheckTest(unittest.TestCase):
|
||||
def ref(self, rel):
|
||||
return {"version": rel["version"], "sha256": rel["sha256"]}
|
||||
|
||||
def capture_logs(self):
|
||||
logs = []
|
||||
orig = agent.log
|
||||
agent.log = logs.append
|
||||
self.addCleanup(setattr, agent, "log", orig)
|
||||
return logs
|
||||
|
||||
def test_incomplete_release_ignored_with_hourly_hint(self):
|
||||
logs = self.capture_logs()
|
||||
for ref in ({"version": NEXT, "sha256": None}, {"version": str(NEXT), "sha256": "a" * 64},
|
||||
{"version": True, "sha256": "a" * 64}, {"version": NEXT, "sha256": "A" * 64},
|
||||
{"version": NEXT, "sha256": "a" * 63}, {"version": NEXT, "sha256": "a" * 64 + "\n"}):
|
||||
self.assertFalse(self.u.check(ref))
|
||||
self.now[0] += 3600
|
||||
self.assertFalse(self.u.check({"version": NEXT, "sha256": None}))
|
||||
self.assertEqual(sum(1 for l in logs if l.startswith("хаб прислал неполную публикацию")), 2)
|
||||
self.assertEqual(self.hub.release_requests, 0)
|
||||
self.assertIsNone(self.u.state())
|
||||
|
||||
def test_incomplete_release_hint_is_bounded(self):
|
||||
"""M-4: подсказка не печатает произвольно длинный мусор от хаба целиком."""
|
||||
logs = self.capture_logs()
|
||||
self.assertFalse(self.u.check({"version": NEXT, "sha256": "x" * 10000}))
|
||||
hints = [l for l in logs if l.startswith("хаб прислал неполную публикацию")]
|
||||
self.assertEqual(len(hints), 1)
|
||||
self.assertLess(len(hints[0]), 300)
|
||||
|
||||
def test_hints_have_separate_timers(self):
|
||||
logs = self.capture_logs()
|
||||
self.u.launcher_version = None
|
||||
self.hub.release = make_release(NEXT)
|
||||
self.assertFalse(self.u.check({"version": NEXT, "sha256": None}))
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertEqual(sum(1 for l in logs if l.startswith("хаб прислал неполную публикацию")), 1)
|
||||
self.assertEqual(logs.count("доступна v%d; для самообновления запускайте через run.py" % NEXT), 1)
|
||||
|
||||
def test_write_error_is_transient(self):
|
||||
self.hub.release = make_release(NEXT)
|
||||
ref = self.ref(self.hub.release)
|
||||
self.u.new_path = os.path.join(self.dir.name, "нет-каталога", "agent.py.new") # open() → FileNotFoundError
|
||||
self.assertFalse(self.u.check(ref))
|
||||
self.assertIsNone(self.u.state())
|
||||
self.assertEqual(self.selftests, [])
|
||||
self.now[0] += 299
|
||||
self.assertFalse(self.u.check(ref))
|
||||
self.assertEqual(self.hub.release_requests, 1) # раньше 5 минут не повторяет
|
||||
|
||||
def test_selftest_os_error_is_transient(self):
|
||||
self.hub.release = make_release(NEXT)
|
||||
def boom(path):
|
||||
raise OSError(28, "No space left on device")
|
||||
self.u.run_selftest = boom
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertIsNone(self.u.state())
|
||||
self.assertFalse(os.path.exists(self.u.new_path))
|
||||
|
||||
def test_staged_write_error_is_transient(self):
|
||||
self.hub.release = make_release(NEXT)
|
||||
self.u.state_path = os.path.join(self.dir.name, "нет-каталога", "update.json")
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release))) # без staged выходить с 75 нельзя
|
||||
self.assertFalse(os.path.exists(self.u.new_path))
|
||||
|
||||
def test_reject_write_error_only_logged(self):
|
||||
rel = make_release(NEXT)
|
||||
self.hub.release = dict(rel, content=rel["content"] + "# подмена\n")
|
||||
self.u.state_path = os.path.join(self.dir.name, "нет-каталога", "update.json")
|
||||
logs = self.capture_logs()
|
||||
self.assertFalse(self.u.check(self.ref(rel))) # не бросает
|
||||
self.assertTrue(any(l.startswith("не удалось записать update.json") for l in logs), logs)
|
||||
|
||||
def test_good_version_is_staged(self):
|
||||
rel = make_release(6)
|
||||
rel = make_release(NEXT)
|
||||
self.hub.release = rel
|
||||
self.assertTrue(self.u.check(self.ref(rel)))
|
||||
with open(self.u.new_path, encoding="utf-8") as f:
|
||||
self.assertEqual(f.read(), rel["content"])
|
||||
st = self.u.state()
|
||||
self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("staged", 5, 6, rel["sha256"]))
|
||||
self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("staged", CUR, NEXT, rel["sha256"]))
|
||||
self.assertEqual(self.selftests, [self.u.new_path])
|
||||
|
||||
def test_not_newer_or_disabled_or_no_release(self):
|
||||
self.hub.release = make_release(5)
|
||||
self.hub.release = make_release(CUR)
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertFalse(self.u.check(None))
|
||||
self.hub.release = make_release(6)
|
||||
self.hub.release = make_release(NEXT)
|
||||
self.u.enabled = False
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertEqual(self.hub.release_requests, 0)
|
||||
|
||||
def test_sha_mismatch_rejected_once(self):
|
||||
rel = make_release(6)
|
||||
rel = make_release(NEXT)
|
||||
self.hub.release = dict(rel, content=rel["content"] + "# подмена\n")
|
||||
ref = self.ref(rel)
|
||||
self.hub.release["sha256"] = rel["sha256"]
|
||||
self.assertFalse(self.u.check(ref))
|
||||
st = self.u.state()
|
||||
self.assertEqual((st["state"], st["to_version"], st["error"]), ("rejected", 6, "sha256 не сходится"))
|
||||
self.assertEqual((st["state"], st["to_version"], st["error"]), ("rejected", NEXT, "sha256 не сходится"))
|
||||
self.assertFalse(os.path.exists(self.u.new_path))
|
||||
self.assertFalse(self.u.check(ref))
|
||||
self.assertEqual(self.hub.release_requests, 1) # отвергнутую версию больше не качает
|
||||
|
||||
def test_syntax_error_and_wrong_version_rejected(self):
|
||||
bad = make_release(6, body="def (:\n")
|
||||
bad = make_release(NEXT, body="def (:\n")
|
||||
self.hub.release = bad
|
||||
self.assertFalse(self.u.check(self.ref(bad)))
|
||||
self.assertIn("не компилируется", self.u.state()["error"])
|
||||
wrong = make_release(7)
|
||||
wrong = dict(wrong, version=8)
|
||||
wrong = make_release(NEXT + 1)
|
||||
wrong = dict(wrong, version=NEXT + 2)
|
||||
self.hub.release = wrong
|
||||
self.assertFalse(self.u.check(self.ref(wrong)))
|
||||
self.assertIn("AGENT_VERSION", self.u.state()["error"])
|
||||
self.assertEqual(self.selftests, [])
|
||||
|
||||
def test_selftest_bad_code_rejected(self):
|
||||
self.hub.release = make_release(6)
|
||||
self.hub.release = make_release(NEXT)
|
||||
self.code = 1
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
st = self.u.state()
|
||||
@ -129,7 +221,7 @@ class UpdaterCheckTest(unittest.TestCase):
|
||||
self.assertFalse(os.path.exists(self.u.new_path))
|
||||
|
||||
def test_selftest_unavailable_is_transient(self):
|
||||
self.hub.release = make_release(6)
|
||||
self.hub.release = make_release(NEXT)
|
||||
self.code = 3
|
||||
ref = self.ref(self.hub.release)
|
||||
self.assertFalse(self.u.check(ref))
|
||||
@ -143,35 +235,35 @@ class UpdaterCheckTest(unittest.TestCase):
|
||||
self.assertTrue(self.u.check(ref))
|
||||
|
||||
def test_selftest_timeout_is_transient(self):
|
||||
self.hub.release = make_release(6)
|
||||
self.hub.release = make_release(NEXT)
|
||||
self.code = None
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertIsNone(self.u.state())
|
||||
|
||||
def test_download_failure_is_transient(self):
|
||||
rel = make_release(6) # в claim есть, а /agent/release — 404
|
||||
rel = make_release(NEXT) # в claim есть, а /agent/release — 404
|
||||
self.assertFalse(self.u.check(self.ref(rel)))
|
||||
self.assertIsNone(self.u.state())
|
||||
|
||||
def test_release_changed_between_requests_is_transient(self):
|
||||
old, new = make_release(6), make_release(7)
|
||||
old, new = make_release(NEXT), make_release(NEXT + 1)
|
||||
self.hub.release = new
|
||||
self.assertFalse(self.u.check(self.ref(old)))
|
||||
self.assertIsNone(self.u.state())
|
||||
self.assertEqual(self.selftests, [])
|
||||
|
||||
def test_rolled_back_same_sha_not_retried(self):
|
||||
rel = make_release(6)
|
||||
rel = make_release(NEXT)
|
||||
self.hub.release = rel
|
||||
agent.write_update_state(self.u.state_path, "rolled_back", 5, 6, rel["sha256"], "код 1 через 3 с")
|
||||
agent.write_update_state(self.u.state_path, "rolled_back", CUR, NEXT, rel["sha256"], "код 1 через 3 с")
|
||||
self.assertFalse(self.u.check(self.ref(rel)))
|
||||
fixed = make_release(6, body="print('fixed')\n")
|
||||
fixed = make_release(NEXT, body="print('fixed')\n")
|
||||
self.hub.release = fixed
|
||||
self.assertTrue(self.u.check(self.ref(fixed))) # тот же номер, другой sha — пробуем
|
||||
|
||||
def test_without_launcher_only_hint(self):
|
||||
self.u.launcher_version = None
|
||||
self.hub.release = make_release(6)
|
||||
self.hub.release = make_release(NEXT)
|
||||
logs = []
|
||||
orig = agent.log
|
||||
agent.log = logs.append
|
||||
@ -182,11 +274,11 @@ class UpdaterCheckTest(unittest.TestCase):
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
finally:
|
||||
agent.log = orig
|
||||
self.assertEqual(logs.count("доступна v6; для самообновления запускайте через run.py"), 2)
|
||||
self.assertEqual(logs.count("доступна v%d; для самообновления запускайте через run.py" % NEXT), 2)
|
||||
self.assertEqual(self.hub.release_requests, 0)
|
||||
|
||||
def test_stop_during_selftest_is_not_reject(self):
|
||||
self.hub.release = make_release(6)
|
||||
self.hub.release = make_release(NEXT)
|
||||
stop = [False]
|
||||
self.u.should_stop = lambda: stop[0]
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user