c8af45ced8
- root_set кодирует метку; parse_root вынесен в wiki/roots.py; PATCH wiki_root "" -> 422 - суррогаты в пачке страниц заменяются; dir_name экранирует ведущую точку; error у вложений - beautifulsoup4 и markdownify<2 объявлены; убраны неиспользуемые импорты и дубль _has_table Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
631 lines
31 KiB
Python
631 lines
31 KiB
Python
"""Синхронизация вики: приём пачек, сборка копии, поиск (wiki-module, spec wiki «Синхронизация вики», «Markdown-копия»).
|
||
Агента имитирует тест: HTTP на порту 0 токеном агента wiki."""
|
||
import json
|
||
import logging
|
||
import sqlite3
|
||
from datetime import timedelta
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from jira_hub.core import db, queue, settings
|
||
from jira_hub.core.client import HubClient, HubError
|
||
from jira_hub.modules.wiki import commands, convert, projects, search, stage, sync, tick
|
||
|
||
|
||
def page(code, title, parent=None, orderno=None, html=None, attachments=None, links=None):
|
||
return {"code": code, "eva_id": "id-" + code, "title": title, "parent_code": parent, "orderno": orderno,
|
||
"url": f"https://eva.example.ru/project/Document/{code}", "html": html if html is not None else f"<p>текст {title}</p>",
|
||
"attachments": attachments or [], "links": links or []}
|
||
|
||
|
||
TREE = [page("DOC-1", "АРМ КА"), page("DOC-2", "Роли/права", "DOC-1", 2), page("DOC-3", "Статусы", "DOC-1", 1,
|
||
"<p>статус «На согласовании» ставит аналитик</p>")]
|
||
CODES = ["DOC-1", "DOC-3", "DOC-2"]
|
||
|
||
|
||
@pytest.fixture
|
||
def api(hub_server, tmp_path):
|
||
hub = hub_server["hub"]
|
||
conn = hub.conn()
|
||
token = settings.tokens(conn)["agents"]["wiki"]
|
||
conn.close()
|
||
wiki = HubClient(hub_server["base"], token)
|
||
hub_server["client"].post("/projects", {"label": "AWPCA", "name": "АРМ КА", "wiki_root": "DOC-1"})
|
||
|
||
def db_rows(sql, *args):
|
||
c = hub.conn()
|
||
try:
|
||
return [dict(r) for r in c.execute(sql, args)]
|
||
finally:
|
||
c.close()
|
||
|
||
def enqueue():
|
||
c = hub.conn()
|
||
try:
|
||
with db.transaction(c):
|
||
return commands.enqueue_sync(c, "AWPCA", "DOC-1", "test")
|
||
finally:
|
||
c.close()
|
||
|
||
def claim():
|
||
return wiki.post("/agent/claim", {"agent_version": 1})["command"]
|
||
|
||
def report(cmd, status="done", result=None, error=None):
|
||
return wiki.post(f"/agent/commands/{cmd['id']}/result", {"status": status, "result": result, "error": error})
|
||
|
||
def run(pages, codes=None, root_title="АРМ КА", status="done", error=None):
|
||
"""Целый обмен: взять команду, прислать страницы, отчитаться. Возвращает команду после отчёта."""
|
||
cmd = claim()
|
||
assert cmd and cmd["type"] == "wiki_sync", cmd
|
||
if pages:
|
||
wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": pages})
|
||
result = {"root_title": root_title, "codes": codes if codes is not None else [p["code"] for p in pages],
|
||
"eva_calls": 3} if status == "done" else None
|
||
return report(cmd, status, result, error)
|
||
|
||
def project():
|
||
return {p["label"]: p for p in hub_server["client"].get("/projects")["projects"]}["AWPCA"]
|
||
|
||
return SimpleNamespace(hub=hub, wiki=wiki, client=hub_server["client"], base=hub_server["base"], rows=db_rows,
|
||
enqueue=enqueue, claim=claim, report=report, run=run, project=project,
|
||
copy=tmp_path / "wiki" / "AWPCA", wiki_dir=tmp_path / "wiki", db_path=tmp_path / "hub.sqlite",
|
||
jira=hub_server["agent"])
|
||
|
||
|
||
def first_sync(api):
|
||
done = api.run(TREE, CODES)
|
||
assert done["status"] == "done"
|
||
return done
|
||
|
||
|
||
def test_first_sync(api):
|
||
first_sync(api)
|
||
assert (api.copy / "АРМ КА" / "index.md").is_file()
|
||
assert (api.copy / "АРМ КА" / "01_Статусы" / "index.md").is_file()
|
||
assert (api.copy / "АРМ КА" / "02_Роли_права" / "index.md").is_file()
|
||
assert [t["code"] for t in json.loads((api.copy / "_tree.json").read_text(encoding="utf-8"))] == CODES
|
||
text = (api.copy / "АРМ КА" / "01_Статусы" / "index.md").read_text(encoding="utf-8")
|
||
assert text.startswith('---\ncode: "DOC-3"\ntitle: "Статусы"\n') and "\n# Статусы\n" in text and "аналитик" in text
|
||
p = api.project()
|
||
assert p["wiki_pages"] == 3 and p["wiki_root_title"] == "АРМ КА" and p["wiki_synced_at"] and p["wiki_sync_error"] is None
|
||
assert api.rows("SELECT COUNT(*) AS n FROM wiki_sync_stage")[0]["n"] == 0 # стадия очищена
|
||
assert {r["code"]: r["path"] for r in api.rows("SELECT code, path FROM wiki_page")} == {
|
||
"DOC-1": "АРМ КА", "DOC-3": "АРМ КА/01_Статусы", "DOC-2": "АРМ КА/02_Роли_права"}
|
||
assert oct(api.copy.stat().st_mode & 0o777) == "0o700" and oct(api.wiki_dir.stat().st_mode & 0o777) == "0o700"
|
||
assert oct((api.copy / "АРМ КА" / "01_Статусы").stat().st_mode & 0o777) == "0o700"
|
||
|
||
|
||
def test_failed_keeps_copy(api):
|
||
first_sync(api)
|
||
api.enqueue()
|
||
api.run(None, status="failed", error="Eva: HTTP 503")
|
||
p = api.project()
|
||
assert p["wiki_sync_error"] == "Eva: HTTP 503" and p["wiki_pages"] == 3
|
||
assert (api.copy / "АРМ КА" / "01_Статусы" / "index.md").is_file()
|
||
conn = api.hub.conn()
|
||
try:
|
||
assert [r["code"] for r in search.query(conn, "AWPCA", "согласов", 10)] == ["DOC-3"]
|
||
finally:
|
||
conn.close()
|
||
assert api.rows("SELECT failed_at FROM wiki_project")[0]["failed_at"]
|
||
|
||
|
||
def test_missing_page_fails(api):
|
||
first_sync(api)
|
||
api.enqueue()
|
||
done = api.run([TREE[0]], ["DOC-1", "DOC-2"])
|
||
assert done["status"] == "failed" and done["error"] == "нет страницы DOC-2"
|
||
assert api.project()["wiki_sync_error"] == "нет страницы DOC-2"
|
||
assert (api.copy / "АРМ КА" / "02_Роли_права" / "index.md").is_file()
|
||
assert api.rows("SELECT COUNT(*) AS n FROM wiki_sync_stage")[0]["n"] == 0
|
||
|
||
|
||
def test_page_removed(api):
|
||
shot = page("DOC-3", "Статусы", "DOC-1", 1, '<p>см. <img src="/files/x/scheme.png"></p>',
|
||
[{"ref": "/files/x/scheme.png", "url": "https://eva.example.ru/files/x/scheme.png", "name": "scheme.png"}])
|
||
api.run([TREE[0], TREE[1], shot], CODES)
|
||
assert len(api.rows("SELECT * FROM wiki_attachment")) == 1
|
||
api.enqueue()
|
||
api.run([TREE[0], page("DOC-3", "Статусы", "DOC-1", 1, "<p>без картинки</p>")], ["DOC-1", "DOC-3"])
|
||
assert not list(api.copy.rglob("*Роли*"))
|
||
assert api.rows("SELECT code FROM wiki_page ORDER BY code") == [{"code": "DOC-1"}, {"code": "DOC-3"}]
|
||
assert api.rows("SELECT * FROM wiki_attachment") == []
|
||
conn = api.hub.conn()
|
||
try:
|
||
assert search.query(conn, "AWPCA", "Роли", 10) == []
|
||
assert api.project()["wiki_pages"] == 2
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _with_file(text="см. файл"):
|
||
att = [{"ref": "/files/x/model.xlsx", "url": "https://eva.example.ru/files/x/model.xlsx", "name": "model.xlsx"}]
|
||
return [TREE[0], page("DOC-3", "Статусы", "DOC-1", 1, f'<p>{text} <a href="/files/x/model.xlsx">model</a></p>', att)]
|
||
|
||
|
||
def test_attachment_id_survives(api):
|
||
api.run(_with_file(), ["DOC-1", "DOC-3"])
|
||
(a,) = api.rows("SELECT id, state FROM wiki_attachment")
|
||
md = (api.copy / "АРМ КА" / "01_Статусы" / "index.md").read_text(encoding="utf-8")
|
||
assert f"[📎 model.xlsx](attachment:{a['id']})" in md and a["state"] == "none"
|
||
c = api.hub.conn()
|
||
c.execute("UPDATE wiki_attachment SET state = 'ready', file = 'ab', size = 5 WHERE id = ?", (a["id"],))
|
||
c.close()
|
||
api.enqueue()
|
||
api.run(_with_file("новый текст"), ["DOC-1", "DOC-3"])
|
||
(b,) = api.rows("SELECT id, state, file FROM wiki_attachment")
|
||
assert (b["id"], b["state"], b["file"]) == (a["id"], "ready", "ab")
|
||
assert f"(attachment:{a['id']})" in (api.copy / "АРМ КА" / "01_Статусы" / "index.md").read_text(encoding="utf-8")
|
||
|
||
|
||
def test_stale_root_discarded(api):
|
||
cmd = api.claim()
|
||
assert cmd["params"]["root"] == "DOC-1"
|
||
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0]]})
|
||
api.client.patch("/projects/AWPCA", {"wiki_root": "DOC-21"})
|
||
done = api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1"]})
|
||
assert done["status"] == "done" and done["result"]["discarded"] == "корень сменился"
|
||
assert not api.copy.exists() and api.rows("SELECT * FROM wiki_page") == []
|
||
p = api.project()
|
||
assert p["wiki_root"] == "DOC-21" and p["wiki_pages"] == 0 and p["wiki_synced_at"] is None
|
||
assert api.rows("SELECT COUNT(*) AS n FROM wiki_sync_stage")[0]["n"] == 0
|
||
assert api.rows("SELECT failed_at FROM wiki_project")[0]["failed_at"] # часы автосинхронизации сдвинуты
|
||
|
||
|
||
def test_redelivery_clears_stage(api):
|
||
cmd = api.claim()
|
||
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0]]})
|
||
assert api.rows("SELECT COUNT(*) AS n FROM wiki_sync_stage")[0]["n"] == 1
|
||
again = api.claim()
|
||
assert again["id"] == cmd["id"] and api.rows("SELECT COUNT(*) AS n FROM wiki_sync_stage")[0]["n"] == 0
|
||
done = api.report(again, "done", {"root_title": "X", "codes": ["DOC-1"]})
|
||
assert done["status"] == "failed" and done["error"] == "нет страницы DOC-1"
|
||
|
||
|
||
def test_pages_wrong_agent(api):
|
||
cmd = api.claim()
|
||
with pytest.raises(HubError) as e:
|
||
api.jira.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0]]})
|
||
assert e.value.status == 401
|
||
|
||
|
||
def test_pages_not_claimed(api):
|
||
(row,) = api.rows("SELECT id FROM command WHERE type = 'wiki_sync'")
|
||
with pytest.raises(HubError) as e:
|
||
api.wiki.post("/agent/wiki/pages", {"command_id": row["id"], "pages": [TREE[0]]})
|
||
assert e.value.status == 409 and e.value.error == "not_claimed"
|
||
|
||
|
||
def test_bad_page(api):
|
||
cmd = api.claim()
|
||
bad = {k: v for k, v in TREE[1].items() if k != "title"}
|
||
with pytest.raises(HubError) as e:
|
||
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0], bad]})
|
||
assert e.value.status == 422 and e.value.error == "bad_pages" and e.value.detail == "DOC-2: title"
|
||
assert api.rows("SELECT COUNT(*) AS n FROM wiki_sync_stage")[0]["n"] == 0 # пачка не принята целиком
|
||
for field, value in (("orderno", "1"), ("parent_code", 5), ("html", None)):
|
||
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 and e.value.detail == f"DOC-1: {field}"
|
||
|
||
|
||
def test_layout_names_safe(tmp_path):
|
||
titles = ["Корень", "../x", "a/b", "😀" * 200, "Статусы", "Статусы", "..", " . ", 'a<b>:"c|?*']
|
||
pages = [{"code": f"D{i}", "title": t, "markdown": "текст\n", "attachments": [], "orderno": i,
|
||
"parent_code": None if i == 0 else "D0", "eva_id": None, "url": None} for i, t in enumerate(titles)]
|
||
target = tmp_path / "t"
|
||
rows = sync.build_copy(pages, [p["code"] for p in pages], target, {})
|
||
paths_ = [r["path"] for r in rows]
|
||
for r in rows:
|
||
assert (target / r["path"] / "index.md").resolve().is_relative_to(target.resolve())
|
||
assert ".." not in r["path"].split("/") and all(len(part.encode()) <= 255 for part in r["path"].split("/"))
|
||
assert "Корень/04_Статусы" in paths_ and "Корень/05_Статусы_2" in paths_
|
||
assert "Корень/06_untitled" in paths_ and "Корень/07_untitled_2" in paths_
|
||
assert "Корень/02_a_b" in paths_ and "Корень/01__x" in paths_
|
||
assert sorted(p.name for p in (target / "Корень").iterdir() if p.is_dir()) == sorted(p.split("/")[1] for p in paths_[1:])
|
||
|
||
|
||
def test_layout_orphans_and_cycles(tmp_path):
|
||
def mk(code, parent, order=None):
|
||
return {"code": code, "title": code, "markdown": "x\n", "attachments": [], "orderno": order, "parent_code": parent,
|
||
"eva_id": None, "url": None}
|
||
pages = [mk("R", None), mk("A", "R", 2), mk("B", "R"), mk("C", "GONE"), mk("X", "Y"), mk("Y", "X")]
|
||
rows = sync.build_copy(pages, [p["code"] for p in pages], tmp_path / "t", {})
|
||
assert {r["code"] for r in rows} == {"R", "A", "B", "C", "X", "Y"} and len(rows) == 6 # никто не потерян
|
||
assert [r["code"] for r in rows if r["depth"] == 1][:2] == ["A", "B"] # orderno, затем None по коду
|
||
|
||
|
||
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 / "нет") # нет каталога — не ошибка
|
||
|
||
|
||
def test_search_prefix(open_db):
|
||
conn = open_db()
|
||
conn.execute("INSERT INTO project(label, name, created_at) VALUES ('AWPCA', 'A', '2026-01-01')")
|
||
assert search.ensure(conn) is True and search.ensure(conn) is True
|
||
search.replace(conn, "AWPCA", [("DOC-3", "Статусы", "статус «На согласовании» ставит аналитик"),
|
||
("DOC-4", "Роли", "роль администратора")])
|
||
search.replace(conn, "OTHER", [("DOC-9", "Чужая", "согласование чужого проекта")])
|
||
conn.execute("INSERT INTO wiki_page(label, code, title, path) VALUES ('AWPCA', 'DOC-3', 'Статусы', 'A/01_Статусы')")
|
||
(hit,) = search.query(conn, "AWPCA", "согласов", 10)
|
||
assert hit["code"] == "DOC-3" and "На согласовании" in hit["snippet"] and hit["path"] == "A/01_Статусы"
|
||
assert search.query(conn, "AWPCA", 'согласов"', 10)[0]["code"] == "DOC-3" # кавычка не роняет
|
||
for q in ('"', "", "* AND (", "NEAR(", "статус ОR"):
|
||
assert isinstance(search.query(conn, "AWPCA", q, 5), list)
|
||
assert [r["code"] for r in search.query(conn, "AWPCA", "роль админ", 10)] == ["DOC-4"]
|
||
assert search.query(conn, "AWPCA", "чужого", 10) == []
|
||
|
||
|
||
def test_search_without_fts_uses_files(api, monkeypatch):
|
||
first_sync(api)
|
||
c = api.hub.conn()
|
||
try:
|
||
c.execute("DROP TABLE wiki_page_fts")
|
||
(hit,) = search.query(c, "AWPCA", "согласов", 10)
|
||
assert hit["code"] == "DOC-3" and "На согласовании" in hit["snippet"]
|
||
finally:
|
||
c.close()
|
||
|
||
|
||
# --- пункты контроллера ---------------------------------------------------------
|
||
|
||
def test_failed_at_prevents_tick_storm(api):
|
||
"""Любая неудача пишет failed_at, и автотик не ставит новый пакет сразу."""
|
||
def tick_once():
|
||
c = api.hub.conn()
|
||
try:
|
||
tick.auto_sync_tick(None, c)
|
||
finally:
|
||
c.close()
|
||
return api.rows("SELECT COUNT(*) AS n FROM batch WHERE project = 'AWPCA'")[0]["n"]
|
||
|
||
first_sync(api)
|
||
assert api.rows("SELECT synced_at, failed_at FROM wiki_project")[0]["synced_at"]
|
||
assert tick_once() == 1 # свежая успешная
|
||
c = api.hub.conn()
|
||
c.execute("UPDATE wiki_project SET synced_at = ?, failed_at = NULL", ((db.utcnow() - timedelta(hours=30)).isoformat(),))
|
||
c.close()
|
||
# failed: пакет выдан, агент отчитался failed
|
||
api.enqueue()
|
||
api.run(None, status="failed", error="Eva: HTTP 503")
|
||
assert tick_once() == 2
|
||
# отброшенная
|
||
c = api.hub.conn()
|
||
c.execute("UPDATE wiki_project SET synced_at = ?, failed_at = NULL", ((db.utcnow() - timedelta(hours=30)).isoformat(),))
|
||
c.close()
|
||
api.enqueue()
|
||
claimed = api.claim()
|
||
api.client.patch("/projects/AWPCA", {"wiki_root": "DOC-21"})
|
||
api.report(claimed, "done", {"root_title": "X", "codes": ["DOC-1"]})
|
||
assert api.rows("SELECT failed_at FROM wiki_project")[0]["failed_at"]
|
||
# отменённая
|
||
c = api.hub.conn()
|
||
c.execute("UPDATE wiki_project SET synced_at = ?, failed_at = NULL", ((db.utcnow() - timedelta(hours=30)).isoformat(),))
|
||
queued = api.rows("SELECT id FROM command WHERE type = 'wiki_sync' AND status = 'queued'")
|
||
for q in queued:
|
||
queue.cancel_command(c, q["id"])
|
||
c.close()
|
||
c = api.hub.conn()
|
||
try:
|
||
tick.housekeeping_tick(None, c)
|
||
assert api.rows("SELECT failed_at FROM wiki_project")[0]["failed_at"]
|
||
before = api.rows("SELECT COUNT(*) AS n FROM batch")[0]["n"]
|
||
tick.auto_sync_tick(None, c)
|
||
assert api.rows("SELECT COUNT(*) AS n FROM batch")[0]["n"] == before
|
||
finally:
|
||
c.close()
|
||
|
||
|
||
def test_fill_ids_error_becomes_failed(api):
|
||
cmd = api.claim()
|
||
bad = {"code": "DOC-1", "eva_id": None, "title": "АРМ КА", "parent_code": None, "orderno": None, "url": None,
|
||
"markdown": "[📎 f](attachment:@5)\n", "attachments": []}
|
||
c = api.hub.conn()
|
||
c.execute("INSERT INTO wiki_sync_stage VALUES (?, 'DOC-1', ?)", (cmd["id"], json.dumps(bad)))
|
||
c.close()
|
||
done = api.report(cmd, "done", {"root_title": "X", "codes": ["DOC-1"]})
|
||
assert done["status"] == "failed" and done["error"].startswith("DOC-1: ") and "вне списка" in done["error"]
|
||
assert api.project()["wiki_sync_error"].startswith("DOC-1: ") and not api.copy.exists()
|
||
assert not list(api.wiki_dir.glob(".*")) # .new убран
|
||
|
||
|
||
def test_long_name_failed_not_crash(hub_server, tmp_path):
|
||
label = "/" * 100 # dir_name даёт 300 символов
|
||
hub_server["client"].post("/projects", {"label": label, "name": "длинная", "wiki_root": "DOC-1"})
|
||
hub = hub_server["hub"]
|
||
conn = hub.conn()
|
||
token = settings.tokens(conn)["agents"]["wiki"]
|
||
conn.close()
|
||
wiki = HubClient(hub_server["base"], token)
|
||
cmd = wiki.post("/agent/claim", {})["command"]
|
||
wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [TREE[0]]})
|
||
done = wiki.post(f"/agent/commands/{cmd['id']}/result", {"status": "done", "result": {"root_title": "X", "codes": ["DOC-1"]}})
|
||
assert done["status"] == "failed" and done["error"].startswith("не удалось собрать копию")
|
||
conn = hub.conn()
|
||
try:
|
||
row = conn.execute("SELECT sync_error, failed_at FROM wiki_project WHERE label = ?", (label,)).fetchone()
|
||
finally:
|
||
conn.close()
|
||
assert row["sync_error"].startswith("не удалось собрать копию") and row["failed_at"]
|
||
|
||
|
||
def test_reset_root_clears_fts_and_dir(api):
|
||
first_sync(api)
|
||
c = api.hub.conn()
|
||
assert c.execute("SELECT COUNT(*) AS n FROM wiki_page_fts").fetchone()["n"] == 3
|
||
c.close()
|
||
api.client.patch("/projects/AWPCA", {"wiki_root": None})
|
||
c = api.hub.conn()
|
||
try:
|
||
assert c.execute("SELECT COUNT(*) AS n FROM wiki_page_fts").fetchone()["n"] == 0
|
||
assert c.execute("SELECT COUNT(*) AS n FROM wiki_page").fetchone()["n"] == 0
|
||
finally:
|
||
c.close()
|
||
assert not api.copy.exists()
|
||
|
||
|
||
def test_drop_copy_logs_rmtree_failure(api, monkeypatch, caplog):
|
||
first_sync(api)
|
||
real = projects.shutil.rmtree
|
||
|
||
def boom(path, *a, **k):
|
||
if "AWPCA" in str(path):
|
||
raise PermissionError("нельзя")
|
||
return real(path, *a, **k)
|
||
|
||
monkeypatch.setattr(projects.shutil, "rmtree", boom)
|
||
with caplog.at_level(logging.WARNING, logger=projects.__name__):
|
||
api.client.patch("/projects/AWPCA", {"wiki_root": None})
|
||
assert "не удалён" in caplog.text and "нельзя" in caplog.text
|
||
|
||
|
||
def test_no_heavy_work_under_write_lock(api, monkeypatch):
|
||
"""Перевод HTML и запись файлов идут вне BEGIN IMMEDIATE: параллельное соединение может открыть запись."""
|
||
seen = []
|
||
|
||
def probe(tag):
|
||
other = sqlite3.connect(api.db_path, timeout=0)
|
||
try:
|
||
other.execute("BEGIN IMMEDIATE")
|
||
other.execute("ROLLBACK")
|
||
seen.append((tag, "free"))
|
||
except sqlite3.OperationalError:
|
||
seen.append((tag, "locked"))
|
||
finally:
|
||
other.close()
|
||
|
||
real_md, real_build = convert.to_markdown, sync.build_copy
|
||
monkeypatch.setattr(stage.convert, "to_markdown", lambda *a, **k: (probe("markdown"), real_md(*a, **k))[1])
|
||
monkeypatch.setattr(sync, "build_copy", lambda *a, **k: (probe("files"), real_build(*a, **k))[1])
|
||
first_sync(api)
|
||
assert ("markdown", "free") in seen and ("files", "free") in seen and not [s for s in seen if s[1] == "locked"]
|
||
|
||
|
||
def test_swap_failure_rolls_back_rows(api, monkeypatch):
|
||
first_sync(api)
|
||
api.enqueue()
|
||
|
||
def boom(cur, new, old):
|
||
raise OSError("диск полон")
|
||
|
||
monkeypatch.setattr(sync, "_swap", boom)
|
||
done = api.run([TREE[0], page("DOC-3", "Новая", "DOC-1", 1)], ["DOC-1", "DOC-3"])
|
||
assert done["status"] == "failed" and "не удалось подменить копию" in done["error"]
|
||
assert api.project()["wiki_pages"] == 3 and "диск полон" in api.project()["wiki_sync_error"]
|
||
assert [r["code"] for r in api.rows("SELECT code FROM wiki_page ORDER BY code")] == ["DOC-1", "DOC-2", "DOC-3"]
|
||
assert (api.copy / "АРМ КА" / "01_Статусы" / "index.md").is_file() and not list(api.wiki_dir.glob(".*"))
|
||
|
||
|
||
def test_batch_limit_and_replace(api):
|
||
cmd = api.claim()
|
||
with pytest.raises(HubError) as e:
|
||
api.wiki.post("/agent/wiki/pages", {"command_id": cmd["id"], "pages": [page(f"DOC-{i}", "t") for i in range(51)]})
|
||
assert e.value.status == 422
|
||
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, '<img src="/files/x/a.png">', 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()
|
||
|
||
|
||
def test_lone_surrogate_in_page_does_not_break_sync(api):
|
||
"""M-8: одиночный суррогат (JSON-эскейп \\ud83d без пары) в html/title не роняет запись стадии (раньше 500)."""
|
||
import urllib.request
|
||
bad = [page("DOC-1", "Корень \ud83d", html="<p>текст \ud83d конец</p>")]
|
||
cmd = api.claim()
|
||
body = json.dumps({"command_id": cmd["id"], "pages": bad}).encode("ascii") # ensure_ascii: эскейп остаётся
|
||
req = urllib.request.Request(api.base + "/agent/wiki/pages", data=body, method="POST",
|
||
headers={"Authorization": "Bearer " + api.wiki.token, "Content-Type": "application/json"})
|
||
with urllib.request.urlopen(req) as r:
|
||
assert r.status == 200
|
||
api.report(cmd, "done", {"root_title": "К", "codes": ["DOC-1"], "eva_calls": 1})
|
||
assert api.rows("SELECT title FROM wiki_page WHERE code = 'DOC-1'")[0]["title"] == "Корень ?" # кодек заменяет суррогат на «?»
|
||
|
||
|
||
|
||
def test_dir_name_escapes_leading_dot():
|
||
"""M-14: метка с ведущей точкой не даёт каталог вида .X.new.5 / .X.old, который уборка при старте сочла бы своим."""
|
||
from jira_hub.modules.wiki import paths
|
||
assert paths.dir_name(".X.new.5") == "%2EX.new.5" and paths.dir_name(".") == "%2E" and paths.dir_name("..") == "%2E."
|
||
assert not paths.dir_name(".X.old").startswith(".") and paths.dir_name("A.B") == "A.B"
|