hermes: restore legacy chat/triage hosts alongside the renamed ones #38
153
testing/tests/test_hermes_public_host_continuity.py
Normal file
153
testing/tests/test_hermes_public_host_continuity.py
Normal file
@ -0,0 +1,153 @@
|
||||
"""Public chat/triage hostnames stay served on every layer during a rename.
|
||||
|
||||
PR #34 renamed the chat and triage public hosts in place: the legacy names left
|
||||
the certificate SANs, the Ingress rules, the CoreDNS overrides and the Keycloak
|
||||
ensure script in one change. The legacy hosts began answering 404 with Traefik's
|
||||
default certificate while the renamed hosts could not finish a login, because
|
||||
Keycloak matches ``redirect_uri`` exactly and still held the old callback. Chat
|
||||
and triage were unreachable on every hostname at once.
|
||||
|
||||
These tests pin the contract that makes that outage impossible to reintroduce
|
||||
silently: a public host is either served by all four layers or by none of them.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
REPO = Path(__file__).parents[2]
|
||||
HERMES = REPO / "services/hermes"
|
||||
KEYCLOAK = REPO / "services/keycloak"
|
||||
COREDNS = REPO / "infrastructure/core/coredns-custom.yaml"
|
||||
ENSURE_SCRIPT = KEYCLOAK / "scripts/hermes_access_oidc_ensure.sh"
|
||||
|
||||
# Every hostname the chat/triage surfaces must answer on, with the oauth2-proxy
|
||||
# backend that serves it. Both the renamed and the legacy names belong here
|
||||
# until a deliberate retirement change removes a row from this table.
|
||||
PUBLIC_HOSTS = {
|
||||
"chat.bstein.dev": "oauth2-proxy-hermes-chat",
|
||||
"chat.hermes.bstein.dev": "oauth2-proxy-hermes-chat",
|
||||
"triage.bstein.dev": "oauth2-proxy-hermes-triage",
|
||||
"triage.hermes.bstein.dev": "oauth2-proxy-hermes-triage",
|
||||
}
|
||||
|
||||
# The agent surface was deliberately untouched by the rename.
|
||||
AGENT_HOST = "agent.hermes.bstein.dev"
|
||||
|
||||
|
||||
def _docs(path: Path) -> list[dict]:
|
||||
return [doc for doc in yaml.safe_load_all(path.read_text()) if doc]
|
||||
|
||||
|
||||
def _named(path: Path, kind: str, name: str) -> dict:
|
||||
for doc in _docs(path):
|
||||
if doc.get("kind") == kind and doc["metadata"]["name"] == name:
|
||||
return doc
|
||||
raise AssertionError(f"{kind}/{name} missing from {path}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def certificate() -> dict:
|
||||
return _named(HERMES / "agent-certificate.yaml", "Certificate", "hermes-sites-tls")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def sites_ingress() -> dict:
|
||||
return _named(HERMES / "agent-ingress.yaml", "Ingress", "hermes-sites")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def coredns_hosts() -> set[str]:
|
||||
block = yaml.safe_load(COREDNS.read_text())["data"]["bstein-dev.server"]
|
||||
return {
|
||||
line.split()[1]
|
||||
for line in block.splitlines()
|
||||
if len(line.split()) == 2 and re.fullmatch(r"[\d.]+", line.split()[0])
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def ensure_script() -> str:
|
||||
return ENSURE_SCRIPT.read_text()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("host", sorted(PUBLIC_HOSTS))
|
||||
def test_host_is_on_the_shared_certificate(host: str, certificate: dict):
|
||||
"""A host without a SAN serves Traefik's default self-signed certificate."""
|
||||
assert host in certificate["spec"]["dnsNames"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("host", sorted(PUBLIC_HOSTS))
|
||||
def test_host_resolves_inside_the_cluster(host: str, coredns_hosts: set[str]):
|
||||
assert host in coredns_hosts
|
||||
|
||||
|
||||
@pytest.mark.parametrize("host", sorted(PUBLIC_HOSTS))
|
||||
def test_host_has_an_ingress_rule_and_tls_entry(host: str, sites_ingress: dict):
|
||||
"""A host without a rule answers 404 even though DNS and TLS look healthy."""
|
||||
spec = sites_ingress["spec"]
|
||||
assert host in {name for entry in spec["tls"] for name in entry["hosts"]}
|
||||
|
||||
rule = next((item for item in spec["rules"] if item["host"] == host), None)
|
||||
assert rule is not None, f"no hermes-sites rule serves {host}"
|
||||
|
||||
backends = {
|
||||
path["backend"]["service"]["name"] for path in rule["http"]["paths"]
|
||||
}
|
||||
assert backends == {PUBLIC_HOSTS[host]}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("host", sorted(PUBLIC_HOSTS))
|
||||
def test_host_is_registered_with_keycloak(host: str, ensure_script: str):
|
||||
"""Keycloak matches redirect_uri exactly, so every served host needs one."""
|
||||
assert f"https://{host}" in ensure_script
|
||||
|
||||
|
||||
def test_agent_surface_was_not_touched_by_the_rename(
|
||||
certificate: dict, coredns_hosts: set[str], ensure_script: str
|
||||
):
|
||||
assert AGENT_HOST in certificate["spec"]["dnsNames"]
|
||||
assert AGENT_HOST in coredns_hosts
|
||||
assert f"https://{AGENT_HOST}" in ensure_script
|
||||
|
||||
|
||||
def test_ensure_script_registers_legacy_and_renamed_origins_together(
|
||||
ensure_script: str,
|
||||
):
|
||||
"""The renamed proxies must carry both origins; the agent proxy only one."""
|
||||
for client, canonical, legacy in (
|
||||
("hermes-chat-proxy", "chat.bstein.dev", "chat.hermes.bstein.dev"),
|
||||
("hermes-triage-proxy", "triage.bstein.dev", "triage.hermes.bstein.dev"),
|
||||
):
|
||||
call = re.search(
|
||||
rf'ensure_proxy_client "{client}".*?(?=\nensure_)',
|
||||
ensure_script,
|
||||
re.DOTALL,
|
||||
)
|
||||
assert call, f"{client} is never ensured"
|
||||
assert f"https://{canonical}" in call.group(0)
|
||||
assert f"https://{legacy}" in call.group(0)
|
||||
|
||||
|
||||
def test_ensure_job_is_rerun_whenever_the_script_changes():
|
||||
"""The Job is immutable, so a stale name silently skips the rerun."""
|
||||
job = _named(
|
||||
KEYCLOAK / "bootstrap-jobs/hermes-access-oidc-client-job.yaml",
|
||||
"Job",
|
||||
# The suffix moves with every rerun; resolve it from the manifest.
|
||||
_job_name(),
|
||||
)
|
||||
assert job["spec"]["template"]["spec"]["containers"][0]["command"] == [
|
||||
"/scripts/hermes_access_oidc_ensure.sh"
|
||||
]
|
||||
|
||||
|
||||
def _job_name() -> str:
|
||||
path = KEYCLOAK / "bootstrap-jobs/hermes-access-oidc-client-job.yaml"
|
||||
name = yaml.safe_load(path.read_text())["metadata"]["name"]
|
||||
assert re.fullmatch(r"hermes-access-oidc-client-ensure-\d+", name), name
|
||||
return name
|
||||
Loading…
x
Reference in New Issue
Block a user