fix: isolate test artifacts from production asset registry
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
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>
This commit is contained in:
@@ -28,26 +28,7 @@ from markitect.assets.deduplicator import AssetDeduplicator
|
||||
from markitect.assets.packager import MarkdownPackager
|
||||
from markitect.assets.exceptions import AssetError, AssetManagerError
|
||||
from markitect.config_manager import ConfigurationManager
|
||||
|
||||
|
||||
@contextmanager
|
||||
def test_workspace():
|
||||
"""Create a test workspace in the project tmp directory."""
|
||||
project_root = Path(__file__).parent.parent
|
||||
temp_dir = project_root / "tmp" / "test_artifacts" / f"test_{int(time.time())}"
|
||||
temp_dir.mkdir(parents=True, exist_ok=True)
|
||||
try:
|
||||
yield temp_dir
|
||||
finally:
|
||||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||||
|
||||
|
||||
def create_test_workspace():
|
||||
"""Create a test workspace in the project tmp directory."""
|
||||
project_root = Path(__file__).parent.parent
|
||||
temp_dir = project_root / "tmp" / "test_artifacts" / f"test_{int(time.time())}"
|
||||
temp_dir.mkdir(parents=True, exist_ok=True)
|
||||
return temp_dir
|
||||
from tests.test_utils import test_workspace as test_workspace_util, create_test_workspace, get_test_asset_config
|
||||
|
||||
|
||||
class TestAssetManagerInitialization:
|
||||
@@ -78,16 +59,31 @@ class TestAssetManagerInitialization:
|
||||
|
||||
def test_manager_initialization_with_defaults(self):
|
||||
"""Test AssetManager initialization with default configuration."""
|
||||
manager = AssetManager()
|
||||
temp_dir = create_test_workspace()
|
||||
|
||||
# Should use reasonable defaults
|
||||
assert manager.storage_path.name == "assets"
|
||||
assert manager.registry_path.name == "asset_registry.json"
|
||||
assert manager.enable_deduplication is True
|
||||
try:
|
||||
# Change to temp directory to test defaults without contaminating production
|
||||
import os
|
||||
original_cwd = os.getcwd()
|
||||
os.chdir(temp_dir)
|
||||
|
||||
manager = AssetManager()
|
||||
|
||||
# Should use reasonable defaults
|
||||
assert manager.storage_path.name == "assets"
|
||||
assert manager.registry_path.name == "asset_registry.json"
|
||||
assert manager.enable_deduplication is True
|
||||
|
||||
# Verify it's using the test directory, not production
|
||||
assert str(manager.registry_path).startswith(str(temp_dir))
|
||||
|
||||
finally:
|
||||
os.chdir(original_cwd)
|
||||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||||
|
||||
def test_manager_creates_required_components(self):
|
||||
"""Test that AssetManager creates required component instances."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -103,12 +99,13 @@ class TestAssetManagerInitialization:
|
||||
|
||||
def test_manager_integration_with_config_manager(self):
|
||||
"""Test AssetManager integration with ConfigurationManager."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
# Create config file
|
||||
config_file = Path(temp_dir) / ".markitect.json"
|
||||
config_data = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "custom_assets"),
|
||||
"registry_path": str(temp_dir / "test_registry.json"),
|
||||
"enable_deduplication": False
|
||||
}
|
||||
}
|
||||
@@ -127,7 +124,7 @@ class TestAssetManagerHighLevelOperations:
|
||||
|
||||
def test_add_asset_with_deduplication(self):
|
||||
"""Test adding asset with automatic deduplication."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -151,7 +148,7 @@ class TestAssetManagerHighLevelOperations:
|
||||
|
||||
def test_add_duplicate_asset_detected(self):
|
||||
"""Test that duplicate assets are properly detected and handled."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -180,7 +177,7 @@ class TestAssetManagerHighLevelOperations:
|
||||
|
||||
def test_list_assets_with_metadata(self):
|
||||
"""Test listing all assets with their metadata."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -210,7 +207,7 @@ class TestAssetManagerHighLevelOperations:
|
||||
|
||||
def test_get_asset_info_by_hash(self):
|
||||
"""Test retrieving detailed asset information by content hash."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -237,7 +234,7 @@ class TestAssetManagerHighLevelOperations:
|
||||
|
||||
def test_remove_asset_by_hash(self):
|
||||
"""Test removing asset by content hash."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -269,7 +266,7 @@ class TestAssetManagerPackaging:
|
||||
|
||||
def test_create_document_package(self):
|
||||
"""Test creating complete document package with assets."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -312,7 +309,7 @@ This document has assets:
|
||||
|
||||
def test_extract_document_package_to_workspace(self):
|
||||
"""Test extracting package to workspace with proper asset linking."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -344,7 +341,7 @@ This document has assets:
|
||||
|
||||
def test_package_with_custom_options(self):
|
||||
"""Test package creation with custom options and exclude patterns."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -388,7 +385,7 @@ class TestAssetManagerErrorHandling:
|
||||
|
||||
def test_add_nonexistent_asset_raises_error(self):
|
||||
"""Test that adding non-existent asset raises appropriate error."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -405,7 +402,7 @@ class TestAssetManagerErrorHandling:
|
||||
|
||||
def test_get_info_for_nonexistent_asset_raises_error(self):
|
||||
"""Test that getting info for non-existent asset raises error."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -420,7 +417,7 @@ class TestAssetManagerErrorHandling:
|
||||
|
||||
def test_manager_logs_operations(self):
|
||||
"""Test that AssetManager logs important operations."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -454,7 +451,7 @@ class TestAssetManagerErrorHandling:
|
||||
def test_configuration_validation_errors(self):
|
||||
"""Test that invalid configuration raises appropriate errors."""
|
||||
# Invalid storage path (file instead of directory)
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
invalid_file = Path(temp_dir) / "not_a_directory.txt"
|
||||
invalid_file.write_text("This is a file")
|
||||
|
||||
@@ -474,7 +471,7 @@ class TestAssetManagerWorkflows:
|
||||
|
||||
def test_complete_document_workflow(self):
|
||||
"""Test complete workflow: add assets, create package, extract elsewhere."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -524,7 +521,7 @@ Assets:
|
||||
|
||||
def test_asset_sharing_between_packages(self):
|
||||
"""Test that assets can be shared between different packages."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
@@ -573,7 +570,7 @@ Assets:
|
||||
|
||||
def test_performance_requirements_met(self):
|
||||
"""Test that operations complete within performance requirements (<100ms)."""
|
||||
with test_workspace() as temp_dir:
|
||||
with test_workspace_util() as temp_dir:
|
||||
config = {
|
||||
"assets": {
|
||||
"storage_path": str(temp_dir / "assets"),
|
||||
|
||||
@@ -8,14 +8,13 @@ Issue #144: Phase 3 - Advanced Features and Performance
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import tempfile
|
||||
import shutil
|
||||
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:
|
||||
@@ -23,9 +22,9 @@ class TestBatchAssetImport:
|
||||
|
||||
def setup_method(self):
|
||||
"""Set up test environment with temporary directories and mock assets."""
|
||||
self.temp_dir = tempfile.mkdtemp()
|
||||
self.source_dir = Path(self.temp_dir) / "source"
|
||||
self.assets_dir = Path(self.temp_dir) / "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()
|
||||
@@ -47,16 +46,14 @@ class TestBatchAssetImport:
|
||||
nested_dir.mkdir(parents=True)
|
||||
(nested_dir / "nested_image.png").write_bytes(b"nested content")
|
||||
|
||||
self.asset_manager = AssetManager(config={
|
||||
'assets': {
|
||||
'storage_path': str(self.assets_dir),
|
||||
'registry_path': str(self.assets_dir / 'registry.json')
|
||||
}
|
||||
})
|
||||
# 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."""
|
||||
shutil.rmtree(self.temp_dir)
|
||||
import shutil
|
||||
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
||||
|
||||
def test_batch_processor_initialization(self):
|
||||
"""Test BatchAssetProcessor can be initialized with AssetManager."""
|
||||
|
||||
@@ -7,8 +7,6 @@ for production environments.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import tempfile
|
||||
import shutil
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
@@ -21,6 +19,7 @@ from markitect.production.error_handler import (
|
||||
RegistryCorruptionError,
|
||||
ResourceExhaustionError
|
||||
)
|
||||
from tests.test_utils import test_workspace
|
||||
|
||||
|
||||
class TestProductionErrorHandler:
|
||||
@@ -29,9 +28,8 @@ class TestProductionErrorHandler:
|
||||
@pytest.fixture
|
||||
def temp_workspace(self):
|
||||
"""Create temporary workspace for testing."""
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
yield Path(temp_dir)
|
||||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||||
with test_workspace("error_handler") as temp_dir:
|
||||
yield temp_dir
|
||||
|
||||
@pytest.fixture
|
||||
def error_handler(self, temp_workspace):
|
||||
|
||||
98
tests/test_utils.py
Normal file
98
tests/test_utils.py
Normal file
@@ -0,0 +1,98 @@
|
||||
"""
|
||||
Test utilities for consistent test workspace and registry management.
|
||||
|
||||
This module provides utilities to ensure all tests use project tmp/ directory
|
||||
and isolated test registries to prevent contamination of production assets.
|
||||
"""
|
||||
|
||||
import time
|
||||
import shutil
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
from markitect.assets.constants import (
|
||||
DEFAULT_TEST_ASSETS_DIR,
|
||||
DEFAULT_TEST_REGISTRY_FILENAME
|
||||
)
|
||||
|
||||
|
||||
def get_project_root() -> Path:
|
||||
"""Get the project root directory."""
|
||||
return Path(__file__).parent.parent
|
||||
|
||||
|
||||
def create_test_workspace(prefix: str = "test") -> Path:
|
||||
"""Create a test workspace in the project tmp directory.
|
||||
|
||||
Args:
|
||||
prefix: Prefix for the test directory name
|
||||
|
||||
Returns:
|
||||
Path to the created test workspace
|
||||
"""
|
||||
project_root = get_project_root()
|
||||
temp_dir = project_root / "tmp" / "test_artifacts" / f"{prefix}_{int(time.time() * 1000000)}"
|
||||
temp_dir.mkdir(parents=True, exist_ok=True)
|
||||
return temp_dir
|
||||
|
||||
|
||||
@contextmanager
|
||||
def test_workspace(prefix: str = "test"):
|
||||
"""Context manager for test workspace that auto-cleans up.
|
||||
|
||||
Args:
|
||||
prefix: Prefix for the test directory name
|
||||
|
||||
Yields:
|
||||
Path to the test workspace
|
||||
"""
|
||||
temp_dir = create_test_workspace(prefix)
|
||||
try:
|
||||
yield temp_dir
|
||||
finally:
|
||||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||||
|
||||
|
||||
def get_test_asset_config(workspace_dir: Path,
|
||||
registry_name: Optional[str] = None) -> Dict[str, Any]:
|
||||
"""Get asset configuration for testing that uses isolated test paths.
|
||||
|
||||
Args:
|
||||
workspace_dir: The test workspace directory
|
||||
registry_name: Optional custom registry filename
|
||||
|
||||
Returns:
|
||||
Configuration dictionary for AssetManager
|
||||
"""
|
||||
if registry_name is None:
|
||||
registry_name = "test_registry.json"
|
||||
|
||||
return {
|
||||
"assets": {
|
||||
"storage_path": str(workspace_dir / "assets"),
|
||||
"registry_path": str(workspace_dir / registry_name),
|
||||
"database_path": str(workspace_dir / "test_assets.db"),
|
||||
"enable_deduplication": True,
|
||||
"default_conflict_resolution": "backup"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def ensure_test_tmp_structure():
|
||||
"""Ensure the test tmp directory structure exists."""
|
||||
project_root = get_project_root()
|
||||
test_artifacts_dir = project_root / "tmp" / "test_artifacts"
|
||||
test_artifacts_dir.mkdir(parents=True, exist_ok=True)
|
||||
return test_artifacts_dir
|
||||
|
||||
|
||||
# Backward compatibility functions for existing tests
|
||||
def temp_workspace():
|
||||
"""Backward compatibility alias for test_workspace()."""
|
||||
return test_workspace()
|
||||
|
||||
|
||||
def create_temp_workspace():
|
||||
"""Backward compatibility alias for create_test_workspace()."""
|
||||
return create_test_workspace()
|
||||
Reference in New Issue
Block a user