railiance-apps/tools/coulomb_image_probe.py

"""Read-only data/schema and view probe; runs inside the existing app container."""

import hashlib
import json
import sys
import uuid
from pathlib import Path

import django

django.setup()
from coulomb_social.apps.members.models import Member, User
from coulomb_social.apps.spaces.models import Space
from django.conf import settings
from django.db import connection
from django.db.models import Count

assert connection.vendor == "postgresql"
assert connection.settings_dict["NAME"] in {
    "coulomb_social_db",
    "coulomb_social_restore",
}
tables = {}
with connection.cursor() as cursor:
    cursor.execute(
        "SELECT tablename FROM pg_tables WHERE schemaname='public' "
        "AND tablename!='django_migrations' ORDER BY tablename"
    )
    names = [row[0] for row in cursor.fetchall()]
    for table in names:
        cursor.execute("SELECT * FROM " + connection.ops.quote_name(table))
        rows = sorted(
            json.dumps(row, default=str, separators=(",", ":"))
            for row in cursor.fetchall()
        )
        tables[table] = {
            "rows": len(rows),
            "sha256": hashlib.sha256("\n".join(rows).encode()).hexdigest(),
        }
    cursor.execute(
        "SELECT conname FROM pg_constraint WHERE conrelid='spaces_space'::regclass"
    )
    constraints = {row[0] for row in cursor.fetchall()}
    cursor.execute(
        "SELECT COUNT(*) FROM django_migrations WHERE app='spaces' AND name='0003_space_slug_global_uniq'"
    )
    applied = cursor.fetchone()[0] == 1
    cursor.execute("SELECT * FROM django_migrations")
    migration_rows = sorted(json.dumps(row, default=str, separators=(",", ":"))
                            for row in cursor.fetchall())
root = Path(settings.CONTENT_ROOT)
files = {
    str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest()
    for p in root.rglob("*")
    if p.is_file()
}
result = {
    "business_tables": len(tables),
    "migration_rows_sha256": hashlib.sha256("\n".join(migration_rows).encode()).hexdigest(),
    "business_rows": sum(t["rows"] for t in tables.values()),
    "business_rows_sha256": hashlib.sha256(
        json.dumps(tables, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest(),
    "content_files": len(files),
    "content_files_sha256": hashlib.sha256(
        json.dumps(files, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest(),
    "global_slug_constraint": "spaces_space_slug_global_uniq" in constraints,
    "tenant_slug_constraint": "spaces_space_tenant_slug_uniq" in constraints,
    "migration_0003_applied": applied,
    "duplicate_slug_groups": Space.objects.values("slug")
    .annotate(n=Count("pk"))
    .filter(n__gt=1)
    .count(),
}
if len(sys.argv) > 1 and sys.argv[1] == "views":
    from coulomb_social.apps.spaces import views
    from coulomb_social.apps.spaces.content import load_space_page
    from django.test import RequestFactory
    from django.urls import reverse

    assert not settings.FLEX_AUTH_BASE_URL
    factory = RequestFactory()
    rendered_count, denied_count = 0, 0
    for space in Space.objects.filter(is_active=True):
        paths = sorted((root / "spaces" / space.slug / "pages").glob("*.md"))
        if not paths:
            continue
        owner = space.created_by
        if owner is None:
            owner = (
                space.memberships.filter(role="owner")
                .select_related("member__user")
                .first()
                .member
            )
        token = "unsaved-release-check-" + uuid.uuid4().hex
        outsider = Member(
            user=User(username=token),
            tenant_id=space.tenant_id,
            subject=token,
            issuer="https://release-check.invalid",
            user_engine_user_id=token,
        )
        for path in paths:
            page = path.stem
            loaded = load_space_page(space, page)
            assert loaded.error is None

            def request(user, slug=space.slug, page_slug=page):
                req = factory.get(
                    reverse("spaces:detail", kwargs={"slug": slug}),
                    {"page": page_slug},
                )
                req.user, req.session = user, {}
                return req

            response = views.space_detail(request(owner.user), slug=space.slug)
            assert (
                response.status_code == 200 and loaded.html.encode() in response.content
            )
            rendered_count += 1
            denied = views.space_detail(request(outsider.user), slug=space.slug)
            assert denied.status_code == 404
            if loaded.html:
                assert loaded.html.encode() not in denied.content
            denied_count += 1
    assert rendered_count > 0 and denied_count == rendered_count
    result.update(
        actual_owner_pages_rendered=rendered_count,
        unsaved_outsider_page_denials=denied_count,
        production_identity_or_grant_created=False,
    )
print(json.dumps(result))