From 3f4c9183b84a060836f3ca126ec90ce054f58386 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sat, 24 Oct 2026 00:59:49 +0400 Subject: [PATCH] =?UTF-8?q?=D0=9E=D1=87=D0=B5=D1=80=D0=B5=D0=B4=D1=8C,=20?= =?UTF-8?q?=D1=80=D0=B5=D0=BB=D0=B8=D0=B7=20=D0=B8=20=D1=81=D1=82=D0=B0?= =?UTF-8?q?=D1=82=D1=83=D1=81=20=D0=BF=D0=BE=20=D0=B0=D0=B3=D0=B5=D0=BD?= =?UTF-8?q?=D1=82=D1=83=20=D0=BC=D0=BE=D0=B4=D1=83=D0=BB=D1=8F=20(#91)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- src/jira_hub/core/api.py | 60 ++++-- src/jira_hub/core/cli.py | 7 +- src/jira_hub/core/paths.py | 3 +- src/jira_hub/core/queue.py | 64 +++--- src/jira_hub/core/release.py | 25 +++ src/jira_hub/modules/jira/fetch.py | 2 +- src/jira_hub/modules/jira/refresh.py | 4 +- tests/fixtures/baseline/routes.expected.json | 6 + tests/test_agents.py | 121 +++++++++++- tests/test_api.py | 10 +- tests/test_cli.py | 6 +- tests/test_db.py | 2 +- tests/test_queue.py | 196 +++++++++---------- tests/test_queue_core.py | 2 +- 14 files changed, 348 insertions(+), 160 deletions(-) diff --git a/src/jira_hub/core/api.py b/src/jira_hub/core/api.py index 0be83fa..350b3be 100644 --- a/src/jira_hub/core/api.py +++ b/src/jira_hub/core/api.py @@ -13,9 +13,8 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, unquote, urlparse -from . import backup, db, files, projects, queue, registry, relay, release, settings +from . import backup, db, files, http, projects, queue, registry, relay, release, settings from .errors import CommandError, HttpMapped -from . import http from .http import ApiError, Raw, int_group, int_param, mapped, not_found from .paths import backups_dir as default_backups_dir from .paths import secure_dir @@ -35,20 +34,21 @@ SWEEP_EVERY = 24 * 3600 # уборка files/ не чаще раза в class Hub: - def __init__(self, db_path, backups_dir=None, release_dir=None, web_dir=None, files_dir=None): + def __init__(self, db_path, backups_dir=None, release_root=None, web_dir=None, files_dir=None): self.db_path = Path(db_path) self.web_dir = Path(web_dir) if web_dir else WEB_APP self.backups_dir = Path(backups_dir) if backups_dir else default_backups_dir() - self.release_dir = Path(release_dir) if release_dir else self.db_path.parent / "release" + self.release_root = Path(release_root) if release_root else self.db_path.parent / "release" self.files_dir = Path(files_dir) if files_dir else self.db_path.parent / "files" self.last_sweep = None # время прошлой уборки files/, в памяти (решение 15) self.module_state: dict[str, dict] = {} # память модулей между тиками (имя модуля → словарь) - # последнее из claim; в памяти (R-3). reported — был ли claim с запуска хаба: до него launcher_version - # неизвестен, а не null (hub-hygiene-d, решение 7) - self.agent_update = {"update_state": None, "launcher_version": None, "reported": False} + # последнее из claim по каждому агенту; в памяти (R-3). reported — был ли claim с запуска хаба: до него + # launcher_version неизвестен, а не null (hub-hygiene-d, решение 7) + self.agent_update: dict[str, dict] = {m.name: self._no_update() for m in registry.agent_modules()} self.claim_wait = 30 secure_dir(self.backups_dir) # каталог копий закрыт при каждом старте, а не только после первой копии secure_dir(self.files_dir) # files/ — 0700 при каждом старте, как backups/ + release.migrate_layout(self.release_root) # release/agent.py → release/jira/ (раскладка по модулям) conn = self.conn() settings.ensure_defaults(conn) conn.close() @@ -89,10 +89,14 @@ class Hub: log.info("уборка файлов: удалено %d", len(removed)) return removed - def note_agent_update(self, body): - """update_state и launcher_version из claim: хранятся в памяти, агент шлёт их в каждом claim (R-3). + @staticmethod + def _no_update() -> dict: + return {"update_state": None, "launcher_version": None, "reported": False} + + def note_agent_update(self, agent, body): + """update_state и launcher_version из claim агента: хранятся в памяти, агент шлёт их в каждом claim (R-3). Значение неверного типа игнорируется — claim из-за него не отвергается; сам claim отмечается (reported).""" - new = dict(self.agent_update) + new = dict(self.agent_update.get(agent) or self._no_update()) new["reported"] = True us = body.get("update_state") if us is None or isinstance(us, dict): @@ -102,7 +106,7 @@ class Hub: lv = body.get("launcher_version") if lv is None or (isinstance(lv, int) and not isinstance(lv, bool)): new["launcher_version"] = lv - self.agent_update = new + self.agent_update[agent] = new def make_backup(self, label): conn = self.conn() @@ -115,17 +119,18 @@ class Hub: # --- обработчики: fn(hub, conn, match, query, body) -> (status, payload) ---------- def h_claim(hub, conn, m, q, body): - hub.note_agent_update(body) + agent = http.agent_of(q) + hub.note_agent_update(agent, body) deadline = time.monotonic() + hub.claim_wait while True: - cmd = queue.claim(conn, body.get("agent_version"), body.get("settings_version"), + cmd = queue.claim(conn, agent, body.get("agent_version"), body.get("settings_version"), body.get("last_call_at", body.get("last_jira_call_at")), - backup=hub.make_backup, files_dir=hub.files_dir, agent=http.agent_of(q)) + backup=hub.make_backup, files_dir=hub.files_dir) if cmd or time.monotonic() >= deadline: break time.sleep(1) try: # команда уже выдана: публикация не должна превращать ответ в 500 (I-1) - rel = release.current(hub.release_dir) + rel = release.current(release.module_dir(hub.release_root, agent)) except Exception: # noqa: BLE001 log.exception("release: чтение публикации при claim") rel = None @@ -134,7 +139,7 @@ def h_claim(hub, conn, m, q, body): def h_agent_release(hub, conn, m, q, body): - rel = release.load(hub.release_dir) + rel = release.load(release.module_dir(hub.release_root, http.agent_of(q))) if rel is None: raise ApiError(404, "no_release", "версия агента не опубликована") return 200, rel @@ -288,14 +293,26 @@ def h_command_cancel(hub, conn, m, q, body): raise ApiError(409, "claimed", e.claimed_at) -def h_agent_get(hub, conn, m, q, body): - view = queue.agent_view(conn) - rel = release.current(hub.release_dir) - au = hub.agent_update +def _agent_view(hub, conn, name) -> dict: + view = queue.agent_view(conn, name) + rel = release.current(release.module_dir(hub.release_root, name)) + au = hub.agent_update.get(name) or hub._no_update() us, lv, reported = au["update_state"], au["launcher_version"], au["reported"] view.update(release=rel, update_state=us, launcher_version=lv, agent_reported=reported, update_hint=release.hint(rel, view, us, lv, reported=reported)) - return 200, view + return view + + +def h_agent_get(hub, conn, m, q, body): + """Статус агента модуля: ?module=<имя>, по умолчанию jira; неизвестный или без агента — 404.""" + name = q.get("module") or "jira" + if name not in [a.name for a in registry.agent_modules()]: + raise ApiError(404, "not_found", f"module {name}") + return 200, _agent_view(hub, conn, name) + + +def h_agents_get(hub, conn, m, q, body): + return 200, {"agents": [_agent_view(hub, conn, a.name) for a in registry.agent_modules()]} def _settings_view(conn) -> dict: @@ -378,6 +395,7 @@ CORE_ROUTES = [ ("POST", r"^/commands/(\d+)/retry$", h_command_retry, "client"), ("POST", r"^/commands/(\d+)/cancel$", h_command_cancel, "client"), ("GET", r"^/agent$", h_agent_get, "client"), + ("GET", r"^/agents$", h_agents_get, "client"), ("GET", r"^/settings$", h_settings_get, "client"), ("PUT", r"^/settings$", h_settings_put, "client"), ("GET", r"^/projects$", h_projects_get, "client"), diff --git a/src/jira_hub/core/cli.py b/src/jira_hub/core/cli.py index 734ea51..f32bc76 100644 --- a/src/jira_hub/core/cli.py +++ b/src/jira_hub/core/cli.py @@ -26,13 +26,18 @@ def main(argv=None) -> int: p = sub.add_parser("agent", help="версия агента для самообновления") agent_sub = p.add_subparsers(dest="agent_cmd", required=True) p = agent_sub.add_parser("publish", help="опубликовать agent.py (версия только вверх)") + p.add_argument("--module", default="jira", help="модуль агента (по умолчанию jira)") p.add_argument("path") args = ap.parse_args(argv) if args.cmd == "agent": from . import release + names = [m.name for m in registry.agent_modules()] + if args.module not in names: + print(f"неизвестный модуль {args.module}; модули с агентами: {', '.join(names)}", file=sys.stderr) + return 2 try: - m = release.publish(paths.release_dir(), args.path) + m = release.publish(release.module_dir(paths.release_root(), args.module), args.path) except release.PublishError as e: print(e, file=sys.stderr) return 1 diff --git a/src/jira_hub/core/paths.py b/src/jira_hub/core/paths.py index b68ff1d..ac4250b 100644 --- a/src/jira_hub/core/paths.py +++ b/src/jira_hub/core/paths.py @@ -18,7 +18,8 @@ def backups_dir() -> Path: return home() / "backups" -def release_dir() -> Path: +def release_root() -> Path: + """Корень публикаций агентов: release/<модуль>/agent.py + manifest.json.""" return home() / "release" diff --git a/src/jira_hub/core/queue.py b/src/jira_hub/core/queue.py index 69aed09..a8adea6 100644 --- a/src/jira_hub/core/queue.py +++ b/src/jira_hub/core/queue.py @@ -110,19 +110,28 @@ def insert_batch(conn, title, author, cmds, project=None) -> dict: return {"id": batch_id, "estimate_seconds": est, "count": len(cmds)} -def has_active(conn) -> bool: - """В очереди есть команда queued или claimed.""" - return conn.execute("SELECT 1 FROM command WHERE status IN ('queued', 'claimed') LIMIT 1").fetchone() is not None +def _own_types(agent: str) -> list[str]: + """Типы команд модуля агента (имя агента = имя модуля); у модуля без команд — пусто.""" + m = registry.module(agent) + return sorted(m.commands.types) if m and m.commands else [] -def _agent(agent: str | None) -> str: - """Имя агента; без имени — первый агент реестра (временно, пока claim не получает агента из запроса).""" - return agent or registry.agent_modules()[0].name +def _in(types: list[str]) -> str: + return ",".join("?" * len(types)) -def agent_version(conn, agent: str | None = None): +def has_active(conn, module: str | None = None) -> bool: + """В очереди есть команда queued или claimed; module — только типы этого модуля, None — любого.""" + if module is None: + return conn.execute("SELECT 1 FROM command WHERE status IN ('queued', 'claimed') LIMIT 1").fetchone() is not None + types = _own_types(module) + return conn.execute(f"SELECT 1 FROM command WHERE status IN ('queued', 'claimed') AND type IN ({_in(types)}) LIMIT 1", + types).fetchone() is not None + + +def agent_version(conn, agent: str): """Версия агента из его последней выдачи; None — агент не отчитывался.""" - row = conn.execute("SELECT agent_version FROM agent_status WHERE agent = ?", (_agent(agent),)).fetchone() + row = conn.execute("SELECT agent_version FROM agent_status WHERE agent = ?", (agent,)).fetchone() return row["agent_version"] if row else None @@ -181,9 +190,11 @@ def _skip_reason(conn, batch_id, definer, unresolved) -> str: return f"создание задачи {cancelled[0]} отменено" if cancelled else f"задача {unresolved[0]} не создана" -def _next(conn) -> dict | None: +def _next(conn, agent: str) -> dict | None: + types = _own_types(agent) rows = conn.execute("SELECT c.* FROM command c JOIN batch b ON b.id = c.batch_id " - "WHERE c.status = 'queued' AND b.cancelled_at IS NULL ORDER BY b.created_at, b.id, c.seq").fetchall() + f"WHERE c.status = 'queued' AND b.cancelled_at IS NULL AND c.type IN ({_in(types)}) " + "ORDER BY b.created_at, b.id, c.seq", types).fetchall() for row in rows: cmd = _cmd(row) h = _hooks(cmd["type"]) @@ -203,11 +214,14 @@ def _next(conn) -> dict | None: return None -def _redeliver(conn, now) -> dict | None: +def _redeliver(conn, now, agent: str) -> dict | None: """Самая старая взятая команда без отчёта — выдать снова. Агент приходит за новой командой, только когда своей - незавершённой нет, значит выданную он не получил или потерял. После MAX_DELIVERIES выдач — failed.""" + незавершённой нет, значит выданную он не получил или потерял. После MAX_DELIVERIES выдач — failed. Берутся только + команды типов модуля агента: взятая другим агентом не довыдаётся.""" + types = _own_types(agent) while True: - row = conn.execute("SELECT * FROM command WHERE status = 'claimed' ORDER BY claimed_at, id LIMIT 1").fetchone() + row = conn.execute(f"SELECT * FROM command WHERE status = 'claimed' AND type IN ({_in(types)}) " + "ORDER BY claimed_at, id LIMIT 1", types).fetchone() if row is None: return None cmd = _cmd(row) @@ -229,16 +243,15 @@ def deliver_params(conn, cmd, files_dir) -> dict: return h.deliver_params(conn, cmd, files_dir) if h else cmd["params"] -def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=None, backup=None, files_dir=None, - agent: str | None = None) -> dict | None: +def claim(conn, agent: str, agent_version=None, settings_version=None, last_call_at=None, backup=None, + files_dir=None) -> dict | None: now = db.now() - agent = _agent(agent) to_backup = None with db.transaction(conn): conn.execute("UPDATE agent_status SET last_seen_at = ?, agent_version = COALESCE(?, agent_version), " "settings_version = COALESCE(?, settings_version), last_call_at = COALESCE(?, last_call_at) WHERE agent = ?", - (now, None if agent_version is None else str(agent_version), settings_version, last_jira_call_at, agent)) - cmd = _redeliver(conn, now) + (now, None if agent_version is None else str(agent_version), settings_version, last_call_at, agent)) + cmd = _redeliver(conn, now, agent) if cmd is not None: pause = _pause(conn, cmd["type"]) conn.execute("UPDATE command SET claimed_at = ?, pause_seconds = ?, deliveries = deliveries + 1 WHERE id = ?", @@ -249,7 +262,7 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non conn.execute("UPDATE agent_status SET current_command_id = ? WHERE agent = ?", (cmd["id"], agent)) log.info("команда %s выдана снова (выдача %s)", cmd["id"], cmd["deliveries"] + 1) else: - cmd = _next(conn) + cmd = _next(conn, agent) if cmd is None: conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE agent = ?", (agent,)) return None @@ -377,16 +390,19 @@ def retry_command(conn, command_id, author, files_dir=None, project=None) -> dic # --- агент и автообновление ---------------------------------------------------- -def agent_view(conn, agent: str | None = None) -> dict: - agent = _agent(agent) +def agent_view(conn, agent: str) -> dict: d = dict(conn.execute("SELECT * FROM agent_status WHERE agent = ?", (agent,)).fetchone()) + d["module"] = agent + types = _own_types(agent) now = db.utcnow() d["online"] = bool(d["last_seen_at"]) and now - db.parse(d["last_seen_at"]) < timedelta(minutes=ONLINE_MINUTES) m = registry.module(agent) min_version = int(settings.get(conn, m.agent_min_version_key) or 0) if m and m.agent_min_version_key else 0 d["warning"] = "обновите агента" if d["agent_version"] and int(d["agent_version"]) < min_version else None - d["rate_limited_24h"] = conn.execute("SELECT COALESCE(SUM(rate_limited), 0) AS n FROM command WHERE finished_at >= ?", - ((now - timedelta(days=1)).isoformat(),)).fetchone()["n"] + d["rate_limited_24h"] = conn.execute( + f"SELECT COALESCE(SUM(rate_limited), 0) AS n FROM command WHERE finished_at >= ? AND type IN ({_in(types)})", + ((now - timedelta(days=1)).isoformat(), *types)).fetchone()["n"] d["redelivered"] = [dict(r) for r in conn.execute( - "SELECT id, type, claimed_at, deliveries FROM command WHERE status = 'claimed' AND deliveries > 1 ORDER BY id")] + "SELECT id, type, claimed_at, deliveries FROM command WHERE status = 'claimed' AND deliveries > 1 " + f"AND type IN ({_in(types)}) ORDER BY id", types)] return d diff --git a/src/jira_hub/core/release.py b/src/jira_hub/core/release.py index e27b3b2..43a35c6 100644 --- a/src/jira_hub/core/release.py +++ b/src/jira_hub/core/release.py @@ -26,6 +26,31 @@ class PublishError(Exception): pass +def module_dir(root, module) -> Path: + """Каталог публикации агента модуля: /<модуль>. Нет каталога — создаётся с правами 0700; существующий не + трогаем (права владельца не переписываем на каждом claim).""" + d = Path(root) / module + if not d.is_dir(): + secure_dir(d) + return d + + +def migrate_layout(root) -> None: + """Раскладка до модулей: agent.py и manifest.json лежали прямо в root. Если root/jira/ нет — переносим их в + root/jira/ (манифест последним: по нему версия считается опубликованной). Сбой — предупреждение в лог, хаб стартует.""" + root = Path(root) + try: + old = [root / n for n in ("agent.py", "manifest.json") if (root / n).exists()] + if not old or (root / "jira").exists(): + return + dest = module_dir(root, "jira") + for f in old: + os.replace(f, dest / f.name) + log.info("release: публикация агента перенесена в %s", dest) + except OSError as e: + log.warning("release: раскладка %s не перенесена в jira/: %s", root, e) + + def parse_version(text: str) -> int: found = VERSION_RE.findall(text) if len(found) != 1: diff --git a/src/jira_hub/modules/jira/fetch.py b/src/jira_hub/modules/jira/fetch.py index 1050eea..79a4a1a 100644 --- a/src/jira_hub/modules/jira/fetch.py +++ b/src/jira_hub/modules/jira/fetch.py @@ -30,7 +30,7 @@ def request_fetch(conn, files_dir, attachment_id: str, author: str | None) -> di raise KeyError(attachment_id) if info["state"] in ("ready", "fetching"): return info - v = queue.agent_version(conn) + v = queue.agent_version(conn, "jira") if v is not None and str(v).isdigit() and int(v) < FETCH_AGENT_VERSION: # версия неизвестна — не блокируем raise AgentTooOld(f"агент v{v} не умеет скачивать вложения — нужен v{FETCH_AGENT_VERSION}") cmds = commands.validate([{"type": "fetch_attachment", "params": {"key": info["key"], "attachment_id": attachment_id}}]) diff --git a/src/jira_hub/modules/jira/refresh.py b/src/jira_hub/modules/jira/refresh.py index 7989a97..a48db99 100644 --- a/src/jira_hub/modules/jira/refresh.py +++ b/src/jira_hub/modules/jira/refresh.py @@ -24,9 +24,9 @@ def maybe_auto_refresh(conn) -> dict | None: if jql is None: return None minutes = int(settings.get(conn, "auto_refresh_minutes") or 0) - if minutes <= 0 or not queue.agent_view(conn)["online"]: + if minutes <= 0 or not queue.agent_view(conn, "jira")["online"]: return None - if queue.has_active(conn): + if queue.has_active(conn, "jira"): return None last = conn.execute("SELECT MAX(started_at) AS t FROM sync_run").fetchone()["t"] if last and db.utcnow() - db.parse(last) < timedelta(minutes=minutes): diff --git a/tests/fixtures/baseline/routes.expected.json b/tests/fixtures/baseline/routes.expected.json index 9f58022..670f122 100644 --- a/tests/fixtures/baseline/routes.expected.json +++ b/tests/fixtures/baseline/routes.expected.json @@ -71,6 +71,12 @@ "agent", "h_agent_state" ], + [ + "GET", + "^/agents$", + "client", + "h_agents_get" + ], [ "GET", "^/app(?:/(.*))?$", diff --git a/tests/test_agents.py b/tests/test_agents.py index 916c475..9024498 100644 --- a/tests/test_agents.py +++ b/tests/test_agents.py @@ -1,15 +1,25 @@ """Агенты модулей: токен определяет модуль, миграция v7 сохраняет агента jira (wiki-module, решения 1–3, 6, 15).""" import pytest -from jira_hub.core import cli, db, queue, settings +from jira_hub.core import api, cli, db, queue, release, settings from jira_hub.core.client import HubError from v6db import make_v6_db def test_v6_upgrade_keeps_jira_agent(tmp_path): make_v6_db(tmp_path / "hub.sqlite") + rel = tmp_path / "release" + rel.mkdir() + (rel / "agent.py").write_text("AGENT_VERSION = 14\n", encoding="utf-8") + release.publish(rel, rel / "agent.py") + hub = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups") + hub.claim_wait = 0 conn = db.connect(tmp_path / "hub.sqlite") try: + tok = settings.tokens(conn)["agents"]["jira"] + assert tok == "T-agent" + code, body = api.h_claim(hub, conn, None, {"_agent": "jira"}, {"agent_version": 14}) + assert code == 200 and body["agent_release"]["version"] == 14 # релиз v14 перенесён в release/jira/ assert db._schema_version(conn) == 7 tok = settings.tokens(conn) assert tok["client"] == "T-client" and tok["agents"]["jira"] == "T-agent" and "agent" not in tok @@ -91,3 +101,112 @@ def test_agent_view_min_version_from_module_key(conn_two_modules): settings.set_(conn_two_modules, "min_agent_version", 5) assert queue.agent_view(conn_two_modules, "jira")["warning"] == "обновите агента" assert queue.agent_view(conn_two_modules, "stub")["warning"] is None + + +# --- очередь, релиз и статус по агенту (Task 2) ------------------------------------------------------------------ + +def _two_batches(conn): + queue.create_batch(conn, "w", "t", [{"type": "stub_cmd", "params": {}}]) + queue.create_batch(conn, "j", "t", [{"type": "ensure_issue", + "params": {"temp_key": "new:1", "issue_type": "task", "summary": "Тема"}}]) + + +def test_wiki_queue_does_not_block_jira(conn_two_modules): + conn = conn_two_modules + _two_batches(conn) + s = queue.claim(conn, "stub") + assert s["type"] == "stub_cmd" + j = queue.claim(conn, "jira") # stub_cmd ещё claimed, но это очередь другого агента + assert j["type"] == "ensure_issue" + assert queue.agent_view(conn, "stub")["current_command_id"] == s["id"] + assert queue.agent_view(conn, "jira")["current_command_id"] == j["id"] + + +def test_redeliver_only_own(conn_two_modules): + conn = conn_two_modules + _two_batches(conn) + s = queue.claim(conn, "stub") + j = queue.claim(conn, "jira") + assert queue.claim(conn, "jira")["id"] == j["id"] # довыдача — своя команда, не stub_cmd + assert queue.claim(conn, "stub")["id"] == s["id"] + assert queue.agent_view(conn, "jira")["redelivered"][0]["id"] == j["id"] + assert [r["id"] for r in queue.agent_view(conn, "stub")["redelivered"]] == [s["id"]] + + +def test_agent_version_per_agent(conn_two_modules): + conn = conn_two_modules + queue.claim(conn, "jira", agent_version=14) + assert queue.agent_version(conn, "jira") == "14" and queue.agent_version(conn, "stub") is None + queue.claim(conn, "stub", agent_version=1) + assert queue.agent_version(conn, "jira") == "14" and queue.agent_version(conn, "stub") == "1" + + +def test_has_active_per_module(conn_two_modules): + conn = conn_two_modules + assert not queue.has_active(conn) + queue.create_batch(conn, "w", "t", [{"type": "stub_cmd", "params": {}}]) + assert queue.has_active(conn) and queue.has_active(conn, "stub") and not queue.has_active(conn, "jira") + + +def _publish(root, module, tmp_path, version): + src = tmp_path / f"{module}{version}.py" + src.write_text(f"AGENT_VERSION = {version}\n", encoding="utf-8") + return release.publish(release.module_dir(root, module), src) + + +def test_release_per_module(hub_server_two_modules, tmp_path): + root = hub_server_two_modules["hub"].release_root + _publish(root, "jira", tmp_path, 14) + _publish(root, "stub", tmp_path, 1) + assert release.module_dir(root, "stub") == root / "stub" and oct(release.module_dir(root, "stub").stat().st_mode & 0o777) == "0o700" + jira, stub = hub_server_two_modules["agents"]["jira"], hub_server_two_modules["agents"]["stub"] + assert jira.post("/agent/claim", {})["agent_release"]["version"] == 14 + assert stub.post("/agent/claim", {})["agent_release"]["version"] == 1 + assert jira.get("/agent/release")["version"] == 14 and stub.get("/agent/release")["version"] == 1 + + +def test_release_layout_migration(tmp_path): + root = tmp_path / "release" + root.mkdir() + (root / "agent.py").write_text("AGENT_VERSION = 14\n", encoding="utf-8") + release.publish(root, root / "agent.py") # старая раскладка: agent.py и manifest.json в корне + hub = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups") + assert not (root / "agent.py").exists() and not (root / "manifest.json").exists() + assert release.current(hub.release_root / "jira")["version"] == 14 + api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups") # повторный старт ничего не ломает + assert release.current(root / "jira")["version"] == 14 + + +def test_migrate_layout_failure_is_warning(tmp_path, monkeypatch, caplog): + root = tmp_path / "release" + root.mkdir() + (root / "agent.py").write_text("AGENT_VERSION = 14\n", encoding="utf-8") + monkeypatch.setattr("os.replace", lambda *a, **k: (_ for _ in ()).throw(OSError(13, "нет прав"))) + release.migrate_layout(root) # без исключения + assert any("release" in r.getMessage() for r in caplog.records) + + +def test_publish_unknown_module(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("JIRA_HUB_HOME", str(tmp_path / "home")) + src = tmp_path / "agent.py" + src.write_text("AGENT_VERSION = 1\n", encoding="utf-8") + assert cli.main(["agent", "publish", "--module", "nope", str(src)]) == 2 + assert "неизвестный модуль nope; модули с агентами: jira" in capsys.readouterr().err + + +def test_publish_module_option(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("JIRA_HUB_HOME", str(tmp_path / "home")) + src = tmp_path / "agent.py" + src.write_text("AGENT_VERSION = 1\n", encoding="utf-8") + assert cli.main(["agent", "publish", "--module", "jira", str(src)]) == 0 + assert (tmp_path / "home" / "release" / "jira" / "manifest.json").exists() + + +def test_agents_endpoint(hub_server_two_modules): + c = hub_server_two_modules["client"] + agents = c.get("/agents")["agents"] + assert [a["module"] for a in agents] == ["jira", "stub"] + assert c.get("/agent")["module"] == "jira" and c.get("/agent", module="stub")["module"] == "stub" + with pytest.raises(HubError) as e: + c.get("/agent", module="nope") + assert e.value.status == 404 and e.value.error == "not_found" diff --git a/tests/test_api.py b/tests/test_api.py index d37b7ee..8746981 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -409,7 +409,7 @@ def test_board_route(hub, add_project): def _publish(hub, tmp_path, version): src = tmp_path / f"agent{version}.py" src.write_text(f"AGENT_VERSION = {version}\nprint('v{version}')\n", encoding="utf-8") - return release.publish(hub["hub"].release_dir, src) + return release.publish(release.module_dir(hub["hub"].release_root, "jira"), src) def test_claim_release_field(hub, tmp_path): @@ -477,8 +477,7 @@ def test_agent_view_without_release(hub): def test_claim_survives_unreadable_release(hub): - rd = hub["hub"].release_dir - rd.mkdir() + rd = release.module_dir(hub["hub"].release_root, "jira") (rd / "manifest.json").write_text('{"version": 6, "sha256": "x"}', encoding="utf-8") os.chmod(rd, 0) try: @@ -500,8 +499,7 @@ def test_claim_survives_unreadable_release(hub): def test_invalid_manifest_one_warning_for_many_claims(hub, monkeypatch, caplog): """M-3/M-7 #25, спека «Негодный манифест»: version = 6.9 — публикации нет; 10 claim и GET /agent — одно предупреждение.""" monkeypatch.setattr(release, "_warned", None) - rd = hub["hub"].release_dir - rd.mkdir() + rd = release.module_dir(hub["hub"].release_root, "jira") (rd / "manifest.json").write_text('{"version": 6.9, "sha256": "' + "a" * 64 + '"}', encoding="utf-8") for _ in range(10): assert hub["agent"].post("/agent/claim", {"agent_version": 5})["agent_release"] is None @@ -516,7 +514,7 @@ def test_agent_release_non_utf8_is_404(hub, tmp_path): """M-5 #25: agent.py не в UTF-8 при совпавшем sha256 — 404 no_release, а не 500.""" import hashlib _publish(hub, tmp_path, 6) - rd = hub["hub"].release_dir + rd = release.module_dir(hub["hub"].release_root, "jira") data = b"AGENT_VERSION = 6\n\xff\n" (rd / "agent.py").write_bytes(data) m = json.loads((rd / "manifest.json").read_text(encoding="utf-8")) diff --git a/tests/test_cli.py b/tests/test_cli.py index 90171e2..7cdafb5 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -37,7 +37,7 @@ def test_agent_publish(tmp_path, monkeypatch, capsys): assert cli.main(["agent", "publish", str(src)]) == 0 out = capsys.readouterr().out assert out.startswith("опубликована v6, sha256 ") and "…, 18 байт" in out - assert (paths.release_dir() / "manifest.json").exists() + assert (release.module_dir(paths.release_root(), "jira") / "manifest.json").exists() assert cli.main(["agent", "publish", str(src)]) == 1 assert "опубликована v6, новая должна быть больше" in capsys.readouterr().err @@ -62,8 +62,8 @@ def test_agent_publish_write_failure(tmp_path, monkeypatch, capsys): err = capsys.readouterr().err assert err.startswith("не удалось записать публикацию: ") and "Traceback" not in err monkeypatch.setattr(os, "replace", real) - assert release.load(paths.release_dir())["version"] == 8 - assert sorted(p.name for p in paths.release_dir().iterdir()) == ["agent.py", "manifest.json"] + assert release.load(release.module_dir(paths.release_root(), "jira"))["version"] == 8 + assert sorted(p.name for p in release.module_dir(paths.release_root(), "jira").iterdir()) == ["agent.py", "manifest.json"] def test_mcp_default_module(monkeypatch): diff --git a/tests/test_db.py b/tests/test_db.py index 39576ed..f3d3729 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -238,7 +238,7 @@ def test_migrated_claimed_redelivered_end_to_end(tmp_path): _v1_database(path) conn = db.connect(path) settings.ensure_defaults(conn) - cmd = queue.claim(conn) + cmd = queue.claim(conn, "jira") assert cmd["id"] == 1 assert queue.get_command(conn, 1)["deliveries"] == 2 conn.close() diff --git a/tests/test_queue.py b/tests/test_queue.py index 23d7d85..e314fcc 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -44,17 +44,17 @@ def test_refresh_only_batch_has_no_tail(conn): def test_claim_order_pause_and_report(conn): b1 = queue.create_batch(conn, "первый", "claude", [fields("A-1")]) b2 = queue.create_batch(conn, "второй", "claude", [fields("A-2")]) - c = queue.claim(conn, agent_version=1, settings_version=1) + c = queue.claim(conn, "jira", agent_version=1, settings_version=1) assert c["type"] == "ensure_fields" and c["params"]["key"] == "A-1" and c["pause_seconds"] == 120 and c["batch_id"] == b1["id"] - again = queue.claim(conn) # ответ потерян — та же команда снова + again = queue.claim(conn, "jira") # ответ потерян — та же команда снова assert again["id"] == c["id"] and again["pause_seconds"] == 120 assert queue.get_command(conn, c["id"])["deliveries"] == 2 - assert queue.agent_view(conn)["current_command_id"] == c["id"] + assert queue.agent_view(conn, "jira")["current_command_id"] == c["id"] done = queue.report_result(conn, c["id"], "done", result={"changed": ["duedate"]}, jira_calls=2) assert done["status"] == "done" and done["jira_calls"] == 2 again = queue.report_result(conn, c["id"], "failed", error="поздно") assert again["status"] == "done" # повторный отчёт не меняет - tail = queue.claim(conn) + tail = queue.claim(conn, "jira") assert tail["type"] == "refresh" and tail["pause_seconds"] == 2 assert conn.execute("SELECT kind, query FROM sync_run WHERE command_id = ?", (tail["id"],)).fetchone()["query"] == "A-1" refresh.record_state(conn, tail["id"], 1, 0) @@ -62,7 +62,7 @@ def test_claim_order_pause_and_report(conn): run = conn.execute("SELECT * FROM sync_run WHERE command_id = ?", (tail["id"],)).fetchone() assert run["finished_at"] and run["issues_count"] == 1 and run["jira_calls"] == 1 assert queue.batch_view(conn, b1["id"])["status"] == "done" - assert queue.claim(conn)["params"]["key"] == "A-2" and queue.batch_view(conn, b2["id"])["status"] == "running" + assert queue.claim(conn, "jira")["params"]["key"] == "A-2" and queue.batch_view(conn, b2["id"])["status"] == "running" def test_claim_is_exclusive(tmp_path, open_db): @@ -76,7 +76,7 @@ def test_claim_is_exclusive(tmp_path, open_db): def worker(c): barrier.wait() - results.append(queue.claim(c)) + results.append(queue.claim(c, "jira")) threads = [threading.Thread(target=worker, args=(c,)) for c in conns] [t.start() for t in threads] @@ -93,7 +93,7 @@ def test_backup_failure_does_not_block_claim(conn, caplog): def broken(label): raise OSError("диск полон") - c = queue.claim(conn, backup=broken) + c = queue.claim(conn, "jira", backup=broken) assert c is not None and c["params"]["key"] == "A-1" assert queue.get_command(conn, c["id"])["status"] == "claimed" assert "не создана" in caplog.text @@ -109,15 +109,15 @@ def test_temp_key_substitution_and_skip(conn, add_project): "labels": ["AWPCA"]}}, {"type": "ensure_sprint", "params": {"key": "new:61", "sprint": "Спринт 2"}}, ]) - c1 = queue.claim(conn) + c1 = queue.claim(conn, "jira") queue.report_result(conn, c1["id"], "done", result={"key": "A-90"}) assert mirror.get_note(conn, "A-90")["fact_status"] == "В работе" and mirror.get_note(conn, "new:60") is None - c2 = queue.claim(conn) + c2 = queue.claim(conn, "jira") assert c2["type"] == "ensure_link" and c2["params"]["source"] == "A-90" queue.report_result(conn, c2["id"], "done") - c3 = queue.claim(conn) + c3 = queue.claim(conn, "jira") queue.report_result(conn, c3["id"], "failed", error="HTTP 400") - tail = queue.claim(conn) # ensure_sprint пропущена, refresh без new:61 + tail = queue.claim(conn, "jira") # ensure_sprint пропущена, refresh без new:61 assert tail["type"] == "refresh" and tail["params"]["keys"] == ["A-90", "A-1"] view = queue.batch_view(conn, c1["batch_id"]) assert [c["status"] for c in view["commands"]][:4] == ["done", "done", "failed", "skipped"] @@ -126,12 +126,12 @@ def test_temp_key_substitution_and_skip(conn, add_project): def test_cancel_and_retry(conn): b = queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") with pytest.raises(ClaimedError): queue.cancel_command(conn, c["id"]) view = queue.cancel_batch(conn, b["id"]) assert [x["status"] for x in view["commands"]] == ["claimed", "cancelled", "cancelled"] - assert queue.claim(conn)["id"] == c["id"] # взятая из отменённого пакета довыдаётся + assert queue.claim(conn, "jira")["id"] == c["id"] # взятая из отменённого пакета довыдаётся forced = queue.cancel_command(conn, c["id"], force=True) assert forced["status"] == "cancelled" after = queue.report_result(conn, c["id"], "done", result={"changed": []}) @@ -144,7 +144,7 @@ def test_cancel_and_retry(conn): def test_report_after_force_cancel_keeps_cancelled(conn): queue.create_batch(conn, "x", "claude", [fields("A-1")]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") queue.cancel_command(conn, c["id"], force=True) assert queue.report_result(conn, c["id"], "failed", error="e")["status"] == "cancelled" @@ -153,7 +153,7 @@ def test_reconcile_confirms_by_state(conn, add_project): add_project(conn) mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "labels": ["AWPCA"]}], []) queue.create_batch(conn, "x", "claude", [{"type": "ensure_sprint", "params": {"key": "A-1", "sprint": "Спринт 2"}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") b = queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) r = queue.batch_view(conn, b["id"])["commands"][0] conn.execute("UPDATE command SET status = 'claimed', claimed_at = ?, deliveries = 1 WHERE id = ?", (db.now(), r["id"])) @@ -164,14 +164,14 @@ def test_reconcile_confirms_by_state(conn, add_project): def test_auto_refresh_needs_projects(conn): - queue.claim(conn, agent_version=1, settings_version=1) + queue.claim(conn, "jira", agent_version=1, settings_version=1) assert refresh.maybe_auto_refresh(conn) is None # агент на связи, очередь пуста, проектов нет def test_auto_refresh_two_projects(conn, add_project): # сценарий «Два проекта» add_project(conn) add_project(conn, "STOCKS_DRRK", board_id=None) - queue.claim(conn, agent_version=1, settings_version=1) + queue.claim(conn, "jira", agent_version=1, settings_version=1) b = refresh.maybe_auto_refresh(conn) assert queue.batch_view(conn, b["id"])["commands"][0]["params"] == {"jql": 'project = AURORA AND labels in ("AWPCA", "STOCKS_DRRK")'} @@ -179,11 +179,11 @@ def test_auto_refresh_two_projects(conn, add_project): # сцен def test_auto_refresh_conditions(conn, add_project): add_project(conn) assert refresh.maybe_auto_refresh(conn) is None # агент не на связи - queue.claim(conn, agent_version=1, settings_version=1) # отметился, очередь пуста + queue.claim(conn, "jira", agent_version=1, settings_version=1) # отметился, очередь пуста b = refresh.maybe_auto_refresh(conn) assert b and queue.batch_view(conn, b["id"])["title"] == "Автообновление" assert refresh.maybe_auto_refresh(conn) is None # очередь занята - c = queue.claim(conn) + c = queue.claim(conn, "jira") queue.report_result(conn, c["id"], "done") assert refresh.maybe_auto_refresh(conn) is None # интервал не прошёл conn.execute("UPDATE sync_run SET started_at = '2020-01-01T00:00:00+00:00'") @@ -196,16 +196,16 @@ def test_auto_refresh_conditions(conn, add_project): def test_backup_called_for_first_write_only(conn, tmp_path): calls = [] queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")]) - c1 = queue.claim(conn, backup=calls.append) + c1 = queue.claim(conn, "jira", backup=calls.append) queue.report_result(conn, c1["id"], "done") - queue.claim(conn, backup=calls.append) + queue.claim(conn, "jira", backup=calls.append) assert calls == [f"batch-{c1['batch_id']}"] def test_no_backup_for_refresh_batch(conn): calls = [] queue.create_batch(conn, "x", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}]) - queue.claim(conn, backup=calls.append) + queue.claim(conn, "jira", backup=calls.append) assert calls == [] @@ -233,7 +233,7 @@ def test_rotation_keeps_schema_migration_copy(tmp_path, open_db): def _claimed_ensure_issue(conn, summary="Тема"): queue.create_batch(conn, "x", "claude", [{"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": summary}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (c["id"],)) return c @@ -254,7 +254,7 @@ def test_reconcile_does_not_overwrite_finished_command(conn, monkeypatch, add_pr add_project(conn) mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "labels": ["AWPCA"]}], []) queue.create_batch(conn, "x", "claude", [{"type": "ensure_sprint", "params": {"key": "A-1", "sprint": "Спринт 2"}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "sprint": "Спринт 2", "labels": ["AWPCA"]}], []) real = commands.check_confirmed @@ -272,7 +272,7 @@ def _two_batches_with_dependent(conn): a = queue.create_batch(conn, "A", "claude", [ {"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": "Тема"}}, {"type": "ensure_link", "params": {"source": "new:1", "target": "A-1", "link_type": "связанные"}}]) - first = queue.claim(conn) + first = queue.claim(conn, "jira") assert first["type"] == "ensure_issue" b = queue.create_batch(conn, "B", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}]) return a, b, first @@ -280,16 +280,16 @@ def _two_batches_with_dependent(conn): def test_redelivered_definer_blocks_dependents(conn): a, b, first = _two_batches_with_dependent(conn) - assert queue.claim(conn)["id"] == first["id"] # довыдача определителя, зависимые ждут + assert queue.claim(conn, "jira")["id"] == first["id"] # довыдача определителя, зависимые ждут conn.execute("UPDATE command SET status = 'done', result = '{\"key\": \"A-9\"}' WHERE id = ?", (first["id"],)) - nxt = queue.claim(conn) + nxt = queue.claim(conn, "jira") assert nxt["batch_id"] == a["id"] and nxt["params"]["source"] == "A-9" def test_auto_refresh_waits_for_claimed_without_report(conn): - queue.claim(conn, agent_version=1, settings_version=1) # агент на связи + queue.claim(conn, "jira", agent_version=1, settings_version=1) # агент на связи queue.create_batch(conn, "y", "claude", [fields("A-3")]) - queue.claim(conn) # выдана, отчёта нет + queue.claim(conn, "jira") # выдана, отчёта нет conn.execute("DELETE FROM command WHERE status = 'queued'") # хвостовой refresh не мешает проверке conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE status = 'claimed'") # давно без отчёта assert refresh.maybe_auto_refresh(conn) is None # claimed без отчёта — очередь занята @@ -297,23 +297,23 @@ def test_auto_refresh_waits_for_claimed_without_report(conn): def test_redelivery_of_lost_claim_before_queue(conn): queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")]) - c = queue.claim(conn) - again = queue.claim(conn) + c = queue.claim(conn, "jira") + again = queue.claim(conn, "jira") assert again["id"] == c["id"] and again["params"] == c["params"] - assert queue.agent_view(conn)["current_command_id"] == c["id"] - view = queue.agent_view(conn)["redelivered"] + assert queue.agent_view(conn, "jira")["current_command_id"] == c["id"] + view = queue.agent_view(conn, "jira")["redelivered"] assert [(x["id"], x["deliveries"]) for x in view] == [(c["id"], 2)] queue.report_result(conn, c["id"], "done") - assert queue.claim(conn)["params"]["key"] == "A-2" - assert queue.agent_view(conn)["redelivered"] == [] + assert queue.claim(conn, "jira")["params"]["key"] == "A-2" + assert queue.agent_view(conn, "jira")["redelivered"] == [] def test_redelivery_limit_fails_command(conn): queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")]) - c = queue.claim(conn) - queue.claim(conn) - queue.claim(conn) # третья выдача - nxt = queue.claim(conn) # четвёртой нет — failed и дальше по очереди + c = queue.claim(conn, "jira") + queue.claim(conn, "jira") + queue.claim(conn, "jira") # третья выдача + nxt = queue.claim(conn, "jira") # четвёртой нет — failed и дальше по очереди assert nxt["params"]["key"] == "A-2" failed = queue.get_command(conn, c["id"]) assert failed["status"] == "failed" and failed["error"] == "агент не отчитался после 3 выдач" and failed["finished_at"] @@ -326,21 +326,21 @@ def test_redelivery_oldest_first(conn): rows = conn.execute("SELECT id FROM command WHERE type = 'ensure_fields' ORDER BY id").fetchall() conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2026-10-13T10:00:00+00:00', deliveries = 1 WHERE id = ?", (rows[1]["id"],)) conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2026-10-13T11:00:00+00:00', deliveries = 1 WHERE id = ?", (rows[0]["id"],)) - assert queue.claim(conn)["id"] == rows[1]["id"] + assert queue.claim(conn, "jira")["id"] == rows[1]["id"] def test_redelivered_refresh_keeps_one_sync_run(conn): queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - r = queue.claim(conn) - assert queue.claim(conn)["id"] == r["id"] + r = queue.claim(conn, "jira") + assert queue.claim(conn, "jira")["id"] == r["id"] assert conn.execute("SELECT COUNT(*) AS n FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["n"] == 1 def test_redelivery_does_not_repeat_backup(conn): queue.create_batch(conn, "x", "claude", [fields("A-1")]) labels = [] - queue.claim(conn, backup=labels.append) - queue.claim(conn, backup=labels.append) + queue.claim(conn, "jira", backup=labels.append) + queue.claim(conn, "jira", backup=labels.append) assert len(labels) == 1 @@ -349,7 +349,7 @@ def test_migrated_claimed_is_redelivered(conn): queue.create_batch(conn, "x", "claude", [fields("A-1")]) cid = conn.execute("SELECT id FROM command WHERE type = 'ensure_fields'").fetchone()["id"] conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2026-10-12T00:00:00+00:00', deliveries = 1 WHERE id = ?", (cid,)) - assert queue.claim(conn)["id"] == cid + assert queue.claim(conn, "jira")["id"] == cid assert queue.get_command(conn, cid)["deliveries"] == 2 @@ -365,7 +365,7 @@ def test_late_report_after_force_cancel_keeps_issue_key(conn): mirror.set_note(conn, "new:7", {"fact_status": "В работе"}, "import") queue.create_batch(conn, "x", "claude", [ {"type": "ensure_issue", "params": {"temp_key": "new:7", "issue_type": "task", "summary": "Тема"}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") cancelled = queue.cancel_command(conn, c["id"], force=True) late = queue.report_result(conn, c["id"], "done", result={"key": "A-20", "created": True}, jira_calls=3) assert late["status"] == "cancelled" and late["result"] == {"key": "A-20", "created": True} and late["jira_calls"] == 3 @@ -378,7 +378,7 @@ def test_late_report_after_force_cancel_keeps_issue_key(conn): def test_late_failed_report_after_force_cancel_changes_nothing(conn): queue.create_batch(conn, "x", "claude", [ {"type": "ensure_issue", "params": {"temp_key": "new:7", "issue_type": "task", "summary": "Тема"}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") queue.cancel_command(conn, c["id"], force=True) late = queue.report_result(conn, c["id"], "failed", error="HTTP 400") assert late["status"] == "cancelled" and late["result"] is None and late["error"] is None @@ -386,14 +386,14 @@ def test_late_failed_report_after_force_cancel_changes_nothing(conn): def _fail_by_limit(conn, cid): for _ in range(queue.MAX_DELIVERIES - 1): - assert queue.claim(conn)["id"] == cid - queue.claim(conn) # четвёртый claim: cid → failed по пределу + assert queue.claim(conn, "jira")["id"] == cid + queue.claim(conn, "jira") # четвёртый claim: cid → failed по пределу assert queue.get_command(conn, cid)["status"] == "failed" def test_error_is_trimmed_to_1500(conn): queue.create_batch(conn, "x", "claude", [fields("A-1")]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") out = queue.report_result(conn, c["id"], "failed", error="е" * 5000) assert out["error"] == "е" * 1500 and queue.ERROR_LIMIT == 1500 @@ -402,7 +402,7 @@ def test_late_done_after_delivery_limit_keeps_issue_key(conn): mirror.set_note(conn, "new:8", {"fact_status": "План"}, "import") queue.create_batch(conn, "x", "claude", [ {"type": "ensure_issue", "params": {"temp_key": "new:8", "issue_type": "task", "summary": "Тема"}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") _fail_by_limit(conn, c["id"]) late = queue.report_result(conn, c["id"], "done", result={"key": "A-30", "created": True}, jira_calls=2) assert late["status"] == "failed" and late["result"] == {"key": "A-30", "created": True} @@ -414,9 +414,9 @@ def test_late_done_after_delivery_limit_keeps_issue_key(conn): def test_cancelled_batch_with_done_command_is_cancelled(conn): b = queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2"), fields("A-3")]) - first = queue.claim(conn) + first = queue.claim(conn, "jira") queue.report_result(conn, first["id"], "done", result={"changed": []}) - second = queue.claim(conn) + second = queue.claim(conn, "jira") queue.cancel_batch(conn, b["id"]) queue.report_result(conn, second["id"], "done", result={"changed": []}) view = queue.batch_view(conn, b["id"]) @@ -430,14 +430,14 @@ def test_empty_full_refresh_keeps_neighbors(conn, add_project): mirror.apply_state(conn, [{"key": "A-1", "summary": "x", "labels": []}], [], fetched_at="2020-01-01T00:00:00+00:00") # прочитана давно — до старта прогона queue.create_batch(conn, "полный", "claude", [{"type": "refresh", "params": {"jql": jql}}]) - r = queue.claim(conn) + r = queue.claim(conn, "jira") queue.report_result(conn, r["id"], "done") # загрузок состояния не было: 0 задач assert conn.execute("SELECT neighbor FROM issue WHERE key = 'A-1'").fetchone()["neighbor"] == 0 def test_cancel_finished_batch_stays_done(conn): b = queue.create_batch(conn, "x", "claude", [fields("A-1")]) - while (c := queue.claim(conn)): + while (c := queue.claim(conn, "jira")): queue.report_result(conn, c["id"], "done", result={"changed": []}) view = queue.cancel_batch(conn, b["id"]) assert view["status"] == "done" @@ -449,10 +449,10 @@ def test_redelivered_full_refresh_marks_neighbors_once(conn, add_project): jql = jira_projects.base_jql(conn) mirror.apply_state(conn, [{"key": "OLD", "summary": "x", "labels": []}], [], fetched_at="2020-01-01T00:00:00+00:00") queue.create_batch(conn, "полный", "claude", [{"type": "refresh", "params": {"jql": jql}}]) - r = queue.claim(conn) + r = queue.claim(conn, "jira") refresh.record_state(conn, r["id"], 1, 0) mirror.apply_state(conn, [{"key": "NEW", "summary": "y", "labels": []}], []) - assert queue.claim(conn)["id"] == r["id"] # отчёт потерян — довыдача + assert queue.claim(conn, "jira")["id"] == r["id"] # отчёт потерян — довыдача assert conn.execute("SELECT deliveries FROM command WHERE id = ?", (r["id"],)).fetchone()["deliveries"] == 2 refresh.record_state(conn, r["id"], 1, 0) mirror.apply_state(conn, [{"key": "NEW", "summary": "y", "labels": []}], []) @@ -464,10 +464,10 @@ def test_redelivered_full_refresh_marks_neighbors_once(conn, add_project): def test_redelivered_refresh_resets_counts(conn): queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - r = queue.claim(conn) + r = queue.claim(conn, "jira") refresh.record_state(conn, r["id"], 150, 40) started = conn.execute("SELECT started_at FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["started_at"] - assert queue.claim(conn)["id"] == r["id"] # отчёт потерян — довыдача + assert queue.claim(conn, "jira")["id"] == r["id"] # отчёт потерян — довыдача refresh.record_state(conn, r["id"], 150, 40) run = conn.execute("SELECT * FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone() assert (run["issues_count"], run["links_count"]) == (150, 40) @@ -476,16 +476,16 @@ def test_redelivered_refresh_resets_counts(conn): def test_refresh_failed_by_limit_finishes_sync_run(conn): queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - r = queue.claim(conn) + r = queue.claim(conn, "jira") for _ in range(queue.MAX_DELIVERIES): - queue.claim(conn) + queue.claim(conn, "jira") assert queue.get_command(conn, r["id"])["status"] == "failed" assert conn.execute("SELECT finished_at FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["finished_at"] def test_cancel_batch_reports_still_claimed(conn): b = queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") view = queue.cancel_batch(conn, b["id"]) assert view["still_claimed"] == [{"id": c["id"], "type": "ensure_fields"}] assert queue.get_command(conn, c["id"])["status"] == "claimed" @@ -496,9 +496,9 @@ def test_dependent_skipped_when_creation_cancelled(conn): b = queue.create_batch(conn, "x", "claude", [ {"type": "ensure_issue", "params": {"temp_key": "new:901", "issue_type": "task", "summary": "Отменённая"}}, {"type": "ensure_comment", "params": {"key": "new:901", "marker": "m", "body": "b"}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") queue.cancel_command(conn, c["id"], force=True) - assert queue.claim(conn) is None # комментарий и хвостовой refresh пропущены + assert queue.claim(conn, "jira") is None # комментарий и хвостовой refresh пропущены cmds = queue.batch_view(conn, b["id"])["commands"] assert [x["status"] for x in cmds] == ["cancelled", "skipped", "skipped"] assert cmds[1]["error"] == "создание задачи new:901 отменено" @@ -510,11 +510,11 @@ def test_dependent_on_failed_and_cancelled_names_cancelled(conn): {"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": "Упавшая"}}, {"type": "ensure_issue", "params": {"temp_key": "new:2", "issue_type": "task", "summary": "Отменённая"}}, {"type": "ensure_link", "params": {"source": "new:1", "target": "new:2", "link_type": "связанные"}}]) - first = queue.claim(conn) + first = queue.claim(conn, "jira") queue.report_result(conn, first["id"], "failed", error="HTTP 400") - second = queue.claim(conn) + second = queue.claim(conn, "jira") queue.cancel_command(conn, second["id"], force=True) - queue.claim(conn) + queue.claim(conn, "jira") link = queue.batch_view(conn, b["id"])["commands"][2] assert link["status"] == "skipped" and link["error"] == "создание задачи new:2 отменено" @@ -522,7 +522,7 @@ def test_dependent_on_failed_and_cancelled_names_cancelled(conn): def test_reconcile_confirms_fields_with_crlf(conn): """O-9 #16, решение 11, спека «Описание с CRLF в зеркале»: потерянный отчёт закрывается по состоянию.""" queue.create_batch(conn, "x", "claude", [{"type": "ensure_fields", "params": {"key": "A-1", "description": "a\nb"}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "description": "a\r\nb", "labels": []}], []) assert queue.reconcile(conn) == [c["id"]] assert queue.get_command(conn, c["id"])["status"] == "confirmed_by_state" @@ -531,15 +531,15 @@ def test_reconcile_confirms_fields_with_crlf(conn): def test_delivery_limit_of_write_command_keeps_open_sync_run(conn): """T5.2 #16: предел выдач у не-refresh команды не закрывает журнал чужой загрузки (finished_at — только для refresh).""" queue.create_batch(conn, "x", "claude", [fields("A-1")]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") r = queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-9"]}}]) rid = queue.batch_view(conn, r["id"])["commands"][0]["id"] conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2099-01-01T00:00:00+00:00', deliveries = 1 " "WHERE id = ?", (rid,)) conn.execute("INSERT INTO sync_run(started_at, kind, query, command_id) VALUES (?, 'keys', 'A-9', ?)", (db.now(), rid)) for _ in range(queue.MAX_DELIVERIES - 1): - assert queue.claim(conn)["id"] == c["id"] - assert queue.claim(conn)["id"] == rid # c — failed по пределу, довыдаётся refresh + assert queue.claim(conn, "jira")["id"] == c["id"] + assert queue.claim(conn, "jira")["id"] == rid # c — failed по пределу, довыдаётся refresh assert queue.get_command(conn, c["id"])["status"] == "failed" assert conn.execute("SELECT finished_at FROM sync_run WHERE command_id = ?", (rid,)).fetchone()["finished_at"] is None @@ -625,16 +625,16 @@ def test_bug_with_attachment_one_batch(conn, store, add_project): bug(), {"type": "ensure_sprint", "params": {"key": "new:1", "sprint": "Спринт 2"}}, attach(key="new:1")], files_dir=store) assert b["count"] == 4 and b["estimate_seconds"] == 120 + 30 + 30 + 2 - c1 = queue.claim(conn, files_dir=store) + c1 = queue.claim(conn, "jira", files_dir=store) assert c1["type"] == "ensure_issue" queue.report_result(conn, c1["id"], "done", result={"key": "AURORA-1650"}) - c2 = queue.claim(conn, files_dir=store) + c2 = queue.claim(conn, "jira", files_dir=store) assert c2["type"] == "ensure_sprint" queue.report_result(conn, c2["id"], "done") - c3 = queue.claim(conn, files_dir=store) + c3 = queue.claim(conn, "jira", files_dir=store) assert c3["type"] == "ensure_attachment" and c3["pause_seconds"] == 30 assert c3["params"] == {"key": "AURORA-1650", "file": SHA, "filename": "ОСВ 60.xlsx", "size": 3} - again = queue.claim(conn, files_dir=store) # довыдача — тоже с size + again = queue.claim(conn, "jira", files_dir=store) # довыдача — тоже с size assert again["id"] == c3["id"] and again["params"]["size"] == 3 stored = queue.get_command(conn, c3["id"])["params"] assert stored == {"key": "AURORA-1650", "file": SHA, "filename": "ОСВ 60.xlsx"} # в базе без size @@ -643,15 +643,15 @@ def test_bug_with_attachment_one_batch(conn, store, add_project): def test_attachment_size_null_when_file_gone(conn, store): queue.create_batch(conn, "вложение", "claude", [attach()], files_dir=store) (store / SHA).unlink() - c = queue.claim(conn, files_dir=store) + c = queue.claim(conn, "jira", files_dir=store) assert c["type"] == "ensure_attachment" and c["params"]["size"] is None - assert queue.claim(conn)["params"]["size"] is None # без хранилища — тоже null, не исключение + assert queue.claim(conn, "jira")["params"]["size"] is None # без хранилища — тоже null, не исключение assert queue.deliver_params(conn, queue.get_command(conn, c["id"]), store)["size"] is None def test_attachment_retry_passes_validation(conn, store): queue.create_batch(conn, "вложение", "claude", [attach()], files_dir=store) - c = queue.claim(conn, files_dir=store) + c = queue.claim(conn, "jira", files_dir=store) queue.report_result(conn, c["id"], "failed", error="HTTP 413") r = queue.retry_command(conn, c["id"], "claude", files_dir=store) assert queue.batch_view(conn, r["id"])["commands"][0]["params"] == {"key": "A-1", "file": SHA, "filename": "ОСВ 60.xlsx"} @@ -663,7 +663,7 @@ def test_attachment_retry_passes_validation(conn, store): def test_attachment_not_confirmed_by_state(conn, store): mirror.apply_state(conn, [{"key": "A-1", "summary": "Т", "labels": []}], []) queue.create_batch(conn, "вложение", "claude", [attach()], files_dir=store) - c = queue.claim(conn, files_dir=store) + c = queue.claim(conn, "jira", files_dir=store) assert queue.reconcile(conn) == [] and queue.get_command(conn, c["id"])["status"] == "claimed" @@ -900,16 +900,16 @@ def test_refresh_delivered_with_boards(conn, add_project): # сце add_project(conn, "STOCKS_DRRK", board_id=5120, board_type=None) add_project(conn, "C", board_id=4216) b = queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": "key = A-1"}}]) - got = queue.claim(conn) + got = queue.claim(conn, "jira") assert got["params"]["boards"] == [{"id": 4216, "type": "scrum"}, {"id": 5120, "type": None}] assert "boards" not in queue.batch_view(conn, b["id"])["commands"][0]["params"] - assert queue.claim(conn)["params"]["boards"] == got["params"]["boards"] # довыдача — тоже с boards + assert queue.claim(conn, "jira")["params"]["boards"] == got["params"]["boards"] # довыдача — тоже с boards def test_refresh_boards_empty_list(conn, add_project): add_project(conn, board_id=None, board_type=None) queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - assert queue.claim(conn)["params"]["boards"] == [] + assert queue.claim(conn, "jira")["params"]["boards"] == [] @pytest.mark.parametrize("result, expected", [ @@ -925,7 +925,7 @@ def test_refresh_boards_empty_list(conn, add_project): def test_refresh_result_board_types(conn, add_project, result, expected): add_project(conn, "STOCKS_DRRK", board_id=5120, board_type=None) queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") queue.report_result(conn, c["id"], "done", result=result) assert projects.get(conn, "STOCKS_DRRK")["board_type"] == expected @@ -933,7 +933,7 @@ def test_refresh_result_board_types(conn, add_project, result, expected): def test_refresh_result_board_type_changes_scrum_to_kanban(conn, add_project): # доска в Jira стала канбаном add_project(conn, "STOCKS_DRRK", board_id=5120, board_type="scrum") queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") queue.report_result(conn, c["id"], "done", result={"boards": {"5120": "kanban"}}) assert projects.get(conn, "STOCKS_DRRK")["board_type"] == "kanban" @@ -941,7 +941,7 @@ def test_refresh_result_board_type_changes_scrum_to_kanban(conn, add_project): def test_refresh_failed_result_records_board_types(conn, add_project): # R-3 / T6.1: тип доски — факт о доске, а не часть выборки add_project(conn, "STOCKS_DRRK", board_id=5120, board_type=None) queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") queue.report_result(conn, c["id"], "failed", result={"boards": {"5120": "kanban"}}, error="e") assert projects.get(conn, "STOCKS_DRRK")["board_type"] == "kanban" @@ -951,7 +951,7 @@ def test_refresh_failed_result_records_board_types(conn, add_project): # R-3 def _full_run(conn, result, started_at=None, status="done"): """Полный refresh: поставить, выдать, отчитаться; started_at — подменить время начала прогона.""" queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") if started_at: conn.execute("UPDATE sync_run SET started_at = ? WHERE command_id = ?", (started_at, c["id"])) queue.report_result(conn, c["id"], status, result=result) @@ -961,7 +961,7 @@ def _full_run(conn, result, started_at=None, status="done"): def test_first_full_refresh_gets_backfill(conn, add_project): # сценарий «Первый refresh после выкатки» add_project(conn) b = queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}]) - got = queue.claim(conn) + got = queue.claim(conn, "jira") assert got["params"]["history"] == { "jql": '(project = AURORA AND labels in ("AWPCA")) AND sprint in openSprints() AND statusCategory != Done'} assert "history" not in queue.batch_view(conn, b["id"])["commands"][0]["params"] @@ -972,7 +972,7 @@ def test_regular_refresh_gets_since(conn, add_project): # «Р _full_run(conn, {"history": {"issues": 1, "changes": 2, "incomplete": []}}, "2026-10-22T09:00:00+00:00") _full_run(conn, {"history": {"error": "HTTP 500"}}, "2026-10-22T10:00:00+00:00") queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}]) - assert queue.claim(conn)["params"]["history"] == {"since": "2026-10-22T08:45:00Z"} + assert queue.claim(conn, "jira")["params"]["history"] == {"since": "2026-10-22T08:45:00Z"} def test_new_project_gets_since_and_backfill(conn, add_project): # «Новый проект» @@ -980,7 +980,7 @@ def test_new_project_gets_since_and_backfill(conn, add_project): # «Н _full_run(conn, {"history": {"issues": 0, "changes": 0, "incomplete": []}}, "2026-10-22T09:00:00+00:00") add_project(conn, "STOCKS_DRRK", board_id=5120) queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}]) - h = queue.claim(conn)["params"]["history"] + h = queue.claim(conn, "jira")["params"]["history"] assert h["since"] == "2026-10-22T08:45:00Z" assert h["jql"] == "(" + jira_projects.base_jql(conn) + ") AND sprint in openSprints() AND statusCategory != Done" assert "STOCKS_DRRK" in h["jql"] @@ -990,7 +990,7 @@ def test_tail_refresh_since_and_no_mark(conn, add_project): # «Х add_project(conn) _full_run(conn, {"history": {"issues": 0, "changes": 0, "incomplete": []}}, "2026-10-22T09:00:00+00:00") queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - c = queue.claim(conn) + c = queue.claim(conn, "jira") assert c["params"]["history"] == {"since": "2026-10-22T08:45:00Z"} queue.report_result(conn, c["id"], "done", result={"history": {"issues": 0, "changes": 0, "incomplete": []}}) assert conn.execute("SELECT history_at FROM sync_run WHERE command_id = ?", (c["id"],)).fetchone()["history_at"] is None @@ -999,19 +999,19 @@ def test_tail_refresh_since_and_no_mark(conn, add_project): # «Х def test_tail_refresh_without_mark_gets_nothing(conn, add_project): # «Хвост пакета до первой отметки» add_project(conn) queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - assert "history" not in queue.claim(conn)["params"] + assert "history" not in queue.claim(conn, "jira")["params"] def test_no_projects_no_history(conn): queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) - assert "history" not in queue.claim(conn)["params"] + assert "history" not in queue.claim(conn, "jira")["params"] def test_redelivery_recomputes_history(conn, add_project): # Review Focus 3 add_project(conn) b = queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}]) - first = queue.claim(conn) - again = queue.claim(conn) # отчёта нет — довыдача + first = queue.claim(conn, "jira") + again = queue.claim(conn, "jira") # отчёта нет — довыдача assert again["id"] == first["id"] and again["params"]["history"] == first["params"]["history"] assert "history" not in queue.batch_view(conn, b["id"])["commands"][0]["params"] @@ -1086,7 +1086,7 @@ def test_fetch_size_limit_boundary(fconn): def test_fetch_report_done_non_dict_result_fails(fconn, tmp_path): """T3.4: done с result не-словарём — failed, вложение не отмечено скачанным.""" queue.create_batch(fconn, "x", "claude", [fetch()]) - c = queue.claim(fconn) + c = queue.claim(fconn, "jira") r = queue.report_result(fconn, c["id"], "done", result="ok", files_dir=tmp_path) assert r["status"] == "failed" assert fconn.execute("SELECT COUNT(*) AS n FROM attachment_file").fetchone()["n"] == 0 @@ -1095,7 +1095,7 @@ def test_fetch_report_done_non_dict_result_fails(fconn, tmp_path): def test_fetch_no_backup(fconn): queue.create_batch(fconn, "x", "claude", [fetch()]) calls = [] - cmd = queue.claim(fconn, backup=calls.append) + cmd = queue.claim(fconn, "jira", backup=calls.append) assert cmd["type"] == "fetch_attachment" and cmd["pause_seconds"] == 2 and calls == [] @@ -1161,7 +1161,7 @@ def test_request_fetch_agent_too_old(fconn, tmp_path): def test_request_fetch_after_failed(fconn, tmp_path): root = tmp_path / "files" first = jira_fetch.request_fetch(fconn, root, "31337", "x") - cmd = queue.claim(fconn) + cmd = queue.claim(fconn, "jira") queue.report_result(fconn, cmd["id"], "failed", error="нет сети", files_dir=root) again = jira_fetch.request_fetch(fconn, root, "31337", "x") assert again["state"] == "fetching" and again["batch_id"] != first["batch_id"] @@ -1169,7 +1169,7 @@ def test_request_fetch_after_failed(fconn, tmp_path): def report_done(conn, root, result): queue.create_batch(conn, "x", "claude", [fetch()]) - cmd = queue.claim(conn) + cmd = queue.claim(conn, "jira") return queue.report_result(conn, cmd["id"], "done", result=result, files_dir=root) diff --git a/tests/test_queue_core.py b/tests/test_queue_core.py index d2eac4e..e882ccb 100644 --- a/tests/test_queue_core.py +++ b/tests/test_queue_core.py @@ -57,7 +57,7 @@ def test_on_result_called(tmp_path, open_db, stub): path = tmp_path / "hub.sqlite" conn = open_db(path) queue.create_batch(conn, "x", "claude", [{"type": "w", "params": {}}]) - cmd = queue.claim(conn) + cmd = queue.claim(conn, "t") seen = {} real = stub.append