cc-ci/tests/unit/test_http.py

"""Unit tests for runner/harness/http.py (Phase 2 §4.2 vendored helpers).

Pure-Python: a tiny in-process http.server on 127.0.0.1 serves rigged responses so we can prove the
parsing + retry semantics deterministically. No network egress.
"""

from __future__ import annotations

import http.server
import json
import os
import socket
import socketserver
import sys
import threading

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "runner"))
from harness import http as harness_http  # noqa: E402


class _State:
    counter = 0
    healthy_after = 0  # set by tests


class _Handler(http.server.BaseHTTPRequestHandler):
    def log_message(self, *_a, **_k):  # silence the test log
        pass

    def _respond(self, code: int, body: bytes, content_type: str = "application/json"):
        self.send_response(code)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == "/ok-json":
            self._respond(200, b'{"hello": "world"}')
            return
        if self.path == "/notfound":
            self._respond(404, b'{"error": "nope"}')
            return
        if self.path == "/text":
            self._respond(200, b"plain", content_type="text/plain")
            return
        if self.path == "/flaky":
            _State.counter += 1
            if _State.counter <= _State.healthy_after:
                self._respond(503, b'{"err": "unhealthy"}')
            else:
                self._respond(200, b'{"ok": true}')
            return
        if self.path.startswith("/echo-headers"):
            self._respond(200, json.dumps(dict(self.headers)).encode())
            return
        self._respond(500, b'{"err": "unknown"}')

    def do_POST(self):
        length = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(length) if length else b""
        ctype = self.headers.get("Content-Type", "")
        if self.path == "/echo-json":
            try:
                parsed = json.loads(body) if body else None
            except Exception:  # noqa: BLE001
                parsed = None
            self._respond(200, json.dumps({"content_type": ctype, "body": parsed}).encode())
            return
        if self.path == "/echo-form":
            self._respond(
                200,
                json.dumps({"content_type": ctype, "raw": body.decode()}).encode(),
            )
            return
        self._respond(500, b'{"err": "unknown"}')


def _free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


@pytest.fixture(scope="module")
def server():
    port = _free_port()
    srv = socketserver.TCPServer(("127.0.0.1", port), _Handler)
    thread = threading.Thread(target=srv.serve_forever, daemon=True)
    thread.start()
    yield f"http://127.0.0.1:{port}"
    srv.shutdown()
    srv.server_close()


@pytest.fixture(autouse=True)
def _reset_state():
    _State.counter = 0
    _State.healthy_after = 0


def test_http_get_parses_json(server):
    status, body = harness_http.http_get(f"{server}/ok-json")
    assert status == 200
    assert body == {"hello": "world"}


def test_http_get_returns_status_on_4xx_with_body(server):
    status, body = harness_http.http_get(f"{server}/notfound")
    assert status == 404
    assert body == {"error": "nope"}


def test_http_get_handles_non_json_body(server):
    status, body = harness_http.http_get(f"{server}/text")
    assert status == 200
    assert body is None  # not JSON => None


def test_http_get_transport_failure_returns_status_zero():
    status, body = harness_http.http_get("http://127.0.0.1:1/nope", timeout=1)
    assert status == 0
    assert body is None


def test_http_get_sends_headers(server):
    status, body = harness_http.http_get(f"{server}/echo-headers", headers={"X-Auth": "token-123"})
    assert status == 200
    assert body["X-Auth"] == "token-123"


def test_http_post_json(server):
    status, body = harness_http.http_post(f"{server}/echo-json", data={"k": "v"})
    assert status == 200
    assert body["content_type"] == "application/json"
    assert body["body"] == {"k": "v"}


def test_http_post_form(server):
    status, body = harness_http.http_post(
        f"{server}/echo-form",
        data={"grant_type": "password", "username": "alice"},
        content_type="application/x-www-form-urlencoded",
    )
    assert status == 200
    assert body["content_type"] == "application/x-www-form-urlencoded"
    assert "grant_type=password" in body["raw"]
    assert "username=alice" in body["raw"]


def test_retry_http_get_converges(server):
    _State.healthy_after = 2  # fail twice, then succeed
    status, body = harness_http.retry_http_get(
        f"{server}/flaky", expect_status=200, max_wait=5, interval=1
    )
    assert status == 200
    assert body == {"ok": True}
    assert _State.counter == 3


def test_retry_http_get_times_out():
    # nonexistent host; expect a RuntimeError after the short max_wait
    with pytest.raises(RuntimeError, match="Did not converge"):
        harness_http.retry_http_get(
            "http://127.0.0.1:1/never",
            expect_status=200,
            max_wait=2,
            interval=1,
            timeout=1,
        )


def test_wait_for_http_returns_status(server):
    status = harness_http.wait_for_http(f"{server}/ok-json", "ok", max_wait=2, interval=1)
    assert status == 200


def test_assert_converges_returns_value():
    n = [0]

    def fn():
        n[0] += 1
        return "done" if n[0] >= 3 else None

    assert harness_http.assert_converges(fn, "fn", max_wait=2, interval=0) == "done"