"""Фейковый хаб для тестов агента: маршруты /agent/* в памяти.""" import base64 import hashlib import json import re import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer class FakeHub: token = "agent-token" def __init__(self, settings): self.settings = dict(settings) self.queue = [] self.commands = {} self.results = {} self.states = [] self.known = [] self.claims = 0 self.claim_bodies = [] self.fail_results = 0 self.reject_results = {} self.in_flight = None self.release = None # {"version", "sha256", "content"} — опубликованная версия агента self.release_requests = 0 self.files = {} # sha256 → содержимое (хранилище файлов вложений хаба) self.file_requests = 0 self.claim_wait = 0 # с: при пустой очереди держать claim, как хаб self.lock = threading.Lock() hub = self class Handler(BaseHTTPRequestHandler): def log_message(self, *a): pass def do_GET(self): hub.handle(self, "GET") def do_POST(self): hub.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 enqueue(self, ctype, params, pause=0): cid = len(self.commands) + 1 self.commands[cid] = {"id": cid, "type": ctype, "params": params, "pause_seconds": pause, "status": "queued"} self.queue.append(cid) return cid def put_file(self, data): sha = hashlib.sha256(data).hexdigest() self.files[sha] = data return sha def _issued(self, c): """Параметры при выдаче: как хаб, к ensure_attachment дописывается size файла (None — файла нет).""" params = dict(c["params"]) if c["type"] == "ensure_attachment": data = self.files.get(params.get("file")) params["size"] = len(data) if data is not None else None return params def _release_ref(self): r = self.release return {"version": r["version"], "sha256": r["sha256"]} if r else None def handle(self, h, method): length = int(h.headers.get("Content-Length") or 0) body = json.loads(h.rfile.read(length).decode("utf-8")) if length else {} if h.headers.get("Authorization") != "Bearer " + self.token: return self._send(h, 401, {"error": "unauthorized"}) path = h.path.split("?")[0] if path == "/agent/claim" and self.claim_wait: end = time.monotonic() + self.claim_wait while time.monotonic() < end: # ждать без блокировки: другие запросы (enqueue) проходят with self.lock: if self.queue: break time.sleep(0.05) with self.lock: code, payload = self.route(method, path, body) self._send(h, code, payload) def _send(self, h, code, payload): data = json.dumps(payload, ensure_ascii=False).encode("utf-8") h.send_response(code) h.send_header("Content-Type", "application/json") h.send_header("Content-Length", str(len(data))) h.end_headers() h.wfile.write(data) def route(self, method, path, body): if path == "/agent/claim": self.claim_bodies.append(body or {}) self.claims += 1 if self.in_flight and self.commands[self.in_flight]["status"] == "claimed": return 200, {"command": None, "settings_version": self.settings["settings_version"], "agent_release": self._release_ref()} if not self.queue: return 200, {"command": None, "settings_version": self.settings["settings_version"], "agent_release": self._release_ref()} cid = self.queue.pop(0) self.commands[cid]["status"] = "claimed" self.in_flight = cid c = self.commands[cid] return 200, {"command": {"id": cid, "type": c["type"], "params": self._issued(c), "pause_seconds": c["pause_seconds"]}, "settings_version": self.settings["settings_version"], "agent_release": self._release_ref()} if path == "/agent/release": self.release_requests += 1 return (200, self.release) if self.release else (404, {"error": "no_release"}) if path == "/agent/settings": return 200, self.settings if path == "/agent/known-keys": return 200, {"keys": list(self.known)} if path == "/agent/state": self.states.append(body) return 200, {"issues": len(body.get("issues") or []), "links": len(body.get("links") or [])} m = re.match(r"^/agent/files/([^/]+)$", path) if m and method == "GET": self.file_requests += 1 data = self.files.get(m.group(1)) if data is None: return 404, {"error": "no_file", "detail": "файла %s нет в хранилище" % m.group(1)} return 200, {"file": m.group(1), "size": len(data), "content": base64.b64encode(data).decode("ascii")} m = re.match(r"^/agent/commands/(\d+)$", path) if m: c = self.commands.get(int(m.group(1))) return (200, dict(c, params=self._issued(c))) if c else (404, {"error": "not_found"}) m = re.match(r"^/agent/commands/(\d+)/result$", path) if m: cid = int(m.group(1)) if self.fail_results > 0: self.fail_results -= 1 return 503, {"error": "busy"} if cid in self.reject_results: return self.reject_results[cid], {"error": "rejected"} if cid not in self.commands: return 404, {"error": "not_found"} if self.commands[cid]["status"] in ("done", "failed", "cancelled"): return 200, self.commands[cid] # терминальный статус не перезаписывается (как живой хаб) self.results[cid] = body self.commands[cid]["status"] = body.get("status") if self.in_flight == cid: self.in_flight = None return 200, self.commands[cid] return 404, {"error": "not_found", "detail": path}