224 lines
11 KiB
Python
224 lines
11 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", "batches"} 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)
|
||
|
||
|
||
def test_batch_module(hub_server):
|
||
"""Модуль пакета — по типу первой команды; пакет без команд — None (list_batches и batch_view)."""
|
||
c, hub = hub_server["client"], hub_server["hub"]
|
||
out = c.post("/projects", {"label": "S", "name": "S", "wiki_root": "DOC-001234"})
|
||
wiki_id = out["batches"][1]["id"]
|
||
conn = hub.conn()
|
||
try:
|
||
cur = conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('пустой', 'api', '2026-10-24T10:00:00')")
|
||
empty_id = cur.lastrowid
|
||
finally:
|
||
conn.close()
|
||
assert c.get(f"/batches/{out['batch_id']}")["module"] == "jira"
|
||
assert c.get(f"/batches/{wiki_id}")["module"] == "wiki"
|
||
assert c.get(f"/batches/{empty_id}")["module"] is None
|
||
assert {b["id"]: b["module"] for b in c.get("/batches")["batches"]} == {
|
||
out["batch_id"]: "jira", wiki_id: "wiki", empty_id: None}
|
||
|
||
|
||
def test_batches_module_filter(hub_server):
|
||
c = hub_server["client"]
|
||
c.post("/projects", {"label": "AWPCA", "name": "A"})
|
||
out = c.patch("/projects/AWPCA", {"wiki_root": "DOC-000777"})
|
||
got = c.get("/batches", module="wiki", project="AWPCA")["batches"]
|
||
assert [b["id"] for b in got] == out["batch_ids"] and {b["module"] for b in got} == {"wiki"}
|
||
assert [b["module"] for b in c.get("/batches", module="jira")["batches"]] == ["jira"]
|
||
with pytest.raises(HubError) as e:
|
||
c.get("/batches", module="nope")
|
||
assert e.value.status == 404 and e.value.error == "not_found"
|
||
|
||
|
||
def test_create_project_batches(hub_server):
|
||
c = hub_server["client"]
|
||
out = c.post("/projects", {"label": "S", "name": "S", "wiki_root": "DOC-001234"})
|
||
ids = [b["id"] for b in out["batches"]]
|
||
assert out["batches"] == [{"module": "jira", "id": ids[0]}, {"module": "wiki", "id": ids[1]}]
|
||
assert out["batch_id"] == ids[0]
|
||
c.post("/projects", {"label": "AWPCA", "name": "A"})
|
||
patched = c.patch("/projects/AWPCA", {"wiki_root": "DOC-000777"})
|
||
assert patched["batches"] == [{"module": "wiki", "id": patched["batch_ids"][0]}]
|
||
|
||
|
||
def test_root_back_while_old_claimed(hub_server):
|
||
"""A → B → A, пока синхронизация A у агента: PATCH не 422, идущий пакет A остаётся (#135, T8.3)."""
|
||
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"})
|
||
c.patch("/projects/AWPCA", {"wiki_root": "DOC-12"})
|
||
assert [(s, p["root"]) for _, s, p in _syncs(hub_server["hub"])] == [("claimed", "DOC-12"), ("cancelled", "DOC-21")]
|
||
|
||
|
||
def test_retry_while_sync_running_refused(hub_server):
|
||
"""Повтор упавшей wiki_sync при идущей того же корня — 422 с номером идущего пакета (#135, T8.3)."""
|
||
c = hub_server["client"]
|
||
c.post("/projects", {"label": "AWPCA", "name": "АРМ КА", "wiki_root": "DOC-1"})
|
||
conn = hub_server["hub"].conn()
|
||
try:
|
||
failed = conn.execute("SELECT id FROM command WHERE type = 'wiki_sync'").fetchone()["id"]
|
||
conn.execute("UPDATE command SET status = 'failed' WHERE id = ?", (failed,))
|
||
finally:
|
||
conn.close()
|
||
running = c.post("/wiki/sync", {"project": "AWPCA"})["batch_id"]
|
||
with pytest.raises(HubError) as e:
|
||
c.post(f"/commands/{failed}/retry", {})
|
||
assert e.value.status == 422 and f"#{running}" in str(e.value)
|
||
assert len(_syncs(hub_server["hub"])) == 2
|