Schedule a demo
Version 4.1 Documentation

Testing OneDrive

Mock OneDrive connections to test services that work with files, folders, sharing, and versions.

Note: If you're new to unit testing with Zato, check the tutorial first.

Getting files

The service:

from zato.server.service import Service

class GetDriveFiles(Service):
    name = 'onedrive.files.get'

    def handle(self):
        conn = self.cloud.ms365.get('ms365.OneDrive').conn

        with conn.client() as client:
            storage = client.impl.storage()
            drive = storage.get_default_drive()
            root_folder = drive.get_root_folder()
            items = root_folder.get_items(limit=25)

        self.response.payload = {
            'items': [{'name': item['name'], 'size': item['size']} for item in items]
        }

The test:

from zato_testing import ServiceTestCase
from myapp.services import GetDriveFiles

class TestGetDriveFiles(ServiceTestCase):

    def test_returns_files(self):

        self.set_response('ms365.OneDrive.storage', {'id': 'storage-123'})
        self.set_response('ms365.OneDrive.storage.get_default_drive', {'id': 'drive-456'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_root_folder', {'id': 'root-789'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_root_folder.get_items', [
            {'name': 'Documents', 'size': 0},
            {'name': 'report.pdf', 'size': 1024000},
            {'name': 'photo.jpg', 'size': 2048000}
        ])

        service = self.invoke(GetDriveFiles)

        self.assertEqual(len(service.response.payload['items']), 3)
        self.assertEqual(service.response.payload['items'][1]['name'], 'report.pdf')

Uploading files

The service:

class UploadFile(Service):
    name = 'onedrive.file.upload'
    input = 'folder', 'file_name', 'content'

    def handle(self):
        conn = self.cloud.ms365.get('ms365.OneDrive').conn

        with conn.client() as client:
            storage = client.impl.storage()
            drive = storage.get_default_drive()
            folder = drive.get_item_by_path(self.request.input.folder)
            uploaded = folder.upload_file(
                item=self.request.input.content,
                item_name=self.request.input.file_name
            )

        self.response.payload = {
            'uploaded': True,
            'file_id': uploaded['id'],
            'web_url': uploaded['web_url']
        }

The test:

class TestUploadFile(ServiceTestCase):

    def test_uploads_file(self):

        self.set_response('ms365.OneDrive.storage', {'id': 'storage-123'})
        self.set_response('ms365.OneDrive.storage.get_default_drive', {'id': 'drive-456'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_item_by_path', {'id': 'folder-001'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_item_by_path.upload_file', {
            'id': 'file-new-123',
            'web_url': 'https://onedrive.com/file-new-123'
        })

        service = self.invoke(UploadFile,
            folder='/Documents',
            file_name='new_report.pdf',
            content=b'PDF content here'
        )

        self.assertEqual(service.response.payload['uploaded'], True)
        self.assertEqual(service.response.payload['file_id'], 'file-new-123')

Downloading files

class TestDownloadFile(ServiceTestCase):

    def test_downloads_file(self):

        self.set_response('ms365.OneDrive.storage', {'id': 'storage-123'})
        self.set_response('ms365.OneDrive.storage.get_default_drive', {'id': 'drive-456'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_item_by_path', {
            'name': 'report.pdf',
            'size': 1024000
        })
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_item_by_path.download', b'PDF file content')

        service = self.invoke(DownloadFile, file_path='/Documents/report.pdf')

        self.assertEqual(service.response.payload['content'], b'PDF file content')

Searching files

class TestSearchFiles(ServiceTestCase):

    def test_searches_files(self):

        self.set_response('ms365.OneDrive.storage', {'id': 'storage-123'})
        self.set_response('ms365.OneDrive.storage.get_default_drive', {'id': 'drive-456'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.search', [
            {'name': 'quarterly_report.pdf', 'path': '/Documents/Reports'},
            {'name': 'annual_report.pdf', 'path': '/Documents/Reports'}
        ])

        service = self.invoke(SearchFiles, query='report')

        self.assertEqual(len(service.response.payload['results']), 2)

Sharing files

class TestShareFile(ServiceTestCase):

    def test_shares_file(self):

        self.set_response('ms365.OneDrive.storage', {'id': 'storage-123'})
        self.set_response('ms365.OneDrive.storage.get_default_drive', {'id': 'drive-456'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_item_by_path', {'id': 'file-001'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_item_by_path.share_with_link', {
            'id': 'perm-001',
            'share_link': 'https://onedrive.com/share/abc123'
        })

        service = self.invoke(ShareFile, file_path='/Documents/report.pdf', share_type='edit')

        self.assertEqual(service.response.payload['share_link'], 'https://onedrive.com/share/abc123')

File versions

class TestGetFileVersions(ServiceTestCase):

    def test_returns_versions(self):

        self.set_response('ms365.OneDrive.storage', {'id': 'storage-123'})
        self.set_response('ms365.OneDrive.storage.get_default_drive', {'id': 'drive-456'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_item_by_path', {'id': 'file-001'})
        self.set_response('ms365.OneDrive.storage.get_default_drive.get_item_by_path.get_versions', [
            {'name': '1.0', 'modified': '2024-01-01T10:00:00Z'},
            {'name': '2.0', 'modified': '2024-02-15T14:30:00Z'},
            {'name': '3.0', 'modified': '2024-03-20T09:15:00Z'}
        ])

        service = self.invoke(GetFileVersions, file_path='/Documents/report.pdf')

        self.assertEqual(len(service.response.payload['versions']), 3)

See also