diff --git a/src/jira_hub/core/api.py b/src/jira_hub/core/api.py index a0e51a5..55c0abd 100644 --- a/src/jira_hub/core/api.py +++ b/src/jira_hub/core/api.py @@ -16,7 +16,7 @@ from typing import NamedTuple from urllib.parse import parse_qs, quote, unquote, urlparse from . import backup, db, files, projects, queue, relay, release, settings -from ..modules.jira import attachments, history, mirror +from ..modules.jira import attachments, fetch, history, mirror, refresh # временно: R-1, до Task 5 from ..modules.jira import projects as jira_projects from .errors import CommandError from .paths import backups_dir as default_backups_dir @@ -78,7 +78,7 @@ class Hub: conn = None try: conn = self.conn() - queue.maybe_auto_refresh(conn) + refresh.maybe_auto_refresh(conn) except Exception: # noqa: BLE001 log.exception("автообновление: сбой тика") try: @@ -216,9 +216,9 @@ def h_agent_state(hub, conn, m, q, body): raise ApiError(422, "bad_state", str(e)) with db.transaction(conn): out = mirror.apply_state(conn, issues, body.get("links") or [], body.get("neighbor_keys") or [], - authoritative=queue.is_full_refresh(conn, body.get("command_id"))) + authoritative=refresh.is_full_refresh(conn, body.get("command_id"))) if body.get("command_id"): - queue.record_state(conn, body["command_id"], out["issues"], out["links"]) + refresh.record_state(conn, body["command_id"], out["issues"], out["links"]) return 200, out @@ -346,7 +346,7 @@ def h_batches_post(hub, conn, m, q, body): try: return 201, queue.create_batch(conn, body["title"], body.get("author") or "api", body.get("commands"), files_dir=hub.files_dir, project=body.get("project")) - except queue.TooLarge as e: # раньше CommandError: TooLarge — его подкласс (M-2) + except fetch.TooLarge as e: # раньше CommandError: TooLarge — его подкласс (M-2) raise ApiError(413, "too_large", str(e)) from None except CommandError as e: raise ApiError(422, "bad_batch", str(e)) @@ -411,12 +411,12 @@ def h_attachment_fetch(hub, conn, m, q, body): if not isinstance(author, str): raise ApiError(422, "bad_request", "author — строка") try: - return 200, queue.request_fetch(conn, hub.files_dir, m.group(1), author) + return 200, fetch.request_fetch(conn, hub.files_dir, m.group(1), author) except KeyError: raise ApiError(404, "no_attachment", f"вложения #{m.group(1)} нет в зеркале — обновите задачу") from None - except queue.TooLarge as e: # раньше CommandError: TooLarge — его подкласс (решение 17) + except fetch.TooLarge as e: # раньше CommandError: TooLarge — его подкласс (решение 17) raise ApiError(413, "too_large", str(e)) from None - except queue.AgentTooOld as e: + except fetch.AgentTooOld as e: raise ApiError(409, "agent_too_old", str(e)) from None except CommandError as e: raise ApiError(422, "bad_batch", str(e)) from None @@ -449,7 +449,7 @@ def h_command_retry(hub, conn, m, q, body): project=body.get("project")) except KeyError: raise ApiError(404, "not_found", "command") - except queue.TooLarge as e: # раньше CommandError: TooLarge — его подкласс (M-2) + except fetch.TooLarge as e: # раньше CommandError: TooLarge — его подкласс (M-2) raise ApiError(413, "too_large", str(e)) from None except CommandError as e: raise ApiError(422, "bad_batch", str(e)) diff --git a/src/jira_hub/core/files.py b/src/jira_hub/core/files.py index e553181..ecb29af 100644 --- a/src/jira_hub/core/files.py +++ b/src/jira_hub/core/files.py @@ -9,6 +9,7 @@ import tempfile import time from pathlib import Path +from . import registry from .paths import secure_dir MAX_SIZE = 10 * 1024 * 1024 # 10 485 760 байт после декодирования (решение 1) @@ -130,15 +131,17 @@ def sweep_downloads(root, now: float | None = None) -> list[str]: def sweep(conn, root, now: float | None = None) -> list[str]: - """Удалить файлы старше KEEP_DAYS (по mtime), на которые не ссылается ни одна ensure_attachment в queued или - claimed; вернуть имена удалённых. Правило одно для всех обычных файлов каталога: брошенные временные .*.tmp уходят + """Удалить файлы старше KEEP_DAYS (по mtime), на которые не ссылается ни одна команда в queued или claimed (ссылки + отдают модули: file_refs); вернуть имена удалённых. Правило одно для всех обычных файлов каталога: брошенные временные .*.tmp уходят тем же путём. Каталога нет — пусто.""" root = Path(root) if not root.is_dir(): return [] cutoff = (time.time() if now is None else now) - KEEP_DAYS * 86400 - active = {r[0] for r in conn.execute("SELECT json_extract(params, '$.file') FROM command " - "WHERE type = 'ensure_attachment' AND status IN ('queued', 'claimed')")} + active = set() + for m in registry.modules(): + if m.file_refs: + active |= m.file_refs(conn) removed = [] for p in sorted(root.iterdir()): if p.name in active or not p.is_file(): diff --git a/src/jira_hub/core/modules.py b/src/jira_hub/core/modules.py index fe824cb..4d9d290 100644 --- a/src/jira_hub/core/modules.py +++ b/src/jira_hub/core/modules.py @@ -21,8 +21,8 @@ class CommandHooks: temp_definer: tuple[str, str, str] | None # (тип, поле params с временным ключом, поле result с ключом) temp_keys_used: Callable[[dict], list[str]] substitute: Callable[[str, dict, dict], dict] - drop_unresolved: Callable[[dict, list[str]], list[str]] # поправить params команды, вернуть неразрешённые - retry_params: Callable[[dict], dict] + drop_unresolved: Callable[[dict, list[str]], list[str]] # (cmd, неразрешённые): поправить cmd["params"], вернуть оставшиеся + retry_params: Callable[[dict], dict] # cmd → params нового пакета check_confirmed: Callable[[Connection, dict], tuple[bool, dict | None]] on_claim: Callable[[Connection, dict, str, bool], None] # cmd, now, redelivery on_failed: Callable[[Connection, dict, str], None] # исчерпаны выдачи diff --git a/src/jira_hub/core/projects.py b/src/jira_hub/core/projects.py index 07562e3..1315aac 100644 --- a/src/jira_hub/core/projects.py +++ b/src/jira_hub/core/projects.py @@ -21,7 +21,7 @@ class ProjectExists(ProjectError): super().__init__("label", f"проект с такой меткой уже есть: {label}") -def _label_error(label) -> str | None: +def label_error(label) -> str | None: if not isinstance(label, str) or not label: return "ожидается непустая строка" if len(label) > MAX_LEN: @@ -33,7 +33,7 @@ def _label_error(label) -> str | None: def validate(label, name, board_id) -> tuple[str, str, int | None]: """Проверить поля проекта; вернуть (метка, название без пробелов по краям, доска).""" - error = _label_error(label) + error = label_error(label) if error: raise ProjectError("label", error) if not isinstance(name, str) or not name.strip(): diff --git a/src/jira_hub/core/queue.py b/src/jira_hub/core/queue.py index 09a939f..0131edc 100644 --- a/src/jira_hub/core/queue.py +++ b/src/jira_hub/core/queue.py @@ -1,16 +1,15 @@ -"""Очередь команд: пакеты, выдача агенту по одной, отчёты, отмена, повтор, сверка с зеркалом, автообновление. -Новых команд хаб сам не создаёт: повтор после failed — новый пакет по запросу человека. Команду, выданную агенту и -оставшуюся без отчёта, хаб довыдаёт при следующем claim — это доставка той же выдачи (bridge-reliability, решение 1).""" +"""Очередь команд: пакеты, выдача агенту по одной, отчёты, отмена, повтор, сверка с зеркалом. Механика — здесь; смысл +типов команд (проверки постановки, паузы, журнал синхронизаций, отчёты) даёт модуль через CommandHooks (hub-modules, +решение 4). Новых команд хаб сам не создаёт: повтор после failed — новый пакет по запросу человека. Команду, выданную +агенту и оставшуюся без отчёта, хаб довыдаёт при следующем claim — это доставка той же выдачи (bridge-reliability, +решение 1).""" import json import logging from datetime import timedelta from . import backup as _backup # noqa: F401 (имя модуля занято параметром backup в claim) -from ..modules.jira import commands as C -from . import db, files, projects, settings -from ..modules.jira import attachments, history, mirror -from ..modules.jira import projects as jira_projects -from ..modules.jira import settings as jira_settings # временно: R-1, до Task 4 +from . import db, registry, settings +from .errors import CommandError log = logging.getLogger(__name__) @@ -18,7 +17,6 @@ TERMINAL = {"done", "failed", "skipped", "cancelled", "confirmed_by_state"} MAX_DELIVERIES = 3 # выдач без отчёта, после которых команда — failed (bridge-reliability, решение 2) ERROR_LIMIT = 1500 # символов ошибки в command.error (spec commands) ONLINE_MINUTES = 5 -READ_TYPES = {"refresh", "fetch_attachment"} # не пишут в Jira: без хвостового refresh и без резервной копии class ClaimedError(Exception): @@ -27,17 +25,6 @@ class ClaimedError(Exception): self.claimed_at = claimed_at -class TooLarge(C.CommandError): - """Вложение больше предела хаба (fetch_attachment) — API отвечает 413 too_large; для POST /batches это 422.""" - - -class AgentTooOld(ValueError): - """Агент ещё не умеет fetch_attachment — API отвечает 409 agent_too_old.""" - - -FETCH_AGENT_VERSION = 14 # первая версия агента с fetch_attachment - - class NotClaimedError(Exception): """Отчёт по команде, которую агенту не выдавали (status = queued) — API отвечает 409.""" @@ -58,163 +45,80 @@ def _dump(value) -> str: return json.dumps(value, ensure_ascii=False) +def _hooks(ctype): + """Обработчики модуля, владеющего типом команды; None — тип никому не известен.""" + m = registry.module_of(ctype) + return m.commands if m else None + + +def _read_types() -> set[str]: + return {t for m in registry.modules() if m.commands for t in m.commands.read_types} + + # --- пакеты ----------------------------------------------------------------- -def _check_setup(conn, cmds, files_dir) -> None: - """Проверки постановки, которым нужны настройки хаба и хранилище файлов (bug-and-attachments, решения 6–7): 422 - до создания пакета. Номер команды — как в присланном пакете (хвостовой refresh ещё не добавлен). - files_dir=None — хранилища нет: любой файл «не загружен».""" - bug = (settings.get(conn, "issue_types") or {}).get("bug") - bug_ok = isinstance(bug, dict) and bool(bug.get("name") or bug.get("id")) - for i, c in enumerate(cmds, 1): - p = c["params"] - if c["type"] == "ensure_issue" and p["issue_type"] == "bug" and not bug_ok: - raise C.CommandError(f"команда {i}: нет настройки issue_types.bug") - if c["type"] == "ensure_attachment": - if not files.is_sha(p["file"]): - raise C.CommandError(f"команда {i}: file — sha256 из POST /files (64 символа 0-9a-f)") - if files_dir is None or not files.exists(files_dir, p["file"]): - raise C.CommandError(f"команда {i}: файл {p['file']} не загружен: POST /files") - if not p["filename"].strip(): - raise C.CommandError(f"команда {i}: filename пустой") - if "/" in p["filename"] or "\\" in p["filename"]: - raise C.CommandError(f"команда {i}: filename без / и \\ — имя вложения, а не путь") - if '"' in p["filename"] or any(ord(ch) < 32 or ord(ch) == 127 for ch in p["filename"]): - # кавычка, CR, LF ломают заголовок части multipart/form-data у агента (R-2) - raise C.CommandError(f"команда {i}: filename без кавычек и управляющих символов (CR, LF и др.)") - if c["type"] == "fetch_attachment": - if not p["attachment_id"].isdigit(): - raise C.CommandError(f"команда {i}: attachment_id — цифры id вложения из issue") - row = conn.execute("SELECT size FROM issue_attachment WHERE attachment_id = ? AND key = ?", - (p["attachment_id"], p["key"])).fetchone() - if row is None: - raise C.CommandError(f"у {p['key']} нет вложения #{p['attachment_id']} в зеркале — обновите задачу") - if row["size"] is not None and row["size"] > files.MAX_SIZE: # size NULL — размер неизвестен, ставим - raise TooLarge(f"файл больше 10 МБ ({attachments.fmt_size(row['size'])}) — скачайте в Jira") - _assign_boards(conn, cmds) - - -def _assign_boards(conn, cmds) -> None: - """Доска проекта задачи в params ensure_sprint (design #7): метки временного ключа — из его ensure_issue в пакете, - иначе из зеркала; 422 — задача вне проектов, разные доски, нет доски, канбан, тип ещё не известен.""" - temp_labels = {c["params"]["temp_key"]: c["params"].get("labels") or [] for c in cmds if c["type"] == "ensure_issue"} - found = projects.light_list(conn) - for i, c in enumerate(cmds, 1): - if c["type"] != "ensure_sprint": - continue - key = c["params"]["key"] - labels = temp_labels[key] if key in temp_labels else (mirror.labels_of(conn, key) or []) - own = [p for p in found if p["label"] in labels] - if not own: - raise C.CommandError(f"команда {i}: {key} не входит ни в один проект — доску для спринта не определить") - if len({p["board_id"] for p in own}) > 1: - raise C.CommandError(f"команда {i}: {key} в проектах {', '.join(p['label'] for p in own)} с разными досками") - p = own[0] - if p["board_id"] is None: - raise C.CommandError(f"команда {i}: у проекта {p['label']} нет доски Jira") - if p["board_type"] == "kanban": - raise C.CommandError(f"команда {i}: у проекта {p['label']} канбан-доска, спринтов нет") - if p["board_type"] is None: - raise C.CommandError(f"команда {i}: тип доски проекта {p['label']} ещё не известен — дождитесь обновления из Jira") - c["params"] = {**c["params"], "board_id": p["board_id"]} - - -def _check_project(conn, cmds, project) -> list[dict]: - """Проверки пакета проекта (spec commands, «Пакет проекта»): новые словари команд — входные params не мутируются. - Номер команды — как в присланном пакете.""" - if projects.get(conn, project) is None: - raise C.CommandError(f"проекта {project} нет") - others = {l.casefold() for l in projects.labels(conn) if l != project} - temps = {c["params"]["temp_key"] for c in cmds if c["type"] == "ensure_issue"} - - def own(key): - return key in temps or project in (mirror.labels_of(conn, key) or []) - - out = [] - for i, c in enumerate(cmds, 1): - t, p = c["type"], dict(c["params"]) - if t == "ensure_issue": - labels = [] - for l in p.get("labels") or []: - if l.casefold() in others: - raise C.CommandError(f"команда {i}: метка другого проекта: {l}") - l = project if l.casefold() == project.casefold() else l - if l not in labels or l != project: - labels.append(l) - if project not in labels: - labels.append(project) - p["labels"] = labels - elif t == "ensure_link": - if not (own(p["source"]) or own(p["target"])): - raise C.CommandError(f"команда {i}: связь {p['source']} → {p['target']} не касается проекта {project}") - elif t not in ("refresh", "fetch_attachment"): # fetch_attachment — не запись в Jira, проекта не касается - if t == "ensure_fields": - if project.casefold() in {l.casefold() for l in p.get("labels_remove") or []}: - raise C.CommandError(f"команда {i}: снять метку своего проекта {project} нельзя") - added = [] - for l in p.get("labels_add") or []: - if l.casefold() in others: - raise C.CommandError(f"команда {i}: метка другого проекта: {l}") - l = project if l.casefold() == project.casefold() else l - if l not in added: - added.append(l) - if "labels_add" in p: - p["labels_add"] = added - if not own(p["key"]): - raise C.CommandError(f"команда {i}: {p['key']} не в проекте {project}") - out.append({**c, "params": p}) - return out +def _batch_hooks(cmds): + """Обработчики модуля пакета. Типы из разных модулей в одном пакете нельзя; ни одного известного типа — первый + модуль реестра: его validate выдаёт ошибку о неизвестном типе.""" + owners = [] + for c in cmds if isinstance(cmds, list) else []: + m = registry.module_of(c["type"]) if isinstance(c, dict) and isinstance(c.get("type"), str) else None + if m is not None and m not in owners: + owners.append(m) + if len(owners) > 1: + raise CommandError("команды разных модулей в одном пакете") + if owners: + return owners[0].commands + return next(m.commands for m in registry.modules() if m.commands) def create_batch(conn, title, author, cmds, files_dir=None, project=None) -> dict: - cmds = C.validate(cmds) - if project is not None: - cmds = _check_project(conn, cmds, project) - _check_setup(conn, cmds, files_dir) - keys = [] - for c in cmds: - keys += [k for k in C.referenced_keys(c) if k not in keys] - if any(c["type"] not in READ_TYPES for c in cmds) and keys: - cmds.append({"type": "refresh", "params": {"keys": keys}}) - return _insert_batch(conn, title, author, cmds, project) + h = _batch_hooks(cmds) + cmds = h.validate(cmds) + cmds = h.check_batch(conn, cmds, project, files_dir) + tail = h.tail(cmds) + if tail is not None: + cmds.append(tail) + return insert_batch(conn, title, author, cmds, project) -def _insert_batch(conn, title, author, cmds, project=None) -> dict: +def estimate(conn, types) -> int: + """Оценка длительности, с: сумма пауз типов (паузу даёт модуль-владелец типа).""" + return sum(_pause(conn, t) for t in types) + + +def _pause(conn, ctype) -> int: + h = _hooks(ctype) + return int(h.pause(conn, ctype)) if h else 0 + + +def insert_batch(conn, title, author, cmds, project=None) -> dict: """Вставка пакета без проверок и хвоста; транзакция своя (вложенная идёт через SAVEPOINT).""" - est = C.estimate([c["type"] for c in cmds], jira_settings.pacing(conn)) + est = estimate(conn, [c["type"] for c in cmds]) with db.transaction(conn): cur = conn.execute("INSERT INTO batch(title, author, created_at, estimate_seconds, project) VALUES (?, ?, ?, ?, ?)", (title, author or "api", db.now(), est, project)) batch_id = cur.lastrowid defined = {} for seq, c in enumerate(cmds, 1): - deps = [defined[k] for k in C.temp_keys_used(c) if k in defined] + h = _hooks(c["type"]) + deps = [defined[k] for k in (h.temp_keys_used(c) if h else []) if k in defined] cur = conn.execute("INSERT INTO command(batch_id, seq, type, params, depends_on) VALUES (?, ?, ?, ?, ?)", (batch_id, seq, c["type"], _dump(c["params"]), deps[0] if deps else None)) - if c["type"] == "ensure_issue": - defined[c["params"]["temp_key"]] = cur.lastrowid + if h and h.temp_definer and c["type"] == h.temp_definer[0]: + defined[c["params"][h.temp_definer[1]]] = cur.lastrowid return {"id": batch_id, "estimate_seconds": est, "count": len(cmds)} -def request_fetch(conn, files_dir, attachment_id: str, author: str | None) -> dict: - """Поставить скачивание вложения. Проверка «уже качается/скачано» и вставка пакета — в одной транзакции с - блокировкой записи (BEGIN IMMEDIATE): два одновременных запроса дают одну команду (решение 7). - Ответ — как attachments.get плюс command_id, batch_id. KeyError — вложения нет в зеркале; TooLarge, AgentTooOld.""" - attachment_id = str(attachment_id) - with db.transaction(conn): - info = attachments.get(conn, files_dir, attachment_id) - if info is None: - raise KeyError(attachment_id) - if info["state"] in ("ready", "fetching"): - return info - agent = conn.execute("SELECT agent_version FROM agent_status WHERE id = 1").fetchone() - v = agent["agent_version"] if agent else None - 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 = C.validate([{"type": "fetch_attachment", "params": {"key": info["key"], "attachment_id": attachment_id}}]) - _check_setup(conn, cmds, files_dir) - _insert_batch(conn, f"Скачать вложение {info['filename']} ({info['key']})", author, cmds) - return attachments.get(conn, files_dir, attachment_id) +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 agent_version(conn): + """Версия агента из его последней выдачи; None — агент не отчитывался.""" + row = conn.execute("SELECT agent_version FROM agent_status WHERE id = 1").fetchone() + return row["agent_version"] if row else None def _status_of(cmds, cancelled_at=None) -> str: @@ -233,11 +137,10 @@ def batch_view(conn, batch_id, with_commands=True) -> dict | None: if b is None: return None cmds = [_cmd(r) for r in conn.execute("SELECT * FROM command WHERE batch_id = ? ORDER BY seq", (batch_id,))] - pacing = jira_settings.pacing(conn) d = dict(b) d["status"] = _status_of(cmds, b["cancelled_at"]) d["counts"] = {s: sum(1 for c in cmds if c["status"] == s) for s in sorted({c["status"] for c in cmds})} - d["remaining_seconds"] = C.estimate([c["type"] for c in cmds if c["status"] in ("queued", "claimed")], pacing) + d["remaining_seconds"] = estimate(conn, [c["type"] for c in cmds if c["status"] in ("queued", "claimed")]) if with_commands: d["commands"] = cmds return d @@ -251,23 +154,25 @@ def list_batches(conn, limit=20, project=None) -> list[dict]: # --- выдача ----------------------------------------------------------------- -def _mapping(conn, batch_id) -> dict: - """temp_key → настоящий ключ по завершённым ensure_issue пакета.""" - rows = conn.execute("SELECT json_extract(params, '$.temp_key') AS t, json_extract(result, '$.key') AS k FROM command " - "WHERE batch_id = ? AND type = 'ensure_issue' AND status IN ('done', 'confirmed_by_state')", - (batch_id,)).fetchall() +def _mapping(conn, batch_id, definer) -> dict: + """temp_key → настоящий ключ по завершённым командам-определителям пакета.""" + ctype, in_field, out_field = definer + rows = conn.execute(f"SELECT json_extract(params, '$.{in_field}') AS t, json_extract(result, '$.{out_field}') AS k " + "FROM command WHERE batch_id = ? AND type = ? AND status IN ('done', 'confirmed_by_state')", + (batch_id, ctype)).fetchall() return {r["t"]: r["k"] for r in rows if r["k"]} -def _definer_status(conn, batch_id, temp_key): - r = conn.execute("SELECT status FROM command WHERE batch_id = ? AND type = 'ensure_issue' " - "AND json_extract(params, '$.temp_key') = ?", (batch_id, temp_key)).fetchone() +def _definer_status(conn, batch_id, definer, temp_key): + ctype, in_field, _ = definer + r = conn.execute(f"SELECT status FROM command WHERE batch_id = ? AND type = ? " + f"AND json_extract(params, '$.{in_field}') = ?", (batch_id, ctype, temp_key)).fetchone() return r["status"] if r else "failed" -def _skip_reason(conn, batch_id, unresolved) -> str: +def _skip_reason(conn, batch_id, definer, unresolved) -> str: """Текст skipped для зависимой команды: отменённое создание называется отменой (O-10 #16, решение 10).""" - cancelled = [k for k in unresolved if _definer_status(conn, batch_id, k) == "cancelled"] + cancelled = [k for k in unresolved if _definer_status(conn, batch_id, definer, k) == "cancelled"] return f"создание задачи {cancelled[0]} отменено" if cancelled else f"задача {unresolved[0]} не создана" @@ -276,19 +181,18 @@ def _next(conn) -> dict | None: "WHERE c.status = 'queued' AND b.cancelled_at IS NULL ORDER BY b.created_at, b.id, c.seq").fetchall() for row in rows: cmd = _cmd(row) - needed = C.temp_keys_used(cmd) + h = _hooks(cmd["type"]) + needed = h.temp_keys_used(cmd) if h and h.temp_definer else [] if needed: - mapping = _mapping(conn, cmd["batch_id"]) + mapping = _mapping(conn, cmd["batch_id"], h.temp_definer) unresolved = [k for k in needed if k not in mapping] - if any(_definer_status(conn, cmd["batch_id"], k) not in TERMINAL for k in unresolved): + if any(_definer_status(conn, cmd["batch_id"], h.temp_definer, k) not in TERMINAL for k in unresolved): return None # создание ещё не завершено — ждём (одна в полёте) - cmd["params"] = C.substitute(cmd["type"], cmd["params"], mapping) - if cmd["type"] == "refresh": - cmd["params"]["keys"] = [k for k in cmd["params"].get("keys", []) if not C.is_temp(k)] - unresolved = [] if cmd["params"]["keys"] else unresolved + cmd["params"] = h.substitute(cmd["type"], cmd["params"], mapping) + unresolved = h.drop_unresolved(cmd, unresolved) if unresolved: conn.execute("UPDATE command SET status = 'skipped', finished_at = ?, error = ? WHERE id = ?", - (db.now(), _skip_reason(conn, cmd["batch_id"], unresolved), cmd["id"])) + (db.now(), _skip_reason(conn, cmd["batch_id"], h.temp_definer, unresolved), cmd["id"])) continue return cmd return None @@ -306,26 +210,18 @@ def _redeliver(conn, now) -> dict | None: return cmd conn.execute("UPDATE command SET status = 'failed', finished_at = ?, error = ? WHERE id = ?", (now, f"агент не отчитался после {MAX_DELIVERIES} выдач", cmd["id"])) - if cmd["type"] == "refresh": # загрузка закончилась ничем — журнал синхронизаций закрыт (M-4 #12) - conn.execute("UPDATE sync_run SET finished_at = ? WHERE command_id = ?", (now, cmd["id"])) + h = _hooks(cmd["type"]) + if h: + h.on_failed(conn, cmd, now) conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1 AND current_command_id = ?", (cmd["id"],)) log.warning("команда %s: агент не отчитался после %s выдач — failed", cmd["id"], MAX_DELIVERIES) def deliver_params(conn, cmd, files_dir) -> dict: - """Параметры команды для агента: ensure_attachment получает size — текущий размер файла в хранилище или None, - если файла уже нет (решение 5); refresh — boards, доски проектов (design #7), и history — отметка журнала - (change-history). В базе params остаются без них: повтор команды проходит проверку как новая.""" - if cmd["type"] == "refresh": - out = {**cmd["params"], "boards": jira_projects.boards(conn)} - hist = history.delivery_param(conn, cmd["params"]) # отметка журнала (change-history, design #5) - if hist is not None: - out["history"] = hist - return out - if cmd["type"] != "ensure_attachment": - return cmd["params"] - size = files.size(files_dir, cmd["params"].get("file")) if files_dir is not None else None - return {**cmd["params"], "size": size} + """Параметры команды для агента: дописывает модуль-владелец типа (в базе params остаются без дописанного: повтор + команды проходит проверку как новая).""" + h = _hooks(cmd["type"]) + 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) -> dict | None: @@ -337,13 +233,12 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non (now, None if agent_version is None else str(agent_version), settings_version, last_jira_call_at)) cmd = _redeliver(conn, now) if cmd is not None: - pause = int(jira_settings.pacing(conn).get(cmd["type"], 0)) + pause = _pause(conn, cmd["type"]) conn.execute("UPDATE command SET claimed_at = ?, pause_seconds = ?, deliveries = deliveries + 1 WHERE id = ?", (now, pause, cmd["id"])) - if cmd["type"] == "refresh": - # повторное выполнение загрузит то же заново: счётчики с нуля; started_at — граница пометки соседей, - # не сдвигается (M-5 #12) - conn.execute("UPDATE sync_run SET issues_count = 0, links_count = 0 WHERE command_id = ?", (cmd["id"],)) + h = _hooks(cmd["type"]) + if h: + h.on_claim(conn, cmd, now, True) conn.execute("UPDATE agent_status SET current_command_id = ? WHERE id = 1", (cmd["id"],)) log.info("команда %s выдана снова (выдача %s)", cmd["id"], cmd["deliveries"] + 1) else: @@ -351,18 +246,18 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non if cmd is None: conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1") return None - pause = int(jira_settings.pacing(conn).get(cmd["type"], 0)) + pause = _pause(conn, cmd["type"]) first = conn.execute("SELECT COUNT(*) AS n FROM command WHERE batch_id = ? AND status != 'queued'", (cmd["batch_id"],)).fetchone()["n"] == 0 - has_writes = conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type NOT IN ('refresh', 'fetch_attachment') LIMIT 1", - (cmd["batch_id"],)).fetchone() is not None + read = sorted(_read_types()) + has_writes = conn.execute(f"SELECT 1 FROM command WHERE batch_id = ? AND type NOT IN ({','.join('?' * len(read))}) LIMIT 1", + (cmd["batch_id"], *read)).fetchone() is not None conn.execute("UPDATE command SET status = 'claimed', claimed_at = ?, pause_seconds = ?, params = ?, deliveries = 1 " "WHERE id = ?", (now, pause, _dump(cmd["params"]), cmd["id"])) conn.execute("UPDATE agent_status SET current_command_id = ? WHERE id = 1", (cmd["id"],)) - if cmd["type"] == "refresh": - p = cmd["params"] - conn.execute("INSERT INTO sync_run(started_at, kind, query, command_id) VALUES (?, ?, ?, ?)", - (now, "jql" if p.get("jql") else "keys", p.get("jql") or ",".join(p.get("keys", [])), cmd["id"])) + h = _hooks(cmd["type"]) + if h: + h.on_claim(conn, cmd, now, False) if first and has_writes: to_backup = f"batch-{cmd['batch_id']}" if to_backup and backup: @@ -376,20 +271,6 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non # --- отчёты ----------------------------------------------------------------- -def is_full_refresh(conn, command_id) -> bool: - """Команда — refresh по базовой JQL (полный прогон): только он вправе сбрасывать флаг neighbor.""" - if not command_id: - return False - cmd = get_command(conn, command_id) - base = jira_projects.base_jql(conn) - return bool(base and cmd and cmd["type"] == "refresh" and cmd["params"].get("jql") == base) - - -def record_state(conn, command_id, issues_count, links_count) -> None: - conn.execute("UPDATE sync_run SET issues_count = issues_count + ?, links_count = links_count + ? WHERE command_id = ?", - (int(issues_count), int(links_count), command_id)) - - def report_result(conn, command_id, status, result=None, error=None, jira_calls=0, rate_limited=0, files_dir=None) -> dict: if status not in ("done", "failed"): raise ValueError("status — done или failed") @@ -400,62 +281,38 @@ def report_result(conn, command_id, status, result=None, error=None, jira_calls= raise KeyError(command_id) if cmd["status"] == "queued": raise NotClaimedError(f"команда {command_id} не выдана") + h = _hooks(cmd["type"]) if cmd["status"] in TERMINAL: - # поздний отчёт о созданной задаче по команде, которую хаб уже закрыл (force-отмена — M-5, предел выдач — - # M-4 #12): ключ не теряем, статус не трогаем - if (cmd["status"] in ("cancelled", "failed") and cmd["type"] == "ensure_issue" and status == "done" - and result and result.get("key") and not cmd["result"]): - conn.execute("UPDATE command SET result = ?, jira_calls = ?, rate_limited = ? WHERE id = ?", - (_dump(result), int(jira_calls or 0), int(rate_limited or 0), command_id)) - mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"]) - log.warning("команда %s уже %s, но агент создал %s — ключ записан в результат", - command_id, cmd["status"], result["key"]) + # поздний отчёт по команде, которую хаб уже закрыл (force-отмена, предел выдач): модуль может принять данные, + # статус не трогаем + if h and h.on_late_result(conn, cmd, status, result, jira_calls, rate_limited): return get_command(conn, command_id) return cmd # идемпотентно; терминальный статус (cancelled/failed) не затирается now = db.now() - fetched = None - if cmd["type"] == "fetch_attachment" and status == "done": - # файл должен лежать в хранилище с заявленным размером; иначе отчёт не принимаем (уборка успела раньше, чужой размер) - sha = result.get("file") if isinstance(result, dict) else None - real = files.size(files_dir, sha) if files_dir is not None else None - if real is None or real != result.get("size"): - status, error = "failed", f"хаб: файл {sha} не найден в хранилище"[:ERROR_LIMIT] - else: - fetched = (cmd["params"]["attachment_id"], sha, real) + if h: + status, why = h.check_result(conn, cmd, status, result, files_dir) + if why is not None: + error = why[:ERROR_LIMIT] conn.execute("UPDATE command SET status = ?, finished_at = ?, result = ?, error = ?, jira_calls = ?, rate_limited = ? WHERE id = ?", (status, now, _dump(result) if result is not None else None, error, int(jira_calls or 0), int(rate_limited or 0), command_id)) conn.execute("UPDATE agent_status SET current_command_id = NULL, last_seen_at = ? WHERE id = 1 AND current_command_id = ?", (now, command_id)) - if cmd["type"] == "ensure_issue" and status == "done" and result and result.get("key"): - mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"]) - if fetched: - conn.execute("INSERT OR REPLACE INTO attachment_file(attachment_id, file, size, fetched_at) VALUES (?, ?, ?, ?)", - (*fetched, now)) - if cmd["type"] == "refresh": - conn.execute("UPDATE sync_run SET finished_at = ?, jira_calls = ?, rate_limited = ? WHERE command_id = ?", - (now, int(jira_calls or 0), int(rate_limited or 0), command_id)) - if isinstance(result, dict) and "boards" in result: - jira_projects.record_board_types(conn, result["boards"]) # в любом статусе; без ключа (агент v11) типы не трогаем - if status == "done" and is_full_refresh(conn, command_id) and history.accepted(result): - conn.execute("UPDATE sync_run SET history_at = ? WHERE command_id = ?", (now, command_id)) - if status == "done" and is_full_refresh(conn, command_id): - # полный прогон не обновил задачу — в базовую JQL она больше не входит: сосед. Пустой прогон (0 задач) — - # почти всегда сбой выборки, а не «все задачи ушли»: соседей не трогаем (M-12) - run = conn.execute("SELECT started_at, issues_count FROM sync_run WHERE command_id = ?", (command_id,)).fetchone() - if run and run["issues_count"] > 0: - conn.execute("UPDATE issue SET neighbor = 1 WHERE fetched_at < ?", (run["started_at"],)) - if cmd["type"] == "refresh" and status == "done": - reconcile(conn) + if h: + h.on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_dir) + if h: + h.after_result(conn, cmd, status) return get_command(conn, command_id) def reconcile(conn) -> list[int]: """Взятые команды без ответа, чей результат уже виден в зеркале → confirmed_by_state.""" - link_types = settings.get(conn, "link_types") or {} changed = [] for row in conn.execute("SELECT * FROM command WHERE status = 'claimed'").fetchall(): cmd = _cmd(row) - ok, result = C.check_confirmed(conn, cmd, link_types) + h = _hooks(cmd["type"]) + if h is None: + continue + ok, result = h.check_confirmed(conn, cmd) if not ok: continue with db.transaction(conn): @@ -504,8 +361,9 @@ def retry_command(conn, command_id, author, files_dir=None, project=None) -> dic raise KeyError(command_id) origin = conn.execute("SELECT project FROM batch WHERE id = ?", (cmd["batch_id"],)).fetchone()["project"] if project is not None and project != origin: - raise C.CommandError(f"команда #{command_id} не из проекта {project}") - params = {k: v for k, v in cmd["params"].items() if not (cmd["type"] == "ensure_sprint" and k == "board_id")} + raise CommandError(f"команда #{command_id} не из проекта {project}") + h = _hooks(cmd["type"]) + params = h.retry_params(cmd) if h else cmd["params"] return create_batch(conn, f"Повтор команды {command_id}", author, [{"type": cmd["type"], "params": params}], files_dir=files_dir, project=origin) @@ -523,18 +381,3 @@ def agent_view(conn) -> dict: 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")] return d - - -def maybe_auto_refresh(conn) -> dict | None: - jql = jira_projects.base_jql(conn) - if jql is None: - return None - minutes = int(settings.get(conn, "auto_refresh_minutes") or 0) - if minutes <= 0 or not agent_view(conn)["online"]: - return None - if conn.execute("SELECT 1 FROM command WHERE status IN ('queued', 'claimed') LIMIT 1").fetchone(): - 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): - return None - return create_batch(conn, "Автообновление", "hub", [{"type": "refresh", "params": {"jql": jql}}]) diff --git a/src/jira_hub/modules/jira/__init__.py b/src/jira_hub/modules/jira/__init__.py index fe1ff7f..2c8e766 100644 --- a/src/jira_hub/modules/jira/__init__.py +++ b/src/jira_hub/modules/jira/__init__.py @@ -1,6 +1,6 @@ """Модуль jira: зеркало задач Jira, очередь команд агенту, вложения, журнал изменений.""" from ...core.modules import Module -from . import schema, settings +from . import hooks, schema, settings MODULE = Module( name="jira", @@ -10,4 +10,6 @@ MODULE = Module( validate_setting=settings.validate_setting, settings_view=settings.settings_view, on_init=settings.on_init, + commands=hooks.HOOKS, + file_refs=hooks.file_refs, ) diff --git a/src/jira_hub/modules/jira/commands.py b/src/jira_hub/modules/jira/commands.py index a924645..eac51d6 100644 --- a/src/jira_hub/modules/jira/commands.py +++ b/src/jira_hub/modules/jira/commands.py @@ -171,10 +171,6 @@ def substitute(cmd_type, params, mapping: dict) -> dict: return out -def estimate(types, pacing: dict) -> int: - return sum(int(pacing.get(t, 0)) for t in types) - - def _norm_text(value): """Значение поля для сравнения с зеркалом: переводы строки CRLF и LF не различаются (описание, правленное в UI Jira, приходит с CRLF, O-9 #16); null и пустая строка равны (решение 11).""" diff --git a/src/jira_hub/modules/jira/fetch.py b/src/jira_hub/modules/jira/fetch.py new file mode 100644 index 0000000..1050eea --- /dev/null +++ b/src/jira_hub/modules/jira/fetch.py @@ -0,0 +1,39 @@ +"""Скачивание вложения из Jira по запросу (fetch_attachment): постановка в очередь и ошибки постановки.""" +from ...core import db, queue +from ...core.errors import CommandError, HttpMapped +from . import attachments, commands + +FETCH_AGENT_VERSION = 14 # первая версия агента с fetch_attachment + + +class TooLarge(CommandError, HttpMapped): + """Вложение больше предела хаба (fetch_attachment) — API отвечает 413 too_large; для POST /batches это 422.""" + status = 413 + error = "too_large" + + +class AgentTooOld(ValueError, HttpMapped): + """Агент ещё не умеет fetch_attachment — API отвечает 409 agent_too_old.""" + status = 409 + error = "agent_too_old" + + +def request_fetch(conn, files_dir, attachment_id: str, author: str | None) -> dict: + """Поставить скачивание вложения. Проверка «уже качается/скачано» и вставка пакета — в одной транзакции с + блокировкой записи (BEGIN IMMEDIATE): два одновременных запроса дают одну команду (решение 7). + Ответ — как attachments.get плюс command_id, batch_id. KeyError — вложения нет в зеркале; TooLarge, AgentTooOld.""" + from . import hooks # внутри функции: hooks импортирует это место ради TooLarge + attachment_id = str(attachment_id) + with db.transaction(conn): + info = attachments.get(conn, files_dir, attachment_id) + if info is None: + raise KeyError(attachment_id) + if info["state"] in ("ready", "fetching"): + return info + v = queue.agent_version(conn) + 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}}]) + hooks.check_setup(conn, cmds, files_dir) + queue.insert_batch(conn, f"Скачать вложение {info['filename']} ({info['key']})", author, cmds) + return attachments.get(conn, files_dir, attachment_id) diff --git a/src/jira_hub/modules/jira/hooks.py b/src/jira_hub/modules/jira/hooks.py new file mode 100644 index 0000000..9c9d8f4 --- /dev/null +++ b/src/jira_hub/modules/jira/hooks.py @@ -0,0 +1,267 @@ +"""Смысл типов команд jira для очереди ядра: проверки постановки, паузы, отчёты, журнал синхронизаций (решения 4, 12).""" +import json +import logging + +from ...core import files, projects, queue, settings +from ...core.errors import CommandError +from ...core.modules import CommandHooks +from . import attachments, commands, history, mirror, refresh +from . import projects as jira_projects +from . import settings as jira_settings +from .fetch import TooLarge + +log = logging.getLogger(__name__) + +READ_TYPES = frozenset({"refresh", "fetch_attachment"}) # не пишут в Jira: без хвостового refresh и без резервной копии + + +# --- постановка --------------------------------------------------------------- + +def check_setup(conn, cmds, files_dir) -> None: + """Проверки постановки, которым нужны настройки хаба и хранилище файлов (bug-and-attachments, решения 6–7): 422 + до создания пакета. Номер команды — как в присланном пакете (хвостовой refresh ещё не добавлен). + files_dir=None — хранилища нет: любой файл «не загружен».""" + bug = (settings.get(conn, "issue_types") or {}).get("bug") + bug_ok = isinstance(bug, dict) and bool(bug.get("name") or bug.get("id")) + for i, c in enumerate(cmds, 1): + p = c["params"] + if c["type"] == "ensure_issue" and p["issue_type"] == "bug" and not bug_ok: + raise CommandError(f"команда {i}: нет настройки issue_types.bug") + if c["type"] == "ensure_attachment": + if not files.is_sha(p["file"]): + raise CommandError(f"команда {i}: file — sha256 из POST /files (64 символа 0-9a-f)") + if files_dir is None or not files.exists(files_dir, p["file"]): + raise CommandError(f"команда {i}: файл {p['file']} не загружен: POST /files") + if not p["filename"].strip(): + raise CommandError(f"команда {i}: filename пустой") + if "/" in p["filename"] or "\\" in p["filename"]: + raise CommandError(f"команда {i}: filename без / и \\ — имя вложения, а не путь") + if '"' in p["filename"] or any(ord(ch) < 32 or ord(ch) == 127 for ch in p["filename"]): + # кавычка, CR, LF ломают заголовок части multipart/form-data у агента (R-2) + raise CommandError(f"команда {i}: filename без кавычек и управляющих символов (CR, LF и др.)") + if c["type"] == "fetch_attachment": + if not p["attachment_id"].isdigit(): + raise CommandError(f"команда {i}: attachment_id — цифры id вложения из issue") + row = conn.execute("SELECT size FROM issue_attachment WHERE attachment_id = ? AND key = ?", + (p["attachment_id"], p["key"])).fetchone() + if row is None: + raise CommandError(f"у {p['key']} нет вложения #{p['attachment_id']} в зеркале — обновите задачу") + if row["size"] is not None and row["size"] > files.MAX_SIZE: # size NULL — размер неизвестен, ставим + raise TooLarge(f"файл больше 10 МБ ({attachments.fmt_size(row['size'])}) — скачайте в Jira") + _assign_boards(conn, cmds) + + +def _assign_boards(conn, cmds) -> None: + """Доска проекта задачи в params ensure_sprint (design #7): метки временного ключа — из его ensure_issue в пакете, + иначе из зеркала; 422 — задача вне проектов, разные доски, нет доски, канбан, тип ещё не известен.""" + temp_labels = {c["params"]["temp_key"]: c["params"].get("labels") or [] for c in cmds if c["type"] == "ensure_issue"} + found = projects.light_list(conn) + for i, c in enumerate(cmds, 1): + if c["type"] != "ensure_sprint": + continue + key = c["params"]["key"] + labels = temp_labels[key] if key in temp_labels else (mirror.labels_of(conn, key) or []) + own = [p for p in found if p["label"] in labels] + if not own: + raise CommandError(f"команда {i}: {key} не входит ни в один проект — доску для спринта не определить") + if len({p["board_id"] for p in own}) > 1: + raise CommandError(f"команда {i}: {key} в проектах {', '.join(p['label'] for p in own)} с разными досками") + p = own[0] + if p["board_id"] is None: + raise CommandError(f"команда {i}: у проекта {p['label']} нет доски Jira") + if p["board_type"] == "kanban": + raise CommandError(f"команда {i}: у проекта {p['label']} канбан-доска, спринтов нет") + if p["board_type"] is None: + raise CommandError(f"команда {i}: тип доски проекта {p['label']} ещё не известен — дождитесь обновления из Jira") + c["params"] = {**c["params"], "board_id": p["board_id"]} + + +def _check_project(conn, cmds, project) -> list[dict]: + """Проверки пакета проекта (spec commands, «Пакет проекта»): новые словари команд — входные params не мутируются. + Номер команды — как в присланном пакете.""" + if projects.get(conn, project) is None: + raise CommandError(f"проекта {project} нет") + others = {l.casefold() for l in projects.labels(conn) if l != project} + temps = {c["params"]["temp_key"] for c in cmds if c["type"] == "ensure_issue"} + + def own(key): + return key in temps or project in (mirror.labels_of(conn, key) or []) + + out = [] + for i, c in enumerate(cmds, 1): + t, p = c["type"], dict(c["params"]) + if t == "ensure_issue": + labels = [] + for l in p.get("labels") or []: + if l.casefold() in others: + raise CommandError(f"команда {i}: метка другого проекта: {l}") + l = project if l.casefold() == project.casefold() else l + if l not in labels or l != project: + labels.append(l) + if project not in labels: + labels.append(project) + p["labels"] = labels + elif t == "ensure_link": + if not (own(p["source"]) or own(p["target"])): + raise CommandError(f"команда {i}: связь {p['source']} → {p['target']} не касается проекта {project}") + elif t not in READ_TYPES: # fetch_attachment — не запись в Jira, проекта не касается + if t == "ensure_fields": + if project.casefold() in {l.casefold() for l in p.get("labels_remove") or []}: + raise CommandError(f"команда {i}: снять метку своего проекта {project} нельзя") + added = [] + for l in p.get("labels_add") or []: + if l.casefold() in others: + raise CommandError(f"команда {i}: метка другого проекта: {l}") + l = project if l.casefold() == project.casefold() else l + if l not in added: + added.append(l) + if "labels_add" in p: + p["labels_add"] = added + if not own(p["key"]): + raise CommandError(f"команда {i}: {p['key']} не в проекте {project}") + out.append({**c, "params": p}) + return out + + +def check_batch(conn, cmds, project, files_dir) -> list[dict]: + if project is not None: + cmds = _check_project(conn, cmds, project) + check_setup(conn, cmds, files_dir) + return cmds + + +def tail(cmds) -> dict | None: + """Хвостовой refresh пишущего пакета: ключи из всех команд, чтобы зеркало увидело результат.""" + keys = [] + for c in cmds: + keys += [k for k in commands.referenced_keys(c) if k not in keys] + if any(c["type"] not in READ_TYPES for c in cmds) and keys: + return {"type": "refresh", "params": {"keys": keys}} + return None + + +def pause(conn, ctype) -> int: + return int(jira_settings.pacing(conn).get(ctype, 0)) + + +# --- выдача ------------------------------------------------------------------- + +def deliver_params(conn, cmd, files_dir) -> dict: + """Параметры команды для агента: ensure_attachment получает size — текущий размер файла в хранилище или None, + если файла уже нет (решение 5); refresh — boards, доски проектов (design #7), и history — отметка журнала + (change-history). В базе params остаются без них: повтор команды проходит проверку как новая.""" + if cmd["type"] == "refresh": + out = {**cmd["params"], "boards": jira_projects.boards(conn)} + hist = history.delivery_param(conn, cmd["params"]) # отметка журнала (change-history, design #5) + if hist is not None: + out["history"] = hist + return out + if cmd["type"] != "ensure_attachment": + return cmd["params"] + size = files.size(files_dir, cmd["params"].get("file")) if files_dir is not None else None + return {**cmd["params"], "size": size} + + +def drop_unresolved(cmd, unresolved) -> list[str]: + """refresh с неразрешёнными временными ключами выполняется по остальным (временные выбрасываются из keys); остался + хоть один ключ — неразрешённых нет. cmd["params"] — уже после подстановки.""" + if cmd["type"] == "refresh": + cmd["params"]["keys"] = [k for k in cmd["params"].get("keys", []) if not commands.is_temp(k)] + return [] if cmd["params"]["keys"] else unresolved + return unresolved + + +def on_claim(conn, cmd, now, redelivery) -> None: + if cmd["type"] != "refresh": + return + if redelivery: + # повторное выполнение загрузит то же заново: счётчики с нуля; started_at — граница пометки соседей, + # не сдвигается (M-5 #12) + conn.execute("UPDATE sync_run SET issues_count = 0, links_count = 0 WHERE command_id = ?", (cmd["id"],)) + else: + p = cmd["params"] + conn.execute("INSERT INTO sync_run(started_at, kind, query, command_id) VALUES (?, ?, ?, ?)", + (now, "jql" if p.get("jql") else "keys", p.get("jql") or ",".join(p.get("keys", [])), cmd["id"])) + + +def on_failed(conn, cmd, now) -> None: + if cmd["type"] == "refresh": # загрузка закончилась ничем — журнал синхронизаций закрыт (M-4 #12) + conn.execute("UPDATE sync_run SET finished_at = ? WHERE command_id = ?", (now, cmd["id"])) + + +# --- отчёты ------------------------------------------------------------------- + +def check_result(conn, cmd, status, result, files_dir) -> tuple[str, str | None]: + """fetch_attachment: файл должен лежать в хранилище с заявленным размером; иначе отчёт не принимаем (уборка успела + раньше, чужой размер). Ошибка None — отчёт принят как есть.""" + if cmd["type"] == "fetch_attachment" and status == "done": + sha = result.get("file") if isinstance(result, dict) else None + real = files.size(files_dir, sha) if files_dir is not None else None + if real is None or real != result.get("size"): + return "failed", f"хаб: файл {sha} не найден в хранилище" + return status, None + + +def on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_dir) -> None: + ctype = cmd["type"] + if ctype == "ensure_issue" and status == "done" and result and result.get("key"): + mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"]) + if ctype == "fetch_attachment" and status == "done": # check_result уже сверил файл и размер + conn.execute("INSERT OR REPLACE INTO attachment_file(attachment_id, file, size, fetched_at) VALUES (?, ?, ?, ?)", + (cmd["params"]["attachment_id"], result["file"], result["size"], now)) + if ctype == "refresh": + conn.execute("UPDATE sync_run SET finished_at = ?, jira_calls = ?, rate_limited = ? WHERE command_id = ?", + (now, int(jira_calls or 0), int(rate_limited or 0), cmd["id"])) + if isinstance(result, dict) and "boards" in result: + jira_projects.record_board_types(conn, result["boards"]) # в любом статусе; без ключа (агент v11) типы не трогаем + full = status == "done" and refresh.is_full_refresh(conn, cmd["id"]) + if full and history.accepted(result): + conn.execute("UPDATE sync_run SET history_at = ? WHERE command_id = ?", (now, cmd["id"])) + if full: + # полный прогон не обновил задачу — в базовую JQL она больше не входит: сосед. Пустой прогон (0 задач) — + # почти всегда сбой выборки, а не «все задачи ушли»: соседей не трогаем (M-12) + run = conn.execute("SELECT started_at, issues_count FROM sync_run WHERE command_id = ?", (cmd["id"],)).fetchone() + if run and run["issues_count"] > 0: + conn.execute("UPDATE issue SET neighbor = 1 WHERE fetched_at < ?", (run["started_at"],)) + + +def after_result(conn, cmd, status) -> None: + if cmd["type"] == "refresh" and status == "done": + queue.reconcile(conn) + + +def on_late_result(conn, cmd, status, result, jira_calls, rate_limited) -> bool: + """Поздний отчёт о созданной задаче по команде, которую хаб уже закрыл (force-отмена — M-5, предел выдач — M-4 #12): + ключ не теряем, статус не трогаем.""" + if (cmd["status"] in ("cancelled", "failed") and cmd["type"] == "ensure_issue" and status == "done" + and result and result.get("key") and not cmd["result"]): + conn.execute("UPDATE command SET result = ?, jira_calls = ?, rate_limited = ? WHERE id = ?", + (json.dumps(result, ensure_ascii=False), int(jira_calls or 0), int(rate_limited or 0), cmd["id"])) + mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"]) + log.warning("команда %s уже %s, но агент создал %s — ключ записан в результат", + cmd["id"], cmd["status"], result["key"]) + return True + return False + + +def check_confirmed(conn, cmd) -> tuple[bool, dict | None]: + return commands.check_confirmed(conn, cmd, settings.get(conn, "link_types") or {}) + + +def retry_params(cmd) -> dict: + """Повтор ensure_sprint берёт доску заново (она в params ставится при постановке).""" + return {k: v for k, v in cmd["params"].items() if not (cmd["type"] == "ensure_sprint" and k == "board_id")} + + +def file_refs(conn) -> set[str]: + """Файлы, на которые ссылаются ensure_attachment в queued или claimed: уборка их не трогает.""" + return {r[0] for r in conn.execute("SELECT json_extract(params, '$.file') FROM command " + "WHERE type = 'ensure_attachment' AND status IN ('queued', 'claimed')")} + + +HOOKS = CommandHooks( + types=frozenset(commands.SPEC), read_types=READ_TYPES, validate=commands.validate, check_batch=check_batch, + tail=tail, pause=pause, deliver_params=deliver_params, temp_definer=("ensure_issue", "temp_key", "key"), + temp_keys_used=commands.temp_keys_used, substitute=commands.substitute, drop_unresolved=drop_unresolved, + retry_params=retry_params, check_confirmed=check_confirmed, on_claim=on_claim, on_failed=on_failed, + check_result=check_result, on_result=on_result, after_result=after_result, on_late_result=on_late_result) diff --git a/src/jira_hub/modules/jira/projects.py b/src/jira_hub/modules/jira/projects.py index 3c97813..66231c0 100644 --- a/src/jira_hub/modules/jira/projects.py +++ b/src/jira_hub/modules/jira/projects.py @@ -3,7 +3,7 @@ import logging import re from ...core import settings -from ...core.projects import ProjectError, _label_error, create, labels, validate +from ...core.projects import ProjectError, label_error, create, labels, validate log = logging.getLogger(__name__) @@ -60,7 +60,7 @@ def parse_legacy_labels(jql) -> list[str] | None: out: list[str] = [] for raw in ([m["one"]] if m["one"] is not None else m["many"].split(",")): label = raw.strip().strip('"') - if _label_error(label): + if label_error(label): return None if label.casefold() not in {x.casefold() for x in out}: out.append(label) diff --git a/src/jira_hub/modules/jira/refresh.py b/src/jira_hub/modules/jira/refresh.py new file mode 100644 index 0000000..2eec202 --- /dev/null +++ b/src/jira_hub/modules/jira/refresh.py @@ -0,0 +1,39 @@ +"""Обновление зеркала из Jira (команда refresh): журнал синхронизаций, полный прогон, автообновление.""" +from datetime import timedelta + +from ...core import db, queue, settings +from . import projects + + +def is_full_refresh(conn, command_id) -> bool: + """Команда — refresh по базовой JQL (полный прогон): только он вправе сбрасывать флаг neighbor.""" + if not command_id: + return False + cmd = queue.get_command(conn, command_id) + base = projects.base_jql(conn) + return bool(base and cmd and cmd["type"] == "refresh" and cmd["params"].get("jql") == base) + + +def record_state(conn, command_id, issues_count, links_count) -> None: + conn.execute("UPDATE sync_run SET issues_count = issues_count + ?, links_count = links_count + ? WHERE command_id = ?", + (int(issues_count), int(links_count), command_id)) + + +def maybe_auto_refresh(conn) -> dict | None: + jql = projects.base_jql(conn) + 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"]: + return None + if queue.has_active(conn): + 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): + return None + return queue.create_batch(conn, "Автообновление", "hub", [{"type": "refresh", "params": {"jql": jql}}]) + + +def auto_refresh_tick(hub, conn) -> None: + """Фоновая работа модуля: автообновление (подключается к tick хаба в Task 5).""" + maybe_auto_refresh(conn) diff --git a/tests/test_commands.py b/tests/test_commands.py index f9d3b57..0e30b7c 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -98,7 +98,6 @@ def test_substitute_and_estimate(): p = commands.substitute("ensure_issue", {"temp_key": "new:60", "issue_type": "task", "summary": "x", "epic_key": "new:2"}, {"new:2": "A-7", "new:60": "A-9"}) assert p["temp_key"] == "new:60" and p["epic_key"] == "A-7" - assert commands.estimate(["ensure_fields", "ensure_comment", "refresh"], {"ensure_fields": 120, "ensure_comment": 30, "refresh": 2}) == 152 @pytest.fixture diff --git a/tests/test_queue.py b/tests/test_queue.py index 8b599c9..4fdebdd 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -3,7 +3,8 @@ import hashlib import pytest from jira_hub.core import db, files, projects, queue, settings -from jira_hub.modules.jira import attachments, mirror +from jira_hub.modules.jira import attachments, commands, mirror, refresh +from jira_hub.modules.jira import fetch as jira_fetch from jira_hub.modules.jira import projects as jira_projects from jira_hub.modules.jira.commands import CommandError from jira_hub.core.queue import ClaimedError @@ -56,7 +57,7 @@ def test_claim_order_pause_and_report(conn): tail = queue.claim(conn) 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" - queue.record_state(conn, tail["id"], 1, 0) + refresh.record_state(conn, tail["id"], 1, 0) queue.report_result(conn, tail["id"], "done", jira_calls=1) 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 @@ -164,32 +165,32 @@ 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) - assert queue.maybe_auto_refresh(conn) is None # агент на связи, очередь пуста, проектов нет + 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) - b = queue.maybe_auto_refresh(conn) + 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")'} def test_auto_refresh_conditions(conn, add_project): add_project(conn) - assert queue.maybe_auto_refresh(conn) is None # агент не на связи + assert refresh.maybe_auto_refresh(conn) is None # агент не на связи queue.claim(conn, agent_version=1, settings_version=1) # отметился, очередь пуста - b = queue.maybe_auto_refresh(conn) + b = refresh.maybe_auto_refresh(conn) assert b and queue.batch_view(conn, b["id"])["title"] == "Автообновление" - assert queue.maybe_auto_refresh(conn) is None # очередь занята + assert refresh.maybe_auto_refresh(conn) is None # очередь занята c = queue.claim(conn) queue.report_result(conn, c["id"], "done") - assert queue.maybe_auto_refresh(conn) is None # интервал не прошёл + assert refresh.maybe_auto_refresh(conn) is None # интервал не прошёл conn.execute("UPDATE sync_run SET started_at = '2020-01-01T00:00:00+00:00'") - assert queue.maybe_auto_refresh(conn) is not None + assert refresh.maybe_auto_refresh(conn) is not None settings.set_(conn, "auto_refresh_minutes", 0) conn.execute("UPDATE sync_run SET started_at = '2020-01-01T00:00:00+00:00'") - assert queue.maybe_auto_refresh(conn) is None + assert refresh.maybe_auto_refresh(conn) is None def test_backup_called_for_first_write_only(conn, tmp_path): @@ -255,14 +256,14 @@ def test_reconcile_does_not_overwrite_finished_command(conn, monkeypatch, add_pr queue.create_batch(conn, "x", "claude", [{"type": "ensure_sprint", "params": {"key": "A-1", "sprint": "Спринт 2"}}]) c = queue.claim(conn) mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "sprint": "Спринт 2", "labels": ["AWPCA"]}], []) - real = queue.C.check_confirmed + real = commands.check_confirmed def racing(*a, **kw): out = real(*a, **kw) queue.cancel_command(conn, c["id"], force=True) # гонка: отмена после проверки return out - monkeypatch.setattr(queue.C, "check_confirmed", racing) + monkeypatch.setattr(commands, "check_confirmed", racing) assert queue.reconcile(conn) == [] assert queue.get_command(conn, c["id"])["status"] == "cancelled" @@ -291,7 +292,7 @@ def test_auto_refresh_waits_for_claimed_without_report(conn): queue.claim(conn) # выдана, отчёта нет 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 queue.maybe_auto_refresh(conn) is None # claimed без отчёта — очередь занята + assert refresh.maybe_auto_refresh(conn) is None # claimed без отчёта — очередь занята def test_redelivery_of_lost_claim_before_queue(conn): @@ -449,11 +450,11 @@ def test_redelivered_full_refresh_marks_neighbors_once(conn, add_project): 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) - queue.record_state(conn, r["id"], 1, 0) + 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 conn.execute("SELECT deliveries FROM command WHERE id = ?", (r["id"],)).fetchone()["deliveries"] == 2 - queue.record_state(conn, r["id"], 1, 0) + refresh.record_state(conn, r["id"], 1, 0) mirror.apply_state(conn, [{"key": "NEW", "summary": "y", "labels": []}], []) queue.report_result(conn, r["id"], "done") nb = {row["key"]: row["neighbor"] for row in conn.execute("SELECT key, neighbor FROM issue")} @@ -464,10 +465,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) - queue.record_state(conn, r["id"], 150, 40) + 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"] # отчёт потерян — довыдача - queue.record_state(conn, r["id"], 150, 40) + 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) assert run["started_at"] == started @@ -1066,7 +1067,7 @@ def test_fetch_checks(fconn): with pytest.raises(CommandError) as e: queue.create_batch(fconn, "x", "claude", [fetch(key="B-1")]) assert str(e.value) == "у B-1 нет вложения #31337 в зеркале — обновите задачу" - with pytest.raises(queue.TooLarge) as e: + with pytest.raises(jira_fetch.TooLarge) as e: queue.create_batch(fconn, "x", "claude", [fetch("31338")]) assert str(e.value) == "файл больше 10 МБ (23,8 МБ) — скачайте в Jira" assert queue.create_batch(fconn, "x", "claude", [fetch("31339")])["count"] == 1 # size NULL — ставится @@ -1077,9 +1078,9 @@ def test_fetch_size_limit_boundary(fconn): mirror.apply_state(fconn, [{"key": "A-1", "summary": "т", "labels": [], "attachments": [ {"id": "41", "filename": "eq.bin", "size": files.MAX_SIZE, "created": "2026-10-20T10:00:00.000+0300"}, {"id": "42", "filename": "over.bin", "size": files.MAX_SIZE + 1, "created": "2026-10-20T10:00:00.000+0300"}]}], []) - assert queue.request_fetch(fconn, None, "41", "claude")["state"] == "fetching" - with pytest.raises(queue.TooLarge): - queue.request_fetch(fconn, None, "42", "claude") + assert jira_fetch.request_fetch(fconn, None, "41", "claude")["state"] == "fetching" + with pytest.raises(jira_fetch.TooLarge): + jira_fetch.request_fetch(fconn, None, "42", "claude") def test_fetch_report_done_non_dict_result_fails(fconn, tmp_path): @@ -1117,7 +1118,7 @@ def test_request_fetch_once(open_db, tmp_path): def run(c): try: barrier.wait() - res.append(queue.request_fetch(c, tmp_path / "files", "31337", "x")) + res.append(jira_fetch.request_fetch(c, tmp_path / "files", "31337", "x")) except Exception as e: # noqa: BLE001 errs.append(e) @@ -1139,30 +1140,30 @@ def put_file(tmp_path, data=b"abc"): def test_request_fetch_ready(fconn, tmp_path): root, sha = put_file(tmp_path) fconn.execute("INSERT INTO attachment_file(attachment_id, file, size, fetched_at) VALUES ('31337', ?, 3, ?)", (sha, db.now())) - r = queue.request_fetch(fconn, root, "31337", "x") + r = jira_fetch.request_fetch(fconn, root, "31337", "x") assert r["state"] == "ready" and r["filename"] == "ОСВ 60.xlsx" assert fconn.execute("SELECT COUNT(*) AS n FROM command").fetchone()["n"] == 0 with pytest.raises(KeyError): - queue.request_fetch(fconn, root, "999", "x") + jira_fetch.request_fetch(fconn, root, "999", "x") def test_request_fetch_agent_too_old(fconn, tmp_path): fconn.execute("UPDATE agent_status SET agent_version = '13' WHERE id = 1") - with pytest.raises(queue.AgentTooOld) as e: - queue.request_fetch(fconn, tmp_path / "files", "31337", "x") + with pytest.raises(jira_fetch.AgentTooOld) as e: + jira_fetch.request_fetch(fconn, tmp_path / "files", "31337", "x") assert str(e.value) == "агент v13 не умеет скачивать вложения — нужен v14" fconn.execute("UPDATE agent_status SET agent_version = '14' WHERE id = 1") - assert queue.request_fetch(fconn, tmp_path / "files", "31337", "x")["state"] == "fetching" + assert jira_fetch.request_fetch(fconn, tmp_path / "files", "31337", "x")["state"] == "fetching" fconn.execute("UPDATE agent_status SET agent_version = NULL WHERE id = 1") - assert queue.request_fetch(fconn, tmp_path / "files", "31337", "x")["state"] == "fetching" + assert jira_fetch.request_fetch(fconn, tmp_path / "files", "31337", "x")["state"] == "fetching" def test_request_fetch_after_failed(fconn, tmp_path): root = tmp_path / "files" - first = queue.request_fetch(fconn, root, "31337", "x") + first = jira_fetch.request_fetch(fconn, root, "31337", "x") cmd = queue.claim(fconn) queue.report_result(fconn, cmd["id"], "failed", error="нет сети", files_dir=root) - again = queue.request_fetch(fconn, root, "31337", "x") + again = jira_fetch.request_fetch(fconn, root, "31337", "x") assert again["state"] == "fetching" and again["batch_id"] != first["batch_id"] @@ -1187,7 +1188,7 @@ def test_report_fetch_done_missing_file(fconn, tmp_path): assert cmd["status"] == "failed" and cmd["error"] == f"хаб: файл {FSHA} не найден в хранилище" assert fconn.execute("SELECT COUNT(*) AS n FROM attachment_file").fetchone()["n"] == 0 # после уборки файла (done → файла нет) и повторной постановки — новый пакет - assert queue.request_fetch(fconn, root, "31337", "x")["state"] == "fetching" + assert jira_fetch.request_fetch(fconn, root, "31337", "x")["state"] == "fetching" def test_report_fetch_wrong_size(fconn, tmp_path): @@ -1201,6 +1202,6 @@ def test_request_fetch_after_done_file_swept(fconn, tmp_path): root, sha = put_file(tmp_path) report_done(fconn, root, {"file": sha, "size": 3}) files.path(root, sha).unlink() # уборка файла - r = queue.request_fetch(fconn, root, "31337", "x") + r = jira_fetch.request_fetch(fconn, root, "31337", "x") assert r["state"] == "fetching" assert fconn.execute("SELECT COUNT(*) AS n FROM command WHERE type = 'fetch_attachment'").fetchone()["n"] == 2 diff --git a/tests/test_queue_core.py b/tests/test_queue_core.py new file mode 100644 index 0000000..976ab95 --- /dev/null +++ b/tests/test_queue_core.py @@ -0,0 +1,84 @@ +"""Ядро очереди на модуле-заглушке: механика не знает типов команд jira (hub-modules, решение 4).""" +import os +import time +from dataclasses import replace + +import pytest + +from jira_hub.core import files, queue, registry +from jira_hub.core.errors import CommandError +from jira_hub.core.modules import CommandHooks, Module +from jira_hub.modules import jira + +SHA = "a" * 64 + + +def _validate(cmds): + return [{"type": c["type"], "params": c.get("params") or {}} for c in cmds] + + +def make_stub(events, refs=()): + hooks = CommandHooks( + types=frozenset({"w", "r"}), read_types=frozenset({"r"}), validate=_validate, + check_batch=lambda conn, cmds, project, files_dir: cmds, + tail=lambda cmds: {"type": "r", "params": {}} if any(c["type"] == "w" for c in cmds) else None, + pause=lambda conn, ctype: 7, deliver_params=lambda conn, cmd, files_dir: cmd["params"], + temp_definer=None, temp_keys_used=lambda cmd: [], substitute=lambda t, p, m: p, + drop_unresolved=lambda params, unresolved: unresolved, retry_params=lambda cmd: cmd["params"], + check_confirmed=lambda conn, cmd: (False, None), on_claim=lambda *a: None, on_failed=lambda *a: None, + check_result=lambda conn, cmd, status, result, files_dir: (status, None), + on_result=lambda conn, cmd, status, *a: events.append(("on_result", cmd["type"], status)), + after_result=lambda conn, cmd, status: events.append(("after_result", cmd["type"], status)), + on_late_result=lambda *a: False) + return Module(name="t", commands=hooks, file_refs=lambda conn: set(refs)) + + +@pytest.fixture +def stub(monkeypatch): + events = [] + monkeypatch.setattr(registry, "_cache", [jira.MODULE, make_stub(events)]) + return events + + +def test_tail_and_estimate(open_db, stub): + conn = open_db() + b = queue.create_batch(conn, "x", "claude", [{"type": "w", "params": {}}]) + assert [c["type"] for c in queue.batch_view(conn, b["id"])["commands"]] == ["w", "r"] + assert b["estimate_seconds"] == 14 and queue.estimate(conn, ["w", "r"]) == 14 + + +def test_mixed_modules(open_db, stub): + conn = open_db() + with pytest.raises(CommandError, match="команды разных модулей в одном пакете"): + queue.create_batch(conn, "x", "claude", [{"type": "w"}, {"type": "refresh", "params": {"jql": "x"}}]) + + +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) + seen = {} + real = stub.append + + def after(conn_, cmd_, status): + other = open_db(path) # новое соединение видит зафиксированное + seen["status"] = other.execute("SELECT status FROM command WHERE id = ?", (cmd_["id"],)).fetchone()["status"] + real(("after_result", cmd_["type"], status)) + + hooks = replace(registry.module("t").commands, after_result=after) + registry._cache[1] = replace(registry.module("t"), commands=hooks) + queue.report_result(conn, cmd["id"], "done", {}) + assert ("on_result", "w", "done") in stub and ("after_result", "w", "done") in stub + assert seen["status"] == "done" + + +def test_sweep_uses_file_refs(tmp_path, open_db, monkeypatch): + monkeypatch.setattr(registry, "_cache", [jira.MODULE, make_stub([], refs={SHA})]) + root = tmp_path / "files" + root.mkdir() + (root / SHA).write_bytes(b"x") + old = time.time() - 40 * 86400 + os.utime(root / SHA, (old, old)) + assert files.sweep(open_db(), root) == [] + assert (root / SHA).exists() diff --git a/tests/test_registry.py b/tests/test_registry.py index ca51eae..1a0a1cf 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -10,7 +10,7 @@ from jira_hub.core.modules import CommandHooks, Module CORE = Path(__file__).resolve().parent.parent / "src" / "jira_hub" / "core" # временно: R-1, сокращается в Task 4–6 -PENDING = {"api.py", "cli.py", "mcp_server.py", "queue.py"} +PENDING = {"api.py", "cli.py", "mcp_server.py"} def _noop(*a, **k):