RAG/backend/tests/integration/test_schema_contract.py

from __future__ import annotations

import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[3]
BACKEND = ROOT / "backend"
sys.path.insert(0, str(BACKEND))

from app.persistence.job_queue_sql import (  # noqa: E402
    CLAIM_JOB_SQL,
    COMPLETE_JOB_SQL,
    FAIL_OR_RETRY_JOB_SQL,
    HEARTBEAT_JOB_SQL,
    REAP_EXPIRED_JOBS_SQL,
)

COMPOSE = (ROOT / "compose.yaml").read_text(encoding="utf-8")
DOCKERFILE = (BACKEND / "Dockerfile").read_text(encoding="utf-8")
FRONTEND_DOCKERFILE = (ROOT / "frontend/Dockerfile").read_text(encoding="utf-8")
NGINX = (ROOT / "frontend/nginx.conf").read_text(encoding="utf-8")
DEMO_API = (BACKEND / "app/api/v1/demo.py").read_text(encoding="utf-8")
BOOTSTRAP = (ROOT / "ops/postgres/init/10-bootstrap-rag.sh").read_text(encoding="utf-8")
MIGRATION = (BACKEND / "migrations/versions/0001_initial_schema.py").read_text(encoding="utf-8")


def _service_block(name: str) -> str:
    pattern = re.compile(rf"(?ms)^  {re.escape(name)}:\n(.*?)(?=^  [a-zA-Z0-9_-]+:\n|\Z)")
    match = pattern.search(COMPOSE)
    assert match is not None, f"missing Compose service: {name}"
    return match.group(1)


