75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
"""Фейковое реле хаба для тестов агента: забирает req/<id>.json из папки и отвечает resp/<id>.json.
|
||
Режимы: forward — переслать в HTTP (FakeHub) и вернуть ответ; silent — забрать и не отвечать;
|
||
garbage — положить нечитаемый ответ; late — ответить через delay секунд."""
|
||
import json
|
||
import os
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
|
||
def write_atomic(directory, name, payload):
|
||
tmp = os.path.join(directory, "." + name + ".tmp")
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
f.write(payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False))
|
||
os.replace(tmp, os.path.join(directory, name))
|
||
|
||
|
||
class FakeRelay:
|
||
def __init__(self, root, target_base_url="", mode="forward", delay=0.0):
|
||
self.root, self.base, self.mode, self.delay = root, target_base_url, mode, delay
|
||
self.requests = []
|
||
self._stop = threading.Event()
|
||
self._thread = threading.Thread(target=self._loop, daemon=True)
|
||
for sub in ("req", "resp"):
|
||
os.makedirs(os.path.join(root, sub), exist_ok=True)
|
||
|
||
def start(self):
|
||
self._thread.start()
|
||
return self
|
||
|
||
def stop(self):
|
||
self._stop.set()
|
||
self._thread.join(5)
|
||
|
||
def _loop(self):
|
||
req_dir, resp_dir = os.path.join(self.root, "req"), os.path.join(self.root, "resp")
|
||
while not self._stop.is_set():
|
||
for name in sorted(os.listdir(req_dir)):
|
||
if name.startswith(".") or not name.endswith(".json"):
|
||
continue
|
||
path = os.path.join(req_dir, name)
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
req = json.load(f)
|
||
os.remove(path)
|
||
except (OSError, ValueError):
|
||
continue
|
||
self.requests.append(req)
|
||
if self.mode == "silent":
|
||
continue
|
||
if self.mode == "garbage":
|
||
write_atomic(resp_dir, name, "{не json")
|
||
continue
|
||
if self.mode == "late":
|
||
time.sleep(self.delay)
|
||
status, body = self._forward(req)
|
||
write_atomic(resp_dir, name, {"id": req["id"], "status": status, "body": body})
|
||
time.sleep(0.02)
|
||
|
||
def _forward(self, req):
|
||
data = json.dumps(req["body"]).encode("utf-8") if req.get("body") is not None else None
|
||
r = urllib.request.Request(self.base + req["path"], data=data, method=req["method"])
|
||
r.add_header("Authorization", req["authorization"])
|
||
if data is not None:
|
||
r.add_header("Content-Type", "application/json")
|
||
try:
|
||
with urllib.request.urlopen(r, timeout=10) as resp:
|
||
return resp.status, resp.read().decode("utf-8")
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
return e.code, e.read().decode("utf-8")
|
||
finally:
|
||
e.close()
|