tests/test_dataroom_search.py — Data Room Toolkit

# By Zero Width Lab (zerowidthlab.com)
"""Subprocess tests for the document-library inventory command."""
from __future__ import annotations
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
__author__ = "Zero Width Lab"
sys.path.insert(0,str(Path(__file__).resolve().parent.parent / 'scripts'))
import dataroom_search as search
import file_history as history
import graph_client as graph
import test_file_history

class DataroomSearchCliTests(unittest.TestCase):
 default_fixture ='search_inventory_basic.json'

 def test_missing_search_path_lookup_and_counted_skips(self):
  client =search.GraphSession('test',60,0,0)
  hit ={'id':'f/a','name':'file'}
  for details,expected in (({'name':'file','parentReference':{'path':'/drives/d/root:/Room'}},1),({'name':'file'},0),({'name':'file','parentReference':{'path':'/drives/d/root:/Other'}},0)):
   skipped,outside =[],[]
   with patch.object(client,'request',side_effect=[(200,{'value':[hit]},{}),(200,details,{})]) as get:
    result =search.content_search(client,'d','Room',['q'],False,skipped,outside)
   self.assertEqual(len(result),expected)
   self.assertEqual(len(skipped) + len(outside),1 - expected)
   self.assertEqual(len(outside),int(details.get('parentReference',{}).get('path','').endswith('/Other')))
   self.assertEqual(get.call_count,2)
   self.assertEqual(get.call_args.args[1],search.GRAPH_ROOT + '/drives/d/items/f%2Fa?$select=parentReference,name')
  with patch.object(client,'request',side_effect=[(200,{'value':[hit]},{}),(403,{},{})]):
   skipped =[]
   self.assertEqual(search.content_search(client,'d','Room',['q'],False,skipped),[])
   self.assertIn('403',skipped[0]['error'])
  for item_id in ('..','%2e%2e','bad\\id'):
   with patch.object(client,'request',return_value=(200,{'value':[{'id':item_id}]},{})) as get:
    skipped =[]
    self.assertEqual(search.content_search(client,'d','Room',['q'],False,skipped),[])
    self.assertEqual(get.call_count,1)
    self.assertEqual(len(skipped),1)

 def test_unresolved_search_hit_marks_cli_incomplete(self):
  from contextlib import redirect_stdout,redirect_stderr
  from io import StringIO
  for flags,code in (([],2),(['--allow-partial'],0)):
   output =StringIO()
   with patch.object(search,'acquire_token',return_value='test'),patch.object(search,'resolve_drive_id',return_value='d'),patch.object(search,'walk_files',return_value=([],[])),patch.object(search.GraphSession,'request',side_effect=[(200,{'value':[{'id':'i','name':'file'}]},{}),(200,{'name':'file'},{})]),redirect_stdout(output),redirect_stderr(StringIO()):
    self.assertEqual(search.main(['--drive-id','d','--content-search','--keywords','q','--json',*flags]),code)
   report =json.loads(output.getvalue())
   self.assertTrue(report['incomplete'])
   self.assertEqual(report['skipped_content_hit_count'],1)
   self.assertIn('cannot be resolved',report['skipped_content_hits'][0]['error'])

 def test_outside_scope_does_not_mark_incomplete(self):
  from contextlib import redirect_stdout
  from io import StringIO
  output =StringIO()
  with patch.object(search,'acquire_token',return_value='test'),patch.object(search,'resolve_drive_id',return_value='d'),patch.object(search,'walk_files',return_value=([],[])),patch.object(search.GraphSession,'request',return_value=(200,{'value':[{'id':'i','name':'file','parentReference':{'path':'/drives/d/root:/Other'}}]},{})),redirect_stdout(output):
   self.assertEqual(search.main(['--drive-id','d','--root-path','Room','--content-search','--keywords','q','--json']),0)
  report =json.loads(output.getvalue())
  self.assertEqual(report['outside_root'],1)
  self.assertEqual(report['skipped_content_hit_count'],0)
  self.assertFalse(report.get('incomplete',False))

 def test_collection_rejects_members_and_bounded_pagination(self):
  client =search.GraphSession('test',60,0,0)
  url =search.GRAPH_ROOT + '/drives/d/root/children'
  for iterate in (search.iter_collection,history.iter_versions):
   for payload,message in (({'value':[None]},'malformed'),({'value':[],'@odata.nextLink':url},'repeated')):
    with patch.object(client,'request',side_effect=[(200,payload,{}),AssertionError('unexpected page request')]),self.assertRaisesRegex(graph.GraphClientError,message):
     list(iterate(client,url))
   with patch.object(graph,'MAX_COLLECTION_PAGES',1),patch.object(client,'request',side_effect=[(200,{'value':[],'@odata.nextLink':url + '?page=2'},{}),AssertionError('unexpected page request')]),self.assertRaisesRegex(graph.GraphClientError,'exceeds 1 pages'):
    list(iterate(client,url))

 def test_search_output_create_only_and_text_matches(self):
  result =self.run_cli('--drive-id','drive-test','--root-path','Records','--keywords','quarterly','--content-search')
  self.assertEqual(result.returncode,0,result.stderr)
  self.assertIn('Content match: Records/Quarterly Notes.txt',result.stdout)
  with tempfile.TemporaryDirectory() as folder:
   folder =str(Path(folder).resolve())
   out =Path(folder) / 'report.json'
   out.write_text('keep')
   result =self.run_cli('--drive-id','drive-test','--out',str(out))
   self.assertEqual(result.returncode,1)
   self.assertEqual(out.read_text(),'keep')
   self.assertIn('local output failed',result.stderr)
   self.assertNotIn('Microsoft Graph request failed',result.stderr)
   result =self.run_cli('--drive-id','drive-test','--out',str(Path(folder) / 'new.json'))
   self.assertEqual(result.returncode,0,result.stderr)

 def test_root_and_remote_names_cannot_escape_transport(self):
  client =search.GraphSession('test',60,0,0)
  with patch.object(search,'acquire_token') as auth:
   self.assertEqual(search.main(['--drive-id','d','--root-path','%2e%2e/Other']),1)
   auth.assert_not_called()
  for name in ('Safe','..','a//b','%2e%2e','C:a','a\\b','a\n'):
   responses =[(200,{'id':'r'},{}),(200,{'value':[{'name':name,'id':'f','folder':{}}]},{}),(200,{'value':[]},{})]
   with self.subTest(name=name),patch.object(graph,'_fake_transport_enabled',return_value=True),patch.object(graph,'_fake_response',side_effect=responses) as transport:
    _,skipped =search.walk_files(client,'d','Room',None,0)
    self.assertEqual(bool(skipped),name != 'Safe')
    for call in transport.call_args_list:
     self.assertTrue(call.args[1].startswith(graph.GRAPH_BASE_URL + '/drives/d/root:/Room:') or call.args[1].startswith(graph.GRAPH_BASE_URL + '/drives/d/root:/Room/Safe:'))

 def test_malformed_pages_keep_partial_inventory(self):
  client =search.GraphSession('test',60,0,0)
  url =graph.GRAPH_BASE_URL + '/drives/d/root:/Room:/children'
  for payload in (None,[],{},{'value':{}},{'value':[],'@odata.nextLink':7}):
   with self.subTest(payload=payload),patch.object(client,'request',side_effect=[(200,{'id':'r'},{}),(200,{'value':[{'name':'file','file':{}}],'@odata.nextLink':url + '?page=2'},{}),(200,payload,{})]):
    files,skipped =search.walk_files(client,'d','Room',None,0)
    self.assertEqual(len(files),1)
    self.assertIn('malformed collection',skipped[0]['error'])

 @classmethod
 def setUpClass(cls) -> None:
  cls.package_root =Path(__file__).resolve().parent.parent
  cls.script =cls.package_root / 'scripts' / 'dataroom_search.py'
  cls.fixtures =cls.package_root / 'tests' / 'fixtures'
 run_cli =test_file_history.run_cli

 def test_inventory_json_and_keyword_ranking(self) -> None:
  result =self.run_cli('--drive-id','drive-test','--keywords','quarterly,notes,overview','--json')
  self.assertEqual(result.returncode,0,result.stderr)
  report =json.loads(result.stdout)
  self.assertEqual(report['file_count'],2)
  self.assertEqual(report['files'][0]['path'],'Records/Quarterly Notes.txt')
  self.assertEqual(report['files'][0]['rank'],2)
  self.assertEqual(report['files'][1]['path'],'Overview.txt')

 def test_redact_users_in_json_and_text(self) -> None:
  json_result =self.run_cli('--drive-id','drive-test','--redact-users','--json')
  self.assertEqual(json_result.returncode,0,json_result.stderr)
  rendered =json.loads(json_result.stdout)
  self.assertEqual(rendered['files'][0]['modifier'],'[redacted]')
  self.assertNotIn('Example Editor',json_result.stdout)
  text_result =self.run_cli('--drive-id','drive-test','--redact-users')
  self.assertEqual(text_result.returncode,0,text_result.stderr)
  self.assertIn('modifier=[redacted]',text_result.stdout)
  self.assertNotIn('Example Editor',text_result.stdout)

 def test_retry_then_success_after_429(self) -> None:
  result =self.run_cli('--drive-id','drive-test','--json',fixture='search_retry_429.json')
  self.assertEqual(result.returncode,0,result.stderr)
  report =json.loads(result.stdout)
  self.assertEqual(report['files'][0]['name'],'Recovered.txt')

 def test_partial_inventory_json_and_text(self):
  for flags,code in ((['--json'],2),(['--json','--allow-partial'],0),([],2)):
   with self.subTest(flags=flags):
    result =self.run_cli('--drive-id','drive-test',*flags,fixture='search_partial_failure.json')
    self.assertEqual(result.returncode,code,result.stderr)
    self.assertIn("skipping folder 'Restricted'",result.stderr)
    if '--json' in flags:
     report =json.loads(result.stdout)
     self.assertTrue(report['incomplete'])
     self.assertEqual(report['file_count'],1)
     self.assertEqual(report['skipped_folders'],[{'path':'Restricted','error':'Graph request failed with HTTP 403'}])
    else:
     self.assertIn('Files: 1',result.stdout)
     self.assertIn('Incomplete inventory: 1 folder(s) skipped (see stderr for details).',result.stdout)

 def test_missing_credentials_target_and_bad_config(self):
  cases =[(['--drive-id','drive-test'],False,'GRAPH_TENANT_ID'),(['--json'],True,'neither drive_id nor site_id')]
  cases += [(['--drive-id','drive-test','--config',path],True,'config') for path in ('tests/fixtures/not-present.json','scripts/dataroom_search.py')]
  for flags,credentials,message in cases:
   with self.subTest(flags=flags):
    result =self.run_cli(*flags,credentials=credentials)
    self.assertEqual(result.returncode,1)
    self.assertIn(message,result.stderr)
    self.assertNotIn('Traceback',result.stderr)

 def test_invalid_numeric_and_date_combinations_use_argparse(self) -> None:
  cases =[('--drive-id','drive-test','--max-depth','-1'),('--drive-id','drive-test','--recent-days','-1'),('--drive-id','drive-test','--recent-days','2','--since','2099-01-01')]
  for arguments in cases:
   with self.subTest(arguments=arguments):
    result =self.run_cli(*arguments)
    self.assertNotEqual(result.returncode,0)
    self.assertIn('error:',result.stderr.lower())

 def test_content_search_rejects_single_quote(self) -> None:
  result =self.run_cli('--drive-id','drive-test','--keywords',"owner's",'--content-search')
  self.assertNotEqual(result.returncode,0)
  self.assertIn('single quote',result.stderr.lower())
  self.assertNotIn('Traceback',result.stderr)

 def test_content_search_respects_root_path(self) -> None:
  result =self.run_cli('--drive-id','drive-test','--root-path','Records','--keywords','quarterly','--content-search','--json')
  self.assertEqual(result.returncode,0,result.stderr)
  report =json.loads(result.stdout)
  self.assertEqual(report['file_count'],1)
  self.assertEqual(report['content_search'][0]['path'],'Records/Quarterly Notes.txt')

 def test_remote_control_names_are_rejected(self) -> None:
  payload =json.loads((self.fixtures / 'search_inventory_basic.json').read_text(encoding='utf-8'))
  for response in payload['responses']:
   for item in (response.get('json') or {}).get('value',[]):
    if item.get('name') =='Overview.txt':
     item['name'] ='Over\nFiles: 99 forged.txt'
     item['lastModifiedBy']['user']['displayName'] ='Editor\rX'
  with tempfile.TemporaryDirectory() as temp_dir:
   fixture =Path(temp_dir) / 'search_newline_name.json'
   fixture.write_text(json.dumps(payload),encoding='utf-8')
   result =self.run_cli('--drive-id','drive-test',fixture=str(fixture))
  self.assertEqual(result.returncode,2,result.stderr)
  lines =result.stdout.splitlines()
  self.assertFalse(any((line.startswith('Files: 99') for line in lines)))
  self.assertIn('invalid remote path',result.stderr)
  self.assertEqual([line for line in lines if line.startswith('Files:')],['Files: 1'])
if __name__ =='__main__':
 unittest.main()

License: MIT (read it).

Back to the listing