42 lines
1.8 KiB
Python
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
|