DataOps-platform/tests/test_local_docker_contract.py

from pathlib import Path


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


def test_local_compose_defines_the_isolated_stack():
    source = COMPOSE.read_text(encoding="utf-8")

    for service in (
        "postgres:",
        "source-postgres:",
        "source-mysql:",
        "neo4j:",
        "minio:",
        "minio-init:",
        "n8n:",
        "backend:",
        "frontend:",
    ):
        assert service in source

    assert "name: dataops-test" in source
    assert "dataops-test-net" in source
    assert "healthcheck:" in source
    assert "DATABASE_URL: postgresql://" in source
    assert "NEO4J_URI: bolt://neo4j:7687" in source
    assert "MINIO_HOST: minio:9000" in source
    assert "N8N_API_URL: http://n8n:5678" in source
    assert "DATASOURCE_CREDENTIAL_MASTER_KEY:" in source
    assert "dataops-test-source-postgres:" in source
    assert "dataops-test-source-mysql:" in source
    assert "import urllib.request" in source


def test_local_compose_and_n8n_client_have_no_production_fallbacks():
    sources = [
        COMPOSE.read_text(encoding="utf-8"),
        (ROOT / "app/config/config.py").read_text(encoding="utf-8"),
        (ROOT / "app/core/data_factory/n8n_client.py").read_text(encoding="utf-8"),
    ]
    forbidden = (
        "n8n.citupro.com",
        "company.citupro.com",
        "192.168.3.143",
        "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9",
    )

    local_sources = "\n".join(sources)
    for value in forbidden:
        assert value not in local_sources


def test_active_source_has_no_hardcoded_nonlocal_service_addresses():
    source_files = [
        *sorted((ROOT / "app").rglob("*.py")),
        *sorted((ROOT / "frontend" / "src").rglob("*.js")),
        *sorted((ROOT / "frontend" / "src").rglob("*.vue")),
        *sorted((ROOT / "frontend").glob(".env.*")),
    ]
    active_source = "\n".join(
        path.read_text(encoding="utf-8") for path in source_files
    )

    for value in (
        "n8n.citupro.com",
        "minio.citupro.com",
        "192.168.3.143",
        "192.168.3.91",
        "115.128.14.14",
    ):
        assert value not in active_source


def test_docker_images_build_from_source_directories_only():
    backend = (ROOT / "deploy/docker/backend.Dockerfile").read_text(encoding="utf-8")
    frontend = (ROOT / "deploy/docker/frontend.Dockerfile").read_text(encoding="utf-8")

    assert "COPY app/" in backend
    assert "COPY migrations/" in backend
    assert "COPY alembic.ini" in backend
    assert "alembic -c alembic.ini upgrade head" in backend
    assert "deployment/app" not in backend
    assert "gcc" in backend
    assert "libc6-dev" in backend
    assert "linux-libc-dev" in backend
    assert "apt-get purge" in backend
    assert "FROM node:24" in frontend
    assert "npm ci" in frontend
    assert (ROOT / "wsgi.py").is_file()


def test_postgres_bootstrap_reuses_source_ddl_without_reversible_local_user():
    bootstrap = (ROOT / "deploy/docker/postgres/init/000-init.sql").read_text(
        encoding="utf-8"
    )
    for ddl in sorted((ROOT / "database").glob("*.sql")):
        assert f"/dataops-schema/{ddl.name}" in bootstrap

    assert "CREATE DATABASE n8n" in bootstrap
    assert "CREATE EXTENSION IF NOT EXISTS vector" in bootstrap
    assert not (ROOT / "deploy/docker/postgres/init/010-local-test-user.sql").exists()
    readme = (ROOT / "deploy/docker/README.md").read_text(encoding="utf-8")
    assert "app.commands.bootstrap_admin" in readme