tests/test_file_history.py — Data Room Toolkit

# By Zero Width Lab (zerowidthlab.com)
"""Subprocess tests for the file version history 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 file_history as history
import graph_client as graph

def run_cli(self,*arguments,fixture=None,credentials=True):
 env =os.environ.copy()
 for name in ('GRAPH_TENANT_ID','GRAPH_CLIENT_ID','GRAPH_CLIENT_SECRET','SHAREPOINT_DATAROOM_CONFIG','GRAPH_FAKE_RESPONSES','GRAPH_FAKE_TRANSPORT_TEST_ACK'):
  env.pop(name,None)
 if credentials:
  env.update({'GRAPH_TENANT_ID':'test-tenant','GRAPH_CLIENT_ID':'test-client','GRAPH_CLIENT_SECRET':'test-secret'})
 fixture =self.default_fixture if fixture is None else fixture
 if fixture:
  env['GRAPH_FAKE_RESPONSES'] =str(self.fixtures / fixture)
  env['GRAPH_FAKE_TRANSPORT_TEST_ACK'] ='zero-width-lab-test-suite-only'
 return subprocess.run([sys.executable,str(self.script),'--throttle','0',*arguments],cwd=self.package_root,env=env,capture_output=True,text=True,check=False,stdin=subprocess.DEVNULL,timeout=10)

class FileHistoryCliTests(unittest.TestCase):
 default_fixture ='history_basic.json'

 def test_rooted_requests(self):
  with patch.object(history,'acquire_token',return_value='test') as auth:
   for flags in (('--root-path','../outside','--path','safe'),('--root-path','Room','--path','safe','--path','%2e%2e/outside')):
    with self.subTest(flags=flags):
     self.assertEqual(history.main(['--drive-id','d',*flags]),1)
     auth.assert_not_called()
   with patch.object(graph,'_fake_transport_enabled',return_value=True),patch.object(graph,'_fake_response',side_effect=[(200,{'id':'f'},{}),(200,{'value':[]},{})]) as transport:
    self.assertEqual(history.main(['--drive-id','d','--root-path','Room','--path','a/file']),0)
    for call in transport.call_args_list:
     self.assertTrue(call.args[1].startswith(graph.GRAPH_BASE_URL + '/drives/d/root:/Room/a/file:'))

 def test_malformed_pages(self):
  url =graph.GRAPH_BASE_URL + '/drives/d/root:/Room/file:/versions'
  client =history.GraphSession('test',60,0,0)
  for payload in (None,{},{'value':'bad'},{'value':[],'@odata.nextLink':[]}):
   with self.subTest(payload=payload),patch.object(client,'request',side_effect=[(200,{'value':[{'id':'1'}],'@odata.nextLink':url + '?page=2'},{}),(200,payload,{})]),self.assertRaisesRegex(graph.GraphClientError,'malformed collection'):
    list(history.iter_versions(client,url))

 @classmethod
 def setUpClass(cls):
  cls.package_root =Path(__file__).resolve().parent.parent
  cls.script =cls.package_root / 'scripts' / 'file_history.py'
  cls.fixtures =cls.package_root / 'tests' / 'fixtures'
 run_cli =run_cli

 def test_pagination_sort(self):
  result =self.run_cli('--drive-id','drive-test','--path','Records/Overview.txt','--json')
  self.assertEqual(result.returncode,0,result.stderr)
  report =json.loads(result.stdout)
  versions =report['results'][0]['versions']
  self.assertEqual([item['id'] for item in versions],['1.0','2.0','3.0'])
  self.assertEqual(report['results'][0]['current']['size'],31)

 def test_redact_users(self):
  for flags in ((),('--json',)):
   with self.subTest(flags=flags):
    result =self.run_cli('--drive-id','drive-test','--path','Records/Overview.txt','--redact-users',*flags)
    self.assertEqual(result.returncode,0,result.stderr)
    self.assertIn('[redacted]',result.stdout)
    self.assertNotIn('Example Editor',result.stdout)
    if flags:
     self.assertEqual(json.loads(result.stdout)['results'][0]['versions'][0]['modifier'],'[redacted]')

 def test_partial_failure(self):
  result =self.run_cli('--drive-id','drive-test','--path','Records/Present.txt','--path','Records/Missing.txt','--json',fixture='history_missing_file.json')
  self.assertNotEqual(result.returncode,0)
  report =json.loads(result.stdout)
  self.assertIn('versions',report['results'][0])
  self.assertEqual(report['results'][1]['status'],404)
  self.assertIn('Records/Missing.txt',result.stderr)

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

 def test_escaped_text(self):
  payload =json.loads((self.fixtures / 'history_basic.json').read_text(encoding='utf-8'))
  for response in payload['responses']:
   for version in (response.get('json') or {}).get('value',[]):
    if version.get('id') =='3.0':
     version['id'] ='3.0\nRecords/Forged.txt'
     version['lastModifiedBy']['user']['displayName'] ='Editor\u2028X'
  with tempfile.TemporaryDirectory() as temp_dir:
   fixture =Path(temp_dir) / 'history_newline_values.json'
   fixture.write_text(json.dumps(payload),encoding='utf-8')
   result =self.run_cli('--drive-id','drive-test','--path','Records/Overview.txt',fixture=str(fixture))
  self.assertEqual(result.returncode,0,result.stderr)
  lines =result.stdout.split('\n')
  self.assertNotIn('Records/Forged.txt',lines)
  self.assertIn('3.0\\u000aRecords/Forged.txt',result.stdout)
  self.assertIn('Editor\\u2028X',result.stdout)
  self.assertNotIn('\u2028',result.stdout)
if __name__ =='__main__':
 unittest.main()

License: MIT (read it).

Back to the listing