diff --git a/openspec/changes/eva-agent-mvp/design.md b/openspec/changes/eva-agent-mvp/design.md index 6400602..9dc2283 100644 --- a/openspec/changes/eva-agent-mvp/design.md +++ b/openspec/changes/eva-agent-mvp/design.md @@ -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). Цена ошибки: нет. diff --git a/src/agent.py b/src/agent.py index 909ffc9..7e8df37 100644 --- a/src/agent.py +++ b/src/agent.py @@ -228,9 +228,16 @@ class Http: """Вызывается на каждый полученный HTTP-ответ (любой код).""" +class EvaError(Exception): + """JSON-RPC Eva вернула `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 /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: diff --git a/tests/fake_eva.py b/tests/fake_eva.py new file mode 100644 index 0000000..28afa24 --- /dev/null +++ b/tests/fake_eva.py @@ -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 diff --git a/tests/test_clients.py b/tests/test_clients.py index 79011ec..1d07005 100644 --- a/tests/test_clients.py +++ b/tests/test_clients.py @@ -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"\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) diff --git a/tests/test_eva_client.py b/tests/test_eva_client.py new file mode 100644 index 0000000..292c1ce --- /dev/null +++ b/tests/test_eva_client.py @@ -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() diff --git a/tests/test_runner.py b/tests/test_runner.py index cffccf1..7a091a9 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -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 = []