471 lines
24 KiB
Python
471 lines
24 KiB
Python
"""MCP вики (wiki-module, spec mcp «MCP вики»): клиент хаба подменяется, живой хаб не нужен."""
|
||
import asyncio
|
||
|
||
import pytest
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from sdlc_hub.core import mcp as core_mcp
|
||
from sdlc_hub.core import registry
|
||
from sdlc_hub.core.client import HubError
|
||
from sdlc_hub.modules.wiki import mcp as wmcp
|
||
|
||
PROJECTS = {"projects": [
|
||
{"label": "AWPCA", "name": "АРМ КА", "wiki_root": "DOC-1", "wiki_root_title": "Корень", "wiki_pages": 3,
|
||
"wiki_synced_at": "2026-10-24T10:00:00+00:00"},
|
||
{"label": "sfm_kedr", "name": "СФМ", "wiki_root": None, "wiki_pages": None, "wiki_synced_at": None},
|
||
{"label": "fresh", "name": "Новый", "wiki_root": "DOC-9", "wiki_pages": 0, "wiki_synced_at": None}]}
|
||
TREE = {"pages": [
|
||
{"code": "DOC-1", "title": "Корень", "depth": 0, "parent_code": None, "size": 100},
|
||
{"code": "DOC-2", "title": "Статусы", "depth": 1, "parent_code": "DOC-1", "size": 200},
|
||
{"code": "DOC-3", "title": "Статистика", "depth": 1, "parent_code": "DOC-1", "size": 300},
|
||
{"code": "DOC-4", "title": "Глубокая", "depth": 2, "parent_code": "DOC-3", "size": 50},
|
||
{"code": "DOC-5", "title": "Ещё глубже", "depth": 3, "parent_code": "DOC-4", "size": 10}]}
|
||
|
||
|
||
class Fake:
|
||
"""Клиент хаба: ответы по пути; callable получает (method, params/body)."""
|
||
def __init__(self, routes):
|
||
self.routes, self.calls = routes, []
|
||
|
||
def _answer(self, method, path, arg):
|
||
self.calls.append((method, path, arg))
|
||
r = self.routes[(method, path)] if (method, path) in self.routes else self.routes[path]
|
||
if isinstance(r, list):
|
||
r = r.pop(0) if len(r) > 1 else r[0]
|
||
if isinstance(r, Exception):
|
||
raise r
|
||
return r(arg) if callable(r) else r
|
||
|
||
def get(self, path, **params):
|
||
return self._answer("GET", path, params)
|
||
|
||
def post(self, path, body=None):
|
||
return self._answer("POST", path, body)
|
||
|
||
def patch(self, path, body):
|
||
return self._answer("PATCH", path, body)
|
||
|
||
def get_bytes(self, path):
|
||
return self._answer("GET", path, None)
|
||
|
||
|
||
@pytest.fixture
|
||
def hub(monkeypatch, tmp_path):
|
||
monkeypatch.setenv("SDLC_HUB_HOME", str(tmp_path / "home"))
|
||
monkeypatch.setattr(wmcp, "POLL_SECONDS", 0.01)
|
||
monkeypatch.setattr(core_mcp, "PROJECT", "AWPCA")
|
||
|
||
def make(routes, project="AWPCA"):
|
||
monkeypatch.setattr(core_mcp, "PROJECT", project)
|
||
fake = Fake({"/projects": PROJECTS, **routes})
|
||
monkeypatch.setattr(core_mcp, "client", lambda: fake)
|
||
return fake
|
||
return make
|
||
|
||
|
||
def paths_of(fake):
|
||
return [p for _, p, _ in fake.calls]
|
||
|
||
|
||
def test_unbound_tools_refuse(monkeypatch):
|
||
monkeypatch.setattr(core_mcp, "PROJECT", None)
|
||
fake = Fake({})
|
||
monkeypatch.setattr(core_mcp, "client", lambda: fake)
|
||
assert wmcp.tree() == core_mcp.NOT_BOUND
|
||
assert wmcp.search("x") == core_mcp.NOT_BOUND
|
||
assert wmcp.status() == core_mcp.NOT_BOUND
|
||
assert wmcp.changes() == core_mcp.NOT_BOUND
|
||
assert wmcp.diff("DOC-1") == core_mcp.NOT_BOUND
|
||
assert wmcp.root_set("DOC-1") == core_mcp.NOT_BOUND
|
||
assert asyncio.run(wmcp.sync()) == core_mcp.NOT_BOUND
|
||
assert fake.calls == []
|
||
|
||
|
||
def test_no_root_message(hub):
|
||
fake = hub({}, project="sfm_kedr")
|
||
msg = "у проекта sfm_kedr нет корня вики: root_set DOC-…"
|
||
assert wmcp.tree() == msg
|
||
assert wmcp.page(code="DOC-1") == msg
|
||
assert wmcp.search("x") == msg
|
||
assert wmcp.changes() == msg and wmcp.diff("DOC-1") == msg
|
||
assert asyncio.run(wmcp.sync()) == msg
|
||
assert asyncio.run(wmcp.attachment(1)) == msg
|
||
assert not any(p.startswith("/wiki/") for p in paths_of(fake))
|
||
|
||
|
||
def test_not_synced_message(hub):
|
||
fake = hub({}, project="fresh")
|
||
msg = "вики проекта ещё не синхронизирована: sync"
|
||
assert wmcp.tree() == msg and wmcp.search("x") == msg and wmcp.page(code="DOC-1") == msg
|
||
assert not any(p.startswith("/wiki/") for p in paths_of(fake))
|
||
|
||
|
||
def test_page_chunks(hub):
|
||
text = "я" * 45000
|
||
page = {"code": "DOC-3", "title": "Статистика", "path": "корень/стат", "url": "https://eva/doc3",
|
||
"synced_at": "2026-10-24T10:00:00+00:00", "markdown": text,
|
||
"attachments": [{"id": 7, "name": "m.xlsx", "state": "none", "size": 5}]}
|
||
hub({"/wiki/pages/DOC-3": page})
|
||
first = wmcp.page(code="DOC-3")
|
||
assert first.endswith("\nпродолжение: offset=20000")
|
||
assert "я" * 20000 in first and "я" * 20001 not in first
|
||
assert "DOC-3" in first and "корень/стат" in first and "https://eva/doc3" in first and "#7 m.xlsx" in first
|
||
last = wmcp.page(code="DOC-3", offset=40000)
|
||
assert "продолжение" not in last and last.endswith("я" * 5000)
|
||
mid = wmcp.page(code="DOC-3", offset=20000)
|
||
assert mid.endswith("продолжение: offset=40000")
|
||
|
||
|
||
def test_page_offset_past_end(hub):
|
||
hub({"/wiki/pages/DOC-3": {"code": "DOC-3", "title": "T", "path": "p", "url": "u", "synced_at": "s",
|
||
"markdown": "abc", "attachments": []}})
|
||
out = wmcp.page(code="DOC-3", offset=10)
|
||
assert "за концом" in out and "длина 3" in out and "продолжение" not in out
|
||
assert "за концом" in wmcp.page(code="DOC-3", offset=3) and "длина 3" in out
|
||
|
||
|
||
def test_page_by_title_candidates(hub):
|
||
fake = hub({"/wiki/tree": TREE})
|
||
out = wmcp.page(title="Стат")
|
||
assert "DOC-2 Статусы" in out and "DOC-3 Статистика" in out and "DOC-1" not in out
|
||
assert not any(p.startswith("/wiki/pages") for p in paths_of(fake))
|
||
|
||
|
||
def test_page_by_title_exact_case_insensitive(hub):
|
||
page = {"code": "DOC-3", "title": "Статистика", "path": "p", "url": "u", "synced_at": "s", "markdown": "тело",
|
||
"attachments": []}
|
||
fake = hub({"/wiki/tree": TREE, "/wiki/pages/DOC-3": page})
|
||
out = wmcp.page(title="статистика")
|
||
assert "тело" in out and "/wiki/pages/DOC-3" in paths_of(fake)
|
||
|
||
|
||
def test_page_by_title_ambiguous_and_missing(hub):
|
||
tree = {"pages": TREE["pages"] + [{"code": "DOC-8", "title": "статистика", "depth": 1, "parent_code": "DOC-1", "size": 1}]}
|
||
hub({"/wiki/tree": tree})
|
||
out = wmcp.page(title="Статистика")
|
||
assert "DOC-3 Статистика" in out and "DOC-8 статистика" in out and "тело" not in out
|
||
assert "нет страницы" in wmcp.page(title="Нет такой")
|
||
assert "code или title" in wmcp.page()
|
||
|
||
|
||
def test_page_candidates_capped(hub):
|
||
tree = {"pages": [{"code": f"DOC-{i}", "title": f"Раздел {i}", "depth": 1, "parent_code": None, "size": 1} for i in range(30)]}
|
||
hub({"/wiki/tree": tree})
|
||
out = wmcp.page(title="Раздел")
|
||
assert len([l for l in out.splitlines() if l.startswith("DOC-")]) == 10
|
||
|
||
|
||
def test_tree_depth_and_subtree(hub):
|
||
fake = hub({"/wiki/tree": TREE})
|
||
out = wmcp.tree()
|
||
lines = out.splitlines()
|
||
assert lines[0].startswith("DOC-1 Корень") and lines[1].startswith(" DOC-2 Статусы")
|
||
assert "DOC-4" in out and "DOC-5" not in out # depth=2 по умолчанию: корень + два уровня
|
||
sub = wmcp.tree(code="DOC-3", depth=1)
|
||
assert sub.splitlines()[0].startswith("DOC-3") and "DOC-4" in sub and "DOC-5" not in sub and "DOC-2" not in sub
|
||
assert "нет страницы DOC-9" in wmcp.tree(code="DOC-9")
|
||
assert fake.calls[-1][2] == {"project": "AWPCA"}
|
||
|
||
|
||
def test_search(hub):
|
||
fake = hub({"/wiki/search": {"results": [{"code": "DOC-2", "title": "Статусы", "path": "к/с", "snippet": "…согласован…"}]}})
|
||
out = wmcp.search("статус согласован", limit=5)
|
||
assert "DOC-2" in out and "Статусы" in out and "к/с" in out and "…согласован…" in out
|
||
assert fake.calls[-1] == ("GET", "/wiki/search", {"project": "AWPCA", "q": "статус согласован", "limit": 5})
|
||
hub({"/wiki/search": {"results": []}})
|
||
assert wmcp.search("zzz") == "ничего не найдено"
|
||
|
||
|
||
def test_root_set(hub):
|
||
fake = hub({("PATCH", "/projects/sfm_kedr"): {"project": {"label": "sfm_kedr", "wiki_root": "DOC-42"}, "batch_ids": [77]}},
|
||
project="sfm_kedr")
|
||
out = wmcp.root_set("https://eva.example/docs/doc-42")
|
||
assert "DOC-42" in out and "#77" in out
|
||
assert fake.calls[-1] == ("PATCH", "/projects/sfm_kedr", {"wiki_root": "DOC-42", "author": "claude"})
|
||
|
||
|
||
def test_root_set_invalid_refused_without_request(hub):
|
||
fake = hub({}, project="sfm_kedr")
|
||
for bad in ("", " ", "страница", "DOCX-1", 5):
|
||
out = wmcp.root_set(bad)
|
||
assert "DOC-" in out and "ошибка хаба" not in out, bad
|
||
assert not any(m == "PATCH" for m, _, _ in fake.calls)
|
||
|
||
|
||
def test_root_set_same_root_no_batch(hub):
|
||
hub({("PATCH", "/projects/AWPCA"): {"project": {}, "batch_ids": []}})
|
||
out = wmcp.root_set("DOC-1")
|
||
assert "DOC-1" in out and "пакета синхронизации не поставлено" in out
|
||
|
||
|
||
def agent_ok():
|
||
return {"online": True, "agent_version": "1", "settings_version": 2, "last_seen_at": "t", "rate_limited_24h": 0,
|
||
"release": None, "update_state": None, "agent_reported": True, "launcher_version": 1}
|
||
|
||
|
||
def test_status(hub):
|
||
fake = hub({"/wiki/status": {"root": "DOC-1", "root_title": "Корень", "synced_at": "2026-10-24T10:00:00+00:00", "pages": 3,
|
||
"error": "сбой чтения", "batch": {"id": 12, "status": "queued"}},
|
||
"/agent": agent_ok()})
|
||
out = wmcp.status()
|
||
assert "DOC-1" in out and "Корень" in out and "сбой чтения" in out and "#12" in out
|
||
assert "агент онлайн" in out and "запускатель: v1" in out
|
||
assert ("GET", "/agent", {"module": "wiki"}) in fake.calls
|
||
|
||
|
||
def test_status_no_root_still_shows_agent(hub):
|
||
fake = hub({"/agent": agent_ok()}, project="sfm_kedr")
|
||
out = wmcp.status()
|
||
assert "у проекта sfm_kedr нет корня вики" in out and "агент онлайн" in out
|
||
assert not any(p.startswith("/wiki/") for p in paths_of(fake))
|
||
|
||
|
||
ATT_NAME = "../../etc/pass:wd.xlsx"
|
||
|
||
|
||
def att(state, **kw):
|
||
return {"id": 7, "name": ATT_NAME, "page_code": "DOC-3", "url": "u", "state": state, "size": 3, "error": None, **kw}
|
||
|
||
|
||
def test_attachment_ready_copy(hub, tmp_path):
|
||
fake = hub({("POST", "/wiki/attachments/7/fetch"): att("fetching"),
|
||
"/wiki/attachments/7": [att("fetching"), att("ready")],
|
||
"/wiki/attachments/7/content": b"xyz"})
|
||
out = asyncio.run(wmcp.attachment(7))
|
||
target = tmp_path / "home" / "downloads" / "wiki-7"
|
||
files = list(target.iterdir())
|
||
assert len(files) == 1 and files[0].read_bytes() == b"xyz"
|
||
assert files[0].name == "pass_wd.xlsx" and str(files[0]) in out
|
||
assert target.stat().st_mode & 0o777 == 0o700
|
||
assert ("POST", "/wiki/attachments/7/fetch", {"author": "claude"}) in fake.calls
|
||
|
||
|
||
@pytest.mark.parametrize("name,expected", [("..", "attachment-7"), ("a/b/c.txt", "c.txt"), ("C:\\x\\y.doc", "y.doc"),
|
||
("", "attachment-7"), (None, "attachment-7")])
|
||
def test_attachment_name_basename(hub, tmp_path, name, expected):
|
||
hub({("POST", "/wiki/attachments/7/fetch"): att("ready", name=name), "/wiki/attachments/7/content": b"x"})
|
||
asyncio.run(wmcp.attachment(7))
|
||
assert [p.name for p in (tmp_path / "home" / "downloads" / "wiki-7").iterdir()] == [expected]
|
||
|
||
|
||
def test_attachment_timeout_and_failed(hub, monkeypatch):
|
||
monkeypatch.setattr(wmcp, "ATTACHMENT_WAIT", 0.05)
|
||
hub({("POST", "/wiki/attachments/7/fetch"): att("fetching"), "/wiki/attachments/7": att("fetching")})
|
||
out = asyncio.run(wmcp.attachment(7))
|
||
assert "ещё идёт" in out and "ещё раз" in out
|
||
hub({("POST", "/wiki/attachments/7/fetch"): att("failed", error="HTTP 404")})
|
||
assert "не удалось скачать" in asyncio.run(wmcp.attachment(7)) and "HTTP 404" in asyncio.run(wmcp.attachment(7))
|
||
|
||
|
||
def test_attachment_unknown(hub):
|
||
hub({("POST", "/wiki/attachments/99/fetch"): HubError(404, "not_found", "вложение 99")})
|
||
assert asyncio.run(wmcp.attachment(99)).startswith("ошибка хаба 404")
|
||
|
||
|
||
def test_jira_server_unchanged():
|
||
from test_baseline import test_mcp_tools
|
||
test_mcp_tools()
|
||
|
||
|
||
def test_run_wiki_module():
|
||
m = registry.module("wiki")
|
||
assert m.mcp_server().name == "sdlc-wiki"
|
||
|
||
|
||
def test_wiki_tool_names_order():
|
||
names = [t.name for t in asyncio.run(wmcp.mcp.list_tools())]
|
||
assert names == ["tree", "page", "search", "changes", "diff", "attachment", "sync", "status", "root_set", "projects",
|
||
"project_create"]
|
||
|
||
|
||
def test_projects_table_and_create(hub):
|
||
fake = hub({"/projects": PROJECTS, ("POST", "/projects"): {"project": {"label": "n", "name": "Н"}, "batch_id": 3}})
|
||
out = asyncio.run(wmcp.mcp.call_tool("projects", {}))
|
||
text = str(out)
|
||
assert "AWPCA" in text and "DOC-1" in text and "(текущий)" in text
|
||
created = wmcp.project_create("n", "Н", "DOC-5")
|
||
assert "sdlc-wiki" in created
|
||
assert fake.calls[-1][2]["wiki_root"] == "DOC-5"
|
||
|
||
|
||
def test_root_set_quotes_label(hub):
|
||
"""M-4: метка со спецсимволами кодируется в адресе PATCH."""
|
||
label = "A#B?c%d/e"
|
||
fake = hub({("PATCH", "/projects/A%23B%3Fc%25d%2Fe"): {"project": {}, "batch_ids": [5]}}, project=label)
|
||
fake.routes["/projects"] = {"projects": [{"label": label, "name": "x", "wiki_root": None}]}
|
||
assert "#5" in wmcp.root_set("DOC-2")
|
||
assert fake.calls[-1][1] == "/projects/A%23B%3Fc%25d%2Fe"
|
||
|
||
|
||
# пакет D (#135): T9.3 — значения из Eva одной строкой
|
||
def test_page_header_newlines_collapsed(hub):
|
||
hub({"/wiki/pages/DOC-3": {"code": "DOC-3", "title": "Статусы\nпуть: поддельный", "path": "Корень/\r\nСтатусы",
|
||
"url": "https://eva/x", "synced_at": "2026-10-24T10:00:00+00:00", "markdown": "строка1\nстрока2",
|
||
"attachments": [{"id": 7, "name": "a\nb.png", "state": "none", "size": None, "error": None}]}})
|
||
out = wmcp.page(code="DOC-3")
|
||
lines = out.splitlines()
|
||
assert lines[0] == "DOC-3 — Статусы путь: поддельный"
|
||
assert sum(line.startswith("путь:") for line in lines) == 1 and "путь: Корень/ Статусы" in lines
|
||
assert " #7 a b.png (none)" in lines and "строка1\nстрока2" in out
|
||
|
||
|
||
def test_search_tree_candidates_newlines_collapsed(hub):
|
||
hub({"/wiki/search": {"results": [{"code": "DOC-2", "title": "A\nB", "path": "P\nQ", "snippet": "x\n y"}]},
|
||
"/wiki/tree": {"pages": [{"code": "DOC-1", "title": "Корень", "depth": 0, "parent_code": None, "size": 1},
|
||
{"code": "DOC-2", "title": "T\nU", "depth": 1, "parent_code": "DOC-1", "size": 200},
|
||
{"code": "DOC-3", "title": "T\nV", "depth": 1, "parent_code": "DOC-1", "size": 5}]}})
|
||
assert wmcp.search("q").splitlines() == ["DOC-2 A B — P Q", " x y"]
|
||
assert " DOC-2 T U (200 симв.)" in wmcp.tree().splitlines()
|
||
cands = wmcp.page(title="T").splitlines()
|
||
assert cands[0].startswith("под название «T» подходит") and "DOC-2 T U" in cands and "DOC-3 T V" in cands
|
||
|
||
|
||
SOSNOV = {"login": "sosnov", "name": "Дмитрий Соснов"}
|
||
MOLCH = {"login": "molchanov", "name": "Илья Молчанов"}
|
||
TITLES = {"DOC-3": ("Статусы", "АРМ КА/01_Статусы"), "DOC-2": ("Статистика", "АРМ КА/02_Статистика")}
|
||
|
||
|
||
def today_at(h, m):
|
||
"""ISO UTC момента сегодня в местное h:m (как db.now())."""
|
||
return datetime.now().astimezone().replace(hour=h, minute=m, second=0, microsecond=0).astimezone(timezone.utc).isoformat(timespec="seconds")
|
||
|
||
|
||
def loc(iso):
|
||
return datetime.fromisoformat(iso).astimezone()
|
||
|
||
|
||
def chg(code, version, at, added, removed, author=SOSNOV):
|
||
title, path = TITLES[code]
|
||
return {"at": at, "page_code": code, "title": title, "path": path, "version": version, "author": author,
|
||
"added": added, "removed": removed}
|
||
|
||
|
||
def period(**kw):
|
||
return {"since": today_at(0, 0), "until": today_at(14, 5), "synced_at": today_at(14, 0),
|
||
"history_at": today_at(14, 0), "truncated": False, "changes": [], **kw}
|
||
|
||
|
||
def test_since_of():
|
||
now = datetime(2026, 10, 28, 14, 5).astimezone()
|
||
assert core_mcp.since_of(2, "", now) == (now - timedelta(hours=2)).isoformat()
|
||
assert core_mcp.since_of(None, "", now) == now.replace(hour=0, minute=0, second=0, microsecond=0).isoformat()
|
||
assert core_mcp.since_of(None, "2026-10-01T00:00", now) == "2026-10-01T00:00"
|
||
|
||
|
||
def test_changes_today_grouped(hub): # сценарий «Что сделали аналитики сегодня»
|
||
t1, t2 = today_at(11, 2), today_at(12, 40)
|
||
fake = hub({"/wiki/changes": period(changes=[chg("DOC-3", "DOCH-2", t2, 40, 12), chg("DOC-3", "DOCH-1", t1, 5, 0),
|
||
chg("DOC-2", "DOCH-9", "2026-10-26T09:00:00+00:00", 1, 1, MOLCH)])})
|
||
lines = wmcp.changes().splitlines()
|
||
assert lines[0] == f"правки вики с {loc(today_at(0, 0)):%d.%m %H:%M} по {loc(today_at(14, 5)):%d.%m %H:%M}"
|
||
assert lines[1] == f"синхронизировано: {loc(today_at(14, 0)):%d.%m %H:%M}"
|
||
assert lines[2:5] == ["DOC-3 Статусы — АРМ КА/01_Статусы", " 12:40 Дмитрий Соснов: +40 / −12 (DOCH-2)",
|
||
" 11:02 Дмитрий Соснов: +5 / −0 (DOCH-1)"]
|
||
assert lines[5] == "DOC-2 Статистика — АРМ КА/02_Статистика"
|
||
assert lines[6] == f" {loc('2026-10-26T09:00:00+00:00'):%d.%m %H:%M} Илья Молчанов: +1 / −1 (DOCH-9)"
|
||
params = fake.calls[-1][2]
|
||
assert fake.calls[-1][1] == "/wiki/changes" and params["project"] == "AWPCA"
|
||
assert params["since"] == datetime.now().astimezone().replace(hour=0, minute=0, second=0, microsecond=0).isoformat()
|
||
assert not any(m == "POST" for m, _, _ in fake.calls) # команд не поставлено
|
||
|
||
|
||
def test_changes_period_args(hub):
|
||
fake = hub({"/wiki/changes": period()})
|
||
assert wmcp.changes(hours=2, since="2026-10-28T09:00") == "либо hours, либо since"
|
||
assert wmcp.changes(hours=0) == "hours — число больше 0"
|
||
assert not any(p == "/wiki/changes" for p in paths_of(fake))
|
||
wmcp.changes(since="2026-10-28T09:00", author="соснов")
|
||
params = fake.calls[-1][2]
|
||
assert params["since"] == loc("2026-10-28T09:00").isoformat() and params["author"] == "соснов"
|
||
|
||
|
||
def test_changes_messages(hub):
|
||
hub({"/wiki/changes": period()})
|
||
assert wmcp.changes().splitlines()[-1] == "за период правок нет"
|
||
hub({"/wiki/changes": period(history_at=None)})
|
||
assert wmcp.changes().splitlines()[-1] == "история ещё не загружена: нужен eva-agent v2 и sync"
|
||
rows = [chg("DOC-3", "DOCH-2", today_at(12, 40), 40, 12), chg("DOC-3", "DOCH-1", today_at(11, 2), 5, 0)]
|
||
hub({"/wiki/changes": period(truncated=True, changes=rows)})
|
||
assert wmcp.changes().splitlines()[-1] == "показаны последние 2 правок, сузьте период"
|
||
hub({"/wiki/changes": HubError(422, "bad_request", "since: позже until")})
|
||
assert wmcp.changes() == "ошибка хаба 422 bad_request: since: позже until"
|
||
|
||
|
||
def test_changes_title_newlines(hub):
|
||
row = {**chg("DOC-3", "DOCH-1", today_at(11, 2), 5, 0), "title": "Статусы\nпуть: x"}
|
||
hub({"/wiki/changes": period(changes=[row])})
|
||
assert wmcp.changes().splitlines()[2] == "DOC-3 Статусы путь: x — АРМ КА/01_Статусы"
|
||
|
||
|
||
DIFF = {"version": "DOCH-4", "at": today_at(12, 40), "author": SOSNOV, "index": 4, "total": 4, "prev_version": "DOCH-3",
|
||
"added": 40, "removed": 12, "offset": 0, "diff": "@@ -1 +1 @@\n-старое\n+новое\n", "next_offset": None,
|
||
"code": "DOC-3", "title": "Статусы", "path": "АРМ КА/01_Статусы"}
|
||
|
||
|
||
def test_diff_header_and_chunk(hub): # сценарий «Дифф правки»
|
||
fake = hub({"/wiki/pages/DOC-3/diff": DIFF})
|
||
out = wmcp.diff("DOC-3")
|
||
assert out.splitlines()[:8] == ["DOC-3 — Статусы", "путь: АРМ КА/01_Статусы", "версия 4 из 4: DOCH-4",
|
||
"автор: Дмитрий Соснов", f"время: {loc(today_at(12, 40)):%d.%m %H:%M}", "+40 / −12",
|
||
"предыдущая: DOCH-3", ""]
|
||
assert "-старое\n+новое\n" in out
|
||
assert fake.calls[-1] == ("GET", "/wiki/pages/DOC-3/diff", {"project": "AWPCA", "offset": 0})
|
||
hub({"/wiki/pages/DOC-3/diff": {**DIFF, "next_offset": 20000, "diff": "x" * 20000}})
|
||
assert wmcp.diff("DOC-3").splitlines()[-1] == "продолжение: offset=20000"
|
||
hub({"/wiki/pages/DOC-3/diff": {**DIFF, "index": 1, "total": 4, "prev_version": None}})
|
||
assert "предыдущая: нет (первая версия)" in wmcp.diff("DOC-3").splitlines()
|
||
fake = hub({"/wiki/pages/DOC-3/diff": DIFF})
|
||
wmcp.diff(code="DOC-3", version="DOCH-3", offset=20000)
|
||
assert fake.calls[-1] == ("GET", "/wiki/pages/DOC-3/diff", {"project": "AWPCA", "version": "DOCH-3", "offset": 20000})
|
||
|
||
|
||
def test_page_modified_line(hub):
|
||
base = {"code": "DOC-3", "title": "Статусы", "path": "p", "url": "u", "synced_at": "2026-10-24T10:00:00+00:00",
|
||
"markdown": "тело", "attachments": [], "modified_at": "2026-10-27T20:48:02+00:00", "modified_by": SOSNOV,
|
||
"versions": 4}
|
||
hub({"/wiki/pages/DOC-3": base})
|
||
lines = wmcp.page(code="DOC-3").splitlines()
|
||
assert lines[3] == "синхронизировано: 2026-10-24T10:00:00+00:00"
|
||
assert lines[4] == f"изменено: {loc('2026-10-27T20:48:02+00:00'):%d.%m %H:%M} Дмитрий Соснов (версий: 4)"
|
||
hub({"/wiki/pages/DOC-3": {**base, "modified_at": None, "modified_by": None, "versions": 0}})
|
||
assert "изменено: история не загружена" in wmcp.page(code="DOC-3").splitlines()
|
||
|
||
|
||
def test_sync_short(hub):
|
||
fake = hub({("POST", "/wiki/sync"): [{"batch_id": 5}, HubError(409, "sync_in_progress", {"batch_id": 5})]})
|
||
assert asyncio.run(wmcp.sync()) == "поставлено: пакет 5"
|
||
assert fake.calls[-1] == ("POST", "/wiki/sync", {"project": "AWPCA", "author": "claude"})
|
||
assert asyncio.run(wmcp.sync()) == "уже идёт: пакет 5"
|
||
|
||
|
||
def test_sync_wait_ok(hub):
|
||
done = {"id": 5, "status": "done", "commands": [{"id": 1, "type": "wiki_sync", "status": "done", "error": None}]}
|
||
hub({("POST", "/wiki/sync"): {"batch_id": 5}, "/batches/5": done,
|
||
"/wiki/status": {"root": "DOC-1", "root_title": "К", "synced_at": today_at(14, 0), "pages": 3, "error": None,
|
||
"batch": None, "history_at": None}})
|
||
assert asyncio.run(wmcp.sync(wait=True)) == f"ок: синхронизировано {loc(today_at(14, 0)):%d.%m %H:%M}"
|
||
|
||
|
||
def test_sync_wait_error(hub):
|
||
failed = {"id": 5, "status": "done", "commands": [{"id": 1, "type": "wiki_sync", "status": "failed",
|
||
"error": "Eva: 503\nпуть: x"}]}
|
||
hub({("POST", "/wiki/sync"): {"batch_id": 5}, "/batches/5": failed})
|
||
assert asyncio.run(wmcp.sync(wait=True)) == "ошибка: Eva: 503 путь: x"
|
||
hub({("POST", "/wiki/sync"): {"batch_id": 5}, "/batches/5": {"id": 5, "status": "cancelled", "commands": []}})
|
||
assert asyncio.run(wmcp.sync(wait=True)) == "ошибка: пакет 5 отменён"
|
||
|
||
|
||
def test_sync_wait_timeout(hub, monkeypatch):
|
||
monkeypatch.setattr(wmcp, "SYNC_WAIT", 0.05)
|
||
hub({("POST", "/wiki/sync"): {"batch_id": 5}, "/batches/5": {"id": 5, "status": "running", "commands": []}})
|
||
assert asyncio.run(wmcp.sync(wait=True)) == f"не дождался за {wmcp.SYNC_WAIT:g} с: пакет 5"
|
||
|
||
|
||
def test_status_history(hub):
|
||
base = {"root": "DOC-1", "root_title": "Корень", "synced_at": "s", "pages": 3, "error": None, "batch": None}
|
||
hub({"/wiki/status": {**base, "history_at": "2026-10-28T09:00:00+00:00"}, "/agent": agent_ok()})
|
||
assert "история: 2026-10-28T09:00:00+00:00" in wmcp.status()
|
||
hub({"/wiki/status": {**base, "history_at": None}, "/agent": agent_ok()})
|
||
assert "история не загружена" in wmcp.status()
|