Клиент Eva JSON-RPC (#1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-24 04:49:50 +04:00
parent 67f1a5a096
commit 7bc59f3f32
6 changed files with 382 additions and 13 deletions

View File

@ -139,3 +139,17 @@ Issue агента — aep/eva-agent#1. Версия агента — 1. Аге
глубже. Цена ошибки: тест тихо пропускается на нестандартной раскладке каталогов.
19. **`.gitignore` дополнен `agent.py.new`, `agent.py.prev`, `__pycache__/`, `*.pyc`** (рабочие файлы самообновления
и кэш). Цена ошибки: нет.
20. **`EvaClient(cfg)` остаётся с сигнатурой каркаса (по всему `cfg`), а не `(cfg_eva, timeout, verify_ssl, ca_bundle)` из брифа.**
Обоснование: `Runner`/`main` уже создают клиента как `EvaClient(cfg)` (решение 14).
Цена ошибки: нет, внутренняя сигнатура.
21. **`EvaClient.call(method, **kwargs)` переопределяет `Http.call(method, path, ...)`.** Транспортный вызов — через
`Http.call(self, ...)`; тесты общего `Http` в `test_clients.py` переведены на `agent.Http.call(client, ...)`.
Обоснование: интерфейс клиента Eva задан брифом (метод JSON-RPC первым аргументом).
Цена ошибки: путаница сигнатур; видна при чтении класса.
22. **Пауза между вызовами — перед вызовом (и перед повтором) по `last_call_at`; первый вызов после старта не
задерживается. `Runner.load_settings` ставит `eva.pacing` из `wiki_pacing` (по умолчанию 0.2).**
Обоснование: `last_call_at` общий с паузой между командами. Цена ошибки: лишняя или недостающая пауза, правится настройкой.
23. **Токен вырезается (`***`) из текстов `HttpError`/`ConnectionError`/`EvaError`, выходящих из `EvaClient`.**
Обоснование: сервер может отразить заголовок в теле ошибки. Цена ошибки: токен в логе.
24. **`download` принимает только адрес под `base` (иначе `ValueError("чужой хост")`) и повторяет те же коды, что `call`.**
Обоснование: токен нельзя отправлять на чужой хост (решение 8). Цена ошибки: нет.

View File

@ -228,9 +228,16 @@ class Http:
"""Вызывается на каждый полученный HTTP-ответ (любой код)."""
class EvaError(Exception):
"""JSON-RPC Eva вернула `error`: текст «<метод>: <error>»."""
class EvaClient(Http):
"""Клиент Eva: счётчики вызовов и 429, время последнего вызова (для паузы между вызовами). Методы JSON-RPC
добавляются в задачах 2–3; здесь — только каркас, общий для цикла и Executor."""
"""Клиент Eva (JSON-RPC 2.2): повторы, пауза между вызовами, счётчики вызовов и 429, время последнего вызова."""
PAGE_SIZE = 200
RETRY_CODES = (429, 502, 503, 504)
ATTEMPTS = 4
MAX_RETRY_AFTER = 60
def __init__(self, cfg: dict):
e = cfg["eva"]
@ -242,8 +249,10 @@ class EvaClient(Http):
elif cfg.get("ca_bundle"):
ctx = ssl.create_default_context(cafile=cfg["ca_bundle"])
super().__init__(e["base_url"], {"Authorization": "Bearer " + e["token"]}, cfg.get("timeout", 60), ctx)
self._token = e["token"]
self.calls = 0
self.rate_limited = 0
self.pacing = 0.0 # с между вызовами; ставит Runner из wiki_pacing
self.last_call_at = None # type: Optional[float] (time.monotonic)
self.last_call_iso = None # type: Optional[str]
self.sleep = time.sleep
@ -255,6 +264,66 @@ class EvaClient(Http):
def reset_counters(self) -> None:
self.calls, self.rate_limited = 0, 0
def _scrub(self, text: str) -> str:
return text.replace(self._token, "***") if self._token else text
def _retrying(self, fn: Callable[[], Any]) -> Any:
"""Вызов с паузой между вызовами и повторами: 429/502/503/504 и сетевые сбои, 4 попытки, ожидание 1, 2, 4 с
(при 429 — Retry-After, не больше 60 с). Тексты ошибок очищаются от токена."""
for attempt in range(self.ATTEMPTS):
if self.pacing > 0 and self.last_call_at is not None:
wait = self.pacing - (time.monotonic() - self.last_call_at)
if wait > 0:
self.sleep(wait)
wait = float(2 ** attempt)
try:
return fn()
except HttpError as e:
if e.code not in self.RETRY_CODES:
raise HttpError(e.code, self._scrub(e.text), e.headers) from None
if e.code == 429:
self.rate_limited += 1
try:
wait = min(max(float(e.headers.get("retry-after", wait)), 0.0), self.MAX_RETRY_AFTER)
except ValueError:
pass
if attempt == self.ATTEMPTS - 1:
raise HttpError(e.code, self._scrub(e.text), e.headers) from None
except ConnectionError as e:
if attempt == self.ATTEMPTS - 1:
raise ConnectionError(self._scrub(str(e))) from None
self.sleep(wait)
raise AssertionError("недостижимо")
def call(self, method: str, **kwargs) -> Any: # type: ignore[override]
"""Метод JSON-RPC: POST <base>/api/?m=<метод>. `error` в ответе — EvaError."""
body = {"jsonrpc": "2.2", "method": method, "callid": str(uuid.uuid4()), "kwargs": kwargs}
data = self._retrying(lambda: Http.call(self, "POST", "/api/", body, {"m": method}))
if isinstance(data, dict) and data.get("error"):
raise EvaError(self._scrub("%s: %s" % (method, data["error"])))
return data.get("result") if isinstance(data, dict) else None
def get_doc(self, code: str, fields: List[str]) -> Optional[dict]:
return self.call("CmfDocument.get", filter=[["code", "==", code]], fields=fields)
def list_docs(self, flt: list, fields: List[str]) -> List[dict]:
out, offset = [], 0 # type: List[dict], int
while True:
chunk = self.call("CmfDocument.list", filter=flt, fields=fields,
slice=[offset, offset + self.PAGE_SIZE]) or []
if isinstance(chunk, dict): # обёрнутый результат
chunk = chunk.get("items") or chunk.get("result") or []
out.extend(chunk)
if len(chunk) < self.PAGE_SIZE:
return out
offset += self.PAGE_SIZE
def download(self, url: str, limit: int) -> bytes:
"""Файл Eva потоком: читается не больше limit + 1 байта, длиннее limit — TooBig. Адрес — только своего хоста."""
if _origin(url) != _origin(self.base) or not url.startswith(self.base):
raise ValueError("чужой хост")
return self._retrying(lambda: self.fetch_bytes(url[len(self.base):], limit))
class FolderTransport:
"""Обмен с хабом через общую папку (hub_url = folder:<путь>), когда сеть до хаба закрыта (VPN на виртуалке).
@ -764,6 +833,7 @@ class Runner:
self.settings = self.hub.settings()
self.executor = Executor(self.eva, self.hub, self.settings)
self.executor.sleep = self.sleep
self.eva.pacing = float(self.settings.get("wiki_pacing", 0.2))
log("настройки хаба v%s" % self.settings.get("settings_version"))
def restore_pace(self) -> None:

134
tests/fake_eva.py Normal file
View File

@ -0,0 +1,134 @@
"""Фейковая Eva для тестов агента: JSON-RPC 2.2 (CmfDocument.get/list) и /files/... на http.server."""
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlsplit
def _field(doc, path):
"""Значение поля по пути `project.code`: ссылка хранится как {"id", "code"}."""
cur = doc
for part in path.split("."):
if not isinstance(cur, dict):
return None
cur = cur.get(part)
return cur
class FakeEva:
token = "eva-secret"
def __init__(self):
self.docs = []
self.files = {} # путь → (байты, с Content-Length?)
self.requests = [] # {"method", "path", "body", "auth"}
self.failures = [] # [status, retry_after] — по одному на ответ
self.rpc_errors = []
self.lock = threading.Lock()
eva = self
class Handler(BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def do_GET(self):
eva.handle(self, "GET")
def do_POST(self):
eva.handle(self, "POST")
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=self.server.serve_forever, daemon=True).start()
@property
def base_url(self):
return "http://127.0.0.1:%d" % self.server.server_port
def stop(self):
self.server.shutdown()
self.server.server_close()
def add_doc(self, id, code, name="", parent=None, tree_parent=None, project=None, orderno=None, text=""):
ref = lambda c: {"id": c, "code": c} if c is not None else None # noqa: E731
self.docs.append({"id": id, "code": code, "name": name, "parent": ref(parent),
"tree_parent": ref(tree_parent), "project": ref(project), "orderno": orderno,
"text": text})
def add_file(self, path, data, length=True):
self.files[path] = (data, length)
def fail_next(self, status, times=1, retry_after=None):
self.failures.extend([[status, retry_after]] * times)
def rpc_error_next(self, text):
self.rpc_errors.append(text)
@property
def api_calls(self):
return [r for r in self.requests if r["path"].startswith("/api/")]
def handle(self, h, verb):
u = urlsplit(h.path)
n = int(h.headers.get("Content-Length") or 0)
raw = h.rfile.read(n) if n else b""
auth = h.headers.get("Authorization")
with self.lock:
self.requests.append({"method": verb, "path": h.path,
"body": json.loads(raw) if raw else None, "auth": auth})
failure = self.failures.pop(0) if self.failures else None
rpc_error = self.rpc_errors.pop(0) if self.rpc_errors else None
if auth != "Bearer " + self.token:
return self._reply(h, 401, b'{"error": "unauthorized"}')
if failure:
extra = {"Retry-After": str(failure[1])} if failure[1] is not None else {}
return self._reply(h, failure[0], b'{"error": "busy"}', extra)
try:
if u.path == "/api/" and verb == "POST":
body = json.loads(raw)
if rpc_error is not None:
out = {"jsonrpc": "2.2", "callid": body.get("callid"), "error": rpc_error}
else:
out = {"jsonrpc": "2.2", "callid": body.get("callid"),
"result": self.rpc(body["method"], body.get("kwargs") or {})}
return self._reply(h, 200, json.dumps(out).encode("utf-8"))
if u.path.startswith("/files/") and verb == "GET":
if u.path not in self.files:
return self._reply(h, 404, b'{"error": "not found"}')
data, with_length = self.files[u.path]
h.send_response(200)
h.send_header("Content-Type", "application/octet-stream")
if with_length:
h.send_header("Content-Length", str(len(data)))
h.end_headers() # без длины HTTP/1.0 отдаёт тело до закрытия соединения
h.wfile.write(data)
return
self._reply(h, 404, b"{}")
except (BrokenPipeError, ConnectionResetError):
pass
def _reply(self, h, status, body, headers=None):
h.send_response(status)
h.send_header("Content-Type", "application/json")
h.send_header("Content-Length", str(len(body)))
for k, v in (headers or {}).items():
h.send_header(k, v)
h.end_headers()
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: {k: d.get(k) for k in fields}) if fields else dict # noqa: E731
if method == "CmfDocument.get":
return pick(found[0]) if found else None
if method == "CmfDocument.list":
lo, hi = kw.get("slice") or [0, len(found)]
return [pick(d) for d in found[lo:hi]]
raise ValueError("метод %s не поддержан" % method)
@staticmethod
def _match(doc, flt):
for field, op, value in flt:
if op != "==" or _field(doc, field) != value:
return False
return True

View File

@ -61,7 +61,7 @@ class ClientsTest(unittest.TestCase):
def test_get_and_counters(self):
self.srv.routes["/x"] = (200, {"Content-Type": "application/json"}, b'{"a": 1}')
self.assertEqual(self.eva.call("GET", "/x"), {"a": 1})
self.assertEqual(agent.Http.call(self.eva, "GET", "/x"), {"a": 1})
self.assertEqual(self.eva.calls, 1)
self.assertIsNotNone(self.eva.last_call_at)
self.assertIsNotNone(self.eva.last_call_iso)
@ -72,21 +72,21 @@ class ClientsTest(unittest.TestCase):
self.assertEqual(self.eva.headers["Authorization"], "Bearer eva-secret")
self.assertNotIn("eva-secret", repr(self.eva))
self.srv.routes["/x"] = (200, {}, b"{}")
self.eva.call("GET", "/x")
agent.Http.call(self.eva, "GET", "/x")
self.assertEqual(self.srv.last_headers["authorization"], "Bearer eva-secret")
def test_secret_not_in_log_output(self):
buf = io.StringIO()
self.srv.routes["/x"] = (200, {}, b"{}")
with contextlib.redirect_stdout(buf):
self.eva.call("GET", "/x")
agent.Http.call(self.eva, "GET", "/x")
agent.log("проверка")
self.assertNotIn("eva-secret", buf.getvalue())
def test_http_error_carries_code_text_headers(self):
self.srv.routes["/x"] = (429, {"Retry-After": "7"}, b"slow down")
with self.assertRaises(agent.HttpError) as cm:
self.eva.call("GET", "/x")
agent.Http.call(self.eva, "GET", "/x")
self.assertEqual(cm.exception.code, 429)
self.assertEqual(cm.exception.headers.get("retry-after"), "7")
self.assertIn("slow down", cm.exception.text)
@ -128,14 +128,14 @@ class ClientsTest(unittest.TestCase):
self.srv.routes["/a"] = (302, {"Location": "http://localhost:%d/b" % port}, b"")
self.srv.routes["/b"] = (200, {}, b"{}")
with self.assertRaises(agent.HttpError) as e:
self.eva.call("GET", "/a")
agent.Http.call(self.eva, "GET", "/a")
self.assertIn("localhost", e.exception.text)
self.assertNotIn("/b", self.srv.requests)
def test_json_call_redirect_same_host_followed(self):
self.srv.routes["/a"] = (302, {"Location": self.srv.base_url + "/b"}, b"")
self.srv.routes["/b"] = (200, {}, b'{"ok": true}')
self.assertEqual(self.eva.call("GET", "/a"), {"ok": True})
self.assertEqual(agent.Http.call(self.eva, "GET", "/a"), {"ok": True})
def test_error_body_read_failure_is_connection_error(self):
import urllib.error
@ -147,14 +147,14 @@ class ClientsTest(unittest.TestCase):
err = urllib.error.HTTPError("http://x", 500, "boom", {}, Broken())
with mock.patch("urllib.request.OpenerDirector.open", side_effect=err):
with self.assertRaises(ConnectionError):
self.eva.call("GET", "/x")
agent.Http.call(self.eva, "GET", "/x")
self.assertEqual(self.eva.calls, 1)
def test_incomplete_read_is_connection_error(self):
import http.client
with mock.patch("urllib.request.OpenerDirector.open", side_effect=http.client.IncompleteRead(b"ab", 10)):
with self.assertRaises(ConnectionError):
self.eva.call("GET", "/x")
agent.Http.call(self.eva, "GET", "/x")
def test_incomplete_read_inside_response_body(self):
srv = socket.socket()
@ -172,19 +172,19 @@ class ClientsTest(unittest.TestCase):
client = agent.EvaClient(eva_cfg("http://127.0.0.1:%d" % srv.getsockname()[1]))
try:
with self.assertRaises(ConnectionError):
client.call("GET", "/x")
agent.Http.call(client, "GET", "/x")
finally:
t.join(2)
srv.close()
def test_unreachable_raises_connection_error(self):
with self.assertRaises(ConnectionError):
agent.EvaClient(eva_cfg("http://127.0.0.1:1")).call("GET", "/x")
agent.Http.call(agent.EvaClient(eva_cfg("http://127.0.0.1:1")), "GET", "/x")
def test_non_json_200_becomes_http_error(self):
self.srv.routes["/x"] = (200, {}, b"<html>\xff\xfe not json")
with self.assertRaises(agent.HttpError) as cm:
self.eva.call("GET", "/x")
agent.Http.call(self.eva, "GET", "/x")
self.assertEqual(cm.exception.code, 200)
self.assertEqual(self.eva.calls, 1)

150
tests/test_eva_client.py Normal file
View File

@ -0,0 +1,150 @@
import io
import unittest
from contextlib import redirect_stderr, redirect_stdout
from tests import _common as c
from tests.fake_eva import FakeEva
agent = c.agent
class EvaClientTest(unittest.TestCase):
def setUp(self):
self.eva = FakeEva()
self.addCleanup(self.eva.stop)
self.cli = agent.EvaClient(c.eva_cfg(self.eva.base_url, timeout=10))
self.sleeps = []
self.cli.sleep = self.sleeps.append
def test_call_ok(self):
self.eva.add_doc(1, "DOC-1", "Корень", project="P")
doc = self.cli.get_doc("DOC-1", ["code", "name"])
self.assertEqual(doc, {"code": "DOC-1", "name": "Корень"})
self.assertIsNone(self.cli.get_doc("DOC-404", ["code"]))
req = self.eva.api_calls[0]
self.assertEqual(req["path"], "/api/?m=CmfDocument.get")
self.assertEqual(req["auth"], "Bearer eva-secret")
self.assertEqual(req["body"]["jsonrpc"], "2.2")
self.assertEqual(req["body"]["kwargs"]["filter"], [["code", "==", "DOC-1"]])
self.assertEqual(self.cli.calls, 2)
def test_retry_429_retry_after(self):
self.eva.fail_next(429, 1, retry_after=7)
self.assertEqual(self.cli.call("CmfDocument.list", filter=[], fields=["code"]), [])
self.assertEqual(self.sleeps, [7])
self.assertEqual(self.cli.rate_limited, 1)
def test_retry_after_capped(self):
self.eva.fail_next(429, 1, retry_after=600)
self.cli.call("CmfDocument.list", filter=[], fields=["code"])
self.assertEqual(self.sleeps, [60])
def test_retry_503_then_ok(self):
self.eva.fail_next(503, 2)
self.cli.call("CmfDocument.list", filter=[], fields=["code"])
self.assertEqual(self.sleeps, [1, 2])
self.assertEqual(self.cli.rate_limited, 0)
def test_gives_up_after_4(self):
self.eva.fail_next(503, 10)
with self.assertRaises(agent.HttpError) as cm:
self.cli.call("CmfDocument.list", filter=[], fields=["code"])
self.assertEqual(cm.exception.code, 503)
self.assertEqual(len(self.eva.api_calls), 4)
self.assertEqual(self.sleeps, [1, 2, 4])
def test_no_retry_on_401(self):
self.cli.headers["Authorization"] = "Bearer wrong"
with self.assertRaises(agent.HttpError):
self.cli.call("CmfDocument.list", filter=[], fields=["code"])
self.assertEqual(len(self.eva.api_calls), 1)
def test_rpc_error(self):
self.eva.rpc_error_next("нет прав")
with self.assertRaises(agent.EvaError) as cm:
self.cli.call("CmfDocument.get", filter=[], fields=["code"])
self.assertEqual(str(cm.exception), "CmfDocument.get: нет прав")
def test_list_pages_200(self):
for i in range(450):
self.eva.add_doc(i + 1, "DOC-%d" % (i + 1), project="P")
docs = self.cli.list_docs([["project.code", "==", "P"]], ["code"])
self.assertEqual(len(docs), 450)
self.assertEqual(len(self.eva.api_calls), 3)
self.assertEqual([r["body"]["kwargs"]["slice"] for r in self.eva.api_calls],
[[0, 200], [200, 400], [400, 600]])
def test_list_unwraps_dict(self):
orig = self.eva.rpc
self.eva.rpc = lambda m, kw: {"items": orig(m, kw)}
self.eva.add_doc(1, "DOC-1")
self.assertEqual(self.cli.list_docs([], ["code"]), [{"code": "DOC-1"}])
def test_tree_parent_filter(self):
self.eva.add_doc(1, "DOC-1")
self.eva.add_doc(2, "DOC-2", tree_parent="DOC-1")
self.eva.add_doc(3, "DOC-3", tree_parent="DOC-2")
got = self.cli.list_docs([["tree_parent.code", "==", "DOC-1"]], ["code"])
self.assertEqual(got, [{"code": "DOC-2"}])
def test_pacing_between_calls(self):
self.cli.pacing = 0.5
self.cli.call("CmfDocument.list", filter=[], fields=["code"])
self.assertEqual(self.sleeps, []) # перед первым вызовом паузы нет
self.cli.call("CmfDocument.list", filter=[], fields=["code"])
self.assertEqual(len(self.sleeps), 1)
self.assertTrue(0 < self.sleeps[0] <= 0.5)
def test_download_ok(self):
self.eva.add_file("/files/a.bin", b"abc")
self.assertEqual(self.cli.download(self.eva.base_url + "/files/a.bin", 10), b"abc")
self.assertEqual(self.eva.requests[-1]["auth"], "Bearer eva-secret")
def test_download_foreign_host(self):
with self.assertRaises(ValueError):
self.cli.download("http://example.invalid/files/a.bin", 10)
self.assertEqual(self.eva.requests, [])
def test_download_limit(self):
limit = 10 * 1024 * 1024
self.eva.add_file("/files/big.bin", b"x" * (limit + 1), length=False)
self.eva.add_file("/files/ok.bin", b"x" * limit, length=False)
with self.assertRaises(agent.TooBig):
self.cli.download(self.eva.base_url + "/files/big.bin", limit)
self.assertEqual(len(self.cli.download(self.eva.base_url + "/files/ok.bin", limit)), limit)
def test_download_404(self):
with self.assertRaises(agent.HttpError) as cm:
self.cli.download(self.eva.base_url + "/files/none", 10)
self.assertEqual(cm.exception.code, 404)
def test_token_not_logged(self):
out, err = io.StringIO(), io.StringIO()
texts = []
with redirect_stdout(out), redirect_stderr(err):
self.eva.fail_next(503, 8)
for call in (lambda: self.cli.call("CmfDocument.list", filter=[], fields=["code"]),
lambda: self.cli.download(self.eva.base_url + "/files/none", 10)):
try:
call()
except Exception as e: # noqa: BLE001
texts.append(str(e))
self.eva.rpc_error_next("сбой eva-secret")
try:
self.cli.call("CmfDocument.get", filter=[], fields=["code"])
except agent.EvaError as e:
texts.append(str(e))
dead = agent.EvaClient(c.eva_cfg("http://127.0.0.1:1"))
dead.sleep = lambda s: None
try:
dead.call("CmfDocument.get", filter=[], fields=["code"])
except ConnectionError as e:
texts.append(str(e))
self.assertEqual(len(texts), 4)
for t in texts + [out.getvalue(), err.getvalue()]:
self.assertNotIn("eva-secret", t)
self.assertNotIn("eva-secret", repr(self.cli))
if __name__ == "__main__":
unittest.main()

View File

@ -350,6 +350,7 @@ class RunnerTest(unittest.TestCase):
self.runner.execute = lambda cmd: (seen.append(self.runner.settings["settings_version"]), real(cmd))
self.runner.loop()
self.assertEqual(seen, [2])
self.assertEqual(self.runner.eva.pacing, 0.5)
def test_hub_down_at_start_keeps_polling(self):
calls = []