cms-SiteForge/tests/test_v818_pg_recovery.py

"""V8.18 recovery safety tests; never touch a real database/server."""
import hashlib
import io
import json
import subprocess
import zipfile
import pytest
from pathlib import Path
from app import backup, db, jobs, enterprise, pg_disaster_recovery as pg

URL='postgresql://backup_user:super-secret@127.0.0.1:5432/empty_restore?sslmode=require'


def test_pg_env_never_passes_secret_in_commands():
    env=pg.connection_env(URL, {'PATH':'/usr/bin'})
    assert env['PGPASSWORD']=='super-secret'
    assert env['PGHOST']=='127.0.0.1'
    assert env['PGSSLMODE']=='require'
    assert env['PGPASSFILE']=='/dev/null'
    with pytest.raises(pg.RecoveryError):pg.connection_env('not a pg url')
    with pytest.raises(pg.RecoveryError):pg.connection_env(URL+'&danger=1')


def test_sqlite_backup_cannot_run_in_postgres_mode(monkeypatch,tmp_path):
    monkeypatch.setattr(db,'POSTGRES',True)
    monkeypatch.setattr(db,'DATA',tmp_path)
    monkeypatch.setattr(db,'DB_FILE',tmp_path/'no-fake.sqlite3')
    with pytest.raises(RuntimeError,match='PostgreSQL'):
        backup.backup()
    assert not (tmp_path/'no-fake.sqlite3').exists()


def _fake_tools(commands):
    def fake_run(args,env,timeout=1800):
        assert 'super-secret' not in ' '.join(args)
        commands.append(args)
        if args[0]=='pg_dump':
            Path(args[-1]).write_bytes(b'PGDMP\0offline-mock')
            return subprocess.CompletedProcess(args,0,b'',b'')
        if args[0]=='psql':
            return subprocess.CompletedProcess(args,0,b'0\n',b'')
        return subprocess.CompletedProcess(args,0,b'',b'')
    return fake_run


def test_backup_verify_and_restore_to_empty_destination(tmp_path,monkeypatch):
    storage=tmp_path/'storage';(storage/'media').mkdir(parents=True)
    (storage/'media'/'logo.png').write_bytes(b'fake-png')
    archive=tmp_path/'external'/'backup.zip';commands=[]
    monkeypatch.setattr(pg,'run_pg',_fake_tools(commands))
    with pytest.raises(pg.RecoveryError,match='Stop all'):
        pg.backup(URL,storage,archive)
    result=pg.backup(URL,storage,archive,confirmed_offline=True)
    assert result['files']==2
    manifest=pg.verify(archive)
    assert manifest['database']=='postgresql'
    assert 'media/logo.png' in manifest['files']
    assert [x[0] for x in commands[:2]]==['pg_dump','pg_restore']
    with pytest.raises(pg.RecoveryError,match='overwrite'):
        pg.backup(URL,storage,archive,confirmed_offline=True)
    target=tmp_path/'restored'
    with pytest.raises(pg.RecoveryError,match='confirmation'):
        pg.restore_new(URL,archive,target)
    restored=pg.restore_new(URL,archive,target,confirmed_offline=True)
    assert restored['media_files']==1
    assert (target/'media'/'logo.png').read_bytes()==b'fake-png'
    assert not (target/'database.dump').exists()
    assert [x[0] for x in commands[-2:]]==['psql','pg_restore']
    with pytest.raises(pg.RecoveryError,match='must not exist'):
        pg.restore_new(URL,archive,target,confirmed_offline=True)


def test_tamper_rejected_before_db_restore(tmp_path,monkeypatch):
    src=tmp_path/'media';src.mkdir()
    z=tmp_path/'good.zip'
    monkeypatch.setattr(pg,'run_pg',_fake_tools([]))
    pg.backup(URL,tmp_path,z,confirmed_offline=True)
    corrupt=tmp_path/'corrupt.zip'
    with zipfile.ZipFile(z) as original, zipfile.ZipFile(corrupt,'w') as out:
        for name in original.namelist():
            payload=original.read(name)
            if name=='database.dump':payload=b'PODMP\0offline-mock'
            out.writestr(name,payload)
    with pytest.raises(pg.RecoveryError,match='Checksum'):
        pg.verify(corrupt)
    assert not (tmp_path/'restored').exists()


