railiance-apps/tools/coulomb_storage_probe.py
tegwick 45d6178b2e Prepare supervised Coulomb storage migration release and guarded node execution
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
2026-10-29 23:35:58 +01:00

130 lines
4.6 KiB
Python

"""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
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),
"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))