import json import os from pathlib import Path import shutil import sqlite3 import subprocess import threading import time import pytest from vrag.backup import snapshot,restore,digest,db_snapshot,retention_keep,stale_status @pytest.fixture def source(tmp_path,monkeypatch): age=os.environ.get('AGE_BIN') or shutil.which('age') if not age:pytest.skip('Install age to run real encrypted restore tests') monkeypatch.setenv('AGE_BIN',age) keygen=str(Path(age).with_name('age-keygen')) key=tmp_path/'identity';r=subprocess.run([keygen,'-o',str(key)],capture_output=True,check=True) recipient=subprocess.run([keygen,'-y',str(key)],capture_output=True,text=True,check=True).stdout.strip() source=tmp_path/'source';data=source/'data';state=source/'state';assets=tmp_path/'object-store' state.mkdir(parents=True);(data/'videos/v1').mkdir(parents=True);(assets/'videos/v1').mkdir(parents=True) (source/'source-snapshot.json').write_text(json.dumps({'snapshot_id':'test-immutable-v1'})) (state/'config-history.json').write_text('[{"version":1}]') (state/'.env').write_text('NEVER_INCLUDE_THIS') (data/'videos/v1/meta.json').write_text('{"id":"v1","title":"Workshop"}') (data/'videos/v1/video.mp4').write_bytes(b'original private video') (assets/'videos/v1/video.mp4').write_bytes(b'original private video') (data/'live').mkdir();(data/'live/raw-camera.jpg').write_bytes(b'NEVER_INCLUDE_THIS') with sqlite3.connect(data/'jobs.sqlite') as db:db.execute('CREATE TABLE test (n INTEGER)');db.execute('INSERT INTO test VALUES (7)') output=tmp_path/'snapshots/one.age' args=dict(data_root=data,state_root=state,assets_root=assets,output=output,recipient=recipient,source_snapshot_id='test-immutable-v1') return args,key def test_actual_age_roundtrip_and_secret_exclusion(source,tmp_path): args,key=source;report=snapshot(**args) assert report['coverage_verified'] and report['asset_count']==1 raw=args['output'].read_bytes();assert raw.startswith(b'age-encryption.org/') assert b'Workshop' not in raw and b'NEVER_INCLUDE_THIS' not in raw staging=tmp_path/'clean' dry=restore(args['output'],key,staging,args['assets_root'],True) assert dry['verified'] and not staging.exists() result=restore(args['output'],key,staging,args['assets_root']) assert staging.stat().st_mode & 0o777 == 0o700 assert result['verified'] and not (staging/'state/.env').exists() and not (staging/'data/live').exists() assert (staging/'data/videos/v1/video.mp4').read_bytes()==b'original private video' with sqlite3.connect(staging/'data/jobs.sqlite') as db:assert db.execute('SELECT n FROM test').fetchone()==(7,) with pytest.raises(ValueError,match='destination must be new'):restore(args['output'],key,staging,args['assets_root']) @pytest.mark.parametrize('failure',['missing','corrupt','secret','oversize']) def test_no_false_success(source,failure): args,_=source if failure=='missing':(args['assets_root']/'videos/v1/video.mp4').unlink() if failure=='corrupt':(args['assets_root']/'videos/v1/video.mp4').write_bytes(b'wrong') if failure=='secret':(args['data_root']/'videos/v1/meta.json').write_text('{"token":"sk-abcdefghijklmnopqrstuvwxyz"}') if failure=='oversize':(args['data_root']/'videos/v1/meta.json').write_bytes(b'x'*(32*1024*1024+1)) with pytest.raises(ValueError):snapshot(**args) assert not args['output'].exists() def test_corruption_and_missing_asset_block_restore(source,tmp_path): args,key=source;snapshot(**args) (args['assets_root']/'videos/v1/video.mp4').unlink() with pytest.raises(ValueError):restore(args['output'],key,tmp_path/'stage',args['assets_root']) args['output'].write_bytes(args['output'].read_bytes()+b'corrupt') with pytest.raises(ValueError,match='integrity'):restore(args['output'],key,tmp_path/'stage',args['assets_root']) assert not (tmp_path/'stage').exists() def test_sqlite_backup_under_concurrent_writes(tmp_path): path=tmp_path/'live.sqlite' with sqlite3.connect(path) as db:db.execute('CREATE TABLE pairs (n INTEGER)');db.execute('INSERT INTO pairs VALUES (0)');db.execute('INSERT INTO pairs VALUES (0)') stop=threading.Event() def writes(): with sqlite3.connect(path) as db: n=0 while not stop.is_set(): n+=1 with db:db.execute('UPDATE pairs SET n=?',(n,)) thread=threading.Thread(target=writes);thread.start() try: db_snapshot(path,tmp_path/'copy.sqlite') with sqlite3.connect(tmp_path/'copy.sqlite') as db:rows=db.execute('SELECT n FROM pairs').fetchall();assert rows[0]==rows[1] finally:stop.set();thread.join() def test_late_run_and_retention_never_remove_newest_good(): assert stale_status({})['stale'] assert not stale_status({'last_verified_success':100},now=101)['stale'] assert stale_status({'last_verified_success':100},now=100+15*3600)['stale'] assert not stale_status({'last_verified_success':100},now=101)['enabled'] rows=[{'created_at':1000,'coverage_verified':True,'ciphertext_sha256':'good'},{'created_at':2000,'coverage_verified':False,'ciphertext_sha256':'bad'}] assert retention_keep(rows,now=100000000)=={'good'} def test_retention_preserves_last_restore_qualified_copy(): rows=[{'created_at':1,'coverage_verified':True,'restore_verified':True,'ciphertext_sha256':'restored'}, {'created_at':10000000,'coverage_verified':True,'ciphertext_sha256':'latest'}] keep=retention_keep(rows,now=10000001) assert {'restored','latest'}<=keep def test_workflow_is_private_template_and_schedule_is_exact(): import yaml root=Path(__file__).resolve().parent.parent source=root/'deploy/backup-repository/inactive-workflows/backup-12h.yml' config=yaml.load(source.read_text(),Loader=yaml.BaseLoader) assert config['on']['schedule'][0]['cron']=='17 0,12 * * *' assert 'workflow_dispatch' in config['on'] assert config['concurrency']['cancel-in-progress']=='false' assert config['jobs']['checkpoint']['if']=="vars.BACKUPS_APPROVED == 'true' && vars.COORDINATOR_VALIDATED == 'true'" assert not (root/'.github/workflows/backup-12h.yml').exists() assert 'Verify private Space reachability and completed checkpoint' in [s.get('name') for s in config['jobs']['checkpoint']['steps']]