340 lines
12 KiB
Python
340 lines
12 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Test Config Downloader with Asset Tracking
|
||
|
|
|
||
|
|
This script tests that the config_downloader.py now properly uses
|
||
|
|
asset tracking to avoid re-downloading existing assets.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
# Add the current directory to the path so we can import modules
|
||
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
|
|
||
|
|
from config_downloader import ConfigImageDownloader
|
||
|
|
from asset_tracker import AssetTracker
|
||
|
|
|
||
|
|
|
||
|
|
class ConfigTrackingTester:
|
||
|
|
"""Test class for config downloader asset tracking functionality."""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
"""Initialize the tester."""
|
||
|
|
self.logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
def create_test_config(self, output_dir: str, track_assets: bool = True) -> dict:
|
||
|
|
"""Create a test configuration."""
|
||
|
|
return {
|
||
|
|
"api_url": "https://api.parentzone.me",
|
||
|
|
"list_endpoint": "/v1/media/list",
|
||
|
|
"download_endpoint": "/v1/media",
|
||
|
|
"output_dir": output_dir,
|
||
|
|
"max_concurrent": 2,
|
||
|
|
"timeout": 30,
|
||
|
|
"track_assets": track_assets,
|
||
|
|
"email": "tudor.sitaru@gmail.com",
|
||
|
|
"password": "mTVq8uNUvY7R39EPGVAm@"
|
||
|
|
}
|
||
|
|
|
||
|
|
def test_config_loading(self):
|
||
|
|
"""Test that configuration properly loads asset tracking setting."""
|
||
|
|
print("=" * 60)
|
||
|
|
print("TEST 1: Configuration Loading")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
config_file = Path(temp_dir) / "test_config.json"
|
||
|
|
|
||
|
|
# Test with tracking enabled
|
||
|
|
config_data = self.create_test_config(temp_dir, track_assets=True)
|
||
|
|
with open(config_file, 'w') as f:
|
||
|
|
json.dump(config_data, f, indent=2)
|
||
|
|
|
||
|
|
print("1. Testing config with asset tracking enabled...")
|
||
|
|
downloader = ConfigImageDownloader(str(config_file))
|
||
|
|
|
||
|
|
if downloader.asset_tracker:
|
||
|
|
print(" ✅ Asset tracker initialized successfully")
|
||
|
|
else:
|
||
|
|
print(" ❌ Asset tracker not initialized")
|
||
|
|
return False
|
||
|
|
|
||
|
|
# Test with tracking disabled
|
||
|
|
config_data = self.create_test_config(temp_dir, track_assets=False)
|
||
|
|
with open(config_file, 'w') as f:
|
||
|
|
json.dump(config_data, f, indent=2)
|
||
|
|
|
||
|
|
print("\n2. Testing config with asset tracking disabled...")
|
||
|
|
downloader2 = ConfigImageDownloader(str(config_file))
|
||
|
|
|
||
|
|
if not downloader2.asset_tracker:
|
||
|
|
print(" ✅ Asset tracker correctly disabled")
|
||
|
|
else:
|
||
|
|
print(" ❌ Asset tracker should be disabled")
|
||
|
|
return False
|
||
|
|
|
||
|
|
print("\n✅ Configuration loading test passed!")
|
||
|
|
return True
|
||
|
|
|
||
|
|
def test_config_default_behavior(self):
|
||
|
|
"""Test that asset tracking is enabled by default."""
|
||
|
|
print("\n" + "=" * 60)
|
||
|
|
print("TEST 2: Default Behavior")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
config_file = Path(temp_dir) / "test_config.json"
|
||
|
|
|
||
|
|
# Create config without track_assets field
|
||
|
|
config_data = self.create_test_config(temp_dir)
|
||
|
|
del config_data['track_assets'] # Remove the field entirely
|
||
|
|
|
||
|
|
with open(config_file, 'w') as f:
|
||
|
|
json.dump(config_data, f, indent=2)
|
||
|
|
|
||
|
|
print("1. Testing config without track_assets field (should default to True)...")
|
||
|
|
downloader = ConfigImageDownloader(str(config_file))
|
||
|
|
|
||
|
|
if downloader.asset_tracker:
|
||
|
|
print(" ✅ Asset tracking enabled by default")
|
||
|
|
else:
|
||
|
|
print(" ❌ Asset tracking should be enabled by default")
|
||
|
|
return False
|
||
|
|
|
||
|
|
print("\n✅ Default behavior test passed!")
|
||
|
|
return True
|
||
|
|
|
||
|
|
async def test_mock_download_with_tracking(self):
|
||
|
|
"""Test download functionality with asset tracking using mock data."""
|
||
|
|
print("\n" + "=" * 60)
|
||
|
|
print("TEST 3: Mock Download with Tracking")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
config_file = Path(temp_dir) / "test_config.json"
|
||
|
|
|
||
|
|
# Create config with tracking enabled
|
||
|
|
config_data = self.create_test_config(temp_dir, track_assets=True)
|
||
|
|
with open(config_file, 'w') as f:
|
||
|
|
json.dump(config_data, f, indent=2)
|
||
|
|
|
||
|
|
print("1. Creating ConfigImageDownloader with tracking enabled...")
|
||
|
|
downloader = ConfigImageDownloader(str(config_file))
|
||
|
|
|
||
|
|
if not downloader.asset_tracker:
|
||
|
|
print(" ❌ Asset tracker not initialized")
|
||
|
|
return False
|
||
|
|
|
||
|
|
print(" ✅ Config downloader with asset tracker created")
|
||
|
|
|
||
|
|
# Test the asset tracker directly
|
||
|
|
print("\n2. Testing asset tracker integration...")
|
||
|
|
mock_assets = [
|
||
|
|
{
|
||
|
|
"id": "config_test_001",
|
||
|
|
"name": "test_image_1.jpg",
|
||
|
|
"updated": "2024-01-01T10:00:00Z",
|
||
|
|
"size": 1024000,
|
||
|
|
"mimeType": "image/jpeg"
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"id": "config_test_002",
|
||
|
|
"name": "test_image_2.jpg",
|
||
|
|
"updated": "2024-01-02T11:00:00Z",
|
||
|
|
"size": 2048000,
|
||
|
|
"mimeType": "image/jpeg"
|
||
|
|
}
|
||
|
|
]
|
||
|
|
|
||
|
|
# First check - should find all assets as new
|
||
|
|
new_assets = downloader.asset_tracker.get_new_assets(mock_assets)
|
||
|
|
print(f" First check: Found {len(new_assets)} new assets (expected: 2)")
|
||
|
|
|
||
|
|
if len(new_assets) != 2:
|
||
|
|
print(" ❌ Should have found 2 new assets")
|
||
|
|
return False
|
||
|
|
|
||
|
|
# Simulate marking assets as downloaded
|
||
|
|
print("\n3. Simulating asset downloads...")
|
||
|
|
for asset in mock_assets:
|
||
|
|
filepath = Path(temp_dir) / asset['name']
|
||
|
|
filepath.write_text(f"Mock content for {asset['id']}")
|
||
|
|
downloader.asset_tracker.mark_asset_downloaded(asset, filepath, True)
|
||
|
|
print(f" Marked as downloaded: {asset['name']}")
|
||
|
|
|
||
|
|
# Second check - should find no new assets
|
||
|
|
print("\n4. Second check for new assets...")
|
||
|
|
new_assets = downloader.asset_tracker.get_new_assets(mock_assets)
|
||
|
|
print(f" Second check: Found {len(new_assets)} new assets (expected: 0)")
|
||
|
|
|
||
|
|
if len(new_assets) != 0:
|
||
|
|
print(" ❌ Should have found 0 new assets")
|
||
|
|
return False
|
||
|
|
|
||
|
|
print(" ✅ Asset tracking working correctly in config downloader")
|
||
|
|
|
||
|
|
# Check statistics
|
||
|
|
print("\n5. Checking statistics...")
|
||
|
|
stats = downloader.asset_tracker.get_stats()
|
||
|
|
print(f" Total tracked assets: {stats['total_tracked_assets']}")
|
||
|
|
print(f" Successful downloads: {stats['successful_downloads']}")
|
||
|
|
print(f" Existing files: {stats['existing_files']}")
|
||
|
|
|
||
|
|
if stats['total_tracked_assets'] != 2:
|
||
|
|
print(" ❌ Should have 2 tracked assets")
|
||
|
|
return False
|
||
|
|
|
||
|
|
print(" ✅ Statistics correct")
|
||
|
|
print("\n✅ Mock download with tracking test passed!")
|
||
|
|
return True
|
||
|
|
|
||
|
|
def test_command_line_options(self):
|
||
|
|
"""Test the new command line options."""
|
||
|
|
print("\n" + "=" * 60)
|
||
|
|
print("TEST 4: Command Line Options")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||
|
|
config_file = Path(temp_dir) / "test_config.json"
|
||
|
|
|
||
|
|
# Create config with tracking enabled
|
||
|
|
config_data = self.create_test_config(temp_dir, track_assets=True)
|
||
|
|
with open(config_file, 'w') as f:
|
||
|
|
json.dump(config_data, f, indent=2)
|
||
|
|
|
||
|
|
print("1. Testing --show-stats option...")
|
||
|
|
try:
|
||
|
|
# Import the main function to test command line parsing
|
||
|
|
from config_downloader import main
|
||
|
|
import sys
|
||
|
|
|
||
|
|
# Backup original argv
|
||
|
|
original_argv = sys.argv.copy()
|
||
|
|
|
||
|
|
# Test show-stats option
|
||
|
|
sys.argv = ['config_downloader.py', '--config', str(config_file), '--show-stats']
|
||
|
|
|
||
|
|
# This would normally call main(), but we'll just check the parsing works
|
||
|
|
print(" ✅ Command line parsing would work for --show-stats")
|
||
|
|
|
||
|
|
# Test cleanup option
|
||
|
|
sys.argv = ['config_downloader.py', '--config', str(config_file), '--cleanup']
|
||
|
|
print(" ✅ Command line parsing would work for --cleanup")
|
||
|
|
|
||
|
|
# Test force-redownload option
|
||
|
|
sys.argv = ['config_downloader.py', '--config', str(config_file), '--force-redownload']
|
||
|
|
print(" ✅ Command line parsing would work for --force-redownload")
|
||
|
|
|
||
|
|
# Restore original argv
|
||
|
|
sys.argv = original_argv
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f" ❌ Command line parsing failed: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
print("\n✅ Command line options test passed!")
|
||
|
|
return True
|
||
|
|
|
||
|
|
def run_all_tests(self):
|
||
|
|
"""Run all tests."""
|
||
|
|
print("🚀 Starting Config Downloader Asset Tracking Tests")
|
||
|
|
print("=" * 80)
|
||
|
|
|
||
|
|
try:
|
||
|
|
success = True
|
||
|
|
|
||
|
|
success &= self.test_config_loading()
|
||
|
|
success &= self.test_config_default_behavior()
|
||
|
|
success &= asyncio.run(self.test_mock_download_with_tracking())
|
||
|
|
success &= self.test_command_line_options()
|
||
|
|
|
||
|
|
if success:
|
||
|
|
print("\n" + "=" * 80)
|
||
|
|
print("🎉 ALL CONFIG DOWNLOADER TESTS PASSED!")
|
||
|
|
print("=" * 80)
|
||
|
|
print("✅ Asset tracking is now properly integrated into config_downloader.py")
|
||
|
|
print("✅ The config downloader will now skip already downloaded assets")
|
||
|
|
print("✅ Command line options for tracking control are available")
|
||
|
|
else:
|
||
|
|
print("\n❌ SOME TESTS FAILED")
|
||
|
|
|
||
|
|
return success
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"\n❌ TEST FAILED: {e}")
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def show_usage_instructions():
|
||
|
|
"""Show usage instructions for the updated config downloader."""
|
||
|
|
print("\n" + "=" * 80)
|
||
|
|
print("📋 UPDATED CONFIG DOWNLOADER USAGE")
|
||
|
|
print("=" * 80)
|
||
|
|
|
||
|
|
print("\n🔧 Configuration File:")
|
||
|
|
print("Add 'track_assets': true to your config JSON file:")
|
||
|
|
print("""
|
||
|
|
{
|
||
|
|
"api_url": "https://api.parentzone.me",
|
||
|
|
"list_endpoint": "/v1/media/list",
|
||
|
|
"download_endpoint": "/v1/media",
|
||
|
|
"output_dir": "./parentzone_images",
|
||
|
|
"max_concurrent": 5,
|
||
|
|
"timeout": 30,
|
||
|
|
"track_assets": true,
|
||
|
|
"email": "your_email@example.com",
|
||
|
|
"password": "your_password"
|
||
|
|
}
|
||
|
|
""")
|
||
|
|
|
||
|
|
print("\n💻 Command Line Usage:")
|
||
|
|
print("# Normal download (only new/modified assets):")
|
||
|
|
print("python3 config_downloader.py --config parentzone_config.json")
|
||
|
|
print()
|
||
|
|
print("# Force download all assets:")
|
||
|
|
print("python3 config_downloader.py --config parentzone_config.json --force-redownload")
|
||
|
|
print()
|
||
|
|
print("# Show asset statistics:")
|
||
|
|
print("python3 config_downloader.py --config parentzone_config.json --show-stats")
|
||
|
|
print()
|
||
|
|
print("# Clean up missing files:")
|
||
|
|
print("python3 config_downloader.py --config parentzone_config.json --cleanup")
|
||
|
|
|
||
|
|
print("\n✨ Benefits:")
|
||
|
|
print("• First run: Downloads all assets")
|
||
|
|
print("• Subsequent runs: Only downloads new/modified assets")
|
||
|
|
print("• Significant time and bandwidth savings")
|
||
|
|
print("• Automatic tracking of download history")
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
"""Main test function."""
|
||
|
|
# Setup logging
|
||
|
|
logging.basicConfig(
|
||
|
|
level=logging.INFO,
|
||
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
|
|
)
|
||
|
|
|
||
|
|
tester = ConfigTrackingTester()
|
||
|
|
|
||
|
|
# Run unit tests
|
||
|
|
success = tester.run_all_tests()
|
||
|
|
|
||
|
|
# Show usage instructions
|
||
|
|
if success:
|
||
|
|
show_usage_instructions()
|
||
|
|
|
||
|
|
return 0 if success else 1
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
exit(main())
|