From e45699f5a7d9791e5374ba0be4865ee45a65c7db Mon Sep 17 00:00:00 2001 From: Marisol Date: Fri, 27 Mar 2026 15:40:58 +0000 Subject: [PATCH] Add Dropbox error handling tests and update documentation --- .gitignore | 6 + README.md | 0 dropbox_upload.py | 330 ++++++++++++++++++++++++++- tests/test_dropbox_error_handling.py | 220 ++++++++++++++++++ 4 files changed, 555 insertions(+), 1 deletion(-) mode change 100644 => 100755 README.md create mode 100755 tests/test_dropbox_error_handling.py diff --git a/.gitignore b/.gitignore index dd30cf6..7dda894 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,9 @@ debug_*.py dist/ build/ *.egg-info/ + +# Auto-added by Marisol pipeline +.pio/ +.gradle/ +*.class +local.properties diff --git a/README.md b/README.md old mode 100644 new mode 100755 diff --git a/dropbox_upload.py b/dropbox_upload.py index ba707f5..641e7f4 100755 --- a/dropbox_upload.py +++ b/dropbox_upload.py @@ -1,5 +1,12 @@ +<<<<<<< Updated upstream """Upload the contents of your Downloads folder to Dropbox. This is an example app for API v2. +======= +"""Upload files to Dropbox with filtering and error handling. + +This module provides modular functions for uploading files to Dropbox +with proper error handling and file filtering for testability. +>>>>>>> Stashed changes """ from __future__ import print_function @@ -11,6 +18,7 @@ import six import sys import time +<<<<<<< Updated upstream import unicodedata if sys.version.startswith('2'): @@ -119,11 +127,214 @@ def main(): else: print('OK, skipping directory:', name) dirs[:] = keep +======= + +import dropbox + +TOKEN = 'YOUR_ACCESS_TOKEN' + + +def parse_args(args=None): + """Parse command-line arguments for Dropbox upload. + + Args: + args: List of command-line arguments (defaults to sys.argv[1:]) + + Returns: + argparse.Namespace: Parsed arguments containing folder, rootdir, token, + yes, no, default, and count flags + """ + parser = argparse.ArgumentParser( + description='Sync directory to Dropbox', + prog='dropbox_upload.py' + ) + parser.add_argument('folder', nargs='?', default='Downloads', + help='Folder name in your Dropbox') + parser.add_argument('rootdir', nargs='?', default='~/Downloads', + help='Local directory to upload') + parser.add_argument('--token', default=TOKEN, + help='Access token (see https://www.dropbox.com/developers/apps)') + parser.add_argument('--yes', '-y', action='store_true', + help='Answer yes to all questions') + parser.add_argument('--no', '-n', action='store_true', + help='Answer no to all questions') + parser.add_argument('--default', '-d', action='store_true', + help='Take default answer on all questions') + parser.add_argument('--count', type=int, default=None, + help='Maximum number of files to upload') + + return parser.parse_args(args) + + +def should_skip_file(filename): + """Determine if a file should be skipped based on its name. + + Filters out: + - Dot files (starting with .) + - Temporary files (starting with @, ~, or ending with .tmp/.temp) + - Generated files (ending with .pyc or .pyo) + - Generated directories (__pycache__) + - Empty filenames + + Note: This check is case-sensitive (e.g., .PYC is not filtered). + + Args: + filename: Name of the file to check (can be string or None) + + Returns: + bool: True if file should be skipped, False otherwise + """ + if filename is None: + return True + + if not isinstance(filename, str): + try: + filename = str(filename) + except (ValueError, TypeError): + return True + + if not filename: + return True + + # Skip dot files (starting with .) + if filename.startswith('.'): + return True + + # Skip temporary files (starting with @ or ~) + if filename.startswith('@') or filename.startswith('~'): + return True + + # Skip temporary file extensions (.tmp, .temp) + if filename.endswith('.tmp') or filename.endswith('.temp'): + return True + + # Skip generated Python bytecode files + if filename.endswith('.pyc') or filename.endswith('.pyo'): + return True + + # Skip __pycache__ directories + if filename == '__pycache__': + return True + + return False + + +def upload(dbx, fullname, folder, subfolder, name, overwrite=False): + """Upload a file to Dropbox with error handling. + + Args: + dbx: Dropbox client instance + fullname: Full path to the local file to upload + folder: Dropbox folder name + subfolder: Subfolder within Dropbox folder (can be empty) + name: Filename to use in Dropbox + overwrite: Whether to overwrite if file exists + + Returns: + dict: Upload result with status information containing: + - status: 'success' or 'error' + - file_path: Dropbox path of uploaded file + - bytes_uploaded: Number of bytes uploaded + - message: Description of result + """ + path = '/%s/%s/%s' % (folder, subfolder.replace(os.path.sep, '/'), name) + while '//' in path: + path = path.replace('//', '/') + + mode = (dropbox.files.WriteMode.overwrite + if overwrite + else dropbox.files.WriteMode.add) + + try: + # Get file size before upload + file_size = os.path.getsize(fullname) + mtime = os.path.getmtime(fullname) + + with open(fullname, 'rb') as f: + data = f.read() + + with stopwatch('upload %d bytes' % len(data)): + try: + res = dbx.files_upload( + data, path, mode, + client_modified=datetime.datetime(*time.gmtime(mtime)[:6]), + mute=True) + print('Uploaded %s (%d bytes) -> %s' % (name, file_size, res.name)) + return { + 'status': 'success', + 'file_path': path, + 'bytes_uploaded': file_size, + 'message': 'File uploaded successfully' + } + except dropbox.exceptions.ApiError as err: + error_msg = str(err) + if 'path' in error_msg and 'not_found' in error_msg: + print('Error: Could not create subfolder, creating...') + # Try creating parent folder first + parent_path = '/'.join(path.split('/')[:-1]) + try: + dbx.files_create_folder_v2(parent_path) + print('Created folder:', parent_path) + # Retry upload + res = dbx.files_upload( + data, path, mode, + client_modified=datetime.datetime(*time.gmtime(mtime)[:6]), + mute=True) + print('Uploaded %s (%d bytes) -> %s' % (name, file_size, res.name)) + return { + 'status': 'success', + 'file_path': path, + 'bytes_uploaded': file_size, + 'message': 'File uploaded successfully after folder creation' + } + except Exception as folder_err: + print('*** Failed to create folder:', folder_err) + return { + 'status': 'error', + 'file_path': path, + 'bytes_uploaded': 0, + 'message': 'Failed to create folder: ' + str(folder_err) + } + else: + print('*** API error uploading %s: %s' % (name, error_msg)) + return { + 'status': 'error', + 'file_path': path, + 'bytes_uploaded': 0, + 'message': 'Upload failed: ' + error_msg + } + except IOError as err: + print('*** IO error reading file %s: %s' % (fullname, err)) + return { + 'status': 'error', + 'file_path': path, + 'bytes_uploaded': 0, + 'message': 'Failed to read file: ' + str(err) + } + except Exception as err: + print('*** Unexpected error uploading %s: %s' % (fullname, err)) + return { + 'status': 'error', + 'file_path': path, + 'bytes_uploaded': 0, + 'message': 'Unexpected error: ' + str(err) + } + +>>>>>>> Stashed changes def list_folder(dbx, folder, subfolder): """List a folder. + Return a dict mapping unicode filenames to FileMetadata|FolderMetadata entries. + + Args: + dbx: Dropbox client instance + folder: Dropbox folder name + subfolder: Subfolder within Dropbox folder (can be empty) + + Returns: + dict: Dict mapping filenames to FileMetadata or FolderMetadata entries """ path = '/%s/%s' % (folder, subfolder.replace(os.path.sep, '/')) while '//' in path: @@ -143,7 +354,17 @@ def list_folder(dbx, folder, subfolder): def download(dbx, folder, subfolder, name): """Download a file. + Return the bytes of the file, or None if it doesn't exist. + + Args: + dbx: Dropbox client instance + folder: Dropbox folder name + subfolder: Subfolder within Dropbox folder (can be empty) + name: Filename to download + + Returns: + bytes|None: File contents if successful, None if download failed """ path = '/%s/%s/%s' % (folder, subfolder.replace(os.path.sep, '/'), name) while '//' in path: @@ -185,6 +406,7 @@ def upload(dbx, fullname, folder, subfolder, name, overwrite=False): def yesno(message, default, args): """Handy helper function to ask a yes/no question. + Command line arguments --yes or --no force the answer; --default to force the default answer. Otherwise a blank line returns the default, and answering @@ -193,6 +415,14 @@ def yesno(message, default, args): Special answers: - q or quit exits the program - p or pdb invokes the debugger + + Args: + message: Question to ask + default: Default answer (True or False) + args: Parsed command-line arguments + + Returns: + bool: User's answer (True or False) """ if args.default: print(message + '? [auto]', 'Y' if default else 'N') @@ -225,7 +455,11 @@ def yesno(message, default, args): @contextlib.contextmanager def stopwatch(message): - """Context manager to print how long a block of code took.""" + """Context manager to print how long a block of code took. + + Args: + message: Description of operation being timed + """ t0 = time.time() try: yield @@ -233,5 +467,99 @@ def stopwatch(message): t1 = time.time() print('Total elapsed time for %s: %.3f' % (message, t1 - t0)) +<<<<<<< Updated upstream +======= + +def main(): + """Main program. + + Parse command line, then iterate over files and directories under + rootdir and upload all files. Skips some temporary files and + directories, and avoids duplicate uploads by comparing size and + mtime with the server. + """ + args = parse_args() + if sum([bool(b) for b in (args.yes, args.no, args.default)]) > 1: + print('At most one of --yes, --no, --default is allowed') + sys.exit(2) + if not args.token: + print('--token is mandatory') + sys.exit(2) + + folder = args.folder + rootdir = os.path.expanduser(args.rootdir) + print('Dropbox folder name:', folder) + print('Local directory:', rootdir) + if not os.path.exists(rootdir): + print(rootdir, 'does not exist on your filesystem') + sys.exit(1) + elif not os.path.isdir(rootdir): + print(rootdir, 'is not a folder on your filesystem') + sys.exit(1) + + dbx = dropbox.Dropbox(args.token) + + file_count = 0 + for dn, dirs, files in os.walk(rootdir): + subfolder = dn[len(rootdir):].strip(os.path.sep) + listing = list_folder(dbx, folder, subfolder) + print('Descending into', subfolder, '...') + + # First do all the files. + for name in files: + if args.count and file_count >= args.count: + print('Reached maximum file count:', args.count) + break + + fullname = os.path.join(dn, name) + if should_skip_file(name): + print('Skipping file:', name) + continue + + nname = name # Already a string, no unicode conversion needed + if nname in listing: + md = listing[nname] + mtime = os.path.getmtime(fullname) + mtime_dt = datetime.datetime(*time.gmtime(mtime)[:6]) + size = os.path.getsize(fullname) + if (isinstance(md, dropbox.files.FileMetadata) and + mtime_dt == md.client_modified and size == md.size): + print(name, 'is already synced [stats match]') + else: + print(name, 'exists with different stats, downloading') + res = download(dbx, folder, subfolder, name) + with open(fullname) as f: + data = f.read() + if res == data: + print(name, 'is already synced [content match]') + else: + print(name, 'has changed since last sync') + if yesno('Refresh %s' % name, False, args): + result = upload(dbx, fullname, folder, subfolder, name, + overwrite=True) + if result['status'] == 'success': + file_count += 1 + elif yesno('Upload %s' % name, True, args): + result = upload(dbx, fullname, folder, subfolder, name) + if result['status'] == 'success': + file_count += 1 + + if args.count and file_count >= args.count: + break + + # Then choose which subdirectories to traverse. + keep = [] + for name in dirs: + if should_skip_file(name): + continue + if yesno('Descend into %s' % name, True, args): + print('Keeping directory:', name) + keep.append(name) + else: + print('OK, skipping directory:', name) + dirs[:] = keep + + +>>>>>>> Stashed changes if __name__ == '__main__': main() diff --git a/tests/test_dropbox_error_handling.py b/tests/test_dropbox_error_handling.py new file mode 100755 index 0000000..4611f3f --- /dev/null +++ b/tests/test_dropbox_error_handling.py @@ -0,0 +1,220 @@ +"""Test dropbox_upload.py error handling and integration. + +Tests upload return values, error scenarios, and integration with main(). +""" +import sys +import os +from unittest.mock import MagicMock + +# Add repo root to path +sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + +# First, set up real exception classes before mocking +class ApiError(Exception): + """Mock Dropbox API error.""" + pass + +class HttpError(Exception): + """Mock Dropbox HTTP error.""" + pass + +# Mock dropbox modules - but set exceptions on them +mock_dropbox = MagicMock() +mock_dropbox.files = MagicMock() +mock_dropbox.exceptions = MagicMock() +mock_dropbox.exceptions.ApiError = ApiError +mock_dropbox.exceptions.HttpError = HttpError + +sys.modules['dropbox'] = mock_dropbox +sys.modules['dropbox.files'] = mock_dropbox.files +sys.modules['dropbox.exceptions'] = mock_dropbox.exceptions + +import dropbox_upload + + +def test_upload_success_response(): + """Test that upload returns a success dict on successful upload.""" + # Create a temporary file for the test + import tempfile + with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.jpg') as f: + f.write('test content') + temp_path = f.name + + try: + mock_dbx = MagicMock() + mock_metadata = MagicMock() + mock_metadata.name = 'test.jpg' + mock_dbx.files_upload.return_value = mock_metadata + + result = dropbox_upload.upload( + mock_dbx, temp_path, 'Photos', '', 'test.jpg', overwrite=False + ) + + assert result['status'] == 'success' + assert result['file_path'] == '/Photos/test.jpg' + assert result['bytes_uploaded'] > 0 + assert result['message'] == 'File uploaded successfully' + finally: + os.unlink(temp_path) + + +def test_upload_error_response(): + """Test that upload returns an error dict on failure.""" + import tempfile + with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.jpg') as f: + f.write('test content') + temp_path = f.name + + try: + mock_dbx = MagicMock() + mock_dbx.files_upload.side_effect = Exception('Connection error') + + result = dropbox_upload.upload( + mock_dbx, temp_path, 'Photos', '', 'test.jpg' + ) + + assert result['status'] == 'error' + assert result['file_path'] == '/Photos/test.jpg' + assert result['bytes_uploaded'] == 0 + assert 'error' in result['message'].lower() + finally: + os.unlink(temp_path) + + +def test_upload_api_error_folder_not_found(): + """Test upload handles folder not found API error.""" + import tempfile + with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.jpg') as f: + f.write('test content') + temp_path = f.name + + try: + mock_dbx = MagicMock() + api_error = MagicMock() + api_error.__str__ = MagicMock(return_value='path:not_found/parent/foobar') + mock_dbx.files_upload.side_effect = api_error + + # Mock the folder creation to succeed + mock_dbx.files_create_folder_v2.return_value = None + + result = dropbox_upload.upload( + mock_dbx, temp_path, 'Photos', 'subdir', 'test.jpg' + ) + + # Should retry and succeed + assert result['status'] == 'success' + finally: + os.unlink(temp_path) + + +def test_upload_io_error(): + """Test upload handles file read errors.""" + mock_dbx = MagicMock() + + # Upload a non-existent file to trigger IOError + result = dropbox_upload.upload( + mock_dbx, '/tmp/nonexistent_file_xyz.jpg', 'Photos', '', 'test.jpg' + ) + + assert result['status'] == 'error' + assert 'Failed to read file' in result['message'] + + +def test_parse_args_count_parameter(): + """Test parse_args handles --count parameter.""" + args = dropbox_upload.parse_args(['--count', '5', 'Albums', '/home/user/Pics']) + assert args.count == 5 + assert args.folder == 'Albums' + assert args.rootdir == '/home/user/Pics' + + +def test_parse_args_conflict_validation(): + """Test that --yes and --no together causes error in main logic.""" + # This test validates the conflict check logic + args = dropbox_upload.parse_args(['--yes', '--no']) + conflict_count = sum([bool(b) for b in (args.yes, args.no, args.default)]) + assert conflict_count == 2 # Both yes and no are True + + +def test_should_skip_file_all_patterns(): + """Comprehensive test of all skip patterns.""" + # These should all be skipped + skipped_patterns = [ + '.gitignore', '.bashrc', '.env', # Dot files + 'file.tmp', 'backup.temp', # Temp extensions + '~backup', '@temp', '@copy', # Start with ~ or @ + 'module.pyc', 'script.pyo', # Python bytecode + '__pycache__', # Python cache dir + '', # Empty string + None, # None value + ] + + for pattern in skipped_patterns: + assert dropbox_upload.should_skip_file(pattern) is True, \ + f"Pattern '{pattern}' should be skipped" + + +def test_should_skip_file_preserved_patterns(): + """Test files that should NOT be skipped.""" + preserved_patterns = [ + 'photo.jpg', 'image.png', 'video.mp4', # Media files + 'document.pdf', 'report.docx', # Documents + 'test.PYC', 'script.PYTHON', # Case-sensitive (not filtered) + 'data_2024.txt', 'backup-2024.bak', # Files with underscores/dashes + ] + + for pattern in preserved_patterns: + assert dropbox_upload.should_skip_file(pattern) is False, \ + f"Pattern '{pattern}' should NOT be skipped" + + +def test_list_folder_empty_result(): + """Test list_folder returns empty dict on API error.""" + mock_dbx = MagicMock() + mock_dbx.files_list_folder.side_effect = ApiError('API error') + + result = dropbox_upload.list_folder(mock_dbx, 'Photos', '') + assert result == {} + + +def test_list_folder_with_entries(): + """Test list_folder correctly builds result dict from API entries.""" + mock_dbx = MagicMock() + + mock_entry1 = MagicMock() + mock_entry1.name = 'photo1.jpg' + mock_entry1.__class__ = MagicMock() + + mock_entry2 = MagicMock() + mock_entry2.name = 'photo2.jpg' + + mock_result = MagicMock() + mock_result.entries = [mock_entry1, mock_entry2] + mock_dbx.files_list_folder.return_value = mock_result + + result = dropbox_upload.list_folder(mock_dbx, 'Photos', '') + + assert 'photo1.jpg' in result + assert 'photo2.jpg' in result + assert len(result) == 2 + + +def test_download_success(): + """Test download returns file bytes on success.""" + mock_dbx = MagicMock() + mock_metadata = MagicMock() + mock_response = MagicMock() + mock_response.content = b'test file content' + mock_dbx.files_download.return_value = (mock_metadata, mock_response) + + result = dropbox_upload.download(mock_dbx, 'Photos', '', 'test.jpg') + assert result == b'test file content' + + +def test_download_failure(): + """Test download returns None on HTTP error.""" + mock_dbx = MagicMock() + mock_dbx.files_download.side_effect = HttpError('HTTP error') + + result = dropbox_upload.download(mock_dbx, 'Photos', '', 'test.jpg') + assert result is None