mdmem/tests/test_deploy.py

"""Smoke tests for the committed Kubernetes overlay (PLAN §13a).

The overlay is not directly buildable: `secret.yaml` is SOPS-encrypted and kept out
of the repo. The image digest and the PocketID client id are filled in, but both are
release-time values — the digest is bumped by hand on every build (§13a.2) — so their
placeholder spellings stay on the allowlist below rather than being treated as a
defect if they reappear.

Nothing else checked that the overlay stayed coherent: a renamed environment
variable, a probe pointing at a port the container does not expose, or a resource
dropped from the kustomization would only surface on a cluster.
"""

from pathlib import Path

import pytest
import yaml

APP = Path("deploy/kubernetes/apps/apps/mdmem/mdmem/app")
DEPLOYMENT = APP / "deployment.yaml"

# Documented, intentional gaps: the overlay cannot render as committed.
ABSENT_BY_DESIGN = {"./secret.yaml"}
RELEASE_PLACEHOLDERS = ("REPLACE_WITH_DIGEST", "REPLACE-WITH-THE-POCKETID-CLIENT-UUID")


def _load(path: Path) -> dict:
    """The overlay has multi-document files; take the first real document."""
    docs = [d for d in yaml.safe_load_all(path.read_text(encoding="utf-8")) if d]
    return docs[0]


def _load_all(path: Path) -> list[dict]:
    return [d for d in yaml.safe_load_all(path.read_text(encoding="utf-8")) if d]


@pytest.fixture(scope="module")
def deployment() -> dict:
    return _load(DEPLOYMENT)


@pytest.fixture(scope="module")
def container(deployment) -> dict:
    return deployment["spec"]["template"]["spec"]["containers"][0]


@pytest.fixture(scope="module")
def kustomization() -> dict:
    return _load(APP / "kustomization.yaml")


@pytest.fixture(scope="module")
def env_literals(kustomization) -> dict[str, str]:
    generator = next(g for g in kustomization["configMapGenerator"] if g["name"] == "mdmem-env")
    out = {}
    for literal in generator["literals"]:
        key, _, value = literal.partition("=")
        out[key] = value
    return out


def test_every_manifest_in_the_overlay_parses():
    for path in sorted(APP.glob("*.yaml")):
        docs = _load_all(path)
        assert docs, f"{path} is empty or unparseable"
        for doc in docs:
            assert doc.get("kind") or "kustomize" in str(doc.get("apiVersion", "")), path


def test_kustomization_resources_exist_apart_from_the_sops_secret(kustomization):
    for resource in kustomization["resources"]:
        if resource in ABSENT_BY_DESIGN:
            # ...but its template must be committed so the gap is fixable
            assert (APP / "secret.example.yaml").exists()
            continue
        assert (APP / resource).exists(), f"{resource} is referenced but missing"


def test_the_only_unrendered_placeholders_are_the_documented_release_values():
    """If a new placeholder creeps in, it must be declared here, not discovered
    during a deploy."""
    found = set()
    for path in sorted(APP.glob("*.yaml")):
        text = path.read_text(encoding="utf-8")
        for token in ("REPLACE_WITH", "REPLACE-WITH", "CHANGEME", "TODO:"):
            if token in text:
                found.update(w for w in text.split() if token in w)
    unexpected = {f for f in found if not any(p in f for p in RELEASE_PLACEHOLDERS)}
    assert not unexpected, f"undocumented placeholders in the overlay: {sorted(unexpected)}"


def test_configmap_env_keys_are_all_read_by_the_application(env_literals):
    """Catches a variable renamed in config.py but not in the overlay (or vice versa):
    a silently ignored EMBED_MODEL would serve lexical-only in production."""
    src = "\n".join(p.read_text(encoding="utf-8") for p in Path("src/mdmem").rglob("*.py"))
    unread = [key for key in env_literals if f'"{key}"' not in src]
    assert not unread, f"the overlay sets variables the app never reads: {unread}"


def test_secret_supplied_variables_are_not_duplicated_in_the_configmap(env_literals):
    """MCP_TOKEN and friends come from Vault via the ExternalSecret; a plaintext
    copy in the ConfigMap would both leak and win or lose unpredictably."""
    secret_keys = {"MCP_TOKEN", "BIFROST_VK", "OIDC_CLIENT_SECRET", "SESSION_SECRET"}
    assert not (secret_keys & set(env_literals))


