republisher/tests/test_entrypoint.py

import io
import logging
from collections.abc import Awaitable, Callable
from types import SimpleNamespace
from typing import Any, cast

from repub.entrypoint import FeedNameFilter, entrypoint, logger, parse_args


def test_feed_name_filter_accepts_matching_item() -> None:
    item = SimpleNamespace(feed_name="nasa")
    feed_filter = FeedNameFilter({"feed_name": "nasa"})

    assert feed_filter.accepts(item) is True


def test_feed_name_filter_rejects_non_matching_item() -> None:
    item = SimpleNamespace(feed_name="gp-pod")
    feed_filter = FeedNameFilter({"feed_name": "nasa"})

    assert feed_filter.accepts(item) is False


def test_parse_args_uses_republisher_host_and_port_env_vars(monkeypatch) -> None:
    monkeypatch.setenv("REPUBLISHER_HOST", "0.0.0.0")
    monkeypatch.setenv("REPUBLISHER_PORT", "9090")

    command, args = parse_args(["serve"])

    assert command == "serve"
    assert args.host == "0.0.0.0"
    assert args.port == "9090"


def test_parse_args_supports_dev_mode_flag() -> None:
    command, args = parse_args(["serve", "--dev-mode"])

    assert command == "serve"
    assert args.dev_mode is True


def test_parse_args_defaults_to_dev_mode_when_no_args() -> None:
    command, args = parse_args([])

    assert command == "serve"
    assert args.dev_mode is True


def test_entrypoint_rejects_invalid_republisher_port(monkeypatch) -> None:
    monkeypatch.setenv("REPUBLISHER_PORT", "not-a-number")
    stream = io.StringIO()
    handlers = [
        cast(logging.StreamHandler[io.StringIO], handler) for handler in logger.handlers
    ]
    original_streams = [handler.stream for handler in handlers]
    for handler in handlers:
        handler.stream = stream

    try:
        exit_code = entrypoint(["serve"])
    finally:
        for handler, original_stream in zip(handlers, original_streams):
            handler.stream = original_stream

    assert exit_code == 2
    assert "Invalid REPUBLISHER_PORT/--port value" in stream.getvalue()


def test_entrypoint_passes_dev_mode_to_create_app(monkeypatch) -> None:
    recorded: dict[str, object] = {}

    class StubApp:
        def __init__(self) -> None:
            self.extensions: dict[str, object] = {}

    def fake_create_app(*, dev_mode: bool) -> StubApp:
        recorded["dev_mode"] = dev_mode
        return StubApp()

    def fake_install_signal_handlers(stop_event: object) -> None:
        recorded["stop_event"] = stop_event

    async def fake_hypercorn_serve(
        app: StubApp,
        config: Any,
        *,
        shutdown_trigger: Callable[[], Awaitable[None]],
    ) -> None:
        recorded["app"] = app
        recorded["host"] = config.bind[0].split(":")[0]
        recorded["port"] = int(config.bind[0].split(":")[1])
        recorded["shutdown_trigger"] = shutdown_trigger
        shutdown_event = cast(Any, app.extensions["repub.shutdown_event"])
        recorded["app_shutdown_event"] = shutdown_event
        shutdown_event.set()
        await shutdown_trigger()

    monkeypatch.setattr("repub.entrypoint.create_app", fake_create_app)
    monkeypatch.setattr(
        "repub.entrypoint._install_signal_handlers", fake_install_signal_handlers
    )
    monkeypatch.setattr("repub.entrypoint.hypercorn_serve", fake_hypercorn_serve)

    exit_code = entrypoint(
        ["serve", "--dev-mode", "--host", "0.0.0.0", "--port", "9090"]
    )

    assert exit_code == 0
    assert recorded["dev_mode"] is True
    assert recorded["host"] == "0.0.0.0"
    assert recorded["port"] == 9090
    assert recorded["stop_event"] is recorded["app_shutdown_event"]
    assert callable(recorded["shutdown_trigger"])