From 4ee2cf5bb08c4f4922250977bfeda58d45673f0a Mon Sep 17 00:00:00 2001 From: user Date: Mon, 16 Feb 2026 19:05:28 +0100 Subject: [PATCH] test: add control API tests 29 tests covering request parsing, JSON response format, all GET/POST handlers with mock context, 404/405 error routing. Co-Authored-By: Claude Opus 4.6 --- tests/test_api.py | 286 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 286 insertions(+) create mode 100644 tests/test_api.py diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..bab6683 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,286 @@ +"""Tests for the control API module.""" + +import asyncio +import json +from unittest.mock import AsyncMock, MagicMock + +from s5p.api import ( + _handle_config, + _handle_metrics, + _handle_pool, + _handle_status, + _json_response, + _parse_request, + _route, +) +from s5p.config import ChainHop, Config, PoolSourceConfig, ProxyPoolConfig +from s5p.metrics import Metrics + +# -- request parsing --------------------------------------------------------- + + +class TestParseRequest: + """Test HTTP request line parsing.""" + + def test_get(self): + assert _parse_request(b"GET /status HTTP/1.1\r\n") == ("GET", "/status") + + def test_post(self): + assert _parse_request(b"POST /reload HTTP/1.1\r\n") == ("POST", "/reload") + + def test_strips_query_string(self): + assert _parse_request(b"GET /pool?foo=bar HTTP/1.1\r\n") == ("GET", "/pool") + + def test_method_uppercased(self): + assert _parse_request(b"get /metrics HTTP/1.1\r\n") == ("GET", "/metrics") + + def test_empty(self): + assert _parse_request(b"") == ("", "") + + def test_garbage(self): + assert _parse_request(b"\xff\xfe") == ("", "") + + def test_incomplete(self): + assert _parse_request(b"GET\r\n") == ("", "") + + +# -- JSON response ----------------------------------------------------------- + + +class TestJsonResponse: + """Test HTTP JSON response builder.""" + + def test_format(self): + writer = MagicMock() + written = bytearray() + writer.write = lambda data: written.extend(data) + + _json_response(writer, 200, {"ok": True}) + text = written.decode() + assert "HTTP/1.1 200 OK\r\n" in text + assert "Content-Type: application/json\r\n" in text + assert "Connection: close\r\n" in text + # body after double newline + body = text.split("\r\n\r\n", 1)[1] + assert json.loads(body) == {"ok": True} + + def test_404(self): + writer = MagicMock() + written = bytearray() + writer.write = lambda data: written.extend(data) + + _json_response(writer, 404, {"error": "not found"}) + text = written.decode() + assert "HTTP/1.1 404 Not Found\r\n" in text + + +# -- helpers ----------------------------------------------------------------- + + +def _make_ctx( + config: Config | None = None, + pool: MagicMock | None = None, +) -> dict: + """Build a mock context dict.""" + return { + "config": config or Config(), + "metrics": Metrics(), + "pool": pool, + "hop_pool": None, + } + + +# -- GET handlers ------------------------------------------------------------ + + +class TestHandleStatus: + """Test GET /status handler.""" + + def test_basic(self): + ctx = _make_ctx() + ctx["metrics"].connections = 10 + ctx["metrics"].success = 8 + status, body = _handle_status(ctx) + assert status == 200 + assert body["connections"] == 10 + assert body["success"] == 8 + assert "uptime" in body + + def test_with_pool(self): + pool = MagicMock() + pool.alive_count = 5 + pool.count = 10 + ctx = _make_ctx(pool=pool) + _, body = _handle_status(ctx) + assert body["pool"] == {"alive": 5, "total": 10} + + def test_with_chain(self): + config = Config(chain=[ChainHop("socks5", "127.0.0.1", 9050)]) + ctx = _make_ctx(config=config) + _, body = _handle_status(ctx) + assert body["chain"] == ["socks5://127.0.0.1:9050"] + + +class TestHandleMetrics: + """Test GET /metrics handler.""" + + def test_returns_dict(self): + ctx = _make_ctx() + ctx["metrics"].connections = 42 + ctx["metrics"].bytes_in = 1024 + status, body = _handle_metrics(ctx) + assert status == 200 + assert body["connections"] == 42 + assert body["bytes_in"] == 1024 + assert "uptime" in body + + +class TestHandlePool: + """Test GET /pool handler.""" + + def test_no_pool(self): + ctx = _make_ctx() + status, body = _handle_pool(ctx) + assert status == 200 + assert body == {"alive": 0, "total": 0, "proxies": {}} + + def test_with_entries(self): + pool = MagicMock() + pool.alive_count = 1 + pool.count = 2 + entry_alive = MagicMock( + alive=True, fails=0, tests=5, + last_ok=100.0, last_test=100.0, last_seen=100.0, + ) + entry_dead = MagicMock( + alive=False, fails=3, tests=5, + last_ok=0.0, last_test=100.0, last_seen=100.0, + ) + pool._proxies = { + "socks5://1.2.3.4:1080": entry_alive, + "socks5://5.6.7.8:1080": entry_dead, + } + ctx = _make_ctx(pool=pool) + _, body = _handle_pool(ctx) + assert len(body["proxies"]) == 2 + assert body["proxies"]["socks5://1.2.3.4:1080"]["alive"] is True + + def test_alive_only(self): + pool = MagicMock() + pool.alive_count = 1 + pool.count = 2 + entry_alive = MagicMock( + alive=True, fails=0, tests=5, + last_ok=100.0, last_test=100.0, last_seen=100.0, + ) + entry_dead = MagicMock( + alive=False, fails=3, tests=5, + last_ok=0.0, last_test=100.0, last_seen=100.0, + ) + pool._proxies = { + "socks5://1.2.3.4:1080": entry_alive, + "socks5://5.6.7.8:1080": entry_dead, + } + ctx = _make_ctx(pool=pool) + _, body = _handle_pool(ctx, alive_only=True) + assert len(body["proxies"]) == 1 + assert "socks5://1.2.3.4:1080" in body["proxies"] + + +class TestHandleConfig: + """Test GET /config handler.""" + + def test_basic(self): + config = Config(timeout=15.0, retries=5, log_level="debug") + ctx = _make_ctx(config=config) + status, body = _handle_config(ctx) + assert status == 200 + assert body["timeout"] == 15.0 + assert body["retries"] == 5 + assert body["log_level"] == "debug" + + def test_with_proxy_pool(self): + pp = ProxyPoolConfig( + sources=[PoolSourceConfig(url="http://api:8081/proxies")], + refresh=600.0, + test_interval=60.0, + max_fails=5, + ) + config = Config(proxy_pool=pp) + ctx = _make_ctx(config=config) + _, body = _handle_config(ctx) + assert body["proxy_pool"]["refresh"] == 600.0 + assert body["proxy_pool"]["sources"][0]["url"] == "http://api:8081/proxies" + + +# -- routing ----------------------------------------------------------------- + + +class TestRouting: + """Test request routing and error responses.""" + + def test_get_status(self): + ctx = _make_ctx() + status, body = asyncio.run(_route("GET", "/status", ctx)) + assert status == 200 + + def test_get_metrics(self): + ctx = _make_ctx() + status, _ = asyncio.run(_route("GET", "/metrics", ctx)) + assert status == 200 + + def test_get_pool(self): + ctx = _make_ctx() + status, _ = asyncio.run(_route("GET", "/pool", ctx)) + assert status == 200 + + def test_get_pool_alive(self): + ctx = _make_ctx() + status, _ = asyncio.run(_route("GET", "/pool/alive", ctx)) + assert status == 200 + + def test_get_config(self): + ctx = _make_ctx() + status, _ = asyncio.run(_route("GET", "/config", ctx)) + assert status == 200 + + def test_post_reload(self): + ctx = _make_ctx() + ctx["reload_fn"] = AsyncMock() + status, body = asyncio.run(_route("POST", "/reload", ctx)) + assert status == 200 + assert body == {"ok": True} + + def test_post_pool_test(self): + pool = MagicMock() + pool._run_health_tests = AsyncMock() + ctx = _make_ctx(pool=pool) + status, body = asyncio.run(_route("POST", "/pool/test", ctx)) + assert status == 200 + assert body == {"ok": True} + + def test_post_pool_refresh(self): + pool = MagicMock() + pool._fetch_all_sources = AsyncMock() + ctx = _make_ctx(pool=pool) + status, body = asyncio.run(_route("POST", "/pool/refresh", ctx)) + assert status == 200 + assert body == {"ok": True} + + def test_404(self): + ctx = _make_ctx() + status, body = asyncio.run(_route("GET", "/nonexistent", ctx)) + assert status == 404 + assert "error" in body + + def test_405_wrong_method(self): + ctx = _make_ctx() + status, body = asyncio.run(_route("POST", "/status", ctx)) + assert status == 405 + assert "GET" in body["error"] + + def test_405_get_on_post_route(self): + ctx = _make_ctx() + status, body = asyncio.run(_route("GET", "/reload", ctx)) + assert status == 405 + assert "POST" in body["error"]