def test_compose_isolates_database_credentials_and_networks() -> None:
    db = _service_block("db")
    migrate = _service_block("migrate")
    upload_init = _service_block("upload-init")
    api = _service_block("api")
    model_gateway = _service_block("model-gateway")
    worker_local = _service_block("worker-local")
    worker_model = _service_block("worker-model")
    gateway = _service_block("gateway")
    web = _service_block("web")
    provider_smoke = _service_block("provider-smoke")
    seed_demo = _service_block("seed-demo")
    seed_demo_offline = _service_block("seed-demo-offline")
    document_smoke = _service_block("document-pipeline-smoke")

    assert "postgres_bootstrap_password" in db
    assert "postgres_migrator_password" in db
    assert "postgres_app_password" in db

    assert "postgres_migrator_password" in migrate
    assert "postgres_bootstrap_password" not in migrate
    assert "postgres_app_password" not in migrate

    assert "network_mode: none" in upload_init
    assert 'user: "0:0"' in upload_init
    assert "uploads_data:/data/uploads" in upload_init
    assert "secrets:" not in upload_init
    assert "CHOWN" in upload_init
    assert "DAC_OVERRIDE" in upload_init

    assert "postgres_app_password" in api
    assert "model_gateway_api_token" in api
    assert "postgres_bootstrap_password" not in api
    assert "postgres_migrator_password" not in api
    assert "bailian_api_key" not in api
    assert "uploads_data:/data/uploads" in api
    assert '"127.0.0.1:8000:8000"' not in api
    assert "      - data" in api
    assert "      - model" in api
    assert "      - edge" not in api
    assert "      - egress" not in api
    assert "read_only: true" in api
    assert "no-new-privileges:true" in api
    assert "cap_drop:" in api and "      - ALL" in api

    assert "bailian_api_key" in model_gateway
    assert "model_gateway_api_token" in model_gateway
    assert "model_gateway_worker_token" in model_gateway
    assert "postgres_" not in model_gateway
    assert "uploads_data" not in model_gateway
    assert "      - model" in model_gateway
    assert "      - egress" in model_gateway
    assert "      - data" not in model_gateway
    assert "      - edge" not in model_gateway
    assert "      - ingress" not in model_gateway
    assert "ports:" not in model_gateway
    assert "read_only: true" in model_gateway
    assert "no-new-privileges:true" in model_gateway
    assert "cap_drop:" in model_gateway and "      - ALL" in model_gateway

    assert "WORKER_CAPABILITIES: document_parse" in worker_local
    assert "postgres_app_password" in worker_local
    assert "model_gateway_" not in worker_local
    assert "bailian_api_key" not in worker_local
    assert "uploads_data:/data/uploads" in worker_local
    assert "      - data" in worker_local
    assert "      - model" not in worker_local
    assert "      - egress" not in worker_local
    assert "read_only: true" in worker_local
    assert "no-new-privileges:true" in worker_local
    assert "stop_grace_period: 150s" in worker_local

    assert "WORKER_CAPABILITIES: embedding" in worker_model
    assert "MODEL_GATEWAY_CALLER: worker" in worker_model
    assert "model_gateway_worker_token" in worker_model
    assert "model_gateway_api_token" not in worker_model
    assert "postgres_app_password" in worker_model
    assert "bailian_api_key" not in worker_model
    assert "uploads_data" not in worker_model
    assert "      - data" in worker_model
    assert "      - model" in worker_model
    assert "      - egress" not in worker_model
    assert "read_only: true" in worker_model
    assert "no-new-privileges:true" in worker_model
    assert "stop_grace_period: 150s" in worker_model

    assert '"127.0.0.1:8000:8000"' not in gateway
    assert "      - ingress" in gateway
    assert "      - data" in gateway
    assert "      - model" not in gateway
    assert "      - edge" not in gateway
    assert "      - egress" not in gateway
    assert "secrets:" not in gateway
    assert "POSTGRES_" not in gateway
    assert "BAILIAN_" not in gateway
    assert "read_only: true" in gateway
    assert "no-new-privileges:true" in gateway

    assert '"127.0.0.1:8000:8080"' in web
    assert "      - edge" in web
    assert "      - ingress" in web
    assert "      - data" not in web
    assert "      - model" not in web
    assert "      - egress" not in web
    assert "secrets:" not in web
    assert "POSTGRES_" not in web
    assert "BAILIAN_" not in web
    assert "read_only: true" in web
    assert "no-new-privileges:true" in web
    assert len(re.findall(r"(?m)^    ports:$", COMPOSE)) == 1

    assert "bailian_api_key" not in provider_smoke
    assert "model_gateway_api_token" in provider_smoke
    assert "postgres_" not in provider_smoke
    assert "      - model" in provider_smoke
    assert "      - egress" not in provider_smoke

    assert "postgres_app_password" in seed_demo
    assert "postgres_bootstrap_password" not in seed_demo
    assert "postgres_migrator_password" not in seed_demo
    assert "bailian_api_key" not in seed_demo
    assert "model_gateway_worker_token" in seed_demo
    assert "MODEL_GATEWAY_CALLER: worker" in seed_demo
    assert "      - model" in seed_demo
    assert "      - egress" not in seed_demo
    assert "./data/samples/public:/demo:ro" in seed_demo

    assert "postgres_app_password" in seed_demo_offline
    assert "bailian_api_key" not in seed_demo_offline
    assert "DEMO_PROVIDER_MODE: fake" in seed_demo_offline
    assert "./data/samples/public:/demo:ro" in seed_demo_offline

    assert "secrets:" not in document_smoke
    assert "POSTGRES_" not in document_smoke
    assert "BAILIAN_" not in document_smoke
    assert "model_gateway_" not in document_smoke
    assert "DOCUMENT_NAMESPACE_MODE:" in document_smoke
    assert "./data/samples/public:/demo:ro" in document_smoke
    assert "      - ingress" in document_smoke
    assert "      - data" not in document_smoke
    assert "      - model" not in document_smoke
    assert "      - egress" not in document_smoke
    assert "read_only: true" in document_smoke

    assert re.search(r"(?ms)^  data:\n.*?^    internal: true$", COMPOSE)
    assert re.search(r"(?ms)^  ingress:\n.*?^    internal: true$", COMPOSE)
    assert re.search(r"(?ms)^  model:\n.*?^    internal: true$", COMPOSE)
    assert re.search(r"(?m)^  edge:$", COMPOSE)
    assert re.search(r"(?m)^  egress:$", COMPOSE)

    assert "USER 10001:10001" in DOCKERFILE
    assert 'CMD ["uvicorn", "app.main:app"' in DOCKERFILE
    assert "USER 101:101" in FRONTEND_DOCKERFILE
    assert "proxy_buffering off" in NGINX
    assert "server gateway:8000" in NGINX


