DataOps-platform/tests/test_phase3_wp04_production_runtime.py

import subprocess
from pathlib import Path
from types import SimpleNamespace

import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from flask import Flask

from app.config.config import ProductionConfig, apply_runtime_env_config
from app.config.database_urls import (
    load_and_validate_database_env,
    validate_database_environment,
    validate_postgresql_url,
)
from app.core.edge_gateway.runtime_roles import provision_runtime_login
from app.core.edge_gateway.service import (
    EdgeGatewayConfigurationError,
    EdgeGatewayService,
)

ROOT = Path(__file__).resolve().parents[1]
PRODUCTION_ENV = {
    "DATABASE_URL": "postgresql://dataops_app:runtime-secret@db.example.test/dataops",
    "NEO4J_URI": "bolt+s://neo4j.example.test:7687",
    "NEO4J_HTTP_URI": "https://neo4j.example.test:7473",
    "NEO4J_USER": "dataops_graph",
    "NEO4J_PASSWORD": "graph-secret",
    "MINIO_HOST": "objects.example.test:9000",
    "MINIO_USER": "dataops_objects",
    "MINIO_PASSWORD": "object-secret",
    "MINIO_BUCKET": "dataops-production",
}


def _production_app(monkeypatch, overrides=None):
    monkeypatch.setenv("FLASK_ENV", "production")
    monkeypatch.setenv("APP_ENV_FILE", "/nonexistent/dataops-test.env")
    for name, value in {**PRODUCTION_ENV, **(overrides or {})}.items():
        if value is None:
            monkeypatch.delenv(name, raising=False)
        else:
            monkeypatch.setenv(name, value)
    app = Flask(__name__)
    app.config.from_object(ProductionConfig)
    return app


@pytest.mark.parametrize(
    ("name", "value"),
    [
        ("DATABASE_URL", None),
        ("DATABASE_URL", "postgresql://dataops_user:replace-password@127.0.0.1/dataops"),
        ("NEO4J_PASSWORD", None),
        ("NEO4J_PASSWORD", "replace-neo4j-password"),
        ("MINIO_USER", None),
        ("MINIO_PASSWORD", "replace-minio-password"),
    ],
)
def test_production_service_configuration_fails_closed(monkeypatch, name, value):
    app = _production_app(monkeypatch, {name: value})
    with pytest.raises(RuntimeError, match=name):
        apply_runtime_env_config(app)


def test_production_configuration_accepts_complete_non_placeholder_values(monkeypatch):
    app = _production_app(monkeypatch)
    apply_runtime_env_config(app)
    assert app.config["SQLALCHEMY_DATABASE_URI"] == PRODUCTION_ENV["DATABASE_URL"]
    assert app.config["NEO4J_PASSWORD"] == PRODUCTION_ENV["NEO4J_PASSWORD"]
    assert app.config["MINIO_PASSWORD"] == PRODUCTION_ENV["MINIO_PASSWORD"]


def test_runtime_and_migrator_images_are_capability_separated():
    backend = (ROOT / "deploy/docker/backend.Dockerfile").read_text()
    runner = (ROOT / "deploy/docker/runner.Dockerfile").read_text()
    compose = (ROOT / "deploy/docker/docker-compose.yml").read_text()
    migration = (ROOT / "migrations/versions/20260809_477_edge_gateway_control_plane.py").read_text()
    provisioner = (ROOT / "app/core/edge_gateway/runtime_roles.py").read_text()

    assert "AS migrator" in backend and "AS runtime" in backend
    runtime = backend.split("AS runtime", 1)[1]
    assert "COPY migrations/" not in runtime and "COPY alembic.ini" not in runtime
    assert "pip uninstall -y alembic" in runtime
    assert "COPY migrations/" not in runner and "COPY alembic.ini" not in runner
    assert "pip uninstall -y alembic" in runner
    assert "target: migrator" in compose and "target: runtime" in compose
    assert "DB_ROLE_INIT_DATABASE_URL" in compose
    assert "DB_ROLE_INIT_DATABASE_URL" in provisioner
    assert "MIGRATION_DATABASE_URL" in provisioner
    role_bootstrap = migration.split("CREATE OR REPLACE FUNCTION", 1)[0]
    assert "CREATE ROLE" not in role_bootstrap and "ALTER ROLE" not in role_bootstrap


