248 lines
8.6 KiB
Python
248 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test progress reporting for deb-bootc-image-builder.
|
|
|
|
This module tests progress reporting functionality, including:
|
|
- Progress tracking
|
|
- Status updates
|
|
- Error reporting
|
|
- Debian-specific progress indicators
|
|
"""
|
|
|
|
import pytest
|
|
import os
|
|
import tempfile
|
|
import shutil
|
|
import json
|
|
import time
|
|
from unittest.mock import Mock, patch
|
|
import logging
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TestProgressReporting:
|
|
"""Test cases for progress reporting functionality."""
|
|
|
|
def test_progress_initialization(self, work_dir):
|
|
"""Test progress tracking initialization."""
|
|
# Initialize progress tracker
|
|
progress = self._create_progress_tracker()
|
|
|
|
assert progress["total_steps"] > 0
|
|
assert progress["current_step"] == 0
|
|
assert progress["status"] == "initialized"
|
|
assert "start_time" in progress
|
|
|
|
def test_progress_step_tracking(self, work_dir):
|
|
"""Test progress step tracking."""
|
|
# Create progress tracker
|
|
progress = self._create_progress_tracker()
|
|
|
|
# Simulate step progression
|
|
steps = [
|
|
"filesystem_setup",
|
|
"package_installation",
|
|
"ostree_integration",
|
|
"bootloader_configuration",
|
|
"image_generation"
|
|
]
|
|
|
|
for i, step in enumerate(steps):
|
|
self._update_progress(progress, step, i + 1)
|
|
|
|
assert progress["current_step"] == i + 1
|
|
assert progress["current_operation"] == step
|
|
assert progress["status"] == "in_progress"
|
|
|
|
# Check progress percentage
|
|
expected_percentage = ((i + 1) / len(steps)) * 100
|
|
assert abs(progress["percentage"] - expected_percentage) < 0.1
|
|
|
|
def test_progress_status_updates(self, work_dir):
|
|
"""Test progress status updates."""
|
|
# Create progress tracker
|
|
progress = self._create_progress_tracker()
|
|
|
|
# Test status transitions
|
|
statuses = ["initialized", "in_progress", "completed", "failed"]
|
|
|
|
for status in statuses:
|
|
self._set_progress_status(progress, status)
|
|
assert progress["status"] == status
|
|
|
|
# Check status-specific properties
|
|
if status == "completed":
|
|
assert progress["end_time"] is not None
|
|
assert progress["percentage"] == 100.0
|
|
elif status == "failed":
|
|
assert progress["error"] is not None
|
|
|
|
def test_debian_specific_progress_indicators(self, work_dir):
|
|
"""Test Debian-specific progress indicators."""
|
|
# Create progress tracker
|
|
progress = self._create_progress_tracker()
|
|
|
|
# Test Debian-specific operations
|
|
debian_operations = [
|
|
"apt_update",
|
|
"package_download",
|
|
"package_installation",
|
|
"initramfs_generation",
|
|
"grub_configuration"
|
|
]
|
|
|
|
for operation in debian_operations:
|
|
self._add_debian_operation(progress, operation)
|
|
assert operation in progress["debian_operations"]
|
|
|
|
# Test Debian package progress
|
|
package_progress = self._track_package_progress(progress, ["linux-image-amd64", "systemd", "ostree"])
|
|
assert package_progress["total_packages"] == 3
|
|
assert package_progress["installed_packages"] == 0
|
|
|
|
def test_error_reporting(self, work_dir):
|
|
"""Test error reporting in progress tracking."""
|
|
# Create progress tracker
|
|
progress = self._create_progress_tracker()
|
|
|
|
# Test error reporting
|
|
error_message = "Package installation failed: network error"
|
|
self._report_progress_error(progress, error_message)
|
|
|
|
assert progress["status"] == "failed"
|
|
assert progress["error"] == error_message
|
|
assert progress["error_time"] is not None
|
|
|
|
# Test error details
|
|
error_details = {
|
|
"operation": "package_installation",
|
|
"step": 2,
|
|
"timestamp": time.time()
|
|
}
|
|
self._add_error_details(progress, error_details)
|
|
|
|
assert "error_details" in progress
|
|
assert progress["error_details"]["operation"] == "package_installation"
|
|
|
|
def test_progress_persistence(self, work_dir):
|
|
"""Test progress persistence and recovery."""
|
|
# Create progress tracker
|
|
progress = self._create_progress_tracker()
|
|
|
|
# Update progress
|
|
self._update_progress(progress, "filesystem_setup", 1)
|
|
self._update_progress(progress, "package_installation", 2)
|
|
|
|
# Save progress
|
|
progress_file = os.path.join(work_dir, "progress.json")
|
|
self._save_progress(progress, progress_file)
|
|
|
|
# Load progress
|
|
loaded_progress = self._load_progress(progress_file)
|
|
|
|
# Verify persistence
|
|
assert loaded_progress["current_step"] == 2
|
|
assert loaded_progress["current_operation"] == "package_installation"
|
|
assert loaded_progress["percentage"] == 40.0
|
|
|
|
def test_progress_cleanup(self, work_dir):
|
|
"""Test progress cleanup and finalization."""
|
|
# Create progress tracker
|
|
progress = self._create_progress_tracker()
|
|
|
|
# Complete all steps
|
|
steps = ["filesystem_setup", "package_installation", "ostree_integration", "bootloader_configuration", "image_generation"]
|
|
for i, step in enumerate(steps):
|
|
self._update_progress(progress, step, i + 1)
|
|
|
|
# Finalize progress
|
|
self._finalize_progress(progress)
|
|
|
|
assert progress["status"] == "completed"
|
|
assert progress["end_time"] is not None
|
|
assert progress["duration"] > 0
|
|
assert progress["percentage"] == 100.0
|
|
|
|
def _create_progress_tracker(self):
|
|
"""Create a progress tracker instance."""
|
|
return {
|
|
"total_steps": 5,
|
|
"current_step": 0,
|
|
"current_operation": None,
|
|
"status": "initialized",
|
|
"start_time": time.time(),
|
|
"end_time": None,
|
|
"percentage": 0.0,
|
|
"error": None,
|
|
"error_time": None,
|
|
"debian_operations": [],
|
|
"package_progress": {}
|
|
}
|
|
|
|
def _update_progress(self, progress, operation, step):
|
|
"""Update progress tracking."""
|
|
progress["current_step"] = step
|
|
progress["current_operation"] = operation
|
|
progress["status"] = "in_progress"
|
|
progress["percentage"] = (step / progress["total_steps"]) * 100
|
|
|
|
def _set_progress_status(self, progress, status):
|
|
"""Set progress status."""
|
|
progress["status"] = status
|
|
|
|
if status == "completed":
|
|
progress["end_time"] = time.time()
|
|
progress["percentage"] = 100.0
|
|
elif status == "failed":
|
|
progress["error_time"] = time.time()
|
|
|
|
def _add_debian_operation(self, progress, operation):
|
|
"""Add Debian-specific operation to progress."""
|
|
if "debian_operations" not in progress:
|
|
progress["debian_operations"] = []
|
|
progress["debian_operations"].append(operation)
|
|
|
|
def _track_package_progress(self, progress, packages):
|
|
"""Track package installation progress."""
|
|
package_progress = {
|
|
"total_packages": len(packages),
|
|
"installed_packages": 0,
|
|
"failed_packages": [],
|
|
"current_package": None
|
|
}
|
|
progress["package_progress"] = package_progress
|
|
return package_progress
|
|
|
|
def _report_progress_error(self, progress, error_message):
|
|
"""Report progress error."""
|
|
progress["status"] = "failed"
|
|
progress["error"] = error_message
|
|
progress["error_time"] = time.time()
|
|
|
|
def _add_error_details(self, progress, error_details):
|
|
"""Add detailed error information."""
|
|
progress["error_details"] = error_details
|
|
|
|
def _save_progress(self, progress, file_path):
|
|
"""Save progress to file."""
|
|
with open(file_path, 'w') as f:
|
|
json.dump(progress, f, indent=2)
|
|
|
|
def _load_progress(self, file_path):
|
|
"""Load progress from file."""
|
|
with open(file_path, 'r') as f:
|
|
return json.load(f)
|
|
|
|
def _finalize_progress(self, progress):
|
|
"""Finalize progress tracking."""
|
|
progress["status"] = "completed"
|
|
progress["end_time"] = time.time()
|
|
progress["duration"] = progress["end_time"] - progress["start_time"]
|
|
progress["percentage"] = 100.0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__])
|