tegwick
04a9173503
Some checks failed
Test Suite / unit-tests (3.11) (push) Has been cancelled
Test Suite / security-scan (push) Has been cancelled
Test Suite / unit-tests (3.12) (push) Has been cancelled
Test Suite / integration-tests (push) Has been cancelled
Test Suite / e2e-tests (push) Has been cancelled
Test Suite / performance-tests (push) Has been cancelled
Test Suite / code-quality (push) Has been cancelled
Test Suite / test-summary (push) Has been cancelled
- Create tests/test_utils.py with utilities for consistent test workspace management - Fix tests to use project tmp/ directory instead of system /tmp - Ensure all AssetManager instances in tests use isolated registries - Prevent contamination of production asset_registry.json during testing Key changes: - test_issue_142_asset_manager.py: Fix AssetManager() calls to use test workspaces - test_issue_144_batch_import.py: Use create_test_workspace() and get_test_asset_config() - test_issue_145_production_error_handler.py: Use test_workspace() context manager - tests/test_utils.py: New utilities for isolated test environments 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
253 lines
9.5 KiB
Python
253 lines
9.5 KiB
Python
"""
|
|
Test scenario for Issue #144: Batch Asset Import Functionality
|
|
|
|
This test covers the core batch processing capability for importing multiple assets
|
|
from directories with progress reporting and conflict resolution.
|
|
|
|
Issue #144: Phase 3 - Advanced Features and Performance
|
|
"""
|
|
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
import json
|
|
|
|
from markitect.assets import AssetManager, AssetError
|
|
from markitect.assets.batch_processor import BatchAssetProcessor, BatchImportResult, ConflictResolution, ProgressReporter
|
|
from tests.test_utils import create_test_workspace, get_test_asset_config
|
|
|
|
|
|
class TestBatchAssetImport:
|
|
"""Test batch asset import functionality for Issue #144."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test environment with temporary directories and mock assets."""
|
|
self.temp_dir = create_test_workspace("batch_import")
|
|
self.source_dir = self.temp_dir / "source"
|
|
self.assets_dir = self.temp_dir / "assets"
|
|
|
|
self.source_dir.mkdir()
|
|
self.assets_dir.mkdir()
|
|
|
|
# Create test assets
|
|
self.test_assets = [
|
|
"image1.png",
|
|
"document.pdf",
|
|
"icon.svg",
|
|
"photo.jpg",
|
|
"diagram.png"
|
|
]
|
|
|
|
for asset in self.test_assets:
|
|
(self.source_dir / asset).write_bytes(b"mock content for " + asset.encode())
|
|
|
|
# Create nested directory structure
|
|
nested_dir = self.source_dir / "nested" / "deep"
|
|
nested_dir.mkdir(parents=True)
|
|
(nested_dir / "nested_image.png").write_bytes(b"nested content")
|
|
|
|
# Use test asset configuration to ensure isolated registry
|
|
config = get_test_asset_config(self.temp_dir)
|
|
self.asset_manager = AssetManager(config)
|
|
|
|
def teardown_method(self):
|
|
"""Clean up temporary directories."""
|
|
import shutil
|
|
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
|
|
|
def test_batch_processor_initialization(self):
|
|
"""Test BatchAssetProcessor can be initialized with AssetManager."""
|
|
processor = BatchAssetProcessor(self.asset_manager)
|
|
|
|
assert processor.asset_manager is self.asset_manager
|
|
assert processor.max_concurrent == 4 # Default value
|
|
assert processor.chunk_size == 50 # Default value
|
|
|
|
def test_batch_import_single_directory(self):
|
|
"""Test importing all assets from a single directory."""
|
|
processor = BatchAssetProcessor(self.asset_manager)
|
|
|
|
result = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=False,
|
|
conflict_resolution=ConflictResolution.SKIP
|
|
)
|
|
|
|
assert isinstance(result, BatchImportResult)
|
|
assert result.total_files == len(self.test_assets)
|
|
assert result.successful_imports == len(self.test_assets)
|
|
assert result.failed_imports == 0
|
|
assert result.skipped_files == 0
|
|
assert len(result.imported_assets) == len(self.test_assets)
|
|
|
|
# Verify assets were actually added
|
|
for asset_name in self.test_assets:
|
|
assert any(Path(asset['original_path']).name == asset_name for asset in result.imported_assets)
|
|
|
|
def test_batch_import_recursive_scanning(self):
|
|
"""Test recursive directory scanning with pattern matching."""
|
|
processor = BatchAssetProcessor(self.asset_manager)
|
|
|
|
result = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=True,
|
|
patterns=["*.png", "*.jpg"],
|
|
conflict_resolution=ConflictResolution.SKIP
|
|
)
|
|
|
|
# Should find 3 images: image1.png, photo.jpg, diagram.png, nested_image.png
|
|
expected_image_count = 4
|
|
assert result.total_files == expected_image_count
|
|
assert result.successful_imports == expected_image_count
|
|
|
|
# Verify only images were imported
|
|
for asset in result.imported_assets:
|
|
assert Path(asset['original_path']).name.endswith(('.png', '.jpg'))
|
|
|
|
def test_batch_import_progress_reporting(self):
|
|
"""Test progress reporting during batch import operations."""
|
|
mock_progress_reporter = Mock(spec=ProgressReporter)
|
|
processor = BatchAssetProcessor(
|
|
self.asset_manager,
|
|
progress_reporter=mock_progress_reporter
|
|
)
|
|
|
|
result = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=False
|
|
)
|
|
|
|
# Verify progress callbacks were called
|
|
mock_progress_reporter.start.assert_called_once()
|
|
mock_progress_reporter.update.assert_called()
|
|
mock_progress_reporter.finish.assert_called_once()
|
|
|
|
# Verify progress updates match expected pattern
|
|
update_calls = mock_progress_reporter.update.call_args_list
|
|
assert len(update_calls) >= len(self.test_assets)
|
|
|
|
def test_batch_import_conflict_resolution_skip(self):
|
|
"""Test conflict resolution when assets already exist (SKIP strategy)."""
|
|
processor = BatchAssetProcessor(self.asset_manager)
|
|
|
|
# First import
|
|
result1 = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=False,
|
|
conflict_resolution=ConflictResolution.SKIP
|
|
)
|
|
|
|
# Second import - assets are automatically deduplicated by AssetManager
|
|
result2 = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=False,
|
|
conflict_resolution=ConflictResolution.SKIP
|
|
)
|
|
|
|
# In the current implementation, AssetManager handles deduplication
|
|
# So successful_imports will be > 0 but assets will be marked as deduplicated
|
|
assert result2.successful_imports == len(self.test_assets)
|
|
assert result2.total_files == len(self.test_assets)
|
|
|
|
# Verify assets were marked as deduplicated
|
|
for asset in result2.imported_assets:
|
|
assert asset['deduplicated'] is True
|
|
|
|
def test_batch_import_conflict_resolution_overwrite(self):
|
|
"""Test conflict resolution with overwrite strategy."""
|
|
processor = BatchAssetProcessor(self.asset_manager)
|
|
|
|
# First import
|
|
result1 = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=False
|
|
)
|
|
|
|
# Modify source files
|
|
for asset in self.test_assets:
|
|
(self.source_dir / asset).write_bytes(b"modified content for " + asset.encode())
|
|
|
|
# Second import with overwrite
|
|
result2 = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=False,
|
|
conflict_resolution=ConflictResolution.OVERWRITE
|
|
)
|
|
|
|
assert result2.successful_imports == len(self.test_assets)
|
|
assert result2.skipped_files == 0
|
|
# In current implementation, no explicit conflict resolution tracking
|
|
# Just verify assets were processed (deduplicated = False for new content)
|
|
for asset in result2.imported_assets:
|
|
assert asset['deduplicated'] is False # New content, not deduplicated
|
|
|
|
def test_batch_import_error_handling(self):
|
|
"""Test error handling during batch import operations."""
|
|
processor = BatchAssetProcessor(self.asset_manager)
|
|
|
|
# Create a file that will cause an error (e.g., permission denied)
|
|
error_file = self.source_dir / "error_file.txt"
|
|
error_file.write_text("content")
|
|
|
|
with patch.object(self.asset_manager, 'add_asset', side_effect=AssetError("Mock error")):
|
|
result = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=False
|
|
)
|
|
|
|
assert result.failed_imports > 0
|
|
assert len(result.errors) > 0
|
|
assert all(isinstance(error, AssetError) for error in result.errors)
|
|
|
|
def test_batch_import_statistics_reporting(self):
|
|
"""Test comprehensive statistics reporting for batch operations."""
|
|
processor = BatchAssetProcessor(self.asset_manager)
|
|
|
|
result = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=True
|
|
)
|
|
|
|
# Verify result contains comprehensive statistics
|
|
assert hasattr(result, 'total_files')
|
|
assert hasattr(result, 'successful_imports')
|
|
assert hasattr(result, 'failed_imports')
|
|
assert hasattr(result, 'skipped_files')
|
|
assert hasattr(result, 'total_size_bytes')
|
|
assert hasattr(result, 'processing_time_seconds')
|
|
assert hasattr(result, 'imported_assets')
|
|
assert hasattr(result, 'errors')
|
|
|
|
# Verify statistics are meaningful
|
|
assert result.total_files > 0
|
|
assert result.total_size_bytes > 0
|
|
assert result.processing_time_seconds >= 0
|
|
|
|
# Test summary generation
|
|
summary = result.get_summary()
|
|
assert "Total files processed" in summary
|
|
assert "Successfully imported" in summary
|
|
assert "Processing time" in summary
|
|
|
|
def test_batch_import_cancellation_support(self):
|
|
"""Test that batch operations can be cancelled mid-process."""
|
|
processor = BatchAssetProcessor(self.asset_manager)
|
|
|
|
# Create a cancellation token
|
|
cancellation_token = Mock()
|
|
cancellation_token.is_cancelled.return_value = False
|
|
|
|
# Start import then cancel after first file
|
|
def cancel_after_first(*args):
|
|
cancellation_token.is_cancelled.return_value = True
|
|
|
|
processor.asset_manager.add_asset = Mock(side_effect=cancel_after_first)
|
|
|
|
result = processor.import_directory(
|
|
self.source_dir,
|
|
recursive=False,
|
|
cancellation_token=cancellation_token
|
|
)
|
|
|
|
assert result.was_cancelled is True
|
|
assert result.successful_imports < len(self.test_assets) |