"""Корень вики проекта: wiki_root в POST/PATCH /projects, сброс копии (wiki-module, spec wiki «Корень вики проекта»).""" import json import pytest from sdlc_hub.core.client import HubError from sdlc_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")] def test_patch_root_empty_string_rejected(hub_server): """M-6: PATCH {wiki_root: ""} — 422, копия цела; сбрасывает только null; POST с "" — проект без корня.""" 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" out = c.post("/projects", {"label": "S", "name": "S", "wiki_root": ""}) assert out["project"]["wiki_root"] is None assert c.patch("/projects/AWPCA", {"wiki_root": None})["project"]["wiki_root"] is None def test_wiki_mcp_imports_no_db_or_queue(): """M-5: mcp.py и roots.py вики не импортируют core.db и core.queue ни прямо, ни через wiki.projects.""" import ast from pathlib import Path base = Path(__file__).parent.parent / "src" / "sdlc_hub" / "modules" / "wiki" for name in ("mcp.py", "roots.py"): tree = ast.parse((base / name).read_text(encoding="utf-8")) mods = [] for n in ast.walk(tree): if isinstance(n, ast.ImportFrom): mods += [f"{'.' * n.level}{n.module or ''}.{a.name}" for a in n.names] elif isinstance(n, ast.Import): mods += [a.name for a in n.names] bad = [m for m in mods if m.endswith((".db", ".queue", ".projects")) or ".core.db" in m or ".core.queue" in m] assert not bad, (name, bad)