51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
import io
|
|
import logging
|
|
from types import SimpleNamespace
|
|
from typing import 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_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()
|