def test_traversal_and_symlink_refused(tmp_path,monkeypatch):
    src=tmp_path/'storage';src.mkdir()
    (src/'media').symlink_to(tmp_path)
    monkeypatch.setattr(pg,'run_pg',_fake_tools([]))
    with pytest.raises(pg.RecoveryError,match='Symlink'):
        pg.backup(URL,src,tmp_path/'bad.zip',confirmed_offline=True)
    evil=tmp_path/'evil.zip'
    with zipfile.ZipFile(evil,'w') as z:
        z.writestr('manifest.json','{}')
        z.writestr('database.dump',b'PGDMP')
        z.writestr('../hijack',b'evil')
    with pytest.raises(pg.RecoveryError,match='Unsafe'):
        pg.verify(evil)


def test_expired_release_is_never_auto_retried(tmp_path,monkeypatch):
    monkeypatch.setattr(db,'POSTGRES',False)
    monkeypatch.setattr(db,'DATA',tmp_path)
    monkeypatch.setattr(db,'DB_FILE',tmp_path/'siteforge.sqlite3')
    db.init(); enterprise.setup()
    db.insert('INSERT INTO sites(id,name,created_at,updated_at) VALUES (?,?,?,?)',
              ('site818','Sample',db.now(),db.now()))
    for jobid,kind in [('release818','remote_deploy'),('roll818','remote_rollback'),('build818','build')]:
        db.insert('''INSERT INTO build_jobs(id,site_id,kind,state,created_by,created_at,
                    attempts,max_attempts,worker_id,lease_until) VALUES (?,?,?,?,?,?,?,?,?,?)''',
                  (jobid,'site818',kind,'running','system',db.now(),1,3,'dead-worker','2020-01-01T00:00:00+00:00'))
    assert jobs.recover_expired()==3
    assert db.one('SELECT state FROM build_jobs WHERE id=?',('release818',))['state']=='failed'
    assert db.one('SELECT state FROM build_jobs WHERE id=?',('roll818',))['state']=='failed'
    assert db.one('SELECT state FROM build_jobs WHERE id=?',('build818',))['state']=='retrying'
    assert 'reconcile' in db.one('SELECT error FROM build_jobs WHERE id=?',('release818',))['error']


def test_unknown_remote_outcome_is_never_retried(tmp_path,monkeypatch):
    from app import worker
    monkeypatch.setattr(db,'POSTGRES',False)
    monkeypatch.setattr(db,'DATA',tmp_path)
    monkeypatch.setattr(db,'DB_FILE',tmp_path/'siteforge.sqlite3')
    db.init();enterprise.setup()
    db.insert('INSERT INTO sites(id,name,created_at,updated_at) VALUES (?,?,?,?)',
              ('release-site','Sample',db.now(),db.now()))
    for jid in ('release-retry','release-cancel'):
        db.insert('''INSERT INTO build_jobs(id,site_id,kind,state,created_by,created_at,
                    attempts,max_attempts,worker_id,lease_until) VALUES (?,?,?,?,?,?,?,?,?,?)''',
                  (jid,'release-site','remote_deploy','running','system',db.now(),1,3,
                   worker.WORKER_ID,'2030-01-01T00:00:00+00:00'))
    old=db.one('SELECT * FROM build_jobs WHERE id=?',('release-retry',))
    worker.complete(old,error='network error',retryable=True)
    assert db.one('SELECT state FROM build_jobs WHERE id=?',('release-retry',))['state']=='failed'
    old=db.one('SELECT * FROM build_jobs WHERE id=?',('release-cancel',))
    worker.complete(old,cancelled=True)
    state=db.one('SELECT state,error FROM build_jobs WHERE id=?',('release-cancel',))
    assert state['state']=='failed' and 'reconcile' in state['error']