atlas-iac/testing/tests/test_av_observer_health.py

42 lines
1.8 KiB
Python

"""Exercise readiness over HTTP without exposing receipt operations or content."""
import importlib.util
from pathlib import Path
import threading
import urllib.error
import urllib.request
import pytest
def test_health_is_read_only_and_does_not_open_other_get_paths(monkeypatch, capfd):
"""A real listener answers health, keeps API GET closed, and emits no logs."""
source = (Path(__file__).resolve().parents[2] / "services/bstein-dev-home"
/ "av-observer/report-service/server.py")
monkeypatch.syspath_prepend(str(source.parent))
spec = importlib.util.spec_from_file_location("observer_health_server", source)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# An object without receipt methods fails if health ever accesses that state.
server = module.Server(("127.0.0.1", 0), "https://example.invalid", object())
worker = threading.Thread(target=server.serve_forever, daemon=True)
worker.start()
client = urllib.request.build_opener(urllib.request.ProxyHandler({}))
root = f"http://127.0.0.1:{server.server_port}"
try:
with client.open(root + "/healthz", timeout=2) as response:
assert response.status == 204
assert response.read() == b""
with pytest.raises(urllib.error.HTTPError) as error:
client.open(root + "/av-test/api/start", timeout=2)
assert error.value.code == 405
with pytest.raises(urllib.error.HTTPError) as error:
client.open(urllib.request.Request(root + "/av-test/api/start", data=b"{}"), timeout=2)
assert error.value.code == 403
finally:
server.shutdown()
server.server_close()
worker.join(timeout=2)
captured = capfd.readouterr()
assert not captured.out and not captured.err