prole/tests/onboarding/conftest.py

"""
tests/onboarding/conftest.py — Phase A fixture for the onboarding TDD suite.

Provides two fixtures:

  k3d_cluster (session-scoped, integration marker)
    Ensures a k3d cluster is available for the Tier 1 / Tier 2 cluster-dependent
    tests.  Behaviour:
      - If KNOE_TEST_CLUSTER env var is set, reuses that kubeconfig context
        (shared-persistent mode for local dev — zero spinup cost).
      - Otherwise, creates a fresh k3d cluster named knoe-test-<uuid8>, yields
        its kubeconfig path, and tears it down on session end (per-run mode for
        CI — deterministic, parallel-safe).

  external_psql_client (function-scoped, integration marker)
    Returns a PsqlClient whose .connect() method simulates either an external
    client (source="external") or an in-cluster client (source="internal").
    The distinction maps directly onto the pg_hba source-IP axis in T1.1.

The one green Tier 1 test in this file (test_hostnossl_reject_precedes_permissive)
is a static analysis test (T1.4) — it requires no live cluster and is always
collected.  It validates that the pg_hba block in deploy/gcp/gke/knoe-db.yaml
has the hostnossl reject rule before any permissive host rule that could match
external IPs, preserving the lesson from the externalTrafficPolicy: Cluster→Local
bug in Phase 1.
"""
from __future__ import annotations

import ipaddress
import os
import subprocess
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Generator

import pytest
import yaml

REPO_ROOT = Path(__file__).resolve().parents[2]
GKE_KNOE_DB_YAML = REPO_ROOT / "deploy" / "gcp" / "gke" / "knoe-db.yaml"

# ---------------------------------------------------------------------------
# k3d_cluster fixture
# ---------------------------------------------------------------------------

@pytest.fixture(scope="session")
def k3d_cluster() -> Generator[str, None, None]:
    """Yield a kubeconfig context name for the test k3d cluster.

    Reuses KNOE_TEST_CLUSTER if set (local dev); otherwise creates and tears
    down an ephemeral cluster (CI).  Marked integration so it is skipped when
    running the fast unit-style subset.
    """
    existing = os.environ.get("KNOE_TEST_CLUSTER", "").strip()
    if existing:
        yield existing
        return

    cluster_name = f"knoe-test-{uuid.uuid4().hex[:8]}"
    subprocess.run(
        ["k3d", "cluster", "create", cluster_name, "--wait"],
        check=True,
        timeout=120,
    )
    try:
        yield f"k3d-{cluster_name}"
    finally:
        subprocess.run(
            ["k3d", "cluster", "delete", cluster_name],
            check=False,
            timeout=60,
        )


# ---------------------------------------------------------------------------
# PsqlClient + external_psql_client fixture
# ---------------------------------------------------------------------------

@dataclass
class PsqlClient:
    """Thin wrapper that attempts a psycopg2 connection and returns the result.

    .connect() returns None on success or raises psycopg2.OperationalError on
    failure, so callers can assert the exact error message.
    """

    host_external: str   # host/IP reachable from the "external" vantage point
    host_internal: str   # host/IP reachable from the "internal" vantage point
    port: int = 5432
    dbname: str = "postgres"
    sslrootcert: str = str(REPO_ROOT / "etc" / "knoe-db-ca.crt")

    def connect(
        self,
        role: str,
        password: str,
        *,
        sslmode: str = "require",
        source: str = "external",
    ) -> None:
        """Attempt a connection.  Returns None on success; raises on failure.

        Args:
            role:     PostgreSQL role name.
            password: Plaintext password for SCRAM-SHA-256.
            sslmode:  "require" or "disable".
            source:   "external" (uses host_external) or "internal" (uses
                      host_internal).  Maps to the pg_hba source-IP axis.
        """
        try:
            import psycopg2  # type: ignore[import]
        except ImportError as exc:
            raise RuntimeError(
                "psycopg2 is required for cluster-dependent tests. "
                "Install it with: pip install psycopg2-binary"
            ) from exc

        host = self.host_external if source == "external" else self.host_internal
        conn_kwargs: dict = dict(
            host=host,
            port=self.port,
            dbname=self.dbname,
            user=role,
            password=password,
            connect_timeout=5,
            sslmode=sslmode,
        )
        if sslmode != "disable" and Path(self.sslrootcert).exists():
            conn_kwargs["sslrootcert"] = self.sslrootcert

        conn = psycopg2.connect(**conn_kwargs)
        conn.close()


@pytest.fixture()
def external_psql_client(k3d_cluster: str) -> PsqlClient:
    """Return a PsqlClient wired to the test cluster.

    host_external is the LoadBalancer / NodePort address that pg_hba sees as
    coming from outside RFC1918 space.  host_internal is the pod IP for
    simulating in-cluster supabase service connections.

    In the k3d fixture the "external" address is the NodePort on 127.0.0.1
    (k3d maps host ports); the NetworkPolicy in the external-client namespace
    ensures the source IP seen by pg_hba is outside 10.0.0.0/8.

    Concrete values are resolved at fixture time from the running cluster.
    """
    # Resolve the NodePort for the CNPG primary service.
    result = subprocess.run(
        [
            "kubectl",
            "--context", k3d_cluster,
            "-n", "knoe-db-test",
            "get", "svc", "knoe-db-rw",
            "-o", "jsonpath={.spec.ports[0].nodePort}",
        ],
        capture_output=True,
        text=True,
        timeout=15,
    )
    node_port = int(result.stdout.strip()) if result.stdout.strip() else 5432

    # Pod IP for internal connections.
    pod_result = subprocess.run(
        [
            "kubectl",
            "--context", k3d_cluster,
            "-n", "knoe-db-test",
            "get", "pod",
            "-l", "cnpg.io/instanceRole=primary",
            "-o", "jsonpath={.items[0].status.podIP}",
        ],
        capture_output=True,
        text=True,
        timeout=15,
    )
    pod_ip = pod_result.stdout.strip() or "127.0.0.1"

    return PsqlClient(
        host_external="127.0.0.1",
        host_internal=pod_ip,
        port=node_port,
    )


# ---------------------------------------------------------------------------
# T1.4 — hostnossl reject precedence (static analysis, no cluster needed)
# ---------------------------------------------------------------------------

def _parse_pg_hba_from_yaml(yaml_path: Path) -> list[str]:
    """Extract the pg_hba list from the first CNPG Cluster document in the file."""
    with yaml_path.open() as fh:
        docs = list(yaml.safe_load_all(fh))
    for doc in docs:
        if not isinstance(doc, dict):
            continue
        if doc.get("kind") == "Cluster":
            pg_hba = (
                doc.get("spec", {})
                .get("postgresql", {})
                .get("pg_hba", [])
            )
            if pg_hba:
                return [line for line in pg_hba if isinstance(line, str)]
    return []


def _is_external_range(cidr: str) -> bool:
    """Return True if the CIDR includes addresses outside RFC1918 / loopback."""
    try:
        net = ipaddress.ip_network(cidr, strict=False)
    except ValueError:
        return False
    rfc1918 = [
        ipaddress.ip_network("10.0.0.0/8"),
        ipaddress.ip_network("172.16.0.0/12"),
        ipaddress.ip_network("192.168.0.0/16"),
        ipaddress.ip_network("127.0.0.0/8"),
    ]
    # If the network is entirely within RFC1918/loopback it is internal-only.
    return not any(net.subnet_of(private) for private in rfc1918)