coulomb-social/scripts/rehearse-storage-migration.py

"""Rehearse migration 0003 on disposable internal-network PostgreSQL.

Uses cached images and read-only source/test dependencies; no published ports,
production credentials, live data, or production migration. Requires Docker,
postgres:16, the deployed dd83c90 app image and the locked local dev environment.
"""

import argparse
import hashlib
import json
import shutil
import subprocess
import sysconfig
import tempfile
import uuid
from pathlib import Path

name = "csoc-migration-" + uuid.uuid4().hex[:10]
root = Path(__file__).resolve().parents[1]
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
    "--app-image",
    default="forgejo.coulomb.social/coulomb/coulomb-social:dd83c90",
)
parser.add_argument(
    "--packaged-source",
    action="store_true",
    help="Use image source/dependencies, injecting only test tools",
)
args = parser.parse_args()


def run(*args):
    return subprocess.run(args, capture_output=True, text=True, check=True).stdout.strip()


app = run(
    "docker",
    "image",
    "inspect",
    args.app_image,
    "--format",
    "{{.Id}}",
)
pg = run("docker", "image", "inspect", "postgres:16", "--format", "{{.Id}}")
code = """import time, json, django
django.setup()
from django.db import connection
for i in range(30):
    try:
        connection.ensure_connection()
        break
    except Exception:
        time.sleep(1)
else:
    raise RuntimeError('isolated_postgres_not_ready')
from django.core.management import call_command
call_command('migrate', verbosity=0)
from tests import test_storage_migration_lifecycle as lifecycle
assert connection.vendor == 'postgresql'
lifecycle.test_forward_refusal_rollback_and_reapply_preserve_rows()
print(json.dumps({
    'status': 'passed', 'database_vendor': connection.vendor,
    'server_version': connection.pg_version, 'forward_collision_refusal': True,
    'refusal_preserves_rows_and_grants': True,
    'rollback_preserves_rows_grants_and_tenant_constraint': True,
    'reapply_enforces_global_constraint': True, 'production_access': False,
}))
"""
tool_directory = None
source_mounts = ["-v", str(root / "coulomb_social") + ":/app/coulomb_social:ro"]
dependency_directory = sysconfig.get_path("purelib")
test_tools = ["pytest", "_pytest", "pluggy", "iniconfig", "pygments", "py.py"]
if args.packaged_source:
    source_mounts = []
    tool_directory = Path(tempfile.mkdtemp(prefix="csoc-migration-test-tools-"))
    for package in test_tools:
        source = Path(dependency_directory) / package
        if source.is_dir():
            shutil.copytree(source, tool_directory / package)
        else:
            shutil.copyfile(source, tool_directory / package)
    # Only public test-tool sources are copied. The image runs as UID10001.
    tool_directory.chmod(0o755)
    dependency_directory = str(tool_directory)
    code = code.replace(
        "django.setup()",
        "assert django.__file__.startswith('/app/.venv/'), 'packaged_django_required'\n"
        "django.setup()",
    )
try:
    run("docker", "network", "create", "--internal", name)
    run(
        "docker",
        "run",
        "-d",
        "--rm",
        "--name",
        name,
        "--network",
        name,
        "--network-alias",
        "storage-db",
        "-e",
        "POSTGRES_HOST_AUTH_METHOD=trust",
        "-e",
        "POSTGRES_DB=storage_rehearsal",
        pg,
    )
    result = run(
        "docker",
        "run",
        "--rm",
        "--network",
        name,
        "--read-only",
        "--tmpfs",
        "/tmp",
        "--entrypoint",
        "/app/.venv/bin/python",
        "-e",
        "DJANGO_SETTINGS_MODULE=coulomb_social.settings.base",
        "-e",
        "DATABASE_URL=postgres://postgres@storage-db/storage_rehearsal",
        "-e",
        "PYTHONDONTWRITEBYTECODE=1",
        *source_mounts,
        "-v",
        str(root / "tests") + ":/app/tests:ro",
        "-v",
        dependency_directory + ":/test-deps:ro",
        "-e",
        "PYTHONPATH=/test-deps",
        app,
        "-c",
        code,
    )
    print(
        json.dumps(
            {
                "app_dependency_image": app,
                "packaged_source_and_django": args.packaged_source,
                "injected_test_tools": test_tools
                if args.packaged_source
                else "locked dev environment",
                "postgres_image": pg,
                "proof": json.loads(result.splitlines()[-1]),
                "source_sha256": {
                    str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest()
                    for p in [
                        root / "tests/test_storage_migration_lifecycle.py",
                        root
                        / "coulomb_social/apps/spaces/migrations/0003_space_slug_global_uniq.py",
                        root / "uv.lock",
                    ]
                },
            },
            indent=2,
        )
    )
except subprocess.CalledProcessError as e:
    print(e.stdout, e.stderr)
    raise
finally:
    subprocess.run(["docker", "rm", "-fv", name], capture_output=True)
    subprocess.run(["docker", "network", "rm", name], capture_output=True)
    if tool_directory is not None:
        shutil.rmtree(tool_directory)