257 lines
11 KiB
Python
257 lines
11 KiB
Python
"""HTTP-слой (Http, EvaClient, HubClient) и конфиг. Перенос из jira-agent: сценарии Http без Jira-специфики."""
|
||
import contextlib
|
||
import io
|
||
import json
|
||
import os
|
||
import socket
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from unittest import mock
|
||
|
||
from _common import agent, eva_cfg
|
||
|
||
|
||
class MiniServer:
|
||
"""Крошечный сервер: routes — путь → (код, заголовки, тело-байты); requests — что пришло."""
|
||
|
||
def __init__(self):
|
||
self.routes = {}
|
||
self.requests = []
|
||
self.last_headers = {}
|
||
srv = self
|
||
|
||
class H(BaseHTTPRequestHandler):
|
||
def log_message(self, *a):
|
||
pass
|
||
|
||
def do_GET(self):
|
||
srv.requests.append(self.path)
|
||
srv.last_headers = {k.lower(): v for k, v in self.headers.items()}
|
||
code, headers, body = srv.routes.get(self.path, (404, {}, b"{}"))
|
||
self.send_response(code)
|
||
for k, v in headers.items():
|
||
self.send_header(k, v)
|
||
if "Content-Length" not in headers:
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
if self.command != "HEAD":
|
||
self.wfile.write(body)
|
||
|
||
self.server = ThreadingHTTPServer(("127.0.0.1", 0), H)
|
||
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()
|
||
|
||
|
||
class ClientsTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.srv = MiniServer()
|
||
self.eva = agent.EvaClient(eva_cfg(self.srv.base_url))
|
||
|
||
def tearDown(self):
|
||
self.srv.stop()
|
||
|
||
def test_get_and_counters(self):
|
||
self.srv.routes["/x"] = (200, {"Content-Type": "application/json"}, b'{"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)
|
||
self.eva.reset_counters()
|
||
self.assertEqual((self.eva.calls, self.eva.rate_limited), (0, 0))
|
||
|
||
def test_bearer_header_and_no_secret_in_repr(self):
|
||
self.assertEqual(self.eva.headers["Authorization"], "Bearer eva-secret")
|
||
self.assertNotIn("eva-secret", repr(self.eva))
|
||
self.srv.routes["/x"] = (200, {}, b"{}")
|
||
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):
|
||
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:
|
||
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)
|
||
self.assertEqual(self.eva.calls, 1)
|
||
|
||
def test_fetch_bytes(self):
|
||
self.srv.routes["/f"] = (200, {}, b"\x89PNG\x00\x01")
|
||
self.assertEqual(self.eva.fetch_bytes("/f", 100), b"\x89PNG\x00\x01")
|
||
self.assertEqual(self.srv.last_headers["authorization"], "Bearer eva-secret")
|
||
self.assertEqual(self.eva.calls, 1)
|
||
|
||
def test_fetch_bytes_limit(self):
|
||
self.srv.routes["/f"] = (200, {}, b"x" * 20)
|
||
with self.assertRaises(agent.TooBig):
|
||
self.eva.fetch_bytes("/f", 10)
|
||
self.assertEqual(self.eva.fetch_bytes("/f", 20), b"x" * 20)
|
||
|
||
def test_fetch_bytes_redirect_same_host(self):
|
||
self.srv.routes["/f"] = (302, {"Location": self.srv.base_url + "/moved"}, b"")
|
||
self.srv.routes["/moved"] = (200, {}, b"abc")
|
||
self.assertEqual(self.eva.fetch_bytes("/f", 100), b"abc")
|
||
|
||
def test_fetch_bytes_redirect_other_host(self):
|
||
port = self.srv.server.server_port
|
||
self.srv.routes["/f"] = (302, {"Location": "http://localhost:%d/moved" % port}, b"")
|
||
self.srv.routes["/moved"] = (200, {}, b"abc")
|
||
with self.assertRaises(agent.RedirectAway) as e:
|
||
self.eva.fetch_bytes("/f", 100)
|
||
self.assertEqual(e.exception.host, "localhost")
|
||
self.assertNotIn("/moved", self.srv.requests)
|
||
|
||
def test_fetch_bytes_cut_body_is_connection_error(self):
|
||
self.srv.routes["/f"] = (200, {"Content-Length": "40"}, b"x" * 10) # обещано 40, отдано 10
|
||
with self.assertRaises(ConnectionError):
|
||
self.eva.fetch_bytes("/f", 1000)
|
||
|
||
def test_json_call_redirect_other_host_refused(self):
|
||
port = self.srv.server.server_port
|
||
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:
|
||
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(agent.Http.call(self.eva, "GET", "/a"), {"ok": True})
|
||
|
||
def test_error_body_read_failure_is_connection_error(self):
|
||
import urllib.error
|
||
|
||
class Broken(io.BytesIO):
|
||
def read(self, *a):
|
||
raise TimeoutError("timed out")
|
||
|
||
err = urllib.error.HTTPError("http://x", 500, "boom", {}, Broken())
|
||
with mock.patch("urllib.request.OpenerDirector.open", side_effect=err):
|
||
with self.assertRaises(ConnectionError):
|
||
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):
|
||
agent.Http.call(self.eva, "GET", "/x")
|
||
|
||
def test_incomplete_read_inside_response_body(self):
|
||
srv = socket.socket()
|
||
srv.bind(("127.0.0.1", 0))
|
||
srv.listen(1)
|
||
|
||
def serve():
|
||
c, _ = srv.accept()
|
||
c.recv(4096)
|
||
c.sendall(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 100\r\n\r\n{\"a\":")
|
||
c.close()
|
||
|
||
t = threading.Thread(target=serve, daemon=True)
|
||
t.start()
|
||
client = agent.EvaClient(eva_cfg("http://127.0.0.1:%d" % srv.getsockname()[1]))
|
||
try:
|
||
with self.assertRaises(ConnectionError):
|
||
agent.Http.call(client, "GET", "/x")
|
||
finally:
|
||
t.join(2)
|
||
srv.close()
|
||
|
||
def test_unreachable_raises_connection_error(self):
|
||
with self.assertRaises(ConnectionError):
|
||
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:
|
||
agent.Http.call(self.eva, "GET", "/x")
|
||
self.assertEqual(cm.exception.code, 200)
|
||
self.assertEqual(self.eva.calls, 1)
|
||
|
||
def test_hub_client_claim_and_wiki_pages_routes(self):
|
||
calls = []
|
||
hub = agent.HubClient(eva_cfg(hub_url="http://x", hub_token="t"))
|
||
hub.call = lambda method, path, body=None, params=None, timeout=None: (calls.append((method, path, body, timeout)), {})[1]
|
||
hub.claim(1, 2, None)
|
||
hub.wiki_pages(7, [{"code": "DOC-1"}])
|
||
hub.upload_file(b"abc")
|
||
hub.wiki_versions(7, [{"code": "DOCH-1"}])
|
||
self.assertEqual(calls[3], ("POST", "/agent/wiki/versions", {"command_id": 7, "versions": [{"code": "DOCH-1"}]}, None))
|
||
self.assertEqual(calls[0][:2], ("POST", "/agent/claim"))
|
||
self.assertEqual(calls[0][2]["last_call_at"], None)
|
||
self.assertGreaterEqual(calls[0][3], 45)
|
||
self.assertEqual(calls[1], ("POST", "/agent/wiki/pages", {"command_id": 7, "pages": [{"code": "DOC-1"}]}, None))
|
||
self.assertEqual(calls[2][:2], ("POST", "/agent/files"))
|
||
self.assertEqual(calls[2][2], {"content": "YWJj"})
|
||
|
||
def test_load_config_requires_fields(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
path = os.path.join(d, "agent.json")
|
||
|
||
def write(cfg):
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(cfg, f)
|
||
|
||
for bad in ({"hub_url": "http://x", "hub_token": "t"},
|
||
{"hub_url": "http://x", "hub_token": "t", "eva": {"base_url": "https://e"}},
|
||
{"hub_url": "http://x", "hub_token": "t", "eva": {"token": "z"}},
|
||
{"hub_url": "http://x", "hub_token": "t", "eva": "https://e"}):
|
||
write(bad)
|
||
with self.assertRaises(SystemExit):
|
||
agent.load_config(path)
|
||
write({"hub_url": "http://x", "hub_token": "t", "eva": {"base_url": "https://e", "token": "z"}})
|
||
cfg = agent.load_config(path)
|
||
self.assertEqual((cfg["timeout"], cfg["poll_seconds"], cfg["verify_ssl"]), (60, 5, True))
|
||
self.assertNotIn("auto_update", cfg) # умолчание true применяет _run: `is not False`
|
||
with self.assertRaises(SystemExit):
|
||
agent.load_config(os.path.join(d, "нет.json"))
|
||
|
||
def test_example_config_loads(self):
|
||
cfg = agent.load_config(os.path.join(agent.HERE, "agent.example.json"))
|
||
self.assertEqual(cfg["auto_update"], True)
|
||
self.assertIn("base_url", cfg["eva"])
|
||
|
||
def test_check_command(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
path = os.path.join(d, "agent.json")
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(eva_cfg(), f)
|
||
out = io.StringIO()
|
||
with contextlib.redirect_stdout(out):
|
||
self.assertEqual(agent.main(["--config", path, "check"]), 3) # хаб и Eva недоступны (порт 1)
|
||
self.assertIn("хаб: НЕДОСТУПЕН", out.getvalue())
|
||
|
||
def test_selftest_command(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
path = os.path.join(d, "agent.json")
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(eva_cfg(), f)
|
||
out = io.StringIO()
|
||
with contextlib.redirect_stdout(out):
|
||
self.assertEqual(agent.main(["--config", path, "selftest"]), 3) # недоступность — 3, а не 1
|
||
self.assertIn("eva-agent v%d" % agent.AGENT_VERSION, out.getvalue())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|