sdlc-hub/tests/test_wiki_projects.py

"""Корень вики проекта: wiki_root в POST/PATCH /projects, сброс копии (wiki-module, spec wiki «Корень вики проекта»)."""
import json

import pytest

from jira_hub.core.client import HubError
from jira_hub.modules.wiki.projects import parse_root


def test_parse_root():
    assert parse_root("https://eva.example.ru/project/Document/doc-001234") == "DOC-001234"
    assert parse_root(" DOC-7 ") == "DOC-7" and parse_root(None) is None and parse_root("") is None
    with pytest.raises(ValueError, match="DOC-"):
        parse_root("страница про статусы")
    with pytest.raises(ValueError):
        parse_root(5)


def _syncs(hub, label=None):
    """Пакеты wiki_sync: [(id, статус команды, params)] по порядку."""
    conn = hub.conn()
    try:
        return [(r["batch_id"], r["status"], json.loads(r["params"])) for r in conn.execute(
            "SELECT batch_id, status, params FROM command WHERE type = 'wiki_sync' ORDER BY id")]
    finally:
        conn.close()


def test_post_project_with_root(hub_server):
    out = hub_server["client"].post("/projects", {"label": "S", "name": "S", "wiki_root": "DOC-001234"})
    assert set(out) == {"project", "batch_id"} and out["project"]["wiki_root"] == "DOC-001234"
    hub = hub_server["hub"]
    conn = hub.conn()
    try:
        first = conn.execute("SELECT type FROM command WHERE batch_id = ?", (out["batch_id"],)).fetchone()
    finally:
        conn.close()
    assert first["type"] == "refresh"                                   # batch_id — пакет первого модуля (jira)
    assert [(s, p) for _, s, p in _syncs(hub)] == [("queued", {"label": "S", "root": "DOC-001234"})]


def test_post_project_without_root_has_no_wiki_batch(hub_server):
    out = hub_server["client"].post("/projects", {"label": "S", "name": "S"})
    assert out["project"]["wiki_root"] is None and out["project"]["wiki_pages"] is None
    assert _syncs(hub_server["hub"]) == []


def test_patch_root_existing(hub_server):
    c = hub_server["client"]
    c.post("/projects", {"label": "AWPCA", "name": "АРМ КА"})
    out = c.patch("/projects/AWPCA", {"wiki_root": "DOC-000777"})
    assert len(out["batch_ids"]) == 1 and out["project"]["wiki_root"] == "DOC-000777"
    assert [p for _, _, p in _syncs(hub_server["hub"])] == [{"label": "AWPCA", "root": "DOC-000777"}]
    got = {p["label"]: p for p in c.get("/projects")["projects"]}
    assert got["AWPCA"]["wiki_root"] == "DOC-000777"
    batch = c.get(f"/batches/{out['batch_ids'][0]}")
    assert batch["title"] == "Вики AWPCA" and batch["project"] == "AWPCA"


def test_patch_root_bad(hub_server):
    c = hub_server["client"]
    c.post("/projects", {"label": "AWPCA", "name": "АРМ КА", "wiki_root": "DOC-1"})
    with pytest.raises(HubError) as e:
        c.patch("/projects/AWPCA", {"wiki_root": "страница"})
    assert e.value.status == 422 and "wiki_root" in e.value.detail
    assert c.get("/projects")["projects"][0]["wiki_root"] == "DOC-1"
    assert len(_syncs(hub_server["hub"])) == 1


def test_patch_root_cancels_queued(hub_server):
    c = hub_server["client"]
    c.post("/projects", {"label": "AWPCA", "name": "АРМ КА", "wiki_root": "DOC-12"})
    c.patch("/projects/AWPCA", {"wiki_root": "DOC-21"})
    assert [(s, p["root"]) for _, s, p in _syncs(hub_server["hub"])] == [("cancelled", "DOC-12"), ("queued", "DOC-21")]


def test_patch_same_root_does_nothing(hub_server):
    c = hub_server["client"]
    c.post("/projects", {"label": "AWPCA", "name": "АРМ КА", "wiki_root": "DOC-12"})
    assert c.patch("/projects/AWPCA", {"wiki_root": "doc-12"})["batch_ids"] == []
    assert len(_syncs(hub_server["hub"])) == 1


def test_reset_root_removes_copy(hub_server, tmp_path):
    c = hub_server["client"]
    c.post("/projects", {"label": "AWPCA", "name": "АРМ КА", "wiki_root": "DOC-1"})
    c.post("/projects", {"label": "OTHER", "name": "O", "wiki_root": "DOC-2"})
    copy = tmp_path / "wiki" / "AWPCA" / "АРМ КА"
    copy.mkdir(parents=True)
    (copy / "index.md").write_text("# т", encoding="utf-8")
    other = tmp_path / "wiki" / "OTHER"
    other.mkdir()
    conn = hub_server["hub"].conn()
    try:
        for label in ("AWPCA", "OTHER"):
            conn.execute("INSERT INTO wiki_page(label, code, title) VALUES (?, 'DOC-1', 'т')", (label,))
            conn.execute("INSERT INTO wiki_attachment(label, page_code, url) VALUES (?, 'DOC-1', 'https://e/f')", (label,))
    finally:
        conn.close()
    out = c.patch("/projects/AWPCA", {"wiki_root": None})
    assert out["batch_ids"] == [] and out["project"]["wiki_root"] is None
    assert not (tmp_path / "wiki" / "AWPCA").exists() and other.exists()
    conn = hub_server["hub"].conn()
    try:
        for table in ("wiki_project", "wiki_page", "wiki_attachment"):
            assert [r["label"] for r in conn.execute(f"SELECT label FROM {table}")] == ["OTHER"], table
    finally:
        conn.close()
    assert [(s, p["root"]) for _, s, p in _syncs(hub_server["hub"])] == [("cancelled", "DOC-1"), ("queued", "DOC-2")]


def test_claimed_sync_is_not_cancelled_by_new_root(hub_server):
    c = hub_server["client"]
    c.post("/projects", {"label": "AWPCA", "name": "АРМ КА", "wiki_root": "DOC-12"})
    conn = hub_server["hub"].conn()
    try:
        conn.execute("UPDATE command SET status = 'claimed' WHERE type = 'wiki_sync'")
    finally:
        conn.close()
    c.patch("/projects/AWPCA", {"wiki_root": "DOC-21"})
    assert [(s, p["root"]) for _, s, p in _syncs(hub_server["hub"])] == [("claimed", "DOC-12"), ("queued", "DOC-21")]