152 lines
7.1 KiB
Python
152 lines
7.1 KiB
Python
"""Корень вики проекта: 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)
|