cms-SiteForge/tests/test_worker_queue_v6.py

"""New V6 async contracts: these tests do not use legacy sync adapters."""
import json
from fastapi.testclient import TestClient
import pytest
from app import db,builder,jobs,worker,enterprise
from app.main import app

@pytest.fixture
def web(tmp_path,monkeypatch):
    monkeypatch.setattr(db,'DATA',tmp_path)
    monkeypatch.setattr(db,'DB_FILE',tmp_path/'siteforge.sqlite3')
    monkeypatch.setenv('SITEFORGE_ADMIN_PASSWORD','test-admin-password-for-queue')
    monkeypatch.setenv('SITEFORGE_SESSION_SECRET','test-session-secret-for-queue-2026-much-longer')
    with TestClient(app) as c:
        assert c.post('/api/auth/login',json={'password':'test-admin-password-for-queue'}).status_code==200
        c.headers['X-SiteForge-CSRF']=c.get('/api/auth/me').json()['csrf']
        sid=c.post('/api/sites',json={'name':'Queued Site','theme':'corporate'}).json()['id']
        yield c,sid

def test_api_enqueue_idempotency_cancel_and_version(web):
    c,sid=web
    from app.version import VERSION
    assert c.get('/api/health').json()['version']==VERSION
    key='same-request-unique-123'
    r=c.post(f'/api/sites/{sid}/builds',json={'mirror_images':False},headers={'Idempotency-Key':key})
    assert r.status_code==202,r.text
    j=r.json()['id']
    duplicate=c.post(f'/api/sites/{sid}/builds',json={'mirror_images':False},headers={'Idempotency-Key':key})
    assert duplicate.json()['id']==j and duplicate.json()['existing']
    conflict=c.post(f'/api/sites/{sid}/builds',json={'mirror_images':True},headers={'Idempotency-Key':key})
    assert conflict.status_code==409
    cancelling=c.post(f'/api/enterprise/sites/{sid}/jobs/{j}/cancel')
    assert cancelling.status_code==200 and cancelling.json()['state']=='cancelled'
    assert worker.run_once(executor=lambda sid,options:builder.build(sid)) is None
    assert c.get(f'/api/enterprise/sites/{sid}/jobs/{j}').json()['state']=='cancelled'

def test_only_worker_generates_static_build(web):
    c,sid=web
    start=c.get(f'/api/sites/{sid}/builds').json()
    submitted=c.post(f'/api/sites/{sid}/builds',json={})
    assert submitted.status_code==202
    assert c.get(f'/api/sites/{sid}/builds').json()==start
    job=submitted.json()['id']
    assert worker.run_once(executor=lambda sid,opts:builder.build(sid,bool(opts.get('mirror_images',False))))==job
    record=c.get(f'/api/enterprise/sites/{sid}/jobs/{job}').json()
    assert record['state']=='done',record['error']
    version=json.loads(record['result_json'])['version']
    assert c.get(f'/preview/{sid}/{version}/').status_code==200

def test_failed_job_retries_and_lease_expiry(web):
    c,sid=web
    job=c.post(f'/api/sites/{sid}/builds',json={}).json()['id']
    attempts=[]
    def fail_once(sid,opts):
        attempts.append(sid)
        if len(attempts)==1:raise TimeoutError('transient timeout')
        return {'version':'ok','pages':1}
    worker.run_once(executor=fail_once)
    row=db.one('SELECT * FROM build_jobs WHERE id=?',(job,))
    assert row['state']=='retrying' and row['attempts']==1
    db.execute('UPDATE build_jobs SET available_at=? WHERE id=?',(db.now(),job))
    worker.run_once(executor=fail_once)
    assert db.one('SELECT state,attempts FROM build_jobs WHERE id=?',(job,))=={'state':'done','attempts':2}
    second=c.post(f'/api/sites/{sid}/builds',json={}).json()['id']
    db.execute("UPDATE build_jobs SET state='running',worker_id='lost',lease_until='2020-01-01T00:00:00+00:00' WHERE id=?",(second,))
    assert jobs.recover_expired()==1
    assert db.one('SELECT state FROM build_jobs WHERE id=?',(second,))['state']=='retrying'

def test_real_subprocess_worker(web):
    c,sid=web
    jid=c.post(f'/api/sites/{sid}/builds',json={}).json()['id']
    worker.run_once()  # real Python subprocess, reads same test SQLite database
    row=db.one('SELECT * FROM build_jobs WHERE id=?',(jid,))
    assert row['state']=='done',row['error']
    report=json.loads(row['result_json'])
    assert report['pages']>=1

def test_cancel_actively_running_subprocess(web,monkeypatch):
    import subprocess,sys,threading,time
    c,sid=web
    jobid=c.post(f'/api/sites/{sid}/builds',json={}).json()['id']
    original=subprocess.Popen
    monkeypatch.setattr(worker.subprocess,'Popen',lambda *args,**kwargs:original(
        [sys.executable,'-c','import time;time.sleep(10)'],**kwargs))
    t=threading.Thread(target=worker.run_once)
    t.start()
    for _ in range(100):
        row=db.one('SELECT state FROM build_jobs WHERE id=?',(jobid,))
        if row and row['state']=='running':break
        time.sleep(0.02)
    r=c.post(f'/api/enterprise/sites/{sid}/jobs/{jobid}/cancel')
    assert r.status_code==200
    t.join(timeout=6)
    assert not t.is_alive()
    assert db.one('SELECT state FROM build_jobs WHERE id=?',(jobid,))['state']=='cancelled'


def test_build_timeout_sets_retrying(web,monkeypatch):
    import subprocess,sys
    c,sid=web
    jobid=c.post(f'/api/sites/{sid}/builds',json={}).json()['id']
    db.execute('UPDATE build_jobs SET timeout_seconds=1 WHERE id=?',(jobid,))
    original=subprocess.Popen
    monkeypatch.setattr(worker.subprocess,'Popen',lambda *args,**kwargs:original(
        [sys.executable,'-c','import time;time.sleep(10)'],**kwargs))
    worker.run_once()
    row=db.one('SELECT state,error FROM build_jobs WHERE id=?',(jobid,))
    assert row['state']=='retrying',row
    assert 'timed out' in row['error']