cms-SiteForge/scripts/verify_pg_bootstrap.py

"""Full PostgreSQL integration smoke test. Requires a disposable PostgreSQL DB.

Run only in CI or a dedicated empty database: creates tables and temporary
site/jobs, and validates actual DB API compatibility and worker claims.
"""
# Allow direct `python scripts/name.py` from a fresh source checkout.
import sys as _sys
from pathlib import Path as _Path
_sys.path.insert(0, str(_Path(__file__).resolve().parents[1]))
import os
import uuid
url=os.getenv('SITEFORGE_DATABASE_URL','')
if not url.startswith(('postgresql://','postgres://')):
    raise SystemExit('Set SITEFORGE_DATABASE_URL to a DISPOSABLE PostgreSQL database')
from app import (db,enterprise,factory_api,migration_audit,reusables,release_api,
    cms_pipeline,v8_workflows,v8_templates,wp_migration_v81,wp_migration_v82,
    wp_migration_v83,connectors_v84,connectors_v85,studio_v86,
    site_design_v87,multilingual,jobs,worker)

# Mirror app.main.lifespan bootstrap so CI will discover any broken DDL/UPSERT.
db.init()
for module in (enterprise,factory_api,migration_audit,reusables,release_api,
    cms_pipeline,v8_workflows,v8_templates,wp_migration_v81,wp_migration_v82,
    wp_migration_v83,connectors_v84,connectors_v85,studio_v86,
    site_design_v87,multilingual):
    module.setup()
assert db.one('SELECT 1 AS ready')['ready']==1
assert db.one('SELECT 1 AS ready')['ready']==1
ledger=db.one('SELECT version FROM schema_migrations WHERE version=?',('20261022_001_p1_limits',))
assert ledger,'PostgreSQL migration ledger not created'
sid='pg-'+uuid.uuid4().hex[:12]
job_id=None
sid2=sid+"-second"
try:
    with db.connect() as c:
        c.execute('INSERT INTO sites(id,name,domain,created_at,updated_at) VALUES (?,?,?,?,?)',
                  (sid,'PG Smoke','','2026-01-01','2026-01-01'))
        c.execute('INSERT INTO sites(id,name,domain,created_at,updated_at) VALUES (?,?,?,?,?)',
                  (sid2,'PG Smoke Second','','2026-01-01','2026-01-01'))
        c.execute('INSERT INTO entries(id,site_id,type,title,route,slug,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?)',
                  (uuid.uuid4().hex[:16],sid,'page','Home','/','home','2026-01-01','2026-01-01'))
    assert db.one('SELECT id FROM sites WHERE id=?',(sid,))['id']==sid
    job=jobs.submit(sid,{'mirror_images':False},'pg-smoke','pg-smoke-'+uuid.uuid4().hex[:14])
    job_id=job['id']
    # PostgreSQL claim exercises FOR UPDATE SKIP LOCKED and numeric COUNT rows.
    claimed=worker.claim()
    assert claimed and claimed['id']==job_id and claimed['state']=='running',claimed
    assert worker.claim() is None,'Same-site job was claimed twice'
    worker.complete(claimed,{'ok':True})
    assert db.one('SELECT state FROM build_jobs WHERE id=?',(job_id,))['state']=='done'
    job2=jobs.submit(sid,{'mirror_images':True},'pg-smoke','pg-smoke-'+uuid.uuid4().hex[:14])
    assert jobs.cancel(sid,job2['id'])['state']=='cancelled'
    # Independent connections racing for work must never receive the same job.
    import threading
    keys=[('pg-claim-a',sid),('pg-claim-b',sid2)]
    submitted=[]
    for label,site in keys:
        submitted.append(jobs.submit(site,{'mirror_images':site==sid},'pg-smoke',
                                     label+'-'+uuid.uuid4().hex[:10])['id'])
    barrier=threading.Barrier(2)
    claimed=[]; failures=[]
    def runner():
        try:
            barrier.wait(timeout=8)
            claimed.append(worker.claim())
        except Exception as ex:failures.append(ex)
    threads=[threading.Thread(target=runner) for _ in range(2)]
    for thread in threads:thread.start()
    for thread in threads:thread.join(timeout=12)
    assert not any(t.is_alive() for t in threads),'Concurrent PG claim deadlock'
    assert not failures,failures
    ids={item['id'] for item in claimed if item}
    assert ids==set(submitted),'Concurrent claims were duplicated or lost'
    for item in claimed:worker.complete(item,{'ok':True})
    print('PASS: PostgreSQL schemas, ledger, CRUD, queue claim/complete/cancel, two concurrent claims')
finally:
    with db.connect() as c:
        c.execute('DELETE FROM sites WHERE id=?',(sid,))
        c.execute('DELETE FROM sites WHERE id=?',(sid2,))