diff --git a/src/agent.py b/src/agent.py
index 2b115c1..82b39ec 100644
--- a/src/agent.py
+++ b/src/agent.py
@@ -327,12 +327,22 @@ class EvaClient(Http):
return self.call("CmfDocument.get", filter=[["code", "==", code]], fields=fields)
def list_docs(self, flt: list, fields: List[str]) -> List[dict]:
+ return self.list_all("CmfDocument", flt, fields)
+
+ def list_versions(self, doc_id: str) -> List[dict]:
+ """Версии страницы (CmfDocumentHistory) по id документа; фильтр parent.code Eva не принимает (-32001)."""
+ return self.list_all("CmfDocumentHistory", [["parent_id", "==", doc_id]], VERSION_FIELDS)
+
+ def get_version(self, code: str) -> Optional[dict]:
+ return self.call("CmfDocumentHistory.get", filter=[["code", "==", code]], fields=VERSION_TEXT_FIELDS)
+
+ def list_all(self, cls: str, flt: list, fields: List[str]) -> List[dict]:
"""Весь список страницами по PAGE_SIZE. Если Eva не соблюдает slice (ответ без новых записей), обход
останавливается на собранном, а не крутится бесконечно (M-4)."""
out, offset = [], 0 # type: List[dict], int
seen = set() # type: set
while True:
- chunk = self.call("CmfDocument.list", filter=flt, fields=fields,
+ chunk = self.call("%s.list" % cls, filter=flt, fields=fields,
slice=[offset, offset + self.PAGE_SIZE]) or []
if isinstance(chunk, dict): # обёрнутый результат
chunk = chunk.get("items") or chunk.get("result") or []
@@ -346,7 +356,7 @@ class EvaClient(Http):
if len(chunk) < self.PAGE_SIZE:
return out
if not fresh:
- log(" Eva: CmfDocument.list повторяет записи (slice не соблюдается?) — останов на %d" % len(out))
+ log(" Eva: %s.list повторяет записи (slice не соблюдается?) — останов на %d" % (cls, len(out)))
return out
offset += self.PAGE_SIZE
@@ -731,6 +741,8 @@ def _parse_at(value) -> float:
MAX_PAGES = 50 # страниц в пачке (предел хаба, stage.MAX_PAGES)
MAX_BATCH_BYTES = 4 * 1024 * 1024 # JSON пачки; предел тела хаба — 16 МБ
TREE_FIELDS = ["id", "code", "name", "parent", "tree_parent", "project", "orderno"]
+VERSION_FIELDS = ["code", "cmf_created_at", "has_published", "cmf_author"] # список версий — без текста
+VERSION_TEXT_FIELDS = ["code", "text"]
def _ref_code(ref) -> Optional[str]:
diff --git a/tests/fake_eva.py b/tests/fake_eva.py
index ab2ee26..b84760c 100644
--- a/tests/fake_eva.py
+++ b/tests/fake_eva.py
@@ -1,4 +1,5 @@
-"""Фейковая Eva для тестов агента: JSON-RPC 2.2 (CmfDocument.get/list) и /files/... на http.server."""
+"""Фейковая Eva для тестов агента: JSON-RPC 2.2 (CmfDocument.get/list, CmfDocumentHistory.get/list) и /files/... на
+http.server."""
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -17,6 +18,10 @@ def _field(doc, path):
return cur
+# автор версии, как его отдаёт Eva в раскрытом поле cmf_author (проба 28.10.2026)
+AUTHOR = {"id": "CmfPerson:6c2b", "class_name": "CmfPerson", "login": "sosnov-dvi@rshbeco.ru", "name": "Дмитрий Соснов"}
+
+
class FakeEva:
token = "eva-secret"
@@ -29,6 +34,8 @@ class FakeEva:
self.requests = [] # {"method", "path", "body", "auth"}
self.failures = [] # [status, retry_after] — по одному на ответ
self.rpc_errors = []
+ self.method_errors = {} # метод → текст разовой ошибки JSON-RPC
+ self.versions = []
self.lock = threading.Lock()
eva = self
@@ -87,6 +94,16 @@ class FakeEva:
def rpc_error_next(self, text):
self.rpc_errors.append(text)
+ def rpc_error_on(self, method: str, text: str):
+ """Разовая ошибка JSON-RPC на первом вызове метода method (очередь rpc_errors не трогает)."""
+ self.method_errors[method] = text
+
+ def add_version(self, parent_id, code, created_at="2026-10-27T17:59:00.123456+03:00", text="", has_published=True,
+ author=AUTHOR):
+ self.versions.append({"id": "CmfDocumentHistory:" + str(code), "parent_id": str(parent_id), "code": code,
+ "cmf_created_at": created_at, "has_published": has_published, "cmf_author": author,
+ "text": text})
+
@property
def api_calls(self):
return [r for r in self.requests if r["path"].startswith("/api/")]
@@ -109,6 +126,9 @@ class FakeEva:
try:
if u.path == "/api/" and verb == "POST":
body = json.loads(raw)
+ if rpc_error is None:
+ with self.lock:
+ rpc_error = self.method_errors.pop(body.get("method"), None)
if rpc_error is not None:
out = {"jsonrpc": "2.2", "callid": body.get("callid"), "error": rpc_error}
else:
@@ -140,12 +160,16 @@ class FakeEva:
h.wfile.write(body)
def rpc(self, method, kw):
- found = [d for d in self.docs if self._match(d, kw.get("filter") or [])]
fields = kw.get("fields")
- pick = lambda d: self._out(d, fields) # noqa: E731
- if method == "CmfDocument.get":
+ if method.startswith("CmfDocumentHistory."):
+ found = [v for v in self.versions if self._match(v, kw.get("filter") or [])]
+ pick = lambda v: {k: v.get(k) for k in fields} if fields else dict(v) # noqa: E731
+ else:
+ found = [d for d in self.docs if self._match(d, kw.get("filter") or [])]
+ pick = lambda d: self._out(d, fields) # noqa: E731
+ if method in ("CmfDocument.get", "CmfDocumentHistory.get"):
return pick(found[0]) if found else None
- if method == "CmfDocument.list":
+ if method in ("CmfDocument.list", "CmfDocumentHistory.list"):
lo, hi = kw.get("slice") or [0, len(found)]
return [pick(d) for d in found[lo:hi]]
raise ValueError("метод %s не поддержан" % method)
diff --git a/tests/test_eva_client.py b/tests/test_eva_client.py
index c3a6002..5b99e6a 100644
--- a/tests/test_eva_client.py
+++ b/tests/test_eva_client.py
@@ -3,6 +3,7 @@ import unittest
from contextlib import redirect_stderr, redirect_stdout
from tests import _common as c
+from tests import fake_eva
from tests.fake_eva import FakeEva
agent = c.agent
@@ -86,6 +87,7 @@ class EvaClientTest(unittest.TestCase):
self.assertEqual(len(docs), 450)
self.assertEqual(len(self.eva.api_calls), 2)
self.assertIn("slice не соблюдается", out.getvalue())
+ self.assertIn("CmfDocument.list повторяет записи", out.getvalue())
def test_list_slice_ignored_same_page(self):
"""M-4: Eva всегда отдаёт первые 200 — без защиты цикл бесконечен."""
@@ -184,6 +186,57 @@ class EvaClientTest(unittest.TestCase):
self.assertNotIn("eva-secret", t)
self.assertNotIn("eva-secret", repr(self.cli))
+ def test_list_versions_filter_and_fields(self):
+ self.eva.add_version("CmfDocument:1", "DOCH-1")
+ self.eva.add_version("CmfDocument:1", "DOCH-2", has_published=False)
+ self.eva.add_version("CmfDocument:2", "DOCH-9")
+ got = self.cli.list_versions("CmfDocument:1")
+ self.assertEqual([v["code"] for v in got], ["DOCH-1", "DOCH-2"])
+ self.assertEqual(got[0], {"code": "DOCH-1", "cmf_created_at": "2026-10-27T17:59:00.123456+03:00",
+ "has_published": True, "cmf_author": fake_eva.AUTHOR})
+ req = self.eva.api_calls[0]
+ self.assertEqual(req["path"], "/api/?m=CmfDocumentHistory.list")
+ self.assertEqual(req["body"]["kwargs"]["filter"], [["parent_id", "==", "CmfDocument:1"]])
+ self.assertEqual(req["body"]["kwargs"]["fields"], ["code", "cmf_created_at", "has_published", "cmf_author"])
+ self.assertEqual(req["body"]["kwargs"]["slice"], [0, 200])
+
+ def test_list_versions_pages_200(self):
+ for i in range(450):
+ self.eva.add_version("CmfDocument:1", "DOCH-%d" % i)
+ self.assertEqual(len(self.cli.list_versions("CmfDocument:1")), 450)
+ self.assertEqual([r["body"]["kwargs"]["slice"] for r in self.eva.api_calls], [[0, 200], [200, 400], [400, 600]])
+
+ def test_list_versions_slice_ignored(self):
+ orig = self.eva.rpc
+ self.eva.rpc = lambda m, kw: orig(m, dict(kw, slice=None))
+ for i in range(450):
+ self.eva.add_version("CmfDocument:1", "DOCH-%d" % i)
+ out = io.StringIO()
+ with redirect_stdout(out):
+ got = self.cli.list_versions("CmfDocument:1")
+ self.assertEqual(len(got), 450)
+ self.assertEqual(len(self.eva.api_calls), 2)
+ self.assertIn("CmfDocumentHistory.list повторяет записи", out.getvalue())
+
+ def test_get_version(self):
+ self.eva.add_version("CmfDocument:1", "DOCH-1", text="v1
")
+ self.assertEqual(self.cli.get_version("DOCH-1"), {"code": "DOCH-1", "text": "v1
"})
+ self.assertIsNone(self.cli.get_version("DOCH-404"))
+ req = self.eva.api_calls[0]
+ self.assertEqual(req["path"], "/api/?m=CmfDocumentHistory.get")
+ self.assertEqual(req["body"]["kwargs"]["filter"], [["code", "==", "DOCH-1"]])
+ self.assertEqual(req["body"]["kwargs"]["fields"], ["code", "text"])
+ self.assertEqual(self.cli.calls, 2)
+
+ def test_rpc_error_on_method(self):
+ self.eva.add_doc(1, "DOC-1", project="P")
+ self.eva.rpc_error_on("CmfDocumentHistory.list", "нет прав")
+ self.assertIsNotNone(self.cli.get_doc("DOC-1", ["code"]))
+ with self.assertRaises(agent.EvaError) as cm:
+ self.cli.list_versions("1")
+ self.assertEqual(str(cm.exception), "CmfDocumentHistory.list: нет прав")
+ self.assertEqual(self.cli.list_versions("1"), []) # ошибка разовая
+
if __name__ == "__main__":
unittest.main()