"""Синхронизация вики: приём пачек, сборка копии, поиск (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 sdlc_hub.core import db, queue, settings
from sdlc_hub.core.client import HubClient, HubError
from sdlc_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 sdlc_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"