refactor: очередь ядра и обработчики команд модуля jira (#87)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
a4e57fac20
commit
c9ac2d5238
@ -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))
|
||||
|
||||
@ -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():
|
||||
|
||||
@ -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] # исчерпаны выдачи
|
||||
|
||||
@ -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():
|
||||
|
||||
@ -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}}])
|
||||
|
||||
@ -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,
|
||||
)
|
||||
|
||||
@ -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)."""
|
||||
|
||||
39
src/jira_hub/modules/jira/fetch.py
Normal file
39
src/jira_hub/modules/jira/fetch.py
Normal file
@ -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)
|
||||
267
src/jira_hub/modules/jira/hooks.py
Normal file
267
src/jira_hub/modules/jira/hooks.py
Normal file
@ -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)
|
||||
@ -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)
|
||||
|
||||
39
src/jira_hub/modules/jira/refresh.py
Normal file
39
src/jira_hub/modules/jira/refresh.py
Normal file
@ -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)
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
84
tests/test_queue_core.py
Normal file
84
tests/test_queue_core.py
Normal file
@ -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()
|
||||
@ -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):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user