DataOps-platform/tests/test_p2_wp11_delivery_contract.py

from __future__ import annotations

import hashlib
import json
import os
import subprocess
import tarfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
OPERATIONS = ROOT / "deployment/compose"


def _package(output: Path) -> Path:
    completed = subprocess.run(
        [
            "bash",
            str(OPERATIONS / "package_offline.sh"),
            "--repo-root",
            str(ROOT),
            "--output-dir",
            str(output),
            "--version",
            "p2-wp11-contract",
            "--source-date-epoch",
            "1775251200",
            "--skip-images",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
    )
    assert completed.returncode == 0, completed.stderr
    return output / "dataops-platform-offline-p2-wp11-contract.tar.gz"


def test_package_is_reproducible_and_contains_trace_evidence(tmp_path: Path):
    first = _package(tmp_path / "first")
    second = _package(tmp_path / "second")

    assert hashlib.sha256(first.read_bytes()).digest() == hashlib.sha256(
        second.read_bytes()
    ).digest()

    with tarfile.open(first, "r:gz") as archive:
        names = set(archive.getnames())

    prefix = "dataops-platform-offline-p2-wp11-contract/"
    expected = {
        "environments/development.json",
        "environments/test.json",
        "environments/preproduction.json",
        "operations/product_engineering.py",
        "operations/reproducible_archive.py",
        "helm/dataops-platform/Chart.yaml",
        "helm/dataops-platform/values.schema.json",
        "enterprise/enterprise_delivery.py",
        "trace/OPENAPI.yaml",
        "trace/backend.cdx.json",
        "trace/frontend.cdx.json",
        "release-manifest.json",
        "release-manifest.txt",
        "checksums.sha256",
    }
    assert {prefix + item for item in expected} <= names


def test_environment_files_match_their_profiles():
    for profile_path in sorted((ROOT / "deployment/environments").glob("*.json")):
        profile = json.loads(profile_path.read_text(encoding="utf-8"))
        env_path = profile_path.with_suffix(".env")
        values = dict(
            line.split("=", 1)
            for line in env_path.read_text(encoding="utf-8").splitlines()
            if line and not line.startswith("#")
        )
        assert values["DATAOPS_NETWORK_NAME"] == profile["network_name"]
        assert values["DATAOPS_ENVIRONMENT"] == profile["environment"]
        assert values["DATAOPS_DATA_NAMESPACE"] == profile["data_namespace"]
        for name, value in profile["ports"].items():
            assert values[name] == str(value)


def test_preflight_and_install_accept_environment_profile():
    for script in ("preflight.sh", "install_offline.sh"):
        help_text = subprocess.run(
            ["bash", str(OPERATIONS / script), "--help"],
            cwd=ROOT,
            capture_output=True,
            text=True,
            check=True,
        ).stdout
        assert "--environment-profile" in help_text


def test_install_rejects_image_head_that_differs_from_release_manifest():
    install = (OPERATIONS / "install_offline.sh").read_text(encoding="utf-8")
    preflight = (OPERATIONS / "preflight.sh").read_text(encoding="utf-8")
    common = (OPERATIONS / "wp13-common.sh").read_text(encoding="utf-8")

    assert "wp13_expected_migration_head" in common
    assert "installed image migration head does not match release manifest" in install
    assert "image migration head does not match release manifest" in preflight


def test_preflight_fails_closed_when_running_image_head_is_stale(tmp_path: Path):
    bundle = tmp_path / "bundle"
    fake_bin = tmp_path / "bin"
    bundle.mkdir()
    fake_bin.mkdir()
    compose = bundle / "docker-compose.yml"
    compose.write_text("services: {}\n", encoding="utf-8")
    (bundle / "release-manifest.json").write_text(
        json.dumps({"migration": {"head": "20260802_460"}}),
        encoding="utf-8",
    )
    fake_docker = fake_bin / "docker"
    fake_docker.write_text(
        """#!/usr/bin/env bash
set -euo pipefail
arguments="$*"
if [[ "${arguments}" == "ps -q --filter label=com.docker.compose.project=stale-project" ]]; then
  printf 'container-id\\n'
elif [[ "${arguments}" == *" config --format json" ]]; then
  printf '{"services":{"backend":{"image":"fake-backend"}}}\\n'
elif [[ "${arguments}" == "run --rm --network none --read-only --entrypoint alembic fake-backend -c alembic.ini heads" ]]; then
  printf '20260730_360 (head)\\n'
elif [[ "${arguments}" == *" alembic -c alembic.ini current" ]]; then
  printf '20260730_360 (head)\\n'
elif [[ "${arguments}" == *" alembic -c alembic.ini heads" ]]; then
  printf '20260730_360 (head)\\n'
fi
""",
        encoding="utf-8",
    )
    fake_docker.chmod(0o755)
    env = os.environ.copy()
    env["PATH"] = f"{fake_bin}:/usr/bin:/bin"

    completed = subprocess.run(
        [
            "bash",
            str(OPERATIONS / "preflight.sh"),
            "--bundle-root",
            str(bundle),
            "--compose-file",
            str(compose),
            "--project-name",
            "stale-project",
            "--skip-checksums",
        ],
        cwd=ROOT,
        env=env,
        capture_output=True,
        text=True,
    )

    assert completed.returncode != 0
    assert "image migration head does not match release manifest" in (
        completed.stdout + completed.stderr
    )


def test_product_engineering_cli_is_read_only_by_default():
    for arguments in (
        ["validate-environments", "--profiles-dir", "deployment/environments"],
        ["manifest", "--repo-root", ".", "--version", "contract", "--source-date-epoch", "1775251200"],
        ["sbom", "--repo-root", ".", "--output-dir", "-"],
    ):
        completed = subprocess.run(
            [str(ROOT / ".venv/bin/python"), str(OPERATIONS / "product_engineering.py"), *arguments],
            cwd=ROOT,
            capture_output=True,
            text=True,
        )
        assert completed.returncode == 0, completed.stderr


def test_rollback_remains_restore_based_and_never_downgrades_database():
    source = (OPERATIONS / "rollback.sh").read_text(encoding="utf-8")
    assert "restore.sh" in source
    assert "alembic downgrade" not in source