def test_demo_queries_pin_governed_offline_identity() -> None:
    normalized = " ".join(DEMO_API.split())

    for required_filter in (
        "chunk.knowledge_base_id = %s",
        "chunk.access_scope_id = %s",
        "chunk.metadata ->> 'source_type' = 'synthetic'",
        "chunk.searchable IS TRUE",
        "chunk.index_status = 'READY'",
        "chunk.approval_status = 'CLOUD_APPROVED'",
        "document.active_version_id = chunk.document_version_id",
        "version.review_state = 'CLOUD_APPROVED'",
        "version.outbound_manifest_sha256 = chunk.outbound_manifest_sha256",
        "version.embedding_profile_hash = chunk.embedding_profile_hash",
        "chunk.embedding_model = %s",
        "chunk.embedding_profile_hash = %s",
    ):
        assert required_filter in normalized

    assert "KNOWLEDGE_BASE_ID" in DEMO_API
    assert "ACCESS_SCOPE_ID" in DEMO_API
    assert "DEMO_FAKE_EMBEDDING_MODEL" in DEMO_API
    assert "offline_embedding_profile_hash" in DEMO_API


def test_database_health_requires_tcp_and_atomic_bootstrap_sentinel() -> None:
    db = _service_block("db")
    assert "pg_isready -h 127.0.0.1 -p 5432" in db
    assert ".rag-bootstrap-complete" in db

    assert "CREATE EXTENSION IF NOT EXISTS vector" in BOOTSTRAP
    assert 'CREATE ROLE :"migrator_user"' in BOOTSTRAP
    assert 'CREATE ROLE :"app_user"' in BOOTSTRAP
    assert "NOSUPERUSER" in BOOTSTRAP
    assert "CREATE SCHEMA rag AUTHORIZATION" in BOOTSTRAP
    assert "ALTER DEFAULT PRIVILEGES" in BOOTSTRAP
    assert BOOTSTRAP.index("COMMIT;") < BOOTSTRAP.index("mv -f")
    assert "\\getenv migrator_password RAG_MIGRATOR_PASSWORD" in BOOTSTRAP
    assert "--set=migrator_password" not in BOOTSTRAP
    assert "--set=app_password" not in BOOTSTRAP
    assert "set -x" not in BOOTSTRAP
    assert "official postgres entrypoint resolves POSTGRES_PASSWORD_FILE" in BOOTSTRAP
    assert 'bootstrap_password="${POSTGRES_PASSWORD:-}"' in BOOTSTRAP


