From b98b963ae100c52d605b58695b9775f04a8ac29d Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Wed, 28 Oct 2026 11:55:06 +0400 Subject: [PATCH] =?UTF-8?q?wiki=5Fsync:=20=D0=BE=D0=BF=D1=83=D0=B1=D0=BB?= =?UTF-8?q?=D0=B8=D0=BA=D0=BE=D0=B2=D0=B0=D0=BD=D0=BD=D1=8B=D0=B5=20=D0=B2?= =?UTF-8?q?=D0=B5=D1=80=D1=81=D0=B8=D0=B8=20=D1=81=D1=82=D1=80=D0=B0=D0=BD?= =?UTF-8?q?=D0=B8=D1=86=20=D1=85=D0=B0=D0=B1=D1=83,=20result.versions=20(#?= =?UTF-8?q?7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- src/agent.py | 72 +++++++++++++- tests/test_wiki_sync.py | 201 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 267 insertions(+), 6 deletions(-) diff --git a/src/agent.py b/src/agent.py index 3fe436b..b1c2a68 100644 --- a/src/agent.py +++ b/src/agent.py @@ -836,6 +836,25 @@ class HubError(Exception): self.retryable = isinstance(cause, ConnectionError) or (isinstance(cause, HttpError) and cause.code >= 500) +def known_versions_ok(value) -> bool: + """known_versions от хаба: {код страницы: [коды версий]}.""" + return isinstance(value, dict) and all(isinstance(v, list) and all(isinstance(x, str) for x in v) + for v in value.values()) + + +def version_author(ref) -> dict: + """Автор версии из раскрытого cmf_author: каждое поле — непустая строка или None (закрытое ACL поле Eva не + отдаёт); не объект (нет поля, строка-id) — оба None (P-1).""" + if not isinstance(ref, dict): + return {"login": None, "name": None} + pick = lambda k: ref.get(k) if isinstance(ref.get(k), str) and ref.get(k) else None # noqa: E731 + return {"login": pick("login"), "name": pick("name")} + + +class _VersionMissing(Exception): + """Версия из списка Eva не читается по коду (P-4).""" + + class _Batcher: """Пачки для стадии хаба: не больше max_items элементов и max_bytes JSON; send получает накопленный список.""" @@ -911,11 +930,15 @@ class Executor: raise HubError(e) from None def do_wiki_sync(self, p: dict, cmd: dict) -> dict: - """Дерево страниц Eva от корня → хаб пачками; отчёт — коды в порядке обхода (корень первым). Eva только читается, - поэтому после перезапуска команда выполняется заново целиком (design 9).""" + """Дерево страниц Eva от корня → хаб пачками; затем, если хаб прислал known_versions, — опубликованные версии + страниц дерева, которых хаб не знает (POST /agent/wiki/versions). Отчёт — коды в порядке обхода (корень первым) и + число версий. Eva только читается, поэтому после перезапуска команда выполняется заново целиком (design 9).""" root_code = str(p.get("root") or "").strip() if not root_code: return self._out("failed", error="wiki_sync: не задан root") + with_versions = "known_versions" in p # хаб без истории ключа не шлёт — шаг версий не выполняется + if with_versions and not known_versions_ok(p["known_versions"]): + return self._out("failed", error="wiki_sync: known_versions — не объект {код страницы: [коды версий]}") root = self.eva.get_doc(root_code, TREE_FIELDS + ["text"]) if not root: return self._out("failed", error="страница %s не найдена или нет прав" % root_code) @@ -923,6 +946,7 @@ class Executor: root_code = root.get("code") or root_code base = self.eva.base codes = [] # type: List[str] + tree = [] # type: List[Tuple[str, Optional[str]]] pages = _Batcher(lambda b: self._send_batch(self.hub.wiki_pages, cmd["id"], b, "страница", "пачка страниц"), MAX_PAGES) for doc, parent_code in walk(root, children): @@ -940,9 +964,51 @@ class Executor: "html": html, "attachments": attachments, "links": links} pages.add(page) codes.append(code) + tree.append((code, page["eva_id"])) pages.flush() title = (root.get("name") or "").strip() or root_code - return self._out("done", {"root_title": title, "codes": codes, "eva_calls": self.eva.calls}) + result = {"root_title": title, "codes": codes} + if with_versions: + try: + result["versions"] = self._sync_versions(cmd["id"], tree, p["known_versions"]) + except _VersionMissing as e: + return self._out("failed", error=str(e)) + result["eva_calls"] = self.eva.calls + return self._out("done", result) + + def _sync_versions(self, command_id: int, tree: List[Tuple[str, Optional[str]]], known: Dict[str, List[str]]) -> int: + """Опубликованные версии страниц дерева, которых нет в known, — хабу пачками; число отправленных (design 2–5).""" + base = self.eva.base + out = _Batcher(lambda b: self._send_batch(self.hub.wiki_versions, command_id, b, "версия", "пачка версий"), + MAX_VERSIONS) + for page_code, doc_id in tree: + if doc_id is None: + log(" у страницы %s нет id — версии не запрошены" % page_code) + continue + seen = set(known.get(page_code, [])) + for v in self.eva.list_versions(doc_id): + if v.get("has_published") is not True: + continue + vcode = v.get("code") + if not isinstance(vcode, str) or not vcode: + log(" версия без code у %s — пропущена" % page_code) + continue + created = v.get("cmf_created_at") + if not isinstance(created, str) or not created: + log(" версия %s у %s без cmf_created_at — пропущена" % (vcode, page_code)) + continue + if vcode in seen: + continue + full = self.eva.get_version(vcode) + if not full: + raise _VersionMissing("версия %s страницы %s не найдена или нет прав" % (vcode, page_code)) + html = full.get("text") if isinstance(full.get("text"), str) else "" + attachments, links = classify_links(html, base) + out.add({"page_code": page_code, "code": vcode, "created_at": created, + "author": version_author(v.get("cmf_author")), "html": html, + "attachments": attachments, "links": links}) + out.flush() + return out.sent def do_wiki_fetch(self, p: dict, cmd: dict) -> dict: """Файл вложения Eva → хабу (POST /agent/files). Eva только читается; адрес — лишь своего хоста (design 8, 24); diff --git a/tests/test_wiki_sync.py b/tests/test_wiki_sync.py index ab12bb8..3caff51 100644 --- a/tests/test_wiki_sync.py +++ b/tests/test_wiki_sync.py @@ -1,7 +1,9 @@ +import io import json import os import tempfile import unittest +from contextlib import redirect_stdout from _common import SETTINGS, agent, eva_cfg, no_sleep from fake_eva import FakeEva @@ -35,12 +37,19 @@ class SyncBase(unittest.TestCase): self.client.sleep = no_sleep self.ex = agent.Executor(self.client, agent.HubClient(self.cfg)) - def sync(self, root="DOC-1", label="tz"): - cid = self.hub.enqueue("wiki_sync", {"label": label, "root": root}) + def sync(self, root="DOC-1", label="tz", **extra): + params = dict({"label": label, "root": root}, **extra) + cid = self.hub.enqueue("wiki_sync", params) self.hub.queue.remove(cid) self.hub.commands[cid]["status"] = "claimed" self.cid = cid - return self.ex.run({"id": cid, "type": "wiki_sync", "params": {"label": label, "root": root}}) + return self.ex.run({"id": cid, "type": "wiki_sync", "params": params}) + + def sent_versions(self): + return [v for body in self.hub.versions for v in body["versions"]] + + def history_calls(self, method): + return [r for r in self.eva.api_calls if r["path"] == "/api/?m=" + method] def sent(self): return [p for body in self.hub.pages for p in body["pages"]] @@ -278,5 +287,191 @@ class WikiSyncRecoverTest(unittest.TestCase): self.assertEqual(journal.unfinished_started(), []) +class WikiVersionsTest(SyncBase): + def tree2(self): + self.eva.add_doc("CmfDocument:1", "DOC-1", "Корень", project="P") + self.eva.add_doc("CmfDocument:2", "DOC-2", "Б", parent="DOC-1", project="P", orderno=1) + + def test_versions_skipped_without_known(self): + self.tree2(); self.eva.add_version("CmfDocument:1", "DOCH-11") + out = self.sync() + self.assertEqual(out["status"], "done", out) + self.assertNotIn("versions", out["result"]) + self.assertEqual(self.hub.versions, []) + self.assertFalse([r for r in self.eva.api_calls if "CmfDocumentHistory" in r["path"]]) + + def test_versions_first_load(self): # сценарий «Первая загрузка истории» + self.tree2(); e = self.eva + e.add_version("CmfDocument:1", "DOCH-11", "2026-10-27T17:59:00.123456+03:00", "

v1

") + e.add_version("CmfDocument:1", "DOCH-12", "2026-10-27T18:02:00+03:00", "

v2

") + e.add_version("CmfDocument:1", "DOCH-13", "2026-10-27T18:05:00+03:00", "

v3

") + e.add_version("CmfDocument:1", "DOCH-19", "2026-10-27T18:10:00+03:00", "

черновик

", has_published=False) + e.add_version("CmfDocument:2", "DOCH-21", "2026-08-01T10:00:00+03:00", "

a

") + e.add_version("CmfDocument:2", "DOCH-22", "2026-08-02T10:00:00+03:00", "

b

") + out = self.sync(known_versions={}) + self.assertEqual(out["status"], "done", out) + self.assertEqual(out["result"]["versions"], 5) + self.assertEqual([v["code"] for v in self.sent_versions()], ["DOCH-11", "DOCH-12", "DOCH-13", "DOCH-21", "DOCH-22"]) + self.assertEqual(self.sent_versions()[0], { + "page_code": "DOC-1", "code": "DOCH-11", "created_at": "2026-10-27T17:59:00.123456+03:00", + "author": {"login": "sosnov-dvi@rshbeco.ru", "name": "Дмитрий Соснов"}, + "html": "

v1

", "attachments": [], "links": []}) + self.assertEqual(self.hub.versions[0]["command_id"], self.cid) + self.assertEqual(self.hub.wiki_posts, ["/agent/wiki/pages", "/agent/wiki/versions"]) + self.assertEqual([r["body"]["kwargs"]["filter"] for r in self.history_calls("CmfDocumentHistory.list")], + [[["parent_id", "==", "CmfDocument:1"]], [["parent_id", "==", "CmfDocument:2"]]]) + self.assertEqual(len(self.history_calls("CmfDocumentHistory.get")), 5) # черновик не читается + self.assertEqual(out["result"]["eva_calls"], self.client.calls) + self.assertEqual(out["result"]["eva_calls"], len(self.eva.api_calls)) + + def test_versions_incremental(self): # сценарий «Догрузка новых версий» + self.tree2() + for code in ("DOCH-10", "DOCH-11", "DOCH-12"): + self.eva.add_version("CmfDocument:1", code) + out = self.sync(known_versions={"DOC-1": ["DOCH-10", "DOCH-11"], "DOC-77": ["DOCH-1"]}) + self.assertEqual(out["result"]["versions"], 1) + self.assertEqual([v["code"] for v in self.sent_versions()], ["DOCH-12"]) + self.assertEqual([r["body"]["kwargs"]["filter"] for r in self.history_calls("CmfDocumentHistory.get")], + [[["code", "==", "DOCH-12"]]]) + + def test_versions_none_in_eva(self): + self.tree2() + out = self.sync(known_versions={}) + self.assertEqual(out["result"]["versions"], 0) + self.assertEqual(self.hub.wiki_posts, ["/agent/wiki/pages"]) + + def test_versions_author_missing(self): + self.tree2(); e = self.eva + e.add_version("CmfDocument:1", "DOCH-1", author=None) + e.add_version("CmfDocument:1", "DOCH-2", author="CmfPerson:6c2b") + e.add_version("CmfDocument:1", "DOCH-3", author={"id": "CmfPerson:1", "_acl_fields": {"login": "deny", "name": "deny"}, "_acl_obj": "deny"}) + e.add_version("CmfDocument:1", "DOCH-4", author={"id": "CmfPerson:2", "login": "x@y.ru", "_acl_fields": {"name": "deny"}}) + out = self.sync(known_versions={}) + self.assertEqual(out["status"], "done", out) + self.assertEqual([v["author"] for v in self.sent_versions()], + [{"login": None, "name": None}] * 3 + [{"login": "x@y.ru", "name": None}]) + + def test_versions_without_code_or_time_skipped(self): + self.tree2(); e = self.eva + e.add_version("CmfDocument:1", None) + e.add_version("CmfDocument:1", "DOCH-31", created_at=None) + e.add_version("CmfDocument:1", "DOCH-32") + buf = io.StringIO() + with redirect_stdout(buf): + out = self.sync(known_versions={}) + self.assertEqual(out["result"]["versions"], 1) + self.assertEqual([v["code"] for v in self.sent_versions()], ["DOCH-32"]) + self.assertIn("версия без code у DOC-1 — пропущена", buf.getvalue()) + self.assertIn("версия DOCH-31 у DOC-1 без cmf_created_at — пропущена", buf.getvalue()) + + def test_has_published_strict(self): + self.tree2() + for i, flag in enumerate(("true", 1, None, False)): + self.eva.add_version("CmfDocument:1", "DOCH-%d" % i, has_published=flag) + out = self.sync(known_versions={}) + self.assertEqual(out["result"]["versions"], 0) + self.assertEqual(self.history_calls("CmfDocumentHistory.get"), []) + + def test_version_html_and_links(self): + self.tree2() + self.eva.add_version("CmfDocument:1", "DOCH-1", text='xСхема') + self.eva.add_version("CmfDocument:1", "DOCH-2", text=None) + self.sync(known_versions={}) + v1, v2 = self.sent_versions() + self.assertEqual(v1["links"], [{"ref": "/project/Document/DOC-7", "code": "DOC-7"}]) + self.assertEqual(v1["attachments"], [{"ref": "/files/a/b.png", "url": self.eva.base_url + "/files/a/b.png", "name": "Схема"}]) + self.assertEqual((v2["html"], v2["attachments"], v2["links"]), ("", [], [])) + + def test_versions_batches_50(self): + self.tree2() + for i in range(120): + self.eva.add_version("CmfDocument:1", "DOCH-%03d" % i) + out = self.sync(known_versions={}) + self.assertEqual([len(b["versions"]) for b in self.hub.versions], [50, 50, 20]) + self.assertEqual(out["result"]["versions"], 120) + + def test_versions_batch_bytes(self): + self.tree2() + big = "

" + "x" * (1500 * 1024) + "

" + for i in (1, 2, 3): + self.eva.add_version("CmfDocument:1", "DOCH-%d" % i, text=big) + self.sync(known_versions={}) + sizes = [len(json.dumps(b, ensure_ascii=False).encode("utf-8")) for b in self.hub.versions] + self.assertTrue(all(s <= 4 * 1024 * 1024 for s in sizes), sizes) + self.assertEqual([len(b["versions"]) for b in self.hub.versions], [2, 1]) + + def test_versions_413(self): + for n, expect in ((1, "версия DOCH-11 больше 16 МБ"), (3, "пачка версий DOCH-11…DOCH-13 (3 шт.) больше 16 МБ")): + with self.subTest(n=n): + self.setUp(); self.tree2() + for i in range(n): + self.eva.add_version("CmfDocument:1", "DOCH-%d" % (11 + i)) + self.hub.reject_versions = (413, {"error": "too_large", "detail": "тело больше 16777216 байт"}) + out = self.sync(known_versions={}) + self.assertEqual(out["status"], "failed") + self.assertIn(expect, out["error"]) + self.assertFalse(out.get("retryable")) + + def test_versions_hub_409(self): + self.tree2(); self.eva.add_version("CmfDocument:1", "DOCH-1") + self.hub.reject_versions = (409, {"error": "not_claimed", "detail": "команда 1 не выдана агенту как wiki_sync"}) + out = self.sync(known_versions={}) + self.assertEqual(out["status"], "failed") + self.assertIn("409", out["error"]) + self.assertIn("не выдана агенту", out["error"]) + + def test_versions_eva_rpc_error(self): + self.tree2(); self.eva.add_version("CmfDocument:1", "DOCH-1") + self.eva.rpc_error_on("CmfDocumentHistory.list", "нет прав") + out = self.sync(known_versions={}) + self.assertEqual(out["status"], "failed") + self.assertEqual(out["error"], "Eva: CmfDocumentHistory.list: нет прав") + self.assertEqual(len(self.hub.pages), 1) # страницы ушли, версии — нет + self.assertEqual(self.hub.versions, []) + + def test_version_vanished(self): + self.tree2(); self.eva.add_version("CmfDocument:1", "DOCH-12") + self.client.get_version = lambda code: None + out = self.sync(known_versions={}) + self.assertEqual(out["status"], "failed") + self.assertEqual(out["error"], "версия DOCH-12 страницы DOC-1 не найдена или нет прав") + + def test_page_without_id(self): + self.eva.add_doc("CmfDocument:1", "DOC-1", "Корень", project="P") + self.eva.add_doc(None, "DOC-2", "Б", parent="DOC-1", project="P") + buf = io.StringIO() + with redirect_stdout(buf): + out = self.sync(known_versions={}) + self.assertEqual(out["status"], "done", out) + self.assertEqual(len(self.history_calls("CmfDocumentHistory.list")), 1) + self.assertIn("у страницы DOC-2 нет id — версии не запрошены", buf.getvalue()) + + def test_known_versions_malformed(self): + for bad in (None, [], {"DOC-1": "DOCH-1"}, {"DOC-1": [1]}): + with self.subTest(bad=bad): + self.setUp(); self.tree2() + out = self.sync(known_versions=bad) + self.assertEqual(out["status"], "failed") + self.assertEqual(out["error"], "wiki_sync: known_versions — не объект {код страницы: [коды версий]}") + self.assertEqual(self.eva.api_calls, []) + self.assertEqual(self.hub.pages, []) + + def test_versions_hub_503_retried_by_runner(self): + self.tree2() + self.eva.add_version("CmfDocument:1", "DOCH-1"); self.eva.add_version("CmfDocument:2", "DOCH-2") + self.hub.fail_versions = 1 + cid = self.hub.enqueue("wiki_sync", {"label": "tz", "root": "DOC-1", "known_versions": {}}) + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + runner = agent.Runner(self.cfg, self.ex.hub, self.client, agent.Journal(os.path.join(tmp.name, "journal.jsonl"))) + runner.sleep = lambda s: None # и пауза RETRY_MIN_WAIT перед повтором (wait_seconds) + runner.idle_limit = 1 + runner.loop() + self.assertEqual(self.hub.results[cid]["status"], "done", self.hub.results[cid]) + self.assertEqual(self.hub.results[cid]["result"]["versions"], 2) + self.assertEqual(len(self.hub.pages), 2) # команда выполнена заново целиком + self.assertEqual([len(b["versions"]) for b in self.hub.versions], [2]) + + if __name__ == "__main__": unittest.main()