nanobot/tests/gateway/test_http_ingress.py

import asyncio
import functools
import random
import socket

import httpx
import pytest

from nanobot.bus.queue import MessageBus
from nanobot.config.schema import WebhookRouteConfig, WebhooksConfig
from nanobot.gateway.http import run_gateway_http_ingress
from nanobot.webhooks import WebhookRouter


def _free_port() -> int:
    for _ in range(100):
        port = random.randint(30_000, 60_000)
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            try:
                sock.bind(("127.0.0.1", port))
            except OSError:
                continue
            return port
    raise RuntimeError("could not find a free localhost port")


async def _request(method: str, url: str, **kwargs) -> httpx.Response:
    return await asyncio.to_thread(
        functools.partial(
            httpx.request,
            method,
            url,
            timeout=5.0,
            trust_env=False,
            **kwargs,
        )
    )


@pytest.mark.asyncio
async def test_gateway_http_ingress_serves_health() -> None:
    port = _free_port()
    task = asyncio.create_task(
        run_gateway_http_ingress(host="127.0.0.1", port=port),
    )
    await asyncio.sleep(0.2)
    try:
        response = await _request("GET", f"http://127.0.0.1:{port}/health")

        assert response.status_code == 200
        assert response.json() == {"status": "ok"}
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


@pytest.mark.asyncio
async def test_gateway_http_ingress_accepts_webhook_and_queues_message() -> None:
    port = _free_port()
    bus = MessageBus()
    router = WebhookRouter(
        WebhooksConfig(
            routes={
                "deploy": WebhookRouteConfig(
                    auth="secret",
                    secret="topsecret",
                    to="telegram:ops",
                    prompt="Deploy {{ event.service }}",
                )
            }
        ),
        bus,
    )
    task = asyncio.create_task(
        run_gateway_http_ingress(host="127.0.0.1", port=port, webhook_router=router),
    )
    await asyncio.sleep(0.2)
    try:
        response = await _request(
            "POST",
            f"http://127.0.0.1:{port}/webhooks/deploy",
            headers={"Authorization": "Bearer topsecret"},
            json={"service": "api"},
        )

        assert response.status_code == 202
        assert response.json()["queued"] is True
        msg = await asyncio.wait_for(bus.consume_inbound(), timeout=1)
        assert msg.channel == "telegram"
        assert msg.chat_id == "ops"
        assert msg.content == "Deploy api"
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


@pytest.mark.asyncio
async def test_gateway_http_ingress_rejects_oversized_webhook_before_queueing() -> None:
    port = _free_port()
    bus = MessageBus()
    router = WebhookRouter(
        WebhooksConfig(
            routes={
                "small": WebhookRouteConfig(
                    auth="none",
                    to="telegram:ops",
                    max_body_bytes=1024,
                )
            }
        ),
        bus,
    )
    task = asyncio.create_task(
        run_gateway_http_ingress(host="127.0.0.1", port=port, webhook_router=router),
    )
    await asyncio.sleep(0.2)
    try:
        reader, writer = await asyncio.open_connection("127.0.0.1", port)
        writer.write(
            b"POST /webhooks/small HTTP/1.1\r\n"
            b"Host: 127.0.0.1\r\n"
            b"Content-Length: 2048\r\n"
            b"\r\n"
        )
        await writer.drain()
        data = await asyncio.wait_for(reader.read(4096), timeout=2)
        writer.close()
        await writer.wait_closed()

        assert data.startswith(b"HTTP/1.0 413 ")
        assert b"Request body too large" in data
        assert bus.inbound_size == 0
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


@pytest.mark.asyncio
async def test_gateway_http_ingress_rejects_chunked_webhook_body() -> None:
    port = _free_port()
    bus = MessageBus()
    router = WebhookRouter(
        WebhooksConfig(
            routes={
                "chunked": WebhookRouteConfig(
                    auth="none",
                    to="telegram:ops",
                )
            }
        ),
        bus,
    )
    task = asyncio.create_task(
        run_gateway_http_ingress(host="127.0.0.1", port=port, webhook_router=router),
    )
    await asyncio.sleep(0.2)
    try:
        reader, writer = await asyncio.open_connection("127.0.0.1", port)
        writer.write(
            b"POST /webhooks/chunked HTTP/1.1\r\n"
            b"Host: 127.0.0.1\r\n"
            b"Transfer-Encoding: chunked\r\n"
            b"\r\n"
            b"2\r\n{}\r\n0\r\n\r\n"
        )
        await writer.drain()
        data = await asyncio.wait_for(reader.read(4096), timeout=2)
        writer.close()
        await writer.wait_closed()

        assert data.startswith(b"HTTP/1.0 501 ")
        assert b"Transfer-Encoding is not supported" in data
        assert bus.inbound_size == 0
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


@pytest.mark.asyncio
async def test_gateway_http_ingress_times_out_partial_request() -> None:
    port = _free_port()
    task = asyncio.create_task(
        run_gateway_http_ingress(host="127.0.0.1", port=port, read_timeout_s=0.1),
    )
    await asyncio.sleep(0.2)
    try:
        reader, writer = await asyncio.open_connection("127.0.0.1", port)
        writer.write(
            b"POST /webhooks/slow HTTP/1.1\r\n"
            b"Host: 127.0.0.1\r\n"
        )
        await writer.drain()
        data = await asyncio.wait_for(reader.read(4096), timeout=2)
        writer.close()
        await writer.wait_closed()

        assert data.startswith(b"HTTP/1.0 408 ")
        assert b"Request timed out" in data
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)