auto-reverse/tests/test_proxy.py

from types import SimpleNamespace

import pytest

from auto_reverse.proxy import ProxyServer, flow_from_mitm
from auto_reverse.store import FlowStore, ScopeFilter


def _fake_mitm_flow():
    request = SimpleNamespace(
        method="POST", pretty_host="ex.com", path="/api/users?role=admin",
        headers={"content-type": "application/json"}, content=b'{"name": "Ada"}',
        query=SimpleNamespace(fields=[("role", "admin")]),
    )
    response = SimpleNamespace(
        status_code=201, headers={"content-type": "application/json"},
        content=b'{"id": 1}',
    )
    return SimpleNamespace(request=request, response=response, timestamp_start=1.5)


def test_flow_from_mitm_maps_fields():
    captured = flow_from_mitm(_fake_mitm_flow())
    assert captured.method == "POST"
    assert captured.host == "ex.com"
    assert captured.path == "/api/users"
    assert captured.query == {"role": ["admin"]}
    assert captured.status == 201
    assert captured.request_json() == {"name": "Ada"}
    assert captured.response_json() == {"id": 1}


def test_flow_from_mitm_handles_missing_response():
    flow = _fake_mitm_flow()
    flow.response = None
    captured = flow_from_mitm(flow)
    assert captured is None


def test_proxy_server_binds_and_reports_real_port(tmp_path):
    store = FlowStore(ScopeFilter(target_hosts={"ex.com"}))
    server = ProxyServer(store, archive_path=tmp_path / "archive.log", port=0)
    try:
        server.start()
    except Exception as exc:
        pytest.skip(f"proxy bind unavailable: {exc}")
    try:
        assert server.port != 0  # real OS-assigned port after binding
    finally:
        server.stop()