diff --git a/src/jira_hub/core/modules.py b/src/jira_hub/core/modules.py index ef4dfd4..cdf88a4 100644 --- a/src/jira_hub/core/modules.py +++ b/src/jira_hub/core/modules.py @@ -32,6 +32,8 @@ class CommandHooks: on_late_result: Callable[[Connection, dict, str, dict | None, int, int], bool] # (conn, cmd, status, result, files_dir): до транзакции отчёта — тяжёлая подготовка (файлы, перевод); None — нет before_result: Callable[[Connection, dict, str, dict | None, Path | None], None] | None = None + # (conn, cmd): отчёт не дошёл до after_result (ранний выход, исключение) — снять подготовку before_result + abort_result: Callable[[Connection, dict], None] | None = None @dataclass(frozen=True) diff --git a/src/jira_hub/core/queue.py b/src/jira_hub/core/queue.py index 3da999b..2f6f6e8 100644 --- a/src/jira_hub/core/queue.py +++ b/src/jira_hub/core/queue.py @@ -296,37 +296,51 @@ def report_result(conn, command_id, status, result=None, error=None, jira_calls= raise ValueError("status — done или failed") error = None if error is None else str(error)[:ERROR_LIMIT] early = get_command(conn, command_id) - if early is not None and early["status"] == "claimed": - eh = _hooks(early["type"]) - if eh and eh.before_result: # вне транзакции отчёта: модуль готовит тяжёлое, не держа блокировку базы - eh.before_result(conn, early, status, result, files_dir) - with db.transaction(conn): - cmd = get_command(conn, command_id) - if cmd is None: - raise KeyError(command_id) - if cmd["status"] == "queued": - raise NotClaimedError(f"команда {command_id} не выдана") - h = _hooks(cmd["type"]) - if cmd["status"] in TERMINAL: - # поздний отчёт по команде, которую хаб уже закрыл (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() + eh = _hooks(early["type"]) if early is not None and early["status"] == "claimed" else None + prepared = bool(eh and eh.before_result) + completed = False + try: + if prepared: # вне транзакции отчёта: модуль готовит тяжёлое, не держа блокировку базы + try: + eh.before_result(conn, early, status, result, files_dir) + except ValueError as e: # не путать с «status — done или failed» (422 bad_status в API) + raise RuntimeError(f"before_result: {e}") from e + with db.transaction(conn): + cmd = get_command(conn, command_id) + if cmd is None: + raise KeyError(command_id) + if cmd["status"] == "queued": + raise NotClaimedError(f"команда {command_id} не выдана") + h = _hooks(cmd["type"]) + if cmd["status"] in TERMINAL: + # поздний отчёт по команде, которую хаб уже закрыл (force-отмена, предел выдач): модуль может принять + # данные, статус не трогаем + if h and h.on_late_result(conn, cmd, status, result, jira_calls, rate_limited): + return get_command(conn, command_id) + return cmd # идемпотентно; терминальный статус не затирается + now = db.now() + 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 current_command_id = ?", + (now, command_id)) + if h: + h.on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_dir) 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 current_command_id = ?", - (now, command_id)) - 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) + h.after_result(conn, cmd, status) + completed = True + return get_command(conn, command_id) + finally: + if prepared and not completed: # ранний выход или исключение: модуль снимает подготовку (abort_result) + ah = eh.abort_result + if ah is not None: + try: + ah(conn, early) + except Exception: # noqa: BLE001 + log.exception("команда %s: abort_result", command_id) def reconcile(conn) -> list[int]: diff --git a/src/jira_hub/modules/wiki/commands.py b/src/jira_hub/modules/wiki/commands.py index 6ba67da..964fcc0 100644 --- a/src/jira_hub/modules/wiki/commands.py +++ b/src/jira_hub/modules/wiki/commands.py @@ -77,4 +77,5 @@ HOOKS = CommandHooks( drop_unresolved=lambda cmd, unresolved: unresolved, retry_params=lambda cmd: cmd["params"], check_confirmed=lambda conn, cmd: (False, None), on_claim=sync.on_claim, on_failed=sync.on_failed, check_result=sync.check_result, on_result=sync.on_result, after_result=sync.after_result, - on_late_result=lambda *a: False, before_result=sync.before_result) + on_late_result=lambda *a: False, before_result=sync.before_result, + abort_result=sync.abort_result) diff --git a/src/jira_hub/modules/wiki/projects.py b/src/jira_hub/modules/wiki/projects.py index 8e39f20..3c71fd6 100644 --- a/src/jira_hub/modules/wiki/projects.py +++ b/src/jira_hub/modules/wiki/projects.py @@ -1,4 +1,5 @@ """Корень вики проекта: поле wiki_root, значения для GET /projects, хуки создания и правки проекта.""" +import glob import logging import re import shutil @@ -74,7 +75,8 @@ def _drop_copy(conn, label: str) -> None: base = paths.project_dir(conn, label) except RuntimeError: # база без файла (тесты в памяти): копии на диске нет return - for d in (base, base.with_name(f".{base.name}.new"), base.with_name(f".{base.name}.old")): + leftovers = sorted(base.parent.glob(glob.escape(f".{base.name}") + ".new*")) if base.parent.is_dir() else [] + for d in (base, *leftovers, base.with_name(f".{base.name}.old")): try: shutil.rmtree(d) except FileNotFoundError: diff --git a/src/jira_hub/modules/wiki/stage.py b/src/jira_hub/modules/wiki/stage.py index b847777..a9334b1 100644 --- a/src/jira_hub/modules/wiki/stage.py +++ b/src/jira_hub/modules/wiki/stage.py @@ -6,6 +6,8 @@ from ...core.http import ApiError from . import convert MAX_PAGES = 50 +MAX_STAGE = 20000 # страниц одной команды в стадии (между пачками) +LIMITS = {"code": 200, "title": 1000, "url": 4000, "parent_code": 200, "eva_id": 200} def _bad(code, field): @@ -16,9 +18,9 @@ def _text(page, key, code, required=True, nullable=False): v = page.get(key) if v is None and (nullable or not required): return None - if not isinstance(v, str) or (required and not v.strip()): + if not isinstance(v, str) or (required and not v.strip()) or len(v) > LIMITS.get(key, 10**9): raise _bad(code, key) - return v + return v if v.strip() else None # пустое необязательное поле (parent_code "") — как отсутствующее def _check(i, page) -> dict: @@ -26,7 +28,7 @@ def _check(i, page) -> dict: if not isinstance(page, dict): raise ApiError(422, "bad_pages", f"страница {i}: не объект") code = page.get("code") - if not isinstance(code, str) or not code.strip(): + if not isinstance(code, str) or not code.strip() or len(code) > LIMITS["code"]: raise ApiError(422, "bad_pages", f"страница {i}: code") title = _text(page, "title", code) html = page.get("html") @@ -36,7 +38,7 @@ def _check(i, page) -> dict: if orderno is not None and (not isinstance(orderno, int) or isinstance(orderno, bool)): raise _bad(code, "orderno") eva_id = page.get("eva_id") - if eva_id is not None and not isinstance(eva_id, str): + if eva_id is not None and (not isinstance(eva_id, str) or len(eva_id) > LIMITS["eva_id"]): raise _bad(code, "eva_id") attachments, links = page.get("attachments") or [], page.get("links") or [] if not isinstance(attachments, list) or not all( @@ -58,12 +60,12 @@ def accept(conn, command_id, pages) -> dict: raise ApiError(422, "bad_request", "command_id — целое") if not isinstance(pages, list) or not pages or len(pages) > MAX_PAGES: raise ApiError(422, "bad_pages", f"pages — список от 1 до {MAX_PAGES} страниц") - checked = [_check(i, p) for i, p in enumerate(pages, 1)] cmd = queue.get_command(conn, command_id) if cmd is None: raise ApiError(404, "not_found", f"command {command_id}") if cmd["type"] != "wiki_sync" or cmd["status"] != "claimed": raise ApiError(409, "not_claimed", f"команда {command_id} не выдана агенту как wiki_sync") + checked = [_check(i, p) for i, p in enumerate(pages, 1)] rows = [] for p in checked: try: @@ -77,5 +79,8 @@ def accept(conn, command_id, pages) -> dict: cur = conn.execute("SELECT status FROM command WHERE id = ?", (command_id,)).fetchone() if cur is None or cur["status"] != "claimed": # состояние могло смениться, пока шёл перевод raise ApiError(409, "not_claimed", f"команда {command_id} уже не в работе") + have = conn.execute("SELECT COUNT(*) AS n FROM wiki_sync_stage WHERE command_id = ?", (command_id,)).fetchone()["n"] + if have + len(rows) > MAX_STAGE: + raise ApiError(422, "bad_pages", f"pages: в команде больше {MAX_STAGE} страниц") conn.executemany("INSERT OR REPLACE INTO wiki_sync_stage(command_id, code, payload) VALUES (?, ?, ?)", rows) return {"accepted": len(rows)} diff --git a/src/jira_hub/modules/wiki/sync.py b/src/jira_hub/modules/wiki/sync.py index 74dd8fd..22b1372 100644 --- a/src/jira_hub/modules/wiki/sync.py +++ b/src/jira_hub/modules/wiki/sync.py @@ -7,6 +7,7 @@ import json import logging import re import shutil +import sqlite3 import threading from dataclasses import dataclass, field from pathlib import Path @@ -38,11 +39,21 @@ class Prepared: kept_ids: set = field(default_factory=set) new_ids: list = field(default_factory=list) # id вложений, добавленных подготовкой new_dir: Path | None = None + old_dir: Path | None = None # прежняя копия после подмены: удаляется в after_result + committed: bool = False root_title: str | None = None _PREPARED: dict[int, Prepared] = {} +_BUSY: set[Path] = set() # каталоги .new идущих подготовок (с начала сборки) _lock = threading.Lock() +_label_locks: dict[str, threading.Lock] = {} # подготовка…применение одного проекта — по одной +_tls = threading.local() # блокировка, взятая этим потоком в before_result + + +def _label_lock(label: str) -> threading.Lock: + with _lock: + return _label_locks.setdefault(label, threading.Lock()) # --- раскладка копии ---------------------------------------------------------- @@ -144,17 +155,18 @@ def build_copy(stage_pages: list[dict], codes: list[str], target: Path, ids: dic # --- подготовка (до транзакции отчёта) --------------------------------------- -def _siblings(conn, label: str) -> tuple[Path, Path, Path]: +def _siblings(conn, label: str, cmd_id: int) -> tuple[Path, Path, Path]: + """(копия, каталог сборки этой команды, каталог прежней копии на время подмены).""" cur = paths.project_dir(conn, label) - return cur, cur.with_name(f".{cur.name}.new"), cur.with_name(f".{cur.name}.old") + return cur, cur.with_name(f".{cur.name}.new.{cmd_id}"), cur.with_name(f".{cur.name}.old") def _allocate_ids(conn, label, pages) -> tuple[dict, list, list]: """id вложений страниц: существующие по (страница, url) сохраняются, новые вставляются короткой транзакцией.""" ids, new_ids, attach = {}, [], [] - existing = {(r["page_code"], r["url"]): r["id"] for r in conn.execute( - "SELECT id, page_code, url FROM wiki_attachment WHERE label = ?", (label,))} with db.transaction(conn): + existing = {(r["page_code"], r["url"]): r["id"] for r in conn.execute( + "SELECT id, page_code, url FROM wiki_attachment WHERE label = ?", (label,))} for page in pages: for a in page["attachments"]: key = (page["code"], a["url"]) @@ -182,6 +194,14 @@ def prepare(conn, cmd: dict, result: dict | None) -> Prepared: """До транзакции: проверить корень и список кодов, выделить id вложений, собрать `.new`. Ошибка — в p.error.""" label, root = cmd["params"]["label"], cmd["params"]["root"] p = Prepared(cmd["id"], label, root) + try: + _, p.new_dir, _ = _siblings(conn, label, cmd["id"]) + except RuntimeError as e: # база без файла + p.error = str(e) + return p + with _lock: # с этого момента ни finish_pending_swaps, ни уборка не тронут .new + _PREPARED[cmd["id"]] = p + _BUSY.add(p.new_dir) row = conn.execute("SELECT root_code FROM wiki_project WHERE label = ?", (label,)).fetchone() if row is None or row["root_code"] != root: p.discarded = "корень сменился" @@ -199,14 +219,17 @@ def prepare(conn, cmd: dict, result: dict | None) -> Prepared: if c not in stage: raise SyncFailed(f"нет страницы {c}") pages = [stage[c] for c in codes] - _, p.new_dir, _ = _siblings(conn, label) + cur, _, old = _siblings(conn, label, cmd["id"]) secure_dir(p.new_dir.parent) shutil.rmtree(p.new_dir, ignore_errors=True) + shutil.rmtree(old, ignore_errors=True) # остаток прошлой подмены: убираем здесь, не под блокировкой базы ids, p.new_ids, p.attachments = _allocate_ids(conn, label, pages) p.kept_ids = set(ids.values()) p.rows = build_copy(pages, codes, p.new_dir, ids) title = (result or {}).get("root_title") p.root_title = title if isinstance(title, str) and title.strip() else p.rows[0]["title"] + except sqlite3.OperationalError: + raise # база занята: 500, агент повторит отчёт; abort_result всё уберёт except SyncFailed as e: p.error = str(e) except ValueError as e: # fill_ids, путь вне копии @@ -222,11 +245,43 @@ def prepare(conn, cmd: dict, result: dict | None) -> Prepared: def before_result(conn, cmd, status, result, files_dir) -> None: + """До транзакции отчёта done: одна подготовка на проект (блокировка до after_result/abort_result).""" if cmd["type"] != "wiki_sync" or status != "done": return - p = prepare(conn, cmd, result) + lock = _label_lock(cmd["params"]["label"]) + lock.acquire() + _tls.held = lock + row = conn.execute("SELECT status FROM command WHERE id = ?", (cmd["id"],)).fetchone() + if row is None or row["status"] != "claimed": # параллельный отчёт уже закрыл команду + return + prepare(conn, cmd, result) + + +def _release(cmd_id: int) -> Prepared | None: with _lock: - _PREPARED[cmd["id"]] = p + p = _PREPARED.pop(cmd_id, None) + if p is not None and p.new_dir is not None: + _BUSY.discard(p.new_dir) + lock = getattr(_tls, "held", None) + if lock is not None: + _tls.held = None + lock.release() + return p + + +def abort_result(conn, cmd) -> None: + """Отчёт не дошёл до конца: подготовка снимается — каталог сборки, добавленные ею вложения, блокировка.""" + p = _release(cmd["id"]) + if p is None: + return + if p.new_dir is not None and not p.committed: + shutil.rmtree(p.new_dir, ignore_errors=True) + if p.new_ids and not p.committed: + try: + with db.transaction(conn): + _abandon(conn, p) + except sqlite3.Error: + log.warning("wiki_sync %s: вложения подготовки не удалены", cmd["id"], exc_info=True) def _prepared(conn, cmd, result) -> Prepared: @@ -239,8 +294,6 @@ def check_result(conn, cmd, status, result, files_dir) -> tuple[str, str | None] if cmd["type"] != "wiki_sync" or status != "done": return status, None p = _prepared(conn, cmd, result) - with _lock: - _PREPARED[cmd["id"]] = p if p.discarded: if isinstance(result, dict): result["discarded"] = p.discarded @@ -252,9 +305,9 @@ def check_result(conn, cmd, status, result, files_dir) -> tuple[str, str | None] # --- применение (в транзакции отчёта) ---------------------------------------- -def _swap(cur: Path, new: Path, old: Path) -> None: - """wiki/<метка> → .old, .new → <метка>, .old удаляется. При сбое прежний каталог возвращается на место.""" - shutil.rmtree(old, ignore_errors=True) +def _swap(cur: Path, new: Path, old: Path) -> bool: + """wiki/<метка> → .old, .new → <метка>; True, если прежняя копия была (её удалит after_result вне транзакции). + При сбое прежний каталог возвращается на место.""" had = cur.exists() try: if had: @@ -264,7 +317,7 @@ def _swap(cur: Path, new: Path, old: Path) -> None: if had and not cur.exists() and old.exists(): old.rename(cur) raise - shutil.rmtree(old, ignore_errors=True) + return had def _fail(conn, cmd, error) -> None: @@ -277,7 +330,7 @@ def on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_di if cmd["type"] != "wiki_sync": return with _lock: - p = _PREPARED.pop(cmd["id"], None) + p = _PREPARED.get(cmd["id"]) if status != "done": error = conn.execute("SELECT error FROM command WHERE id = ?", (cmd["id"],)).fetchone()["error"] on_failed_or_reported_failed(conn, cmd, error or "синхронизация не удалась") @@ -298,6 +351,7 @@ def on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_di try: with db.transaction(conn): _commit(conn, p, now) + p.committed = True except OSError as e: _abandon(conn, p) _fail(conn, cmd, f"не удалось подменить копию: {e}") @@ -319,8 +373,9 @@ def _commit(conn, p: Prepared, now: str) -> None: conn.execute("UPDATE wiki_project SET synced_at = ?, pages_count = ?, root_title = ?, sync_error = NULL, failed_at = NULL " "WHERE label = ?", (now, len(p.rows), p.root_title, p.label)) conn.execute("DELETE FROM wiki_sync_stage WHERE command_id = ?", (p.cmd_id,)) - cur, new, old = _siblings(conn, p.label) - _swap(cur, new, old) # последним: сбой откатывает строки (savepoint) + cur, new, old = _siblings(conn, p.label, p.cmd_id) + if _swap(cur, new, old): # последним: сбой откатывает строки (savepoint) + p.old_dir = old def on_failed_or_reported_failed(conn, cmd: dict, error: str | None) -> None: @@ -345,33 +400,42 @@ def on_claim(conn, cmd, now, redelivery) -> None: def after_result(conn, cmd, status) -> None: - with _lock: - _PREPARED.pop(cmd["id"], None) + """Вне транзакции: прежняя копия после подмены удаляется, блокировка проекта снимается.""" + p = _release(cmd["id"]) + if p is not None and p.old_dir is not None: + try: + shutil.rmtree(p.old_dir) + except OSError as e: + log.warning("вики: прежняя копия %s не удалена (уберёт следующий старт): %s", p.old_dir, e) # --- уборка ---------------------------------------------------------------- +_PENDING = re.compile(r"^\.(.+)\.(new(?:\.\d+)?|old)$", re.DOTALL) + + def finish_pending_swaps(root: Path) -> None: - """При старте: `.<имя>.new` без каталога `<имя>` — переименовать (подмена прервана); `.new` рядом с `<имя>` — - остаток неудачной сборки, удаляется; `.<имя>.old` — удаляется. Каталоги, которые собирает идущая подготовка, не трогаются.""" + """При старте: `.<имя>.new[.]` с `_tree.json` (сборка закончена) без каталога `<имя>` — переименовать (подмена + прервана); без `_tree.json` или рядом с `<имя>` — остаток сборки, удаляется; `.<имя>.old` — удаляется. Каталоги + идущих подготовок не трогаются.""" root = Path(root) if not root.is_dir(): return with _lock: - busy = {p.new_dir for p in _PREPARED.values() if p.new_dir} + busy = set(_BUSY) for entry in sorted(root.iterdir()): - n = entry.name - if not n.startswith(".") or entry in busy: + m = _PENDING.match(entry.name) + if m is None or entry in busy: continue try: - if n.endswith(".new"): - target = root / n[1:-4] - if target.exists(): - shutil.rmtree(entry, ignore_errors=True) - else: - entry.rename(target) - elif n.endswith(".old"): + if m.group(2) == "old": shutil.rmtree(entry) + continue + target = root / m.group(1) + if target.exists() or not (entry / "_tree.json").is_file(): + shutil.rmtree(entry, ignore_errors=True) + else: + entry.rename(target) except OSError as e: log.warning("вики: не удалось довести подмену %s: %s", entry, e) diff --git a/src/jira_hub/modules/wiki/tick.py b/src/jira_hub/modules/wiki/tick.py index 1a5cbeb..3d0c413 100644 --- a/src/jira_hub/modules/wiki/tick.py +++ b/src/jira_hub/modules/wiki/tick.py @@ -2,7 +2,7 @@ from datetime import timedelta from ...core import db, settings -from . import commands, paths, sync +from . import commands, paths, search, sync def auto_sync_tick(hub, conn) -> None: @@ -34,6 +34,7 @@ def housekeeping_tick(hub, conn) -> None: if root is not None and str(root) not in _swept: _swept.add(str(root)) sync.finish_pending_swaps(root) + search.ensure(conn) # нет FTS5 — одно предупреждение в лог, поиск пойдёт по файлам sync.housekeeping(conn) diff --git a/tests/test_registry.py b/tests/test_registry.py index 5fc7bea..efc9200 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -51,9 +51,8 @@ def test_route_conflict(): def test_routes_assembled(): from jira_hub.core import api - from jira_hub.modules import jira, wiki pairs = lambda routes: {(m, p) for m, p, _, _ in routes} # noqa: E731 - mods = jira.MODULE.routes + wiki.MODULE.routes + mods = tuple(r for m in registry.modules() for r in m.routes) assert pairs(api.ROUTES) == pairs(api.CORE_ROUTES) | pairs(mods) assert not pairs(api.CORE_ROUTES) & pairs(mods) assert len(api.ROUTES) == len(api.CORE_ROUTES) + len(mods) diff --git a/tests/test_wiki_sync.py b/tests/test_wiki_sync.py index c7de22b..3283b8e 100644 --- a/tests/test_wiki_sync.py +++ b/tests/test_wiki_sync.py @@ -239,12 +239,26 @@ def test_layout_orphans_and_cycles(tmp_path): def test_finish_pending_swap(tmp_path): root = tmp_path / "wiki" (root / ".AWPCA.new" / "x").mkdir(parents=True) + (root / ".AWPCA.new" / "_tree.json").write_text("[]") (root / ".AWPCA.old").mkdir() (root / ".B.new").mkdir() (root / "B").mkdir() # остаток неудачной сборки рядом с живой копией + (root / ".C.new.7").mkdir() # сборка без _tree.json прервана: не продвигается, удаляется + (root / ".D.new.8").mkdir() + (root / ".D.new.8" / "_tree.json").write_text("[]") # каталог с суффиксом команды, сборка закончена sync.finish_pending_swaps(root) assert (root / "AWPCA" / "x").is_dir() and not (root / ".AWPCA.old").exists() and not (root / ".AWPCA.new").exists() assert (root / "B").is_dir() and not (root / ".B.new").exists() + assert not (root / "C").exists() and not (root / ".C.new.7").exists() and (root / "D" / "_tree.json").is_file() + busy = root / ".E.new.9" + busy.mkdir() + (busy / "_tree.json").write_text("[]") + sync._BUSY.add(busy) + try: + sync.finish_pending_swaps(root) # идущая подготовка не трогается + finally: + sync._BUSY.discard(busy) + assert busy.is_dir() and not (root / "E").exists() sync.finish_pending_swaps(root / "нет") # нет каталога — не ошибка @@ -433,3 +447,162 @@ def test_batch_limit_and_replace(api): api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0]]}) api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0] | {"title": "Новый"}]}) # повтор — замена assert [json.loads(r["payload"])["title"] for r in api.rows("SELECT payload FROM wiki_sync_stage")] == ["Новый"] + + +# --- правки ревью (I1–I3, M1–M9) ------------------------------------------------- + +def _clean(api): + """Ни подготовок, ни блокировки проекта, ни остатков сборки, ни вложений-сирот.""" + assert sync._PREPARED == {} and sync._BUSY == set() + assert not sync._label_lock("AWPCA").locked() + assert not [p.name for p in api.wiki_dir.glob(".*")] if api.wiki_dir.exists() else True + + +def test_old_copy_removed_outside_transaction(api, monkeypatch): + first_sync(api) + api.enqueue() + seen = [] + real = sync.shutil.rmtree + + def probe(path, *a, **k): + if str(path).endswith(".old"): + other = sqlite3.connect(api.db_path, timeout=0) + try: + other.execute("BEGIN IMMEDIATE") + other.execute("ROLLBACK") + seen.append("free") + except sqlite3.OperationalError: + seen.append("locked") + finally: + other.close() + return real(path, *a, **k) + + monkeypatch.setattr(sync.shutil, "rmtree", probe) + api.run(TREE, CODES) + assert seen and set(seen) == {"free"} + _clean(api) + + +def _attach_pages(): + att = [{"ref": "/files/x/a.png", "url": "https://eva.example.ru/files/x/a.png", "name": "a.png"}] + return [TREE[0], page("DOC-3", "Статусы", "DOC-1", 1, '', att)] + + +def test_exception_in_before_result_cleans_up(api, monkeypatch): + cmd = api.claim() + api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": _attach_pages()}) + real = sync.build_copy + + def boom(*a, **k): + real(*a, **k) + raise sqlite3.OperationalError("database is locked") + + monkeypatch.setattr(sync, "build_copy", boom) + with pytest.raises(HubError) as e: + api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1", "DOC-3"]}) + assert e.value.status == 500 # не failed: агент повторит отчёт + assert api.rows("SELECT status FROM command WHERE id = ?", cmd["id"])[0]["status"] == "claimed" + assert api.rows("SELECT * FROM wiki_attachment") == [] # сирота убран + _clean(api) + monkeypatch.setattr(sync, "build_copy", real) + assert api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1", "DOC-3"]})["status"] == "done" + assert len(api.rows("SELECT * FROM wiki_attachment")) == 1 + _clean(api) + + +def test_value_error_from_hook_is_not_422(api, monkeypatch): + cmd = api.claim() + monkeypatch.setattr(sync, "prepare", lambda *a: (_ for _ in ()).throw(ValueError("случайная"))) + with pytest.raises(HubError) as e: + api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1"]}) + assert e.value.status == 500 and e.value.error != "bad_status" + _clean(api) + + +def test_terminal_command_late_report_cleans_up(api, monkeypatch): + """Команда закрыта между подготовкой и транзакцией (force-отмена): подготовка снята, копия не тронута.""" + cmd = api.claim() + api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": _attach_pages()}) + real = sync.before_result + + def cancel_after(conn, c, status, result, files_dir): + real(conn, c, status, result, files_dir) + other = api.hub.conn() + queue.cancel_command(other, c["id"], force=True) + other.close() + + monkeypatch.setattr(sync, "before_result", lambda *a: cancel_after(*a)) + # хуки заданы при импорте: подменяем и в самом наборе хуков + hooks = commands.HOOKS + object.__setattr__(hooks, "before_result", sync.before_result) + try: + out = api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1", "DOC-3"]}) + finally: + object.__setattr__(hooks, "before_result", real) + assert out["status"] == "cancelled" and not api.copy.exists() + assert api.rows("SELECT * FROM wiki_attachment") == [] and api.rows("SELECT * FROM wiki_page") == [] + _clean(api) + + +def test_repeat_done_is_idempotent(api): + cmd = api.claim() + api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": TREE}) + res = {"root_title": "АРМ КА", "codes": CODES} + assert api.report(cmd, "done", res)["status"] == "done" + again = api.report(cmd, "done", res) + assert again["status"] == "done" and api.project()["wiki_pages"] == 3 + _clean(api) + assert len(api.rows("SELECT * FROM wiki_page")) == 3 + + +def test_concurrent_done_reports(api): + import threading + cmd = api.claim() + api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": _attach_pages()}) + res = {"root_title": "АРМ КА", "codes": ["DOC-1", "DOC-3"]} + out, errs = [], [] + + def go(): + try: + out.append(api.report(cmd, "done", res)["status"]) + except Exception as e: # noqa: BLE001 + errs.append(e) + + ts = [threading.Thread(target=go) for _ in range(3)] + [t.start() for t in ts] + [t.join(30) for t in ts] + assert not errs and out == ["done"] * 3 + assert api.project()["wiki_pages"] == 2 + ((att),) = api.rows("SELECT id FROM wiki_attachment") + md = (api.copy / "АРМ КА" / "01_Статусы" / "index.md").read_text(encoding="utf-8") + assert f"(attachment:{att['id']})" in md + _clean(api) + + +def test_bad_pages_after_state_check(api): + (row,) = api.rows("SELECT id FROM command WHERE type = 'wiki_sync'") + with pytest.raises(HubError) as e: # команда в queued: 409, а не 422 за форму + api.wiki.post("/agent/wiki/pages", {"command_id": row["id"], "pages": [{"code": "X"}]}) + assert e.value.status == 409 + + +def test_page_limits_and_free_code_format(api): + cmd = api.claim() + for field, value in (("title", "т" * 1001), ("url", "u" * 4001), ("code", "c" * 201)): + with pytest.raises(HubError) as e: + api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0] | {field: value}]}) + assert e.value.status == 422 + api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0] | {"code": "ABC-9"}]}) # формат не навязан + + +def test_empty_parent_code_is_null(api): + run = api.run([TREE[0] | {"parent_code": ""}], ["DOC-1"]) + assert run["status"] == "done" + assert api.rows("SELECT parent_code FROM wiki_page")[0]["parent_code"] is None + + +def test_housekeeping_ensures_fts(open_db, tmp_path): + conn = open_db(tmp_path / "h.sqlite") + assert not conn.execute("SELECT 1 FROM sqlite_master WHERE name = 'wiki_page_fts'").fetchone() + tick.housekeeping_tick(None, conn) + assert conn.execute("SELECT 1 FROM sqlite_master WHERE name = 'wiki_page_fts'").fetchone()