coulomb-social/scripts/rehearse-import-inspection.py

"""Run synthetic corpus/receipt tests on disposable isolated PostgreSQL."""

import hashlib
import json
import subprocess
import sysconfig
import uuid
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
IMAGE = (
    "forgejo.coulomb.social/coulomb/coulomb-social@sha256:"
    "522bcd7fce92436765a60c1e4afa2598f3cc40913352153ff1c4319a99ca92ed"
)
CODE = """import json,os,time
os.environ['DJANGO_SETTINGS_MODULE']='coulomb_social.settings.test'
from django.conf import settings
settings.DATABASES={'default':{'ENGINE':'django.db.backends.postgresql',
 'HOST':'inspection-db','NAME':'inspection','USER':'postgres','PORT':'5432'}}
import django
django.setup()
from django.db import connection
for attempt in range(30):
 try:
  connection.ensure_connection();break
 except Exception:time.sleep(1)
else:raise RuntimeError('disposable_database_not_ready')
version=connection.pg_version
connection.close()
from coulomb_social.apps.spaces import import_inspection
original=import_inspection.observe_unit
checks=[]
def observe(*args):
 with connection.cursor() as cursor:
  cursor.execute('SHOW transaction_read_only');assert cursor.fetchone()[0]=='on'
  cursor.execute('SHOW transaction_isolation');assert cursor.fetchone()[0]=='repeatable read'
 checks.append(True)
 return original(*args)
import_inspection.observe_unit=observe
import pytest
code=pytest.main(['-q','-p','no:cacheprovider','/app/tests/test_import_receipts.py',
 '/app/tests/test_import_inspection.py','/app/tests/test_import_corpus.py',
 '/app/tests/test_import_reconciliation.py'])
assert code==0 and checks
print(json.dumps({'status':'passed','database_vendor':'postgresql','server_version':version,
 'inspection_transaction_mode_checks':len(checks),'source_overlay':True,
 'locked_development_dependencies':True,'production_data_used':False}))
"""


def docker(*args):
    return subprocess.run(
        ["docker", *args], capture_output=True, text=True, check=True, timeout=180
    ).stdout.strip()


def main():
    app = docker("image", "inspect", IMAGE, "--format", "{{.Id}}")
    pg = docker("image", "inspect", "postgres:16", "--format", "{{.Id}}")
    network = database = runner = None
    name = "csoc-inspection-" + uuid.uuid4().hex[:12]
    try:
        network = docker("network", "create", "--internal", name)
        database = docker(
            "create",
            "--name",
            name + "-db",
            "--network",
            network,
            "--network-alias",
            "inspection-db",
            "-e",
            "POSTGRES_HOST_AUTH_METHOD=trust",
            "-e",
            "POSTGRES_DB=inspection",
            pg,
        )
        docker("start", database)
        runner = docker(
            "create",
            "--name",
            name + "-app",
            "--network",
            network,
            "--read-only",
            "--tmpfs",
            "/tmp",
            "--entrypoint",
            "/app/.venv/bin/python",
            "-e",
            "PYTHONDONTWRITEBYTECODE=1",
            "-e",
            "PYTHONPATH=/test-deps",
            "-v",
            str(ROOT / "coulomb_social") + ":/app/coulomb_social:ro",
            "-v",
            str(ROOT / "tests") + ":/app/tests:ro",
            "-v",
            sysconfig.get_path("purelib") + ":/test-deps:ro",
            app,
            "-c",
            CODE,
        )
        for identity in (database, runner):
            metadata = json.loads(docker("inspect", identity))[0]
            if metadata["HostConfig"].get("PortBindings"):
                raise RuntimeError("published_ports_refused")
        if not json.loads(docker("network", "inspect", network))[0]["Internal"]:
            raise RuntimeError("internal_network_required")
        result = docker("start", "--attach", runner)
        if docker("inspect", runner, "--format", "{{.State.ExitCode}}") != "0":
            raise RuntimeError("rehearsal_failed")
        proof = json.loads(result.splitlines()[-1])
    finally:
        # Delete only IDs returned by this invocation, including their anonymous
        # test volumes. Retained recovery containers and networks are unrelated.
        for identity in (runner, database):
            if identity:
                docker("rm", "--force", "--volumes", identity)
        if network:
            docker("network", "rm", network)
    files = [
        Path(__file__),
        ROOT / "uv.lock",
        ROOT / "coulomb_social/apps/spaces/import_tree.py",
        ROOT / "coulomb_social/apps/spaces/import_receipts.py",
        ROOT / "coulomb_social/apps/spaces/import_inspection.py",
        ROOT / "coulomb_social/apps/spaces/import_reconciliation.py",
        ROOT / "coulomb_social/apps/spaces/management/commands/inspect_content_corpus.py",
        ROOT / "coulomb_social/apps/spaces/management/commands/reconcile_content_import.py",
        *(
            ROOT / "tests" / name
            for name in (
                "test_import_corpus.py",
                "test_import_receipts.py",
                "test_import_inspection.py",
                "test_import_reconciliation.py",
            )
        ),
    ]
    print(
        json.dumps(
            {
                "schema": "coulomb.corpus-inspection-rehearsal.v1",
                "proof": proof,
                "app_dependency_image": app,
                "postgres_image": pg,
                "internal_network": True,
                "published_ports": False,
                "disposable_resources_removed": True,
                "source_sha256": {
                    str(p.relative_to(ROOT)): hashlib.sha256(p.read_bytes()).hexdigest()
                    for p in files
                },
            },
            indent=2,
        )
    )


if __name__ == "__main__":
    main()