tests/test_mirror_dataroom.py — Data Room Toolkit

# By Zero Width Lab (zerowidthlab.com)
from __future__ import annotations
import argparse
import contextlib
from io import StringIO
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
__author__ = "Zero Width Lab"

class MirrorDataroomCliTests(unittest.TestCase):
 SESSION ='https://contoso.sharepoint.com/upload-session-token'
 DRY ='mirror_dry_run.json'
 SMALL ='mirror_live_small_put.json'
 CHUNK ='mirror_live_chunked_upload.json'


 def setUp(self):
  self.package_root =Path(__file__).resolve().parent.parent
  self.script =self.package_root / 'scripts' / 'mirror_dataroom.py'
  self.fixtures =Path(__file__).resolve().parent / 'fixtures'
  self.temp_dir =tempfile.TemporaryDirectory()
  self.addCleanup(self.temp_dir.cleanup)
  self.temp_path =Path(self.temp_dir.name).resolve()
  self.source =self.temp_path / 'source'
  self.source.mkdir()
  self.dry_fixture =self.fixtures / self.DRY
  self.small_fixture =self.fixtures / self.SMALL
  self.chunk_fixture =self.fixtures / self.CHUNK
  self.args =argparse.Namespace(request_timeout=1,max_retries=0,throttle=0,chunk_size=327680,verify='none')

 def environment(self,fixture=None):
  env =os.environ.copy()
  for name in ('GRAPH_TENANT_ID','GRAPH_CLIENT_ID','GRAPH_CLIENT_SECRET','GRAPH_FAKE_RESPONSES','GRAPH_FAKE_TRANSPORT_TEST_ACK','SHAREPOINT_DATAROOM_CONFIG','SHAREPOINT_DATAROOM_OPERATOR'):
   env.pop(name,None)
  env.update({'GRAPH_TENANT_ID':'tenant-placeholder','GRAPH_CLIENT_ID':'client-placeholder','GRAPH_CLIENT_SECRET':'secret-placeholder'})
  for proxy in ('HTTPS_PROXY','https_proxy','HTTP_PROXY','http_proxy'):
   env[proxy] ='http://127.0.0.1:9'
  for name in ('NO_PROXY','no_proxy'):
   env.pop(name,None)
  if fixture is not None:
   env['GRAPH_FAKE_RESPONSES'] =str(fixture)
   env['GRAPH_FAKE_TRANSPORT_TEST_ACK'] ='zero-width-lab-test-suite-only'
  return env

 def run_cli(self,*arguments,fixture=None,env=None):
  selected_env =env if env is not None else self.environment(fixture)
  return subprocess.run([sys.executable,str(self.script),'--source',str(self.source),'--dest-path','vault',*arguments],cwd=self.package_root,env=selected_env,stdin=subprocess.DEVNULL,capture_output=True,text=True,check=False,timeout=10)

 def drive(self,*args,**kwargs):
  return self.run_cli('--drive-id','drive-placeholder',*args,**kwargs)

 def check_result(self,result,code=0):
  self.assertEqual(result.returncode,code,result.stderr or result.stdout)

 def live(self,approval,*extra,fixture,flag='--plan-digest'):
  return self.drive('--live','--yes',flag,approval,'--json','--throttle','0',*extra,fixture=fixture)

 def check_fields(self,actual,**expected):
  for key,value in expected.items():
   self.assertEqual(actual[key],value)

 def upload(self,result):
  return json.loads(result.stdout)['uploads'][0]

 def write_alpha(self):
  (self.source / 'alpha.txt').write_text('alpha\n',encoding='utf-8')

 def fixture_copy(self,name,transform):
  payload =json.loads((self.fixtures / name).read_text(encoding='utf-8'))
  transform(payload)
  output =self.temp_path / name
  output.write_text(json.dumps(payload),encoding='utf-8')
  return output

 def approved_digest(self,*options,fixture=None):
  result =self.drive('--dry-run','--json',*options,fixture=fixture or self.dry_fixture)
  self.check_result(result)
  return str(json.loads(result.stdout)['plan_digest'])

 def test_upload_chunk_contract(self):
  module =self.load_module()
  gc =module.graph_client
  self.assertIn('Data Room Toolkit 0.1.5',(self.package_root / 'README.md').read_text())
  self.assertEqual(module.DEFAULT_CHUNK_SIZE,5242880)
  self.assertEqual(gc.CONFIG_DEFAULTS['upload_chunk_bytes'],5242880)
  self.assertEqual(module.DEFAULT_CHUNK_SIZE % 327680,0)
  example =json.loads((self.package_root / 'examples/dataroom.config.example.json').read_text())
  self.assertEqual(example['upload_chunk_bytes'],5242880)
  self.assertIn('self-created temporary fixtures',(self.package_root / 'README.md').read_text())
  with mock.patch.object(gc,'acquire_token') as auth,mock.patch.object(gc,'request') as request:
   for size in (4194304,0,-327680,1,62914560,63242240,True,327680.0,'327680'):
    with self.subTest(size=size),self.assertRaises(gc.ConfigError):
     gc.load_config(None,overrides={'drive_id':'drive','upload_chunk_bytes':size},env={})
   for size in (4194304,0,-1,1,62914560):
    for use_config in (False,True):
     config =self.temp_path / 'chunks.json'
     config.write_text(json.dumps({'drive_id':'drive','upload_chunk_bytes':size}))
     flags =['--config',str(config)] if use_config else ['--drive-id','drive','--chunk-size',str(size)]
     with contextlib.redirect_stderr(StringIO()):
      try:
       result =module.main(['--source',str(self.source),'--dest-path','vault',*flags])
      except SystemExit as exc:
       result =exc.code
      self.assertIn(result,(1,2))
   auth.assert_not_called()
   request.assert_not_called()
  module,local,entry =self.local_entry(b'a' * (327680 * 2 + 3),'ranges.bin')
  def respond(call,calls):
   if call['method'] =='POST':
    return (200,{'uploadUrl':self.SESSION},{})
   body =call['data']
   final =call['headers']['Content-Range'].endswith('/655363') and len(body)==3
   if not final:
    self.assertEqual(len(body) % 327680,0)
   return (201 if final else 202,{'size':655363},{})
  calls =self.recorder(module,respond)
  self.assertEqual(self.apply_entry(module,local,entry)['status'],'uploaded')
  self.assertEqual([len(c['data']) for c in calls if c['method']=='PUT'],[327680,327680,3])

 def test_default_read_only_plan(self):
  self.write_alpha()
  plan_path =self.temp_path / 'default-plan.json'
  result =self.drive('--json','--plan-file',str(plan_path),fixture=self.dry_fixture)
  self.check_result(result)
  audit =json.loads(result.stdout)
  plan =json.loads(plan_path.read_text())
  self.assertEqual(plan['plan_digest'],audit['plan_digest'])
  self.check_fields(plan['target'],drive_id='drive-placeholder',site_id='site-placeholder',destination_path='vault')
  self.check_fields(plan['files'][0],action='create')
  self.check_fields(plan['options'],verify='sha256')
  methods ={(r['method'],r['url_prefix'].split('/')[2]) for r in json.loads(self.dry_fixture.read_text())['responses']}
  self.assertEqual(methods,{('POST','login.microsoftonline.com'),('GET','graph.microsoft.com')})
  self.assertEqual(audit['mode'],'dry-run')
  self.assertEqual(len(audit['plan_digest']),64)
  self.assertEqual(audit['uploads'][0]['status'],'dry-run')
  self.assertEqual(audit['failures'],[])
  changed =json.loads(json.dumps(plan))
  changed['target']['drive_id'] ='drive-other'
  self.assertNotEqual(self.load_module().plan_digest(changed),audit['plan_digest'])

 def test_small_upload_plan_and_audit(self):
  self.write_alpha()
  plan_path,plan =self.dry_run_plan(self.dry_fixture,'--drive-id','drive-placeholder')
  self.assertEqual(plan['files'][0]['source_path'],'alpha.txt')
  audit_path =self.temp_path / 'audit.json'
  result =self.live(str(plan_path),'--audit-log',str(audit_path),flag='--plan-file',fixture=self.small_fixture)
  self.check_result(result)
  audit =json.loads(result.stdout)
  self.check_fields(audit['uploads'][0],action='create',status='uploaded',upload_method='direct')
  self.assertEqual(audit['uploads'][0]['verification']['status'],'passed')
  self.assertEqual(audit['hosts_contacted'],['graph.microsoft.com','login.microsoftonline.com'])
  self.assertEqual(json.loads(audit_path.read_text(encoding='utf-8')),audit)

 def test_chunked_create_and_replace_uploads(self):
  (self.source / 'large.bin').write_bytes(b'abcdefgh')
  options =('--small-threshold','4','--chunk-size','327680')
  for replace in (False,True):
   with self.subTest(replace=replace):
    def transform(payload):
     payload['responses'].insert(1,self.existing_item('vault/large.bin'))
     for response in payload['responses']:
      if response.get('method') =='POST' and 'createUploadSession' in response['url_prefix']:
       response.pop('match_absent_headers',None)
       response.update(match_headers={'If-Match':self.ETAG},match_json={'item':{'@microsoft.graph.conflictBehavior':'replace'}})
    dry =self.with_existing(self.DRY,'vault/large.bin') if replace else self.dry_fixture
    fixture =self.fixture_copy(self.CHUNK,transform) if replace else self.chunk_fixture
    result =self.live(self.approved_digest(*options,fixture=dry),*options,fixture=fixture)
    self.check_result(result)
    upload =self.upload(result)
    self.check_fields(upload,action='replace' if replace else 'create',status='uploaded',upload_method='chunked')
    self.assertEqual(upload['verification']['status'],'passed')
    self.assertEqual(json.loads(result.stdout)['hosts_contacted'],['contoso.sharepoint.com','graph.microsoft.com','login.microsoftonline.com'])

 def test_nested_folder_created_once(self):
  nested =self.source / 'nested'
  nested.mkdir()
  (nested / 'alpha.txt').write_text('alpha\n',encoding='utf-8')
  (nested / 'beta.txt').write_text('beta!\n',encoding='utf-8')
  approved_digest =self.approved_digest()
  result =self.live(approved_digest,fixture=self.small_fixture)
  self.check_result(result)
  uploads =json.loads(result.stdout)['uploads']
  self.assertEqual(len(uploads),2)
  self.assertEqual(uploads[0]['created_folders'],['vault/nested'])
  self.assertEqual(uploads[1]['created_folders'],[])
  self.assertTrue(all((entry['status'] =='uploaded' for entry in uploads)))

 def test_configuration_and_approval_refusals(self):
  self.write_alpha()
  invalid =self.temp_path / 'invalid.json'
  invalid.write_text('{')
  env =self.environment(self.dry_fixture)
  env.pop('GRAPH_CLIENT_SECRET')
  cases =[
   (('--drive-id','drive-placeholder'),env,1,'GRAPH_CLIENT_SECRET'),
   (('--config',str(self.temp_path / 'missing.json')),None,1,'config'),
   (('--config',str(invalid)),None,1,'config'),
   ((),None,1,'neither drive_id nor site_id'),
   (('--dry-run','--live'),None,2,'not allowed with argument'),
   (('--live','--yes'),None,2,'--live requires --plan-digest or --plan-file'),
   (('--live','--plan-digest',self.approved_digest()),None,1,'not interactive'),
  ]
  for args,environment,code,message in cases:
   with self.subTest(message=message):
    runner =self.drive if args and args[0] in ('--live','--dry-run') else self.run_cli
    result =runner(*args,env=environment)
    self.check_result(result,code)
    self.assertIn(message,result.stderr)
    self.assertNotIn('FAKE TRANSPORT ENGAGED',result.stderr)

 def test_plain_file_sha256_failure(self):
  self.write_alpha()
  approved_digest =self.approved_digest()

  def mismatch(payload):
   for response in payload['responses']:
    if response.get('method') =='GET' and 'root:/vault/alpha.txt:/content' in response.get('url_prefix',''):
     response['body'] ='different'
  fixture =self.fixture_copy(self.SMALL,mismatch)
  result =self.live(approved_digest,fixture=fixture)
  self.assertNotEqual(result.returncode,0)
  audit =json.loads(result.stdout)
  self.assertEqual(audit['uploads'][0]['status'],'failed')
  self.assertEqual(audit['uploads'][0]['verification']['status'],'failed')
  self.assertEqual(len(audit['failures']),1)

 def test_include_exclude_plan(self):
  (self.source / 'keep.txt').write_text('keep',encoding='utf-8')
  (self.source / 'skip.txt').write_text('skip',encoding='utf-8')
  result =self.drive('--json','--include','*.txt','--exclude','skip*',fixture=self.dry_fixture)
  self.check_result(result)
  uploads =json.loads(result.stdout)['uploads']
  self.assertEqual([entry['source_path'] for entry in uploads],['keep.txt'])

 def test_changed_source_before_network(self):
  source_file =self.source / 'alpha.txt'
  source_file.write_text('alpha\n',encoding='utf-8')
  (self.source / 'unchanged.txt').write_text('same\n',encoding='utf-8')
  plan_path =self.temp_path / 'plan.json'
  dry_run =self.drive('--dry-run','--plan-file',str(plan_path),'--json',fixture=self.dry_fixture)
  self.check_result(dry_run)
  source_file.write_text('changed\n',encoding='utf-8')
  live_run =self.drive('--live','--yes','--plan-file',str(plan_path),'--json',fixture=self.dry_fixture)
  self.check_result(live_run,1)
  self.assertIn('plan digest mismatch',live_run.stderr)
  self.assertIn('Changed files:',live_run.stderr)
  self.assertIn('old size=6',live_run.stderr)
  self.assertIn('new size=8',live_run.stderr)
  self.assertNotIn('unchanged.txt:',live_run.stderr)
  self.assertNotIn('FAKE TRANSPORT ENGAGED',live_run.stderr)
  self.assertNotIn('uploaded',live_run.stdout.lower())

 GRAPH ='https://graph.microsoft.com/v1.0/drives/drive-placeholder'
 ETAG ='"{11111111-2222-3333-4444-555555555555},7"'

 def existing_item(self,path,etag=None):
  return {'method':'GET','url_prefix':f'{self.GRAPH}/root:/{path}?$select=id,eTag,folder,file','status':200,'json':{'id':'existing-item-placeholder','eTag':etag or self.ETAG,'file':{'mimeType':'text/plain'}}}

 def with_existing(self,name,path,etag=None):

  def transform(payload):
   payload['responses'].insert(1,self.existing_item(path,etag))
  return self.fixture_copy(name,transform)

 def replace_put_fixture(self):

  def transform(payload):
   payload['responses'].insert(1,self.existing_item('vault/alpha.txt'))
   payload['responses'].insert(1,{'method':'POST','url_prefix':f'{self.GRAPH}/root:/vault/alpha.txt:/createUploadSession','status':200,'json':{'uploadUrl':self.SESSION},'match_headers':{'If-Match':self.ETAG},'match_json':{'item':{'@microsoft.graph.conflictBehavior':'replace'}}})
   for response in payload['responses']:
    if response.get('method') =='PUT' and response['url_prefix'].endswith('vault/alpha.txt:/content?@microsoft.graph.conflictBehavior=fail'):
     response['url_prefix'] =self.SESSION
     response.pop('match_absent_headers',None)
  return self.fixture_copy(self.SMALL,transform)

 def dry_run_plan(self,fixture,*options):
  plan_path =self.temp_path / f'plan-{len(list(self.temp_path.iterdir()))}.json'
  result =self.run_cli(*options,'--dry-run','--plan-file',str(plan_path),'--json',fixture=fixture)
  self.check_result(result)
  return (plan_path,json.loads(plan_path.read_text(encoding='utf-8')))

 def test_changed_site_drive_refusal(self):
  self.write_alpha()

  def site_drive(drive_id):

   def transform(payload):
    payload['responses'].insert(1,{'method':'GET','url_prefix':'https://graph.microsoft.com/v1.0/sites/site-placeholder/drive','status':200,'json':{'id':drive_id}})
    if drive_id != 'drive-placeholder':
     payload['responses'].insert(2,{'method':'GET','url_prefix':f'https://graph.microsoft.com/v1.0/drives/{drive_id}?','status':200,'json':{'id':drive_id,'sharePointIds':{'siteId':'site-placeholder'}}})
     payload['responses'].insert(3,{'method':'GET','url_prefix':f'https://graph.microsoft.com/v1.0/drives/{drive_id}/root:/vault?','status':200,'json':{'id':'vault','folder':{}}})
     payload['responses'].insert(4,{'method':'GET','url_prefix':f'https://graph.microsoft.com/v1.0/drives/{drive_id}/root:/vault/','status':404,'json':{}})
   return transform
  approved_fixture =self.fixture_copy(self.DRY,site_drive('drive-placeholder'))
  plan_path,_ =self.dry_run_plan(approved_fixture,'--site-id','site-placeholder')
  moved =self.fixture_copy(self.SMALL,site_drive('drive-moved'))
  result =self.run_cli('--site-id','site-placeholder','--live','--yes','--plan-file',str(plan_path),'--json','--throttle','0',fixture=moved)
  self.check_result(result,1)
  self.assertIn('plan digest mismatch',result.stderr)
  self.assertIn('Target changed',result.stderr)
  self.assertIn('Nothing was written',result.stderr)
  self.assertEqual(result.stdout,'')

 def test_all_replacements_listed(self):
  self.write_alpha()
  (self.source / 'beta.txt').write_text('beta\n',encoding='utf-8')
  fixture =self.with_existing(self.DRY,'vault/alpha.txt')
  result =self.drive(fixture=fixture)
  self.check_result(result)
  self.assertIn('REPLACE existing item existing...',result.stdout)
  self.assertIn('create (no item at this path): beta.txt',result.stdout)
  self.assertIn('Planned folder creates: 0; planned creates: 1; planned replacements: 1',result.stdout)
  self.assertIn('Existing destination files that a live run would replace:\n  vault/alpha.txt\n',result.stdout)
  json_result =self.drive('--json',fixture=fixture)
  audit =json.loads(json_result.stdout)
  self.check_fields(audit,planned_replacements=['vault/alpha.txt'],planned_creates=1)
  self.assertEqual([entry['action'] for entry in audit['uploads']],['replace','create'])

 def test_create_race_json_and_text(self):
  self.write_alpha()
  approved =self.approved_digest()

  def conflict(payload):
   for response in payload['responses']:
    if response.get('method') =='PUT':
     response['status'] =409
     response['json'] ={'error':{'code':'nameAlreadyExists'}}
  result =self.live(approved,fixture=self.fixture_copy(self.SMALL,conflict))
  self.check_result(result,2)
  upload =self.upload(result)
  self.assertEqual(upload['status'],'failed')
  self.assertIn('approved action was create',upload['error'])
  text =self.drive('--live','--yes','--plan-digest',approved,'--throttle','0',fixture=self.fixture_copy(self.SMALL,conflict))
  self.check_result(text,2)
  self.assertIn('failed (create): alpha.txt',text.stdout)
  self.assertIn('error: alpha.txt: upload returned HTTP 409: an item now exists',text.stderr)

 def test_replace_success_and_failures(self):
  self.write_alpha()
  approved =self.approved_digest(fixture=self.with_existing(self.DRY,'vault/alpha.txt'))
  for failure in (None,412,500,429,503):
   with self.subTest(failure=failure):
    fixture =self.replace_put_fixture()
    payload =json.loads(fixture.read_text())
    if failure ==412:
     response =next(r for r in payload['responses'] if 'createUploadSession' in r['url_prefix'])
     response.update(status=412,json={'error':{'code':'preconditionFailed'}})
    elif failure is not None:
     put =next(r for r in payload['responses'] if r.get('method') =='PUT')
     payload['responses'].insert(payload['responses'].index(put),dict(put,status=failure,once=True,json={},headers={'Retry-After':'0'}))
    fixture.write_text(json.dumps(payload))
    result =self.live(approved,fixture=fixture)
    self.check_result(result,0 if failure is None else 2)
    upload =self.upload(result)
    self.check_fields(upload,action='replace',status='uploaded' if failure is None else 'failed')
    if failure ==412:
     self.assertIn('eTag no longer matches',upload['error'])
    if failure ==500:
     self.assertEqual(upload['outcome'],'unknown')
     self.assertIn('outcome unknown: the write may have completed',upload['error'])
     self.assertNotIn('nothing was replaced',upload['error'])

 def test_digest_binds_options(self):
  self.write_alpha()
  approved =self.approved_digest()
  result =self.live(approved,'--verify','none',fixture=self.small_fixture)
  self.check_result(result,1)
  self.assertIn('plan digest mismatch',result.stderr)

 def load_module(self):
  sys.path.insert(0,str(self.package_root / 'scripts'))
  import mirror_dataroom
  return mirror_dataroom

 def recorder(self,module,respond):
  """Patch Graph requests with a local recorder; real network is refused."""
  calls:list[dict] =[]

  def fake_request(method,url,**kwargs):
   call ={'method':method,'url':url,**kwargs}
   calls.append(call)
   return respond(call,calls)

  def refuse_network(*_args,**_kwargs):
   raise AssertionError('a test attempted a real network request')
  stack =contextlib.ExitStack()
  stack.enter_context(mock.patch.object(module.graph_client,'request',fake_request))
  stack.enter_context(mock.patch.object(module.graph_client,'_network_response',refuse_network))
  self.addCleanup(stack.close)
  return calls

 def planned_entry(self,module,content,method,path):
  return {'source_path':path,'destination_path':f'vault/{path}','size_bytes':len(content),'sha256':module.sha256_bytes(content),'upload_method':method,'action':'create'}

 def local_entry(self,content,name,method='chunked'):
  module =self.load_module()
  local =self.source / name
  local.write_bytes(content)
  return module,local,self.planned_entry(module,content,method,name)

 def apply_entry(self,module,local,entry,folders=None,args=None):
  return module.upload_one(args or self.args,'token','drive',local,entry,'vault',folders or set(),set())

 def test_same_size_mutation_no_requests(self):
  module =self.load_module()
  nested =self.source / 'nested'
  nested.mkdir()
  for method in ('direct','chunked'):
   with self.subTest(method=method):
    local =nested / f'{method}.bin'
    local.write_bytes(b'abcdefgh')
    planned =self.planned_entry(module,b'abcdefgh',method,f'nested/{method}.bin')
    local.write_bytes(b'abcdXXXX')
    calls =self.recorder(module,lambda call,calls:(500,{},{}))
    entry =self.apply_entry(module,local,planned,{'vault/nested'})
    self.assertEqual(entry['status'],'failed')
    self.assertIn('local file changed after approval',entry['error'])
    self.assertEqual(entry['created_folders'],[])
    self.assertEqual(calls,[])

 def test_mid_session_mutation_cancels(self):
  module,local,planned =self.local_entry(b'a' * 327680 + b'efgh','large.bin')
  upload_url =self.SESSION

  def respond(call,calls):
   if call['method'] =='POST':
    with local.open('r+b') as handle:
     handle.seek(327680)
     handle.write(b'XXXX')
    return (200,{'uploadUrl':upload_url},{})
   if call['method'] =='PUT':
    return (202,{'nextExpectedRanges':['4-']},{})
   if call['method'] =='DELETE':
    return (204,None,{})
   raise AssertionError(f"unexpected request {call['method']}")
  calls =self.recorder(module,respond)
  entry =self.apply_entry(module,local,planned)
  self.assertEqual(entry['status'],'failed')
  self.assertIn('local file changed after approval',entry['error'])
  self.assertEqual([call['method'] for call in calls],['POST','PUT','DELETE'])
  chunk_ranges =[call['headers']['Content-Range'] for call in calls if call['method'] =='PUT']
  self.assertEqual(chunk_ranges,['bytes 0-327679/327684'])
  delete =calls[-1]
  self.assertEqual(delete['url'],upload_url)
  self.assertTrue(delete['graph_issued_url'])
  self.assertNotIn('token',delete)
  self.assertEqual(entry['upload_session_cancel'],{'status':'cancelled','http_status':204})

 def test_local_integrity_entrypoints(self):
  module =self.load_module()
  local =self.source / 'alpha.txt'
  local.write_text('alphX\n',encoding='utf-8')
  entry =self.planned_entry(module,b'alpha\n','direct','alpha.txt')
  calls =self.recorder(module,lambda call,calls:(500,{},{}))
  with self.assertRaises(module.LocalFileChanged):
   module.upload_small(self.args,'token','drive',local,entry)
  with self.assertRaises(module.LocalFileChanged):
   module.check_local_integrity(local,entry)
  local.write_text('changed\n',encoding='utf-8')
  with self.assertRaises(module.LocalFileChanged):
   module.upload_large(self.args,'token','drive',local,entry)
  self.assertEqual(calls,[])

 def test_only_approved_folders_created(self):
  module =self.load_module()
  calls =self.recorder(module,lambda call,calls:(201,{'folder':{}},{}))
  with self.assertRaises(module.UnapprovedFolder):
   module.create_folder(self.args,'token','drive','vault/other',{'vault/nested'})
  self.assertEqual(calls,[])
  created =module.create_planned_folders(self.args,'token','drive','vault/nested/deeper',{'vault/nested','vault/nested/deeper'},set())
  self.assertEqual(created,['vault/nested','vault/nested/deeper'])
  self.assertEqual([call['method'] for call in calls],['POST','POST'])
  self.recorder(module,lambda call,calls:(201,{'folder':{}},{}) if len(calls) ==1 else (500,{},{}))
  recorded:list[str] =[]
  with self.assertRaises(RuntimeError):
   module.create_planned_folders(self.args,'token','drive','vault/a/b',{'vault/a','vault/a/b'},set(),recorded)
  self.assertEqual(recorded,['vault/a'])

 def test_read_only_folder_inventory(self):
  module =self.load_module()
  existing ={'vault'}

  def respond(call,calls):
   self.assertEqual(call['method'],'GET')
   path =call['url'].split('root:/',1)[1].split('?',1)[0]
   return (200,{'folder':{}},{}) if path in existing else (404,{},{})
  calls =self.recorder(module,respond)
  missing =module.missing_folders(self.args,'token','drive',['vault/a/b','vault/a','vault','vault/c'],destination_root='vault')
  self.assertEqual(missing,['vault/a','vault/a/b','vault/c'])
  checked =[call['url'].split('root:/',1)[1].split('?',1)[0] for call in calls]
  self.assertEqual(checked,['vault','vault/a','vault/a/b','vault/c'])

 def test_missing_folder_plan_revalidation(self):
  nested =self.source / 'nested'
  nested.mkdir()
  (nested / 'alpha.txt').write_text('alpha\n',encoding='utf-8')
  plan_path,plan =self.dry_run_plan(self.dry_fixture,'--drive-id','drive-placeholder')
  self.assertEqual(plan['folders_to_create'],['vault/nested'])
  text =self.drive(fixture=self.dry_fixture)
  self.check_result(text)
  self.assertIn('  create folder (missing): vault/nested\n',text.stdout)
  self.assertIn('Planned folder creates: 1;',text.stdout)

  def nested_exists(payload):
   payload['responses'].insert(1,{'method':'GET','url_prefix':f'{self.GRAPH}/root:/vault/nested?$select=id,name,folder','status':200,'json':{'id':'nested','name':'nested','folder':{}}})
  result =self.drive('--live','--yes','--plan-file',str(plan_path),'--throttle','0',fixture=self.fixture_copy(self.SMALL,nested_exists))
  self.check_result(result,1)
  self.assertIn('plan digest mismatch',result.stderr)
  self.assertIn('Folders to create changed: old=vault/nested; new=none',result.stderr)
  self.assertEqual(result.stdout,'')

 def test_unsafe_local_name_is_identified_without_invisible_output(self):
  for name in ('evil\nPlan digest: forged.txt','joined\u200dname.txt'):
   with self.subTest(name=ascii(name)):
    (self.source / name).write_text('x',encoding='utf-8')
    result =self.drive('--include',name,fixture=self.dry_fixture)
    self.check_result(result,1)
    self.assertIn(self.load_module().graph_client.display_text(ascii(name)),result.stderr)
    self.assertNotIn(name,result.stderr)
    self.assertIn('control',result.stderr.lower())

 def test_destination_must_exist(self):
  module =self.load_module()
  for status,payload in ((404,{}),(200,{'file':{}})):
   with self.subTest(status=status):
    calls =self.recorder(module,lambda call,calls:(status,payload,{}))
    with self.assertRaises(RuntimeError):
     module.missing_folders(self.args,'token','drive',['top/vault/a/b'],destination_root='top/vault')
    self.assertEqual([c['method'] for c in calls],['GET'])
    self.assertIn('/root:/top/vault?',calls[0]['url'])

 def test_selected_destination_boundary(self):
  module =self.load_module()
  root ='top/vault'

  def respond(call,calls):
   self.assertIn('/root:/top/vault',call['url'])
   if call['method'] =='GET':
    return (200,{'folder':{}},{}) if '/root:/top/vault?' in call['url'] else (404,{},{})
   self.assertEqual(call['method'],'POST')
   return (201,{'folder':{}},{})
  calls =self.recorder(module,respond)
  missing =module.missing_folders(self.args,'token','drive',[root + '/a/b'],destination_root=root)
  self.assertEqual(missing,[root + '/a',root + '/a/b'])
  created =module.create_planned_folders(self.args,'token','drive',root + '/a/b',set(missing),set(),destination_root=root)
  self.assertEqual(created,missing)
  self.assertEqual([c['method'] for c in calls],['GET','GET','GET','POST','POST'])
  before =len(calls)
  with self.assertRaises(module.UnapprovedFolder):
   module.create_planned_folders(self.args,'token','drive',root + '/a',{'top',root,root + '/a'},set(),destination_root=root)
  self.assertEqual(len(calls),before)

 def test_replacement_commit_guards(self):
  module =self.load_module()
  session =self.SESSION
  for content in (b'abc',b'a' * 327680 + b'efgh'):
   for outcome in ('unchanged','changed','precondition','check-failed','missing-tag','changed-id'):
    with self.subTest(size=len(content),outcome=outcome):
     module,local,entry =self.local_entry(content,'replacement.bin')
     entry.update(action='replace',remote_etag=self.ETAG,remote_item_id='item')

     def respond(call,calls):
      if call['method'] =='POST':
       self.assertEqual(call['headers']['If-Match'],self.ETAG)
       self.assertEqual(call['json_body']['item']['@microsoft.graph.conflictBehavior'],'replace')
       return (412,{},{}) if outcome =='precondition' else (200,{'uploadUrl':session},{})
      if call['method'] =='GET':
       self.assertIn('/root:/vault/replacement.bin?',call['url'])
       return (500,{},{}) if outcome =='check-failed' else (200,{'id':'other' if outcome =='changed-id' else 'item','file':{},'eTag':None if outcome =='missing-tag' else self.ETAG if outcome in ('unchanged','changed-id') else 'changed'},{})
      if call['method'] =='DELETE':
       self.assertEqual(call['url'],session)
       self.assertTrue(call['graph_issued_url'])
       return (204,None,{})
      self.check_fields(call,method='PUT',url=session)
      final =call['headers']['Content-Range'].endswith(f'-{len(content) - 1}/{len(content)}')
      self.assertEqual(call['max_retries'],0 if final else 2)
      return (201,{'size':len(content)},{}) if final else (202,{'nextExpectedRanges':['4-']},{})
     calls =self.recorder(module,respond)
     args =self.args
     args.max_retries =2
     result =self.apply_entry(module,local,entry,args=args)
     methods =[c['method'] for c in calls]
     self.assertEqual(result['status'],'uploaded' if outcome =='unchanged' else 'failed')
     if outcome =='precondition':
      self.assertEqual(methods,['POST'])
      self.assertIn('412',result['error'])
     elif outcome =='unchanged':
      self.assertEqual(methods[-2:],['GET','PUT'])
      self.assertNotIn('DELETE',methods)
     else:
      self.assertEqual(methods[-2:],['GET','DELETE'])
      self.assertEqual(sum((c['method'] =='PUT' for c in calls)),1 if len(content) > 327680 else 0)
      self.assertEqual(result['upload_session_cancel']['status'],'cancelled')

 def test_empty_replace_refuses_before_requests(self):
  module,local,entry =self.local_entry(b'','empty.bin')
  entry.update(action='replace',remote_etag=self.ETAG,remote_item_id='item')
  calls =self.recorder(module,lambda call,calls:(500,{},{}))
  result =self.apply_entry(module,local,entry)
  self.assertEqual(result['status'],'failed')
  self.assertIn('zero-byte',result['error'])
  self.assertEqual(calls,[])

 def test_empty_replace_is_a_counted_planning_refusal(self):
  self.write_alpha()
  (self.source / 'alpha.txt').write_bytes(b'')
  fixture =self.with_existing(self.DRY,'vault/alpha.txt')
  plan_path =self.temp_path / 'refused.json'
  dry =self.drive('--plan-file',str(plan_path),'--json',fixture=fixture)
  self.check_result(dry,2)
  audit =json.loads(dry.stdout)
  self.assertEqual(len(audit['failures']),1)
  self.assertIn('zero-byte replacement refused',audit['failures'][0]['error'])
  self.assertEqual(audit['planned_replacements'],[])
  self.assertEqual(json.loads(plan_path.read_text())['files'][0]['action'],'refuse')
  live =self.live(str(plan_path),flag='--plan-file',fixture=fixture)
  self.check_result(live,2)
  self.assertEqual(len(json.loads(live.stdout)['failures']),1)
  text =self.drive(fixture=fixture)
  self.assertIn('REFUSE: zero-byte',text.stdout)
  text =self.drive('--live','--yes','--plan-file',str(plan_path),fixture=fixture)
  self.assertIn('No file was uploaded: the plan contains refused actions',text.stdout)
  self.assertIn('failures: 1',text.stdout)

 def test_audit_reserved_before_upload_and_retains_failure(self):
  module =self.load_module()
  self.write_alpha()
  with mock.patch.dict(os.environ,self.environment(self.dry_fixture),clear=True):
   args =module.build_parser().parse_args(['--source',str(self.source),'--dest-path','vault','--drive-id','drive-placeholder'])
   args.small_threshold =module.DEFAULT_THRESHOLD
   args.chunk_size =module.DEFAULT_CHUNK_SIZE
   args.throttle,args.max_retries,args.request_timeout =0.5,4,60
   current,files =module.compute_plan(args,'test',module.graph_client.load_config(None,overrides={'drive_id':'drive-placeholder'}),self.source,'vault')
  current['files'].append(dict(current['files'][0],source_path='second'))
  files.append((files[0][0],'second'))
  audit =self.temp_path / 'audit'
  flags =['--source',str(self.source),'--dest-path','vault','--drive-id','drive-placeholder','--live','--yes','--plan-digest',module.plan_digest(current),'--audit-log',str(audit)]
  from io import StringIO
  def upload(*args):
   self.assertTrue(audit.exists())
   self.assertEqual(audit.stat().st_mode & 511,384)
   if args[4]['source_path'] =='second':
    raise RuntimeError('synthetic interruption')
   return dict(module.dry_run_entry(args[4],'none'),status='uploaded')
  with mock.patch.object(module.graph_client,'acquire_token',return_value='test'),mock.patch.object(module,'compute_plan',return_value=(current,files)),mock.patch.object(module,'upload_one',side_effect=upload) as put,contextlib.redirect_stderr(StringIO()):
   self.assertEqual(module.main(flags),1)
   record =json.loads(audit.read_text())
   self.assertEqual([e['status'] for e in record['uploads']],['uploaded','failed'])
   self.assertIn('synthetic interruption',record['failures'][0]['error'])
   put.reset_mock()
   self.assertEqual(module.main(flags),1)
   put.assert_not_called()
   link =self.temp_path / 'audit-link'
   link.symlink_to(audit)
   for refused in (link,self.temp_path / 'missing' / 'audit'):
    self.assertEqual(module.main([*flags[:-1],str(refused)]),1)
    put.assert_not_called()

 def test_plan_and_audit_outputs_are_new_private_files_only(self):
  module =self.load_module()
  for writer in (lambda p:module.write_plan(p,{},'digest'),lambda p:module.write_audit(p,{})):
   with self.subTest(writer=writer):
    path =self.temp_path / ('record-' + str(len(list(self.temp_path.iterdir()))))
    writer(path)
    self.assertEqual(path.stat().st_mode & 511,384)
    before =path.read_bytes()
    with self.assertRaisesRegex(RuntimeError,'exist'):
     writer(path)
    self.assertEqual(path.read_bytes(),before)
    link =self.temp_path / (path.name + '-link')
    link.symlink_to(path)
    with self.assertRaises(RuntimeError):
     writer(link)
    missing =self.temp_path / (path.name + '-missing')
    with self.assertRaises(RuntimeError):
     writer(missing / 'record')
    self.assertFalse(missing.exists())
    directory =self.temp_path / (path.name + '-parent')
    directory.symlink_to(self.temp_path,target_is_directory=True)
    with self.assertRaises(RuntimeError):
     writer(directory / 'record')
    self.assertFalse((self.temp_path / 'record').exists())

 def test_chunked_create_conflicts(self):
  module,local,entry =self.local_entry(b'a' * 327680 + b'efgh','create.bin')
  for conflict_at in ('session','final'):
   with self.subTest(conflict_at=conflict_at):

    def respond(call,calls):
     if call['method'] =='POST':
      self.assertNotIn('If-Match',call['headers'])
      self.assertEqual(call['json_body']['item']['@microsoft.graph.conflictBehavior'],'fail')
      return (409,{},{}) if conflict_at =='session' else (200,{'uploadUrl':self.SESSION},{})
     self.assertEqual(call['method'],'PUT')
     return (202,{},{}) if len(calls) ==2 else (409,{},{})
    calls =self.recorder(module,respond)
    result =self.apply_entry(module,local,entry)
    self.assertEqual(result['status'],'failed')
    self.assertIn('approved action was create',result['error'])
    self.assertEqual([c['method'] for c in calls],['POST'] if conflict_at =='session' else ['POST','PUT','PUT'])

 def test_old_plan_rejected(self):
  self.write_alpha()
  old_plan =self.temp_path / 'old-plan.json'
  old_plan.write_text(json.dumps({'destination_root':'vault','files':[],'plan_digest':'0' * 64}),encoding='utf-8')
  result =self.drive('--live','--yes','--plan-file',str(old_plan),fixture=self.small_fixture)
  self.check_result(result,1)
  self.assertIn('rerun the dry run',result.stderr)
  self.assertNotIn('FAKE TRANSPORT ENGAGED',result.stderr)

 def test_changed_remote_state_refusal(self):
  self.write_alpha()
  for before,after,changed in ((False,True,False),(True,False,False),(True,True,True)):
   with self.subTest(before=before,after=after,changed=changed):
    dry =self.with_existing(self.DRY,'vault/alpha.txt') if before else self.dry_fixture
    plan_path,plan =self.dry_run_plan(dry,'--drive-id','drive-placeholder')
    self.assertEqual(plan['files'][0]['action'],'replace' if before else 'create')
    live =self.with_existing(self.SMALL,'vault/alpha.txt','changed' if changed else self.ETAG) if after else self.small_fixture
    result =self.live(str(plan_path),flag='--plan-file',fixture=live)
    self.check_result(result,1)
    self.assertIn('plan digest mismatch',result.stderr)
    self.assertEqual(result.stdout,'')
    if not changed:
     self.assertIn('action=replace',result.stderr)
    if not before:
     self.assertIn('old size=6',result.stderr)
     self.assertIn('action=create',result.stderr)

 def test_office_verification_information(self):
  (self.source / 'record.docx').write_bytes(b'local-office')

  def transform(payload):
   for response in payload['responses']:
    response['url_prefix'] =response['url_prefix'].replace('alpha.txt','record.docx')
    if response.get('method') =='PUT' and 'record.docx' in response['url_prefix']:
     response['json'].update(name='record.docx',size=20)
    if response.get('method') =='GET' and '/record.docx:/content' in response['url_prefix']:
     response['body'] ='remote-office'
  fixture =self.fixture_copy(self.SMALL,transform)
  for verify in ('sha256','size'):
   with self.subTest(verify=verify):
    approved =self.approved_digest('--verify',verify)
    result =self.live(approved,'--verify',verify,fixture=fixture)
    self.check_result(result)
    upload =self.upload(result)
    self.assertEqual(upload['status'],'uploaded')
    self.assertEqual(upload['verification']['status'],'informational')
if __name__ =='__main__':
 unittest.main()

License: MIT (read it).

Back to the listing