def test_enterprise_edge_mtls_proxy_is_dedicated_and_fail_closed():
    compose = (ROOT / "deploy/docker/docker-compose.yml").read_text()
    container_nginx = (ROOT / "deploy/docker/edge-mtls-nginx.conf").read_text()
    host_nginx = (ROOT / "deployment/config/nginx-dataops-platform.conf").read_text()

    assert "edge-mtls-proxy:" in compose
    edge_service = compose.split("edge-mtls-proxy:", 1)[1].split(
        "\n  lightrag-neo4j:", 1
    )[0]
    assert "enterprise-edge" in edge_service
    assert "edge-control-net" in edge_service
    assert "ipv4_address: 172.31.0.10" in edge_service
    assert "EDGE_MTLS_TRUSTED_PROXY_IPS: 172.31.0.10" in compose
    assert "EDGE_GATEWAY_SIGNING_PRIVATE_KEY_FILE: /run/secrets/edge_gateway_signing_private_key" in compose
    assert "EDGE_GATEWAY_SIGNING_PRIVATE_KEY:" not in compose
    assert "edge_gateway_signing_private_key:" in compose
    assert "/run/secrets/edge_gateway_signing_private_key" in compose
    assert "ssl_verify_client on;" in container_nginx
    assert "ssl_client_certificate /etc/nginx/edge-mtls/client-ca.pem;" in container_nginx
    assert "ssl_crl /etc/nginx/edge-mtls/client-ca.crl;" in container_nginx
    assert "location /api/datasource/edge/" in container_nginx
    assert (
        "proxy_set_header X-DataOps-Edge-Client-Cert $ssl_client_escaped_cert;"
        in container_nginx
    )
    assert (
        "proxy_set_header X-DataOps-Edge-Client-Verify $ssl_client_verify;"
        in container_nginx
    )
    assert "listen 18443 ssl;" in host_nginx
    assert "ssl_verify_client on;" in host_nginx
    assert "edge-client-ca.crt" in host_nginx
    assert "edge-client-ca.crl" in host_nginx


def test_production_edge_signer_uses_secure_file_or_provider_and_rejects_inline(tmp_path):
    private_key = Ed25519PrivateKey.generate()
    private_path = tmp_path / "edge-signing.key"
    private_path.write_bytes(
        private_key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        )
    )
    private_path.chmod(0o600)
    public_key = private_key.public_key().public_bytes(
        serialization.Encoding.Raw,
        serialization.PublicFormat.Raw,
    ).hex()
    repository = SimpleNamespace(rollback=lambda: None)
    service = EdgeGatewayService(
        repository,
        signing_private_key_file=str(private_path),
        signing_public_key=public_key,
        signing_key_id="edge-key-2026-08",
        production=True,
    )
    assert service._require_signer()[1] == "edge-key-2026-08"
    provider_service = EdgeGatewayService(
        repository,
        signing_key_provider=lambda: private_path.read_bytes(),
        signing_public_key=public_key,
        signing_key_id="edge-key-2026-08",
        production=True,
    )
    assert provider_service._require_signer()[0].public_key().public_bytes(
        serialization.Encoding.Raw,
        serialization.PublicFormat.Raw,
    ).hex() == public_key
    with pytest.raises(EdgeGatewayConfigurationError, match="inline"):
        EdgeGatewayService(
            repository,
            signing_private_key=private_key,
            signing_public_key=public_key,
            signing_key_id="edge-key-2026-08",
            production=True,
        )
    private_path.chmod(0o640)
    with pytest.raises(EdgeGatewayConfigurationError, match="permissions"):
        EdgeGatewayService(
            repository,
            signing_private_key_file=str(private_path),
            signing_public_key=public_key,
            signing_key_id="edge-key-2026-08",
            production=True,
        )
    private_path.chmod(0o600)
    with pytest.raises(EdgeGatewayConfigurationError, match="public key"):
        EdgeGatewayService(
            repository,
            signing_private_key_file=str(private_path),
            signing_public_key="0" * 64,
            signing_key_id="edge-key-2026-08",
            production=True,
        )


def _template_env() -> dict[str, str]:
    values = {}
    for raw_line in (ROOT / "deployment/dataops.env").read_text(encoding="utf-8-sig").splitlines():
        line = raw_line.strip()
        if line and not line.startswith("#") and "=" in line:
            name, value = line.split("=", 1)
            values[name] = value
    return values