def test_embed_model_carries_its_provider_prefix(env_literals):
    """A bare alias 404s at Bifrost and looks like an mdmem bug (§7, §14.14)."""
    assert "/" in env_literals["EMBED_MODEL"]


def test_listener_addresses_match_the_declared_container_ports(env_literals, container):
    ports = {p["name"]: p["containerPort"] for p in container["ports"]}
    assert env_literals["MCP_ADDR"].endswith(f":{ports['mcp']}")
    assert env_literals["UI_ADDR"].endswith(f":{ports['ui']}")
    # the MCP SDK checks Host, so every allowed host must name the MCP port
    for host in env_literals["MCP_ALLOWED_HOSTS"].split(","):
        assert host.strip().endswith(f":{ports['mcp']}"), host


def test_probes_point_at_ports_the_container_actually_exposes(container):
    names = {p["name"] for p in container["ports"]}
    for probe in ("livenessProbe", "readinessProbe", "startupProbe"):
        assert container[probe]["httpGet"]["port"] in names


def test_mdmem_deployment_has_startup_probe_budget(container):
    """§14.17: the pod must be allowed to come up in lexical-only mode while the
    embedding backend is unreachable, instead of being restarted mid-startup."""
    probe = container["startupProbe"]
    assert probe["httpGet"] == {"path": "/healthz", "port": "ui"}
    assert probe["periodSeconds"] * probe["failureThreshold"] >= 180


def test_liveness_is_dumb_and_readiness_never_depends_on_bifrost(container):
    assert container["livenessProbe"]["httpGet"]["path"] == "/healthz"
    assert container["readinessProbe"]["httpGet"]["path"] == "/readyz"


def test_the_pod_stays_locked_down(deployment, container):
    pod_sc = deployment["spec"]["template"]["spec"]["securityContext"]
    assert pod_sc["runAsNonRoot"] is True
    sc = container["securityContext"]
    assert sc["readOnlyRootFilesystem"] is True
    assert sc["allowPrivilegeEscalation"] is False
    assert sc["capabilities"]["drop"] == ["ALL"]


def test_read_only_root_filesystem_has_somewhere_to_write(deployment, container):
    """readOnlyRootFilesystem only holds if HOME/XDG_CACHE_HOME point at a mounted
    writable path."""
    mounts = {m["name"]: m["mountPath"] for m in container["volumeMounts"]}
    env = {e["name"]: e["value"] for e in container["env"]}
    volumes = {v["name"] for v in deployment["spec"]["template"]["spec"]["volumes"]}
    assert mounts["tmp"] == "/tmp" and "tmp" in volumes
    assert env["HOME"] == "/tmp" and env["XDG_CACHE_HOME"] == "/tmp"
    assert mounts["data"] == "/data" and "data" in volumes


def test_every_persistent_path_lives_on_the_data_volume(env_literals, container):
    mounts = {m["name"]: m["mountPath"] for m in container["volumeMounts"]}
    for key in ("VAULT_DIR", "INDEX_DB", "CACHE_DB", "HISTORY_DIR", "AUDIT_LOG"):
        assert env_literals[key].startswith(mounts["data"] + "/"), key


def test_sqlite_requires_a_single_writer(deployment):
    assert deployment["spec"]["replicas"] == 1
    assert deployment["spec"]["strategy"]["type"] == "Recreate"


def test_the_servicemonitor_scrapes_the_ui_port_where_metrics_live():
    monitor = _load(APP / "servicemonitor.yaml")
    service = _load(APP / "service.yaml")
    endpoint = monitor["spec"]["endpoints"][0]
    assert endpoint["path"] == "/metrics"
    assert endpoint["port"] in {p["name"] for p in service["spec"]["ports"]}
    assert monitor["spec"]["selector"]["matchLabels"] == service["metadata"]["labels"]


def test_the_service_targets_the_containers_named_ports(container):
    service = _load(APP / "service.yaml")
    names = {p["name"] for p in container["ports"]}
    for port in service["spec"]["ports"]:
        assert port["targetPort"] in names