........................................................................ [ 8%] ........................................F............................... [ 17%] ........................................................................ [ 26%] ....F................................................................... [ 35%] ........................................................................ [ 44%] ........................................................................ [ 52%] ........................................................................ [ 61%] ........................................................................ [ 70%] ........................................................................ [ 79%] ..................................................F..................... [ 88%] ........................................................................ [ 96%] .......................... [100%] =================================== FAILURES =================================== _____ test_cancellation_and_correction_receipts_survive_format12_recovery ______ app = tmp_path = PosixPath('/private/var/folders/kn/1_8nby0s6m7g8b2k6grh24wm0000gn/T/pytest-of-sandyrestall/pytest-73/test_cancellation_and_correcti0') def test_cancellation_and_correction_receipts_survive_format12_recovery(app,tmp_path): c,j,op,p=prepared(app);action(c,j,op,p);edit=proposal(p);action(c,j,op,edit,'apply-correction') cancelled=str(uuid.uuid4());action(c,j,cancelled,{'binding':p['binding']},'cancel') never=proposal(p,2);action(c,j,op,{'binding':p['binding'],'edit_id':never['edit_id']},'resolve-correction') with app.state.store.tenant('oak') as db:before=[tuple(r) for r in db.execute('SELECT * FROM operations ORDER BY key')] archive=tmp_path/'lifecycle.zip';b=recovery.backup(app.state.store.root,archive);assert recovery.verify(archive,b['sha256'])['valid'] target=tmp_path/'restore';recovery.restore(archive,b['sha256'],target,confirm_fresh=True);restored=create_app(target) with restored.state.store.tenant('oak') as db: assert [tuple(r) for r in db.execute('SELECT * FROM operations ORDER BY key')]==before assert db.execute('SELECT COUNT(*) FROM field_logs').fetchone()[0]==1 assert db.execute('SELECT COUNT(*) FROM field_log_history').fetchone()[0]==2 > assert json.loads((target/'RESTORE-REPORT.json').read_text())['recovery_format']==13 E assert 14 == 13 tests/test_device_note_lifecycle.py:147: AssertionError __ test_format_eleven_migrates_without_fabricated_intake_or_photo_permission ___ tmp_path = PosixPath('/private/var/folders/kn/1_8nby0s6m7g8b2k6grh24wm0000gn/T/pytest-of-sandyrestall/pytest-73/test_format_eleven_migrates_wi0') def test_format_eleven_migrates_without_fabricated_intake_or_photo_permission(tmp_path): root=tmp_path/'old';application=create_app(root);owner=client(application);j=job(owner);p=reviewed(owner,j) before=table_rows(root,'oak');original=tmp_path/'original.zip';recovery.backup(root,original) def old_ledger(db): assert db.execute('SELECT COUNT(*) FROM customer_transfer_jobs').fetchone()[0]==0 pinned=json.loads((Path(__file__).resolve().parents[1]/'docs/recovery-schema-v11.json').read_text()) sql=next(r[3] for r in pinned['tenant'] if r[1]=='customer_transfer_jobs') db.execute('DROP TABLE customer_transfer_jobs');db.execute(sql) def make_old(files,meta): meta.update(version=11,prototype_version='0.14.0') for name,data in list(files.items()): if not name.startswith('businesses/') or not name.endswith('.sqlite3'):continue path=tmp_path/(name.split('/')[1]+'.sqlite3');path.write_bytes(data) with sqlite3.connect(path) as db: for table in TABLES:db.execute('DROP TABLE '+table) old_ledger(db) files[name]=path.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest()) > old=tmp_path/'format11.zip';sha=rewrite_archive(original,old,make_old);assert recovery.verify(old,sha)['valid'] ^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_enquiry_recovery.py:120: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ backend/recovery.py:442: in verify meta,summaries=unpack_validated(archive,sha256,Path(temporary)) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ backend/recovery.py:428: in unpack_validated summary=tenant_summary(stage/prefix/'business.sqlite3',bid,meta['dataset_id'],sanitised=True,format_version=meta['version']) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ backend/recovery.py:128: in tenant_summary check_db(db,expected_schemas(format_version)[1]) backend/recovery.py:117: in check_db require(schema(db)==expected,'Database schema is unsupported or has unexpected tables, views, indexes or triggers.') _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ condition = False message = 'Database schema is unsupported or has unexpected tables, views, indexes or triggers.' def require(condition,message): > if not condition:raise RecoveryError(message) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ E backend.recovery.RecoveryError: Database schema is unsupported or has unexpected tables, views, indexes or triggers. backend/recovery.py:43: RecoveryError _________ test_four_business_reports_and_exports_survive_disappearance _________ tmp_path = PosixPath('/private/var/folders/kn/1_8nby0s6m7g8b2k6grh24wm0000gn/T/pytest-of-sandyrestall/pytest-73/test_four_business_reports_and0') def test_four_business_reports_and_exports_survive_disappearance(tmp_path): > result = run_reporting_drill(tmp_path) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_reporting_recovery.py:80: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ tmp_path = PosixPath('/private/var/folders/kn/1_8nby0s6m7g8b2k6grh24wm0000gn/T/pytest-of-sandyrestall/pytest-73/test_four_business_reports_and0') def run_reporting_drill(tmp_path): source = tmp_path / 'fictional'; application = create_app(source); owner = client(application) clients = {}; ids = {} for n, slug in enumerate(('oak-and-vale', 'harbour-homeworks', 'willow-reporting', 'cedar-reporting')): if n < 2: c = client(application, n); bid = DEMO_BUSINESSES[n]['id'] else: business = new_business(owner, slug).json(); bid = business['business_id'] switched = write(owner, '/api/business/switch', dict(membership_id=business['membership_id'])) owner.headers['x-csrf-token'] = switched.json()['csrf']; c = owner j, people, clock, tid = captured(c, application) entry(c, j, people[0], work_date='2026-10-15', duration_minutes=30) material(c, j, quantity='2', cost_price='12.34', status='received', received_date='2026-10-15') q = quote(c, j, 10000) assert act(c, 'quote', q, 'share').status_code == 200 assert act(c, 'quote', q, 'accept').status_code == 200 i = act(c, 'quote', q, 'invoice').json() assert act(c, 'invoice', i, 'issue').status_code == 200 assert write(c, '/api/invoices/' + i['id'] + '/payments', dict(amount_pence=1000, paid_on='2026-10-15')).status_code == 200 edit(c, j, status='completed', completed_date='2026-10-15', tags=['reviewed', 'repeat']) target = job(c, name='Retained Fictional Customer'); twin = job(c) assert confirm(c, preview(c, selection(c, j, target))).status_code == 200 clients[bid] = c; ids[bid] = dict(original=j['customer_id'], retained=target['customer_id'], distinct=twin['customer_id']) observation = now(); filters = Filters(start='2026-01-01', end='2026-12-31'); before = {}; files = {} for bid in clients: with application.state.store.tenant(bid) as db: db.execute('PRAGMA query_only=ON'); db.execute('BEGIN') result = build_report(db, filters, generated_at=observation) assert result['totals']['issued_invoice_value_pence'] == 10000 assert result['totals']['recorded_receipts_pence'] == 1000 assert result['totals']['issued_invoice_balance_pence'] == 9000 assert result['totals']['person_minutes'] == 32 and result['totals']['internal_labour_cost_pence'] == 1067 assert result['totals']['received_material_cost_pence'] == 2468 assert len(result['options']['customers']) == 3 and len(result['groups']['customer']) == 2 choice = ExportFilters(**filters.model_dump(), snapshot=result['snapshot'], dimension='customer') before[bid] = result; files[bid] = export_csv(result, choice, bid) archive = tmp_path / 'reporting.zip'; saved = recovery.backup(source, archive) > assert recovery.verify(archive, saved['sha256'])['valid'] and recovery.VERSION == 13 E assert (True and 14 == 13) E + where 14 = recovery.VERSION tests/test_reporting_recovery.py:56: AssertionError =========================== short test summary info ============================ FAILED tests/test_device_note_lifecycle.py::test_cancellation_and_correction_receipts_survive_format12_recovery FAILED tests/test_enquiry_recovery.py::test_format_eleven_migrates_without_fabricated_intake_or_photo_permission FAILED tests/test_reporting_recovery.py::test_four_business_reports_and_exports_survive_disappearance 3 failed, 815 passed, 2 warnings in 274.19s (0:04:34)