stack/tests/test_compose_mounts.py

"""Tests for compose.yml — verify all bind mount sources exist on disk.

refs #250
"""

from __future__ import annotations

from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[1]
COMPOSE = ROOT / "compose.yml"


def _load_compose() -> dict:
    return yaml.safe_load(COMPOSE.read_text())


def _extract_bind_mounts(compose: dict) -> list[tuple[str, str, str]]:
    """Return (service, host_path, full_mount) for every bind mount."""
    mounts: list[tuple[str, str, str]] = []
    for svc, cfg in compose.get("services", {}).items():
        for vol in cfg.get("volumes", []):
            if isinstance(vol, str):
                host = vol.split(":")[0]
            elif isinstance(vol, dict):
                host = vol.get("source", "")
            else:
                continue
            # Only check relative paths (./...) — skip named volumes,
            # env-var-only paths, and absolute system paths
            if host.startswith("./"):
                mounts.append((svc, host, vol if isinstance(vol, str) else str(vol)))
    return mounts


class TestBindMountSourcesExist:
    """Every relative bind mount in compose.yml must point to an existing path."""

    def test_all_relative_bind_mounts_exist(self):
        compose = _load_compose()
        mounts = _extract_bind_mounts(compose)
        assert mounts, "Expected to find bind mounts in compose.yml"

        missing = []
        for svc, host_path, full in mounts:
            # ./data/ is gitignored runtime state — populated by service
            # bootstrap on first run, not present in a fresh checkout (CI).
            if host_path.startswith("./data/"):
                continue
            resolved = ROOT / host_path
            if not resolved.exists():
                missing.append(f"  {svc}: {host_path}")

        assert missing == [], "Bind mount sources not found on disk:\n" + "\n".join(
            missing
        )

    def test_compose_parses_cleanly(self):
        """compose.yml must be valid YAML with services defined."""
        compose = _load_compose()
        assert "services" in compose
        assert len(compose["services"]) >= 10


class TestNoHardcodedPaths:
    """Bind mounts should not contain user-specific absolute paths as sources."""

    def test_no_absolute_home_paths_in_sources(self):
        """Mount sources should use ./ or env vars, not /home/user/..."""
        compose = _load_compose()
        for svc, cfg in compose.get("services", {}).items():
            for vol in cfg.get("volumes", []):
                if isinstance(vol, str):
                    host = vol.split(":")[0]
                else:
                    continue
                # Absolute /home paths as source are fragile
                # Allow /home/kert/.local/share/docker (promtail needs it)
                if host.startswith("/home/") and "docker/containers" not in host:
                    raise AssertionError(
                        f"{svc}: hardcoded absolute path as mount source: {host}"
                    )


class TestReadOnlyMounts:
    """Config files and assets should be mounted read-only."""

    def test_infra_config_mounts_are_ro(self):
        """infra/ mounts should be :ro."""
        compose = _load_compose()
        not_ro = []
        for svc, cfg in compose.get("services", {}).items():
            for vol in cfg.get("volumes", []):
                if not isinstance(vol, str):
                    continue
                host = vol.split(":")[0]
                # infra/ config files should be read-only
                if host.startswith("./infra/") and host.endswith(
                    (".yml", ".yaml", ".conf", ".toml", ".xml", ".css", ".js")
                ):
                    if not vol.endswith(":ro"):
                        not_ro.append(f"  {svc}: {vol}")

        assert not_ro == [], "infra/ config mounts should be :ro:\n" + "\n".join(not_ro)

    def test_asset_mounts_are_ro(self):
        """assets/ mounts should be :ro."""
        compose = _load_compose()
        not_ro = []
        for svc, cfg in compose.get("services", {}).items():
            for vol in cfg.get("volumes", []):
                if not isinstance(vol, str):
                    continue
                host = vol.split(":")[0]
                if host.startswith("./assets/") and not vol.endswith(":ro"):
                    not_ro.append(f"  {svc}: {vol}")

        assert not_ro == [], "assets/ mounts should be :ro:\n" + "\n".join(not_ro)


class TestSecurityOpts:
    """Services should have security_opt configured."""

    def test_services_have_no_new_privileges(self):
        """Most services should have no-new-privileges."""
        # Exceptions: privileged, ephemeral, or upstream images without secopt
        exempt = {
            "wire",  # ephemeral bootstrap, profile=tools
            "zotero",  # GPU + display server needs
            "coredns",  # upstream, no secopt in image
            "gitea",  # rootless image handles its own security
            "notebooks",  # GPU + dev environment
            "webdav",  # rclone upstream
            "docs",  # static site
            "promtail",  # needs host log access
            "cloudflared",  # tunnel agent
            "nvidia-exporter",  # upstream dcgm-exporter, GPU runtime + SYS_ADMIN
        }
        compose = _load_compose()
        missing = []
        for svc, cfg in compose.get("services", {}).items():
            if svc in exempt:
                continue
            sec = cfg.get("security_opt", [])
            if "no-new-privileges:true" not in sec:
                missing.append(svc)

        assert missing == [], (
            f"Services missing no-new-privileges: {', '.join(missing)}"
        )


class TestNetworkAssignment:
    """Every service should be assigned to at least one network."""

    def test_all_services_have_networks(self):
        # Exceptions: ephemeral/profile-only services
        exempt = {"wire"}
        compose = _load_compose()
        missing = []
        for svc, cfg in compose.get("services", {}).items():
            if svc in exempt:
                continue
            if "networks" not in cfg:
                missing.append(svc)

        assert missing == [], (
            f"Services without network assignment: {', '.join(missing)}"
        )