gitea_admin/sientia-dataops-model-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1350: Refactor: Replace AsyncMock with MagicMock in cleanup tests for synchronous calls

Browse Source
This commit is contained in:
Bruno Domingues
2025-11-25 16:54:45 -03:00
parent 33480fa108
commit 823df2d65d
1 changed files with 7 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
tests/activities/test_cleanup.py
View File

@@ -226,7 +226,7 @@ def test_cleanup_minio_files_delete_error(
metrics_controller=mock_metrics_controller, metrics_controller=mock_metrics_controller,
) )
cleanup._emit_metrics = AsyncMock() cleanup._emit_metrics = AsyncMock()
cleanup.error = AsyncMock() cleanup.error = MagicMock()
old_ts = int((datetime.now(UTC) - timedelta(hours=48)).timestamp() * 1000) old_ts = int((datetime.now(UTC) - timedelta(hours=48)).timestamp() * 1000)
mock_storage_repository.list_bucket_objects.return_value = [f'{old_ts}-old-file.txt'] mock_storage_repository.list_bucket_objects.return_value = [f'{old_ts}-old-file.txt']
@@ -253,7 +253,7 @@ def test_cleanup_minio_files_exception_handling(
notification_handler=mock_notification_handler, notification_handler=mock_notification_handler,
metrics_controller=mock_metrics_controller, metrics_controller=mock_metrics_controller,
) )
cleanup.send_notification = AsyncMock() cleanup.send_notification = MagicMock()
cleanup._emit_metrics = AsyncMock() cleanup._emit_metrics = AsyncMock()
mock_storage_repository.list_bucket_objects.side_effect = Exception('Connection Error') mock_storage_repository.list_bucket_objects.side_effect = Exception('Connection Error')
@@ -284,7 +284,7 @@ def test_cleanup_temp_directories_nonexistent_path(
metrics_controller=mock_metrics_controller, metrics_controller=mock_metrics_controller,
) )
cleanup._emit_metrics = AsyncMock() cleanup._emit_metrics = AsyncMock()
cleanup.warning = AsyncMock() cleanup.warning = MagicMock()
asyncio.run( asyncio.run(
cleanup.cleanup_temp_directories({'temp_path': '/nonexistent/path', 'metadata': {}}) cleanup.cleanup_temp_directories({'temp_path': '/nonexistent/path', 'metadata': {}})
@@ -382,7 +382,7 @@ def test_cleanup_temp_directories_delete_error(
metrics_controller=mock_metrics_controller, metrics_controller=mock_metrics_controller,
) )
cleanup._emit_metrics = AsyncMock() cleanup._emit_metrics = AsyncMock()
cleanup.error = AsyncMock() cleanup.error = MagicMock()
old_time = (datetime.now() - timedelta(hours=48)).strftime('%Y%m%d_%H%M%S_000000') old_time = (datetime.now() - timedelta(hours=48)).strftime('%Y%m%d_%H%M%S_000000')
old_dir = os.path.join(temp_dir, f'old_dir_{old_time}') old_dir = os.path.join(temp_dir, f'old_dir_{old_time}')
@@ -447,7 +447,7 @@ def test_cleanup_temp_directories_with_files_and_unmatched_dirs(
metrics_controller=mock_metrics_controller, metrics_controller=mock_metrics_controller,
) )
cleanup._emit_metrics = AsyncMock() cleanup._emit_metrics = AsyncMock()
cleanup.debug = AsyncMock() cleanup.debug = MagicMock()
# Create a file and a directory with a non-matching name # Create a file and a directory with a non-matching name
with open(os.path.join(temp_dir, 'a_file.txt'), 'w') as f: with open(os.path.join(temp_dir, 'a_file.txt'), 'w') as f:
@@ -483,7 +483,7 @@ def test_cleanup_temp_directories_invalid_timestamp_format(
metrics_controller=mock_metrics_controller, metrics_controller=mock_metrics_controller,
) )
cleanup._emit_metrics = AsyncMock() cleanup._emit_metrics = AsyncMock()
cleanup.error = AsyncMock() cleanup.error = MagicMock()
# Create a directory with a malformed timestamp that matches the regex but fails parsing # Create a directory with a malformed timestamp that matches the regex but fails parsing
malformed_dir_name = 'dir_20239999_999999_999999' malformed_dir_name = 'dir_20239999_999999_999999'
@@ -515,7 +515,7 @@ def test_cleanup_temp_directories_generic_exception(
metrics_controller=mock_metrics_controller, metrics_controller=mock_metrics_controller,
) )
cleanup._emit_metrics = AsyncMock() cleanup._emit_metrics = AsyncMock()
cleanup.send_notification = AsyncMock() cleanup.send_notification = MagicMock()
with patch('os.listdir', side_effect=Exception('Unexpected OS Error')): with patch('os.listdir', side_effect=Exception('Unexpected OS Error')):
with pytest.raises(Exception, match='Unexpected OS Error'): with pytest.raises(Exception, match='Unexpected OS Error'):