DataOps-platform/tests/test_wp13_delivery_contract.py

from __future__ import annotations

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

import pytest

ROOT = Path(__file__).resolve().parents[1]
COMPOSE = ROOT / "deploy/docker/docker-compose.yml"
OPERATIONS = ROOT / "deployment/compose"


def _compose_config() -> dict:
    completed = subprocess.run(
        [
            "docker",
            "compose",
            "-f",
            str(COMPOSE),
            "config",
            "--format",
            "json",
        ],
        cwd=ROOT,
        check=True,
        capture_output=True,
        text=True,
    )
    return json.loads(completed.stdout)


def _tree_hashes(root: Path) -> dict[str, str]:
    hashes: dict[str, str] = {}
    for path in sorted(root.rglob("*")):
        if not path.is_file():
            continue
        relative = path.relative_to(root)
        if "__pycache__" in relative.parts or path.suffix in {".pyc", ".pyo"}:
            continue
        hashes[relative.as_posix()] = hashlib.sha256(path.read_bytes()).hexdigest()
    return hashes


def test_release_copy_matches_all_sources_of_truth():
    for source, release in (
        (ROOT / "app", ROOT / "deployment/app"),
        (ROOT / "database", ROOT / "deployment/database"),
        (ROOT / "migrations", ROOT / "deployment/migrations"),
    ):
        assert _tree_hashes(release) == _tree_hashes(source)

    for filename in ("requirements.txt", "gunicorn_config.py", "__init__.py", "alembic.ini"):
        assert (ROOT / f"deployment/{filename}").read_bytes() == (
            ROOT / filename
        ).read_bytes()


def test_locally_built_services_have_stable_offline_image_names():
    services = _compose_config()["services"]
    expected = {
        "backend": "dataops-platform-backend:local",
        "runner": "dataops-platform-runner:local",
        "frontend": "dataops-platform-frontend:local",
    }
    for service, image in expected.items():
        assert services[service]["image"] == image


def test_published_ports_can_be_isolated_for_recovery_rehearsal():
    source = COMPOSE.read_text(encoding="utf-8")
    expected_overrides = {
        "DATAOPS_POSTGRES_PORT": "15432",
        "SOURCE_POSTGRES_PORT": "25432",
        "SOURCE_MYSQL_PORT": "23306",
        "NEO4J_HTTP_PORT": "17474",
        "NEO4J_BOLT_PORT": "17687",
        "MINIO_API_PORT": "19000",
        "MINIO_CONSOLE_PORT": "19001",
        "N8N_PORT": "15678",
        "KESTRA_PORT": "18080",
        "RUNNER_PORT": "15600",
        "BACKEND_PORT": "15500",
        "FRONTEND_PORT": "18183",
    }
    for name, default in expected_overrides.items():
        assert f"${{{name}:-{default}}}" in source


@pytest.mark.parametrize(
    "script_name",
    (
        "preflight.sh",
        "install_offline.sh",
        "package_offline.sh",
        "backup.sh",
        "restore.sh",
        "rollback.sh",
    ),
)
def test_wp13_commands_publish_help_without_mutating_state(script_name: str):
    completed = subprocess.run(
        ["bash", str(OPERATIONS / script_name), "--help"],
        cwd=ROOT,
        capture_output=True,
        text=True,
    )
    assert completed.returncode == 0, completed.stderr
    assert "Usage:" in completed.stdout


def test_offline_package_contains_runtime_payload_and_checksums(tmp_path: Path):
    completed = subprocess.run(
        [
            "bash",
            str(OPERATIONS / "package_offline.sh"),
            "--repo-root",
            str(ROOT),
            "--output-dir",
            str(tmp_path),
            "--version",
            "contract-test",
            "--skip-images",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
    )
    assert completed.returncode == 0, completed.stderr

    archive = tmp_path / "dataops-platform-offline-contract-test.tar.gz"
    assert archive.is_file()
    with tarfile.open(archive, "r:gz") as package:
        names = set(package.getnames())
        checksum_member = package.extractfile(
            "dataops-platform-offline-contract-test/checksums.sha256"
        )
        assert checksum_member is not None
        checksums = checksum_member.read().decode("utf-8")

    prefix = "dataops-platform-offline-contract-test/"
    for relative in (
        "deploy/docker/docker-compose.yml",
        "deploy/docker/.env.example",
        "database/create_data_orders_table.sql",
        "migrations/env.py",
        "operations/install_offline.sh",
        "operations/preflight.sh",
        "operations/backup.sh",
        "operations/restore.sh",
        "release-manifest.txt",
        "checksums.sha256",
    ):
        assert prefix + relative in names

    plugin_migration = "migrations/versions/20260818_559_plugin_three_identity_gate.py"
    assert prefix + plugin_migration in names
    digest = hashlib.sha256((ROOT / plugin_migration).read_bytes()).hexdigest()
    assert f"{digest}  {plugin_migration}" in checksums


def test_preflight_rejects_modified_bundle_before_docker_access(tmp_path: Path):
    bundle = tmp_path / "bundle"
    bundle.mkdir()
    payload = bundle / "release-manifest.txt"
    payload.write_text("version=contract-test\n", encoding="utf-8")
    digest = hashlib.sha256(payload.read_bytes()).hexdigest()
    (bundle / "checksums.sha256").write_text(
        f"{digest}  release-manifest.txt\n",
        encoding="utf-8",
    )
    payload.write_text("version=tampered\n", encoding="utf-8")

    env = os.environ.copy()
    env["PATH"] = "/usr/bin:/bin"
    completed = subprocess.run(
        [
            "bash",
            str(OPERATIONS / "preflight.sh"),
            "--bundle-root",
            str(bundle),
            "--checksums-only",
        ],
        cwd=ROOT,
        env=env,
        capture_output=True,
        text=True,
    )
    assert completed.returncode != 0
    assert "checksum" in (completed.stdout + completed.stderr).lower()


def test_restore_requires_exact_project_confirmation_before_file_or_docker_access():
    completed = subprocess.run(
        [
            "bash",
            str(OPERATIONS / "restore.sh"),
            "--backup-dir",
            "/path/that/does/not/exist",
            "--project-name",
            "dataops-wp13-restore",
            "--confirm-project",
            "wrong-project",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
    )
    assert completed.returncode != 0
    assert "confirmation" in (completed.stdout + completed.stderr).lower()


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