def test_checked_in_production_template_is_rejected_everywhere(monkeypatch):
    template = _template_env()
    app = _production_app(monkeypatch, template)
    with pytest.raises(RuntimeError, match="DATABASE_URL"):
        apply_runtime_env_config(app)
    for name, value in template.items():
        monkeypatch.setenv(name, value)
    with pytest.raises(RuntimeError, match="DB_ROLE_INIT_DATABASE_URL"):
        provision_runtime_login()
    with pytest.raises(RuntimeError, match="DB_ROLE_INIT_DATABASE_URL"):
        load_and_validate_database_env(ROOT / "deployment/dataops.env")
    result = subprocess.run(
        [
            ".venv/bin/python", "app/config/database_urls.py", "--env-file",
            "deployment/dataops.env",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
    )
    assert result.returncode != 0
    assert "DB_ROLE_INIT_DATABASE_URL" in result.stderr
    deploy = (ROOT / "deployment/deploy_dataops.sh").read_text()
    source_deploy = (ROOT / "scripts/deploy_dataops.sh").read_text()
    assert "database_urls.py" in deploy and "database_urls.py" in source_deploy


@pytest.mark.parametrize(
    ("name", "bad_value"),
    [
        ("DB_ROLE_INIT_DATABASE_URL", "postgresql://role_admin:replace-role-init-password@db/dataops"),
        ("MIGRATION_DATABASE_URL", "postgresql://migrator:replace-migration-password@db/dataops"),
        ("DATABASE_URL", "postgresql://dataops_app:replace-runtime-password@db/dataops"),
        ("DATAOPS_RUNTIME_PASSWORD", "replace-runtime-password"),
    ],
)
def test_role_init_rejects_each_database_template_secret(monkeypatch, name, bad_value):
    valid = {
        "DB_ROLE_INIT_DATABASE_URL": "postgresql://role_admin:Strong%40Role9@db/dataops",
        "MIGRATION_DATABASE_URL": "postgresql://migrator:Strong%40Migration9@db/dataops",
        "DATABASE_URL": "postgresql://dataops_app:Strong%40Runtime9@db/dataops",
        "DATAOPS_MIGRATOR_USER": "dataops_migrator",
        "DATAOPS_RUNTIME_USER": "dataops_app",
        "DATAOPS_RUNTIME_PASSWORD": "Strong@Runtime9",
    }
    valid[name] = bad_value
    for key, value in valid.items():
        monkeypatch.setenv(key, value)
    with pytest.raises(RuntimeError, match=name):
        provision_runtime_login()


def test_database_url_validation_decodes_components_without_substring_false_positive(tmp_path):
    validate_postgresql_url(
        "postgresql://dataops_app:S7replace%40middle%21@db.internal/dataops",
        "DATABASE_URL",
    )
    with pytest.raises(RuntimeError, match="DATABASE_URL"):
        validate_postgresql_url(
            "postgresql://dataops_app:replace%2Druntime%2Dpassword@db.internal/dataops",
            "DATABASE_URL",
        )
    env_file = tmp_path / "dataops.env"
    env_file.write_text(
        "\n".join([
            "DB_ROLE_INIT_DATABASE_URL=postgresql://role_admin:Strong%40Role9@db.internal/dataops",
            "MIGRATION_DATABASE_URL=postgresql://migrator:Strong%40Migration9@db.internal/dataops",
            "DATABASE_URL=postgresql://dataops_app:Strong%40Runtime9@db.internal/dataops",
            "DATAOPS_RUNTIME_PASSWORD=Strong@Runtime9",
        ]),
    )
    load_and_validate_database_env(env_file)


@pytest.mark.parametrize(
    "bad_url",
    [
        "postgresql://:Strong%40Pass9@db.internal/dataops",
        "postgresql://replace-user:Strong%40Pass9@db.internal/dataops",
        "postgresql://dataops_app:@db.internal/dataops",
        "postgresql://dataops_app:Strong%40Pass9@/dataops",
        "postgresql://dataops_app:Strong%40Pass9@replace-host/dataops",
        "postgresql://dataops_app:Strong%40Pass9@db.internal/",
        "postgresql://dataops_app:Strong%40Pass9@db.internal/database_name",
        "postgresql://dataops_app:replace%2Dpassword@db.internal/dataops",
    ],
)
def test_database_url_validation_rejects_each_empty_or_template_component(bad_url):
    with pytest.raises(RuntimeError, match="DATABASE_URL"):
        validate_postgresql_url(bad_url, "DATABASE_URL")


def test_shared_environment_validation_compares_decoded_runtime_password():
    values = {
        "DB_ROLE_INIT_DATABASE_URL": "postgresql://role_admin:Strong%40Role9@db.internal/dataops",
        "MIGRATION_DATABASE_URL": "postgresql://migrator:Strong%40Migration9@db.internal/dataops",
        "DATABASE_URL": "postgresql://dataops_app:Strong%40Runtime9@db.internal/dataops",
        "DATAOPS_RUNTIME_PASSWORD": "Strong@Runtime9",
    }
    validate_database_environment(values)
    values["DATAOPS_RUNTIME_PASSWORD"] = "Different@Runtime9"
    with pytest.raises(RuntimeError, match="DATAOPS_RUNTIME_PASSWORD.*match"):
        validate_database_environment(values)