eva-agent v2: версии страниц в wiki_sync (#7) #8
72
src/agent.py
72
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);
|
||||
|
||||
@ -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", "<p>v1</p>")
|
||||
e.add_version("CmfDocument:1", "DOCH-12", "2026-10-27T18:02:00+03:00", "<p>v2</p>")
|
||||
e.add_version("CmfDocument:1", "DOCH-13", "2026-10-27T18:05:00+03:00", "<p>v3</p>")
|
||||
e.add_version("CmfDocument:1", "DOCH-19", "2026-10-27T18:10:00+03:00", "<p>черновик</p>", has_published=False)
|
||||
e.add_version("CmfDocument:2", "DOCH-21", "2026-08-01T10:00:00+03:00", "<p>a</p>")
|
||||
e.add_version("CmfDocument:2", "DOCH-22", "2026-08-02T10:00:00+03:00", "<p>b</p>")
|
||||
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": "<p>v1</p>", "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='<a href="/project/Document/DOC-7">x</a><img src="/files/a/b.png" alt="Схема">')
|
||||
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 = "<p>" + "x" * (1500 * 1024) + "</p>"
|
||||
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()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user