railiance-platform/tests/test_root_custody_finish.py

import configparser
import importlib.util
import json
from pathlib import Path
import sys
import tempfile
import subprocess
import os
import shutil
import unittest
from unittest.mock import patch

ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'scripts'))
import root_custody_finish as finish
import root_grant_attended as owner
spec=importlib.util.spec_from_file_location('validator',ROOT.parent/'net-kingdom/tools/custody-record/validate.py')
validator=importlib.util.module_from_spec(spec);spec.loader.exec_module(validator)

class FinishTests(unittest.TestCase):
    def test_retained_source_copy_contains_latest_valid_history(self):
        path=ROOT/'docs/evidence/root-custody-history/baseline.json'
        state,bundle=finish.checked_bundle(path,validator)
        self.assertEqual(state['document']['entries'],[])
        self.assertIn(state,json.loads(bundle)['history'])

    def test_current_historical_share_is_classified_without_values(self):
        with tempfile.TemporaryDirectory() as tmp:
            path=Path(tmp)/'rclone.conf'
            path.write_text('[nextcloud-backup]\nurl=https://example.invalid/public.php/dav/filesdrop/TEST-SHARE-VALUE\nuser=TEST-SHARE-VALUE\n')
            result=finish.historical_config_comparison(path,{'NC_WEBDAV_TOKEN':'TEST-SHARE-VALUE'})
            self.assertTrue(result['historical_share_matches_current_lane'])
            self.assertNotIn('TEST-SHARE-VALUE',json.dumps(result))
            result=finish.historical_config_comparison(path,{'NC_WEBDAV_TOKEN':'CURRENT-DIFFERENT'})
            self.assertFalse(result['historical_share_matches_current_lane'])
            self.assertEqual(result['historical_credential_status'],'historical-owner-and-revocation-unverified')

    def test_unverified_upload_destination_refused_before_network(self):
        lane={'data':{'NC_WEBDAV_URL':'https://unapproved.invalid/private','NC_WEBDAV_TOKEN':'TEST'}}
        with patch.object(finish,'request') as network:
            with self.assertRaises(ValueError):
                finish.roundtrip(b'{}',lane,{'BACKUP_USERNAME':'Backup','BACKUP_PASSWORD':'TEST'},network=network)
            network.assert_not_called()

    def test_actual_age_roundtrip_with_disposable_key_and_memory_provider(self):
        key=subprocess.run(['age-keygen'],capture_output=True,check=True).stdout.decode()
        lane={'data':{'NC_WEBDAV_URL':finish.HOST+'/public.php/dav/filesdrop/DISPOSABLE-SHARE',
                     'NC_WEBDAV_TOKEN':'DISPOSABLE-SHARE','AGE_PRIVATE_KEY':key},'metadata':{'version':9}}
        stored={}
        def memory_provider(url,method='GET',body=None,auth=None,headers=None):
            if method=='PUT':
                self.assertEqual(headers['If-None-Match'],'*')
                self.assertEqual(headers['X-Requested-With'],'XMLHttpRequest')
                stored['encrypted']=body
                return 201,b''
            return 200,stored['encrypted']
        with tempfile.TemporaryDirectory() as tmp:
            escrow=Path(tmp)/'escrow.age.key'
            escrow.write_text(key);escrow.chmod(0o600)
            with patch.object(finish,'INDEPENDENT_ESCROW',escrow):
                receipt=finish.roundtrip(b'{"nonsecret":"receipt-fixture"}',lane,
                    {'BACKUP_USERNAME':'Backup','BACKUP_PASSWORD':'DISPOSABLE-PASSWORD'},
                    network=memory_provider,get_quota=lambda auth:{'available_bytes':1048576})
        self.assertTrue(receipt['decrypted_history_matches'])
        self.assertNotIn(key,json.dumps(receipt))
        self.assertNotIn('DISPOSABLE-SHARE',json.dumps(receipt))

    def test_missing_latest_history_cannot_make_backup(self):
        with tempfile.TemporaryDirectory() as tmp:
            path=Path(tmp)/'baseline.json'
            path.write_text((ROOT/'docs/evidence/root-custody-history/baseline.json').read_text())
            (path.parent/'history').mkdir()
            with self.assertRaises(ValueError):
                finish.checked_bundle(path,validator)