def test_initial_migration_has_vector_and_activation_contracts() -> None:
    normalized = " ".join(MIGRATION.lower().split())

    assert "embedding vector(1024)" in normalized
    assert "using hnsw (embedding vector_cosine_ops)" in normalized
    assert "where searchable" in normalized
    assert "chunks_searchable_requires_ready_approved_embedding" in normalized
    assert "check (embedding_dimension = 1024)" in normalized
    assert "active_version_id uuid" in normalized
    assert "documents_active_version_fk" in normalized
    assert "deferrable initially deferred" in normalized
    assert "foreign key (id, active_version_id)" in normalized
    assert "references rag.document_versions (document_id, id)" in normalized
    assert "local_parsed_pending_cloud_review" in normalized
    assert "cloud_approved" in normalized
    assert "outbound_manifest_sha256" in normalized
    assert "cloud_processing_allowed" not in normalized
    assert "document_versions_cloud_approval_bound" in normalized
    assert "chunks_approval_binding_fk" in normalized
    assert "create table rag.outbound_manifest_items" in normalized
    assert "chunks_manifest_item_binding_fk" in normalized
    assert "outbound_manifest_items_immutable_after_approval" in normalized
    assert "chunks_guard_approved_mutation" in normalized
    assert "chunks_guard_ready_vector_update" in normalized
    assert "new.access_scope_id is distinct from old.access_scope_id" in normalized
    assert "new.document_version_id is distinct from old.document_version_id" in normalized
    assert "cloud_text text not null" in normalized
    assert "cloud_text_sha256 char(64) not null" in normalized
    assert "embedding_text = embedding_prefix || cloud_text" in normalized
    assert "sha256(convert_to(cloud_text, 'utf8'))" in normalized
    assert "sha256(convert_to(embedding_text, 'utf8'))" in normalized
    assert "approval_status = 'cloud_approved'" in normalized
    assert "document_versions_revoke_cloud_approval" in normalized
    assert "new.review_state is distinct from old.review_state" in normalized
    assert "set searchable = false" in normalized
    assert "approval_status = revoked_state" in normalized
    assert "embedding = null" in normalized
    assert "embedding_profile_hash = null" in normalized
    assert "where document_version_id = old.id" in normalized
    assert "chunks_enforce_active_search_projection" in normalized
    assert "document.active_version_id = new.document_version_id" in normalized
    assert "version.status = 'ready'" in normalized
    assert "documents_enforce_activation" in normalized


def test_downgrade_drops_manifest_items_before_document_versions() -> None:
    normalized = " ".join(MIGRATION.lower().split())

    manifest_drop = normalized.index("drop table if exists rag.outbound_manifest_items")
    version_drop = normalized.index("drop table if exists rag.document_versions")
    assert manifest_drop < version_drop


def test_initial_migration_has_fenced_job_schema_and_indexes() -> None:
    normalized = " ".join(MIGRATION.lower().split())

    assert "create table rag.background_jobs" in normalized
    assert "lease_owner text" in normalized
    assert "lease_token uuid" in normalized
    assert "lease_until timestamptz" in normalized
    assert "attempt <= max_attempts" in normalized
    assert "background_jobs_lease_consistent" in normalized
    assert "unique (job_type, idempotency_key)" in normalized
    assert "background_jobs_claim_queued" in normalized
    assert "background_jobs_reap_expired" in normalized


def test_job_claim_and_terminal_updates_are_fenced() -> None:
    claim = " ".join(CLAIM_JOB_SQL.lower().split())
    heartbeat = " ".join(HEARTBEAT_JOB_SQL.lower().split())
    complete = " ".join(COMPLETE_JOB_SQL.lower().split())
    failure = " ".join(FAIL_OR_RETRY_JOB_SQL.lower().split())

    assert "for update skip locked" in claim
    assert "lease_token = gen_random_uuid()" in claim
    assert "attempt = job.attempt + 1" in claim

    for statement in (heartbeat, complete, failure):
        assert "job.status = 'running'" in statement
        assert "job.lease_owner = :worker_id" in statement
        assert "job.lease_token = :lease_token" in statement
        assert "returning" in statement

    assert "when job.attempt < job.max_attempts then 'queued'" in failure
    assert "else 'failed'" in failure


def test_reaper_uses_advisory_lock_and_handles_exhausted_attempts() -> None:
    reaper = " ".join(REAP_EXPIRED_JOBS_SQL.lower().split())

    assert "pg_try_advisory_xact_lock" in reaper
    assert "job.status = 'running'" in reaper
    assert "job.lease_until < now()" in reaper
    assert "for update of job skip locked" in reaper
    assert "when job.attempt < job.max_attempts then 'queued'" in reaper
    assert "else 'failed'" in reaper
    assert "lease_owner = null" in reaper
    assert "lease_token = null" in reaper
    assert "lease_until = null" in reaper