teleo-infrastructure/tests/test_bootstrap_clone_kb_gate.py

from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any

import pytest

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "scripts"))

import bootstrap_clone_kb_gate as bootstrap  # noqa: E402


def _identity(name: str, identifier: str, *, disposable: bool) -> dict[str, Any]:
    return {
        "id": identifier,
        "name": name,
        "image": "postgres:16",
        "labels": (
            {bootstrap.bound.DISPOSABLE_LABEL_KEY: bootstrap.bound.DISPOSABLE_LABEL_VALUE}
            if disposable
            else {}
        ),
        "network_mode": "none" if disposable else "bridge",
        "ports": {},
        "mount_sources": [f"/{name}-data"],
        "started_at": f"{name}-start",
    }


def test_bootstrap_installs_gate_verifies_logins_and_writes_only_private_secret_files(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    tmp_path.chmod(0o700)
    target = _identity("clone", "clone-id", disposable=True)
    production = _identity(bootstrap.bound.PRODUCTION_CONTAINER, "production-id", disposable=False)
    monkeypatch.setattr(
        bootstrap.bound,
        "container_identity",
        lambda name: dict(production if name == bootstrap.bound.PRODUCTION_CONTAINER else target),
    )
    monkeypatch.setattr(
        bootstrap.bound,
        "validate_disposable_target_identity",
        lambda observed, live: {"all": observed["id"] != live["id"], "network_none": True},
    )
    passwords = iter(("review-test-secret", "apply-test-secret"))
    monkeypatch.setattr(bootstrap.secrets, "token_urlsafe", lambda _length: next(passwords))
    sql_calls: list[dict[str, Any]] = []
    monkeypatch.setattr(
        bootstrap,
        "_psql",
        lambda container, database, sql, **kwargs: sql_calls.append(
            {"container": container, "database": database, "sql": sql, **kwargs}
        ),
    )
    login_calls: list[tuple[str, str, str, str]] = []
    monkeypatch.setattr(
        bootstrap,
        "_verify_login",
        lambda container, database, role, password: login_calls.append((container, database, role, password)),
    )
    manifest = {
        "approval_table": "kb_stage.kb_proposal_approvals",
        "review_principals_table": "kb_stage.kb_review_principals",
        "review_role_exists": True,
        "apply_role_exists": True,
    }
    monkeypatch.setattr(bootstrap.bound, "_psql_json", lambda *_args, **_kwargs: dict(manifest))

    result = bootstrap.bootstrap("clone", "teleo", tmp_path)

    assert result["status"] == "ready"
    assert result["bound_container_id"] == "clone-id"
    assert result["gate_manifest"] == manifest
    assert result["secret_values_or_hashes_recorded"] is False
    assert login_calls == [
        ("clone-id", "teleo", "kb_review", "review-test-secret"),
        ("clone-id", "teleo", "kb_apply", "apply-test-secret"),
    ]
    assert len(sql_calls) == 3
    assert sql_calls[-1]["secret_values"] == ("review-test-secret", "apply-test-secret")
    review_file = tmp_path / "kb-review.env"
    apply_file = tmp_path / "kb-apply.env"
    assert review_file.read_text(encoding="utf-8") == "KB_REVIEW_PASSWORD=review-test-secret\n"
    assert apply_file.read_text(encoding="utf-8") == "KB_APPLY_DB_PASSWORD=apply-test-secret\n"
    assert review_file.stat().st_mode & 0o777 == 0o600
    assert apply_file.stat().st_mode & 0o777 == 0o600
    serialized = json.dumps(result, sort_keys=True)
    assert "review-test-secret" not in serialized
    assert "apply-test-secret" not in serialized


@pytest.mark.parametrize("mode", [0o755, 0o750])
def test_bootstrap_rejects_nonprivate_output_directory(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    mode: int,
) -> None:
    tmp_path.chmod(mode)
    target = _identity("clone", "clone-id", disposable=True)
    production = _identity(bootstrap.bound.PRODUCTION_CONTAINER, "production-id", disposable=False)
    monkeypatch.setattr(
        bootstrap.bound,
        "container_identity",
        lambda name: dict(production if name == bootstrap.bound.PRODUCTION_CONTAINER else target),
    )
    monkeypatch.setattr(bootstrap.bound, "validate_disposable_target_identity", lambda *_args: {"all": True})

    with pytest.raises(bootstrap.bound.CheckpointError, match="private directory"):
        bootstrap.bootstrap("clone", "teleo", tmp_path)


def test_bootstrap_refuses_existing_or_symlinked_secret_paths(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    tmp_path.chmod(0o700)
    target = _identity("clone", "clone-id", disposable=True)
    production = _identity(bootstrap.bound.PRODUCTION_CONTAINER, "production-id", disposable=False)
    monkeypatch.setattr(
        bootstrap.bound,
        "container_identity",
        lambda name: dict(production if name == bootstrap.bound.PRODUCTION_CONTAINER else target),
    )
    monkeypatch.setattr(bootstrap.bound, "validate_disposable_target_identity", lambda *_args: {"all": True})
    (tmp_path / "kb-review.env").symlink_to(tmp_path / "missing-target")

    with pytest.raises(bootstrap.bound.CheckpointError, match="already exist"):
        bootstrap.bootstrap("clone", "teleo", tmp_path)


def test_write_secret_is_exclusive_and_does_not_follow_symlinks(tmp_path: Path) -> None:
    target = tmp_path / "target"
    target.write_text("unchanged\n", encoding="utf-8")
    link = tmp_path / "secret.env"
    link.symlink_to(target)

    with pytest.raises(FileExistsError):
        bootstrap._write_secret(link, "SECRET", "value")
    assert target.read_text(encoding="utf-8") == "unchanged\n"


def test_main_rejects_production_without_calling_bootstrap(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    capsys: pytest.CaptureFixture[str],
) -> None:
    called = False

    def unexpected(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
        nonlocal called
        called = True
        return {}

    monkeypatch.setattr(bootstrap, "bootstrap", unexpected)
    result = bootstrap.main(
        ["--container", bootstrap.bound.PRODUCTION_CONTAINER, "--db", "teleo", "--output-dir", str(tmp_path)]
    )

    assert result == 2
    assert called is False
    output = json.loads(capsys.readouterr().out)
    assert output["status"] == "rejected"
    assert "non-production" in output["error"]