2026-10-23 12:31:03 +08:00
|
|
|
"""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]))
|
2026-10-22 16:49:25 +08:00
|
|
|
import os
|
|
|
|
|
import uuid
|
2026-10-23 12:31:03 +08:00
|
|
|
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)
|
2026-10-22 16:49:25 +08:00
|
|
|
|
2026-10-23 12:31:03 +08:00
|
|
|
# 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,))
|