class MainIntegrationTests(unittest.TestCase):
    def setUp(self):
        self.temp=tempfile.TemporaryDirectory()
        self.directory=Path(self.temp.name)
        shutil.copytree(ROOT/'docs/evidence/root-custody-history',self.directory/'holder')
        self.state=self.directory/'holder/baseline.json'
        self.receipt=self.directory/'receipt.json'
        self.args=['finish','--state',str(self.state),'--net-kingdom',str(ROOT.parent/'net-kingdom'),'--receipt',str(self.receipt)]
        self.baseline=json.loads(self.state.read_text())
        key=subprocess.run(['/usr/bin/age-keygen'],capture_output=True,check=True).stdout.decode()
        self.lane={'data':{'NC_WEBDAV_URL':finish.HOST+'/public.php/dav/filesdrop/TEST-SHARE',
                   'NC_WEBDAV_TOKEN':'TEST-SHARE','AGE_PRIVATE_KEY':key},'metadata':{'version':4}}
        self.operator={'data':{'BACKUP_USERNAME':'Backup','BACKUP_PASSWORD':'TEST-PASSWORD'}}
        self.escrow=self.directory/'disposable-escrow.key'
        self.escrow.write_text(key);self.escrow.chmod(0o600)
        self.stored={}
    def tearDown(self):self.temp.cleanup()
    def read(self,path):
        if path==finish.LANE:return self.lane
        if path==finish.OPERATOR_PATH:return self.operator
        return {'data':{'document':self.baseline['document']},'metadata':{'version':1}}
    def provider(self,url,method='GET',body=None,auth=None,headers=None):
        if method=='PUT':self.stored['encrypted']=body;return 201,b''
        return 200,self.stored['encrypted']
    def invoke(self,reader=None,provider=None,quota_failure=None,comparison=None):
        # The actual attended-envelope predicate runs with private fake home;
        # only owner I/O boundaries are doubled. System deps and age stay real.
        actor=self.directory/'.warden-attended-login/operator'
        with patch.object(sys,'argv',self.args), patch.object(owner.Path,'home',return_value=actor), \
             patch.dict(os.environ,{'BAO_ADDR':'http://127.0.0.1:18200','PATH':'/usr/bin:/bin'},clear=True), \
             patch.object(finish.Bao,'read',side_effect=reader or self.read), \
             patch.object(finish,'request',side_effect=provider or self.provider), \
             patch.object(finish,'INDEPENDENT_ESCROW',self.escrow), \
             patch.object(finish,'quota',side_effect=quota_failure,return_value={'available_bytes':1048576}), \
             patch.object(finish,'historical_config_comparison',return_value=comparison or {'historical_config_present':False,'historical_share_matches_current_lane':None}):
            return finish.main()
    def test_full_attended_main_success_real_crypto_fake_owner_and_provider(self):
        self.assertEqual(self.invoke(),0)
        receipt=json.loads(self.receipt.read_text())
        self.assertTrue(receipt['ok']);self.assertEqual(receipt['stage'],'complete')
        self.assertTrue(receipt['current_live_baseline_matches'])
        self.assertTrue(receipt['decrypted_history_matches'])
        self.assertEqual(receipt['upload_http_status'],201)
        self.assertNotIn('TEST-PASSWORD',self.receipt.read_text())
    def test_permission_or_backend_error_writes_safe_exact_stage(self):
        def failure(path):
            if path==finish.OPERATOR_PATH:raise RuntimeError('DO-NOT-LOG-CREDENTIAL-OR-BACKEND-TEXT')
            return self.read(path)
        self.assertEqual(self.invoke(reader=failure),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertFalse(receipt['ok']);self.assertEqual(receipt['stage'],'backup_operator_read')
        self.assertTrue(receipt['current_live_baseline_matches'])
        self.assertNotIn('DO-NOT-LOG',self.receipt.read_text())
    def test_upload_failure_has_status_and_artifact_without_raw_error(self):
        self.assertEqual(self.invoke(provider=lambda *args,**kwargs:(403,b'NEVER-LOG-RESPONSE')),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'provider_upload');self.assertEqual(receipt['upload_http_status'],403)
        self.assertIn('artifact',receipt);self.assertNotIn('NEVER-LOG',self.receipt.read_text())
        self.assertEqual(receipt['lane_kv_version'],4)
    def test_download_failure_preserves_successful_created_artifact(self):
        def failure(url,method='GET',body=None,auth=None,headers=None):
            return (201,b'') if method=='PUT' else (401,b'RAW-SECRET')
        self.assertEqual(self.invoke(provider=failure),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'provider_download');self.assertEqual(receipt['upload_http_status'],201)
        self.assertEqual(receipt['readback_http_status'],401);self.assertIn('artifact',receipt)
    def test_quota_failure_has_safe_stage_and_no_secret(self):
        self.assertEqual(self.invoke(quota_failure=RuntimeError('PRIVATE-QUOTA-URL')),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'provider_quota')
        self.assertNotIn('PRIVATE-QUOTA-URL',self.receipt.read_text())
    def test_missing_crypto_binary_is_identified_without_exception_text(self):
        with patch.object(finish.subprocess,'run',side_effect=FileNotFoundError('SECRET-ENV-PATH')):
            self.assertEqual(self.invoke(),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'derive_age_recipient')
        self.assertEqual(receipt['error_code'],'crypto_binary_unavailable')
        self.assertNotIn('SECRET-ENV',self.receipt.read_text())
    def test_ciphertext_mismatch_never_decrypts_or_claims_backup(self):
        def bad_download(url,method='GET',body=None,auth=None,headers=None):
            return (201,b'') if method=='PUT' else (200,b'WRONG-CIPHERTEXT')
        self.assertEqual(self.invoke(provider=bad_download),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'provider_ciphertext_check')
        self.assertEqual(receipt['error_code'],'provider_ciphertext_mismatch')
        self.assertEqual(receipt['upload_http_status'],201)
    def test_decryption_failure_reports_fixed_code_after_matching_download(self):
        original=subprocess.run
        def protected_process(command,*args,**kwargs):
            if command[0]=='age' and command[1]=='-d':
                return subprocess.CompletedProcess(command,1,b'',b'RAW-KEY-MUST-NOT-LOG')
            return original(command,*args,**kwargs)
        with patch.object(finish.subprocess,'run',side_effect=protected_process):
            self.assertEqual(self.invoke(),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'decrypt_and_compare_history')
        self.assertEqual(receipt['error_code'],'crypto_command_failed')
        self.assertEqual(receipt['upload_http_status'],201)
        self.assertNotIn('RAW-KEY-MUST-NOT-LOG',self.receipt.read_text())

    def test_current_incident_stops_before_any_provider_action(self):
        comparison={'historical_config_present':True,'historical_share_matches_current_lane':True,
                    'historical_credential_status':'current-lane-incident'}
        provider=unittest.mock.Mock()
        self.assertEqual(self.invoke(provider=provider,comparison=comparison),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'historical_share_comparison')
        self.assertEqual(receipt['error_code'],'current_backup_lane_incident')
        provider.assert_not_called()
    def test_independent_escrow_missing_does_not_upload(self):
        self.escrow.unlink()
        provider=unittest.mock.Mock()
        self.assertEqual(self.invoke(provider=provider),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'independent_key_escrow')
        self.assertEqual(receipt['error_code'],'independent_key_escrow_missing')
        provider.assert_not_called()

    def test_failed_attended_envelope_writes_safe_receipt(self):
        with patch.object(sys,'argv',self.args),patch.object(finish,'require_attended',side_effect=RuntimeError('TOKEN-DO-NOT-LOG')):
            self.assertEqual(finish.main(),1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'attended_envelope')
        self.assertEqual(receipt['error_code'],'attended_envelope_required')
        self.assertNotIn('TOKEN-DO-NOT-LOG',self.receipt.read_text())
    def test_old_success_receipt_is_preserved(self):
        self.receipt.write_text('{"ok":true}')
        self.assertEqual(self.invoke(),1)
        self.assertEqual(self.receipt.read_text(),'{"ok":true}')
    def test_bootstrap_dependency_import_failure_still_writes_receipt(self):
        entry=self.directory/'root_custody_finish_entry.py'
        shutil.copy2(ROOT/'scripts/root_custody_finish_entry.py',entry)
        (self.directory/'root_custody_finish.py').write_text('raise RuntimeError("IMPORT-MUST-NOT-LOG")')
        result=subprocess.run(['/usr/bin/python3',str(entry),*self.args[1:]],capture_output=True)
        self.assertEqual(result.returncode,1)
        receipt=json.loads(self.receipt.read_text())
        self.assertEqual(receipt['stage'],'helper_import');self.assertEqual(receipt['error_code'],'helper_startup_failed')
        self.assertEqual(result.stdout,b'');self.assertEqual(result.stderr,b'')
        self.assertNotIn('IMPORT-MUST-NOT-LOG',self.receipt.read_text())


if __name__=='__main__':unittest.main()

class PublicDavWireTests(unittest.TestCase):
    def test_public_upload_marker_reaches_http_server_owner_get_unchanged(self):
        from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
        import threading
        import migrate_nextcloud_backup_account as migration
        seen=[]
        class Handler(BaseHTTPRequestHandler):
            def log_message(self,*args):pass
            def do_PUT(self):
                seen.append(dict(self.headers))
                self.rfile.read(int(self.headers['Content-Length']))
                self.send_response(201 if self.headers.get('X-Requested-With')=='XMLHttpRequest' else 401)
                self.end_headers()
            def do_GET(self):
                seen.append(dict(self.headers));self.send_response(200);self.end_headers()
        server=ThreadingHTTPServer(('127.0.0.1',0),Handler)
        worker=threading.Thread(target=server.serve_forever,daemon=True);worker.start()
        origin='http://127.0.0.1:'+str(server.server_port)
        try:
            with patch.object(migration,'HOST',origin):
                code,_=migration.request(origin+'/public.php/dav/filesdrop/FAKE/forgejo/new.age','PUT',b'fake-encrypted-fixture',('FAKE',''))
                self.assertEqual(code,201)
                code,_=migration.request(origin+'/remote.php/dav/files/Backup/fixture','GET',auth=('Backup','FAKE-PASSWORD'))
                self.assertEqual(code,200)
            self.assertEqual(seen[0]['X-Requested-With'],'XMLHttpRequest')
            self.assertNotIn('X-Requested-With',seen[1])
        finally:
            server.shutdown();server.server_close();worker.join()

class ShareInspectionTests(unittest.TestCase):
    def fixture(self):
        return ({'data':{'NC_WEBDAV_URL':finish.HOST+'/public.php/dav/filesdrop/TEST-TOKEN','NC_WEBDAV_TOKEN':'TEST-TOKEN'},'metadata':{'version':3}},
                {'BACKUP_USERNAME':'Backup','BACKUP_PASSWORD':'DO-NOT-OUTPUT'},
                [{'label':finish.LABEL,'token':'TEST-TOKEN','uid_owner':'Backup','path':finish.FOLDER,'permissions':4,'share_type':3,'password':None,'expiration':None}])
    def test_existing_share_scope_and_probe_routes_no_mutation_or_secret_output(self):
        lane,operator,shares=self.fixture();calls=[]
        def probe(url,method,body,auth,headers):
            calls.append((url,method,auth));self.assertEqual(method,'PROPFIND')
            self.assertEqual(headers['Depth'],'0');self.assertEqual(headers['X-Requested-With'],'XMLHttpRequest')
            return (401,b'RAW-MUST-NOT-OUTPUT') if '/filesdrop/' in url else (403,b'RAW-MUST-NOT-OUTPUT')
        result=finish.inspect_share(lane,operator,inventory=lambda *args:shares,network=probe)
        self.assertTrue(result['current_share_create_only']);self.assertTrue(result['current_share_owner_matches'])
        self.assertEqual(result['current_route_classification'],'authentication_rejected')
        self.assertEqual(result['documented_route_classification'],'capability_denied')
        self.assertIsNone(calls[-1][2]);self.assertFalse(result['provider_mutations_performed'])
        for secret in ('TEST-TOKEN','DO-NOT-OUTPUT','RAW-MUST-NOT-OUTPUT'):
            self.assertNotIn(secret,json.dumps(result))
    def test_missing_and_ambiguous_share_do_not_assume_authority(self):
        lane,operator,shares=self.fixture()
        for inventory in ([],shares+shares):
            result=finish.inspect_share(lane,operator,inventory=lambda *args:inventory,network=lambda *args:(401,b''))
            self.assertFalse(result['current_token_unique']);self.assertNotIn('current_share_create_only',result)
    def test_password_and_expiry_indicators_do_not_emit_values(self):
        lane,operator,shares=self.fixture();shares[0].update(password='PRIVATE-HASH',expiration='2000-01-01T00:00:00+02:00')
        result=finish.inspect_share(lane,operator,inventory=lambda *args:shares,network=lambda *args:(401,b''))
        self.assertTrue(result['current_share_password_indicator']);self.assertTrue(result['current_share_expired'])
        self.assertTrue(result['public_route_probes_skipped'])
        self.assertNotIn('PRIVATE-HASH',json.dumps(result))
    def test_main_diagnostic_cannot_call_upload(self):
        case=MainIntegrationTests('test_full_attended_main_success_real_crypto_fake_owner_and_provider')
        case.setUp()
        try:
            case.args.append('--inspect-share')
            lane,operator,shares=self.fixture()
            calls=[]
            def network(*args):calls.append(args);return (403,b'')
            with patch.object(finish,'ocs',return_value=shares),patch.object(finish,'roundtrip',side_effect=AssertionError('UPLOAD-FORBIDDEN')):
                self.assertEqual(case.invoke(provider=network),0)
            receipt=json.loads(case.receipt.read_text())
            self.assertTrue(receipt['ok']);self.assertFalse(receipt['backup_roundtrip_proved'])
            self.assertEqual(receipt['scope'],'read-only-current-governed-share-inspection')
            self.assertTrue(all(x[1]=='PROPFIND' for x in calls))
        finally:case.tearDown()

    def test_naive_expiry_and_missing_password_remain_unknown(self):
        lane,operator,shares=self.fixture();shares[0].pop('password')
        shares[0]['expiration']='2000-01-01 00:00:00'
        result=finish.inspect_share(lane,operator,inventory=lambda *args:shares,network=lambda *args:(403,b''))
        self.assertIsNone(result['current_share_expired'])
        self.assertFalse(result['current_share_expiry_clock_basis_known'])
        self.assertIsNone(result['current_share_password_indicator'])
        self.assertFalse(result['current_share_password_indicator_known'])
    def test_clearly_protected_or_expired_share_skips_public_auth_retries(self):
        for protection in ('password','expiration'):
            lane,operator,shares=self.fixture()
            shares[0][protection]='PRIVATE' if protection=='password' else '2000-01-01T00:00:00-02:00'
            network=unittest.mock.Mock(side_effect=AssertionError('NO-AUTH-RETRY'))
            result=finish.inspect_share(lane,operator,inventory=lambda *args:shares,network=network)
            self.assertTrue(result['public_route_probes_skipped']);network.assert_not_called()

class ShareRepairTests(unittest.TestCase):
    def setUp(self):
        self.case=MainIntegrationTests('test_full_attended_main_success_real_crypto_fake_owner_and_provider');self.case.setUp()
        self.share={'id':'42','label':finish.LABEL,'token':'NEWTESTTOKEN','uid_owner':'Backup','path':finish.FOLDER,'permissions':4,'share_type':3,'password':None,'expiration':None}
        self.existing=[];self.ocs_calls=[];self.write_calls=[]
    def tearDown(self):self.case.tearDown()
    def inventory(self,auth,suffix='',method='GET',fields=None):
        self.ocs_calls.append((suffix,method,fields))
        if method=='POST':self.existing=[self.share];return dict(self.share)
        if suffix:return [dict(self.share)]
        return self.existing
    def provider(self,url,method='GET',body=None,auth=None,headers=None):
        if method=='PROPFIND':return 207,b''
        if method=='PUT':
            self.assertIn('/public.php/dav/files/NEWTESTTOKEN/',url)
            self.assertEqual(headers['X-Requested-With'],'XMLHttpRequest')
        return self.case.provider(url,method,body,auth,headers)
    def write(self,path,payload):
        self.write_calls.append(payload)
        self.assertEqual(payload['options']['cas'],4)
        self.assertEqual(payload['data']['AGE_PRIVATE_KEY'],self.case.lane['data']['AGE_PRIVATE_KEY'])
        self.case.lane={'data':payload['data'],'metadata':{'version':5}}
        return {'data':{'version':5}}
    def invoke(self,provider=None):
        self.case.args += ['--repair-share','--expected-lane-version','4','--confirm-repair','RESTORE Backup CREATE ONLY']
        with patch.object(finish,'ocs',side_effect=self.inventory),patch.object(finish.Bao,'write',side_effect=self.write):
            return self.case.invoke(provider=provider or self.provider)
    def test_full_main_create_prove_then_coordinates_only_cas_real_crypto(self):
        self.assertEqual(self.invoke(),0)
        receipt=json.loads(self.case.receipt.read_text())
        self.assertTrue(receipt['provider_roundtrip_verified_before_cas']);self.assertEqual(receipt['published_lane_kv_version'],5)
        self.assertTrue(receipt['escrow_and_unrelated_fields_preserved']);self.assertFalse(receipt['runtime_consumers_refreshed'])
        self.assertEqual(len(self.write_calls),1)
        self.assertFalse(any(x[1]=='DELETE' for x in self.ocs_calls))
        self.assertNotIn('NEWTESTTOKEN',json.dumps(receipt))
    def test_provider_failure_preserves_created_share_refuses_cas(self):
        def bad_provider(*args):return (207,b'') if args[1]=='PROPFIND' else (401,b'')
        self.assertEqual(self.invoke(provider=bad_provider),1)
        receipt=json.loads(self.case.receipt.read_text())
        self.assertEqual(receipt['replacement_share_id'],42);self.assertTrue(receipt['replacement_share_created'])
        self.assertEqual(receipt['upload_http_status'],401);self.assertEqual(self.write_calls,[])
        self.assertEqual(len(self.existing),1)
    def test_existing_candidate_reused_and_same_coordinates_cas_noop(self):
        self.existing=[self.share]
        self.case.lane['data'].update(NC_WEBDAV_TOKEN='NEWTESTTOKEN',NC_WEBDAV_URL=finish.HOST+'/public.php/dav/files/NEWTESTTOKEN')
        self.assertEqual(self.invoke(),0)
        receipt=json.loads(self.case.receipt.read_text());self.assertTrue(receipt['lane_cas_noop'])
        self.assertEqual(self.write_calls,[]);self.assertFalse(any(x[1]=='POST' for x in self.ocs_calls))
    def test_ambiguous_share_and_stale_version_refuse_mutations(self):
        self.existing=[self.share,self.share]
        self.assertEqual(self.invoke(),1);self.assertEqual(self.write_calls,[])
        self.assertFalse(any(x[1]=='POST' for x in self.ocs_calls))
    def test_cas_failure_keeps_successful_provider_receipt(self):
        with patch.object(finish.Bao,'write',side_effect=RuntimeError('PRIVATE')):
            # invoke's inner patch would shadow ours; patch bounded callback instead.
            original=self.write
            self.write=lambda *args:(_ for _ in ()).throw(RuntimeError('PRIVATE'))
            self.assertEqual(self.invoke(),1)
            self.write=original
        receipt=json.loads(self.case.receipt.read_text())
        self.assertEqual(receipt['stage'],'lane_cas');self.assertTrue(receipt['provider_roundtrip_verified_before_cas'])
        self.assertTrue(receipt['decrypted_history_matches']);self.assertNotIn('PRIVATE',json.dumps(receipt))
    def test_partial_cas_readback_failure_retains_written_version(self):
        original=self.case.read
        def read(path):
            if path==finish.LANE and self.write_calls:raise RuntimeError('PRIVATE')
            return original(path)
        self.case.read=read
        self.assertEqual(self.invoke(),1)
        receipt=json.loads(self.case.receipt.read_text());self.assertEqual(receipt['stage'],'lane_readback')
        self.assertEqual(receipt['published_lane_kv_version'],5);self.assertTrue(receipt['provider_roundtrip_verified_before_cas'])

    def test_stale_expected_version_refuses_before_provider_or_share_mutation(self):
        inventory=unittest.mock.Mock();network=unittest.mock.Mock();bao=unittest.mock.Mock()
        with self.assertRaises(finish.FinishFailure) as raised:
            finish.repair_share(b'{}',self.case.lane,self.case.operator['data'],bao,3,inventory=inventory,network=network)
        self.assertEqual(raised.exception.code,'lane_version_changed')
        inventory.assert_not_called();network.assert_not_called();bao.write.assert_not_called()

class ShareRepairWireTests(unittest.TestCase):
    def test_actual_ocs_form_and_singleton_reread_and_canonical_public_put(self):
        from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
        from urllib.parse import parse_qs
        import threading
        import migrate_nextcloud_backup_account as migration
        calls=[]
        share={'id':'42','token':'DISPOSABLE'}
        class Handler(BaseHTTPRequestHandler):
            def log_message(self,*args):pass
            def respond(self,value):
                self.send_response(200);self.end_headers()
                self.wfile.write(json.dumps({'ocs':{'meta':{'statuscode':100},'data':value}}).encode())
            def do_POST(self):
                fields=parse_qs(self.rfile.read(int(self.headers['Content-Length'])).decode(),keep_blank_values=True)
                calls.append(('POST',self.path,fields,dict(self.headers)));self.respond(share)
            def do_GET(self):
                calls.append(('GET',self.path,None,dict(self.headers)));self.respond([share])
            def do_PUT(self):
                self.rfile.read(int(self.headers['Content-Length']))
                calls.append(('PUT',self.path,None,dict(self.headers)))
                self.send_response(201);self.end_headers()
        server=ThreadingHTTPServer(('127.0.0.1',0),Handler)
        worker=threading.Thread(target=server.serve_forever,daemon=True);worker.start()
        origin='http://127.0.0.1:'+str(server.server_port)
        try:
            with patch.object(migration,'HOST',origin):
                created=migration.ocs(('Backup','FAKE'),method='POST',fields={'path':finish.FOLDER,'shareType':3,'permissions':4,'label':finish.LABEL,'sendMail':'false','expireDate':''})
                self.assertEqual(created,share)
                self.assertEqual(migration.ocs(('Backup','FAKE'),'/42'),[share])
                code,_=migration.request(origin+'/public.php/dav/files/DISPOSABLE/forgejo/new.age','PUT',b'cipher',('DISPOSABLE',''),{'If-None-Match':'*'})
                self.assertEqual(code,201)
            self.assertEqual(calls[0][2],{'path':[finish.FOLDER],'shareType':['3'],'permissions':['4'],'label':[finish.LABEL],'sendMail':['false'],'expireDate':['']})
            self.assertEqual({k.lower():v for k,v in calls[0][3].items()}['ocs-apirequest'],'true')
            self.assertTrue(calls[1][1].endswith('/shares/42'))
            self.assertEqual(calls[2][3]['X-Requested-With'],'XMLHttpRequest')
            self.assertEqual(calls[2][3]['If-None-Match'],'*')
        finally:server.shutdown();server.server_close();worker.join()