151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for backup_script.py functionality
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
from backup_script import PVCBackupManager
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def test_pvc_verification():
|
|
"""Test PVC verification functionality"""
|
|
print("=== Testing PVC Verification ===")
|
|
|
|
with patch('backup_script.client.CoreV1Api') as mock_core_api:
|
|
# Mock successful PVC check
|
|
mock_api_instance = Mock()
|
|
mock_core_api.return_value = mock_api_instance
|
|
mock_api_instance.read_namespaced_persistent_volume_claim.return_value = Mock()
|
|
|
|
backup_manager = PVCBackupManager()
|
|
result = backup_manager.verify_pvc_exists("test-pvc")
|
|
print(f"PVC verification result: {result}")
|
|
assert result == True
|
|
|
|
def test_snapshot_name_generation():
|
|
"""Test snapshot name generation"""
|
|
print("=== Testing Snapshot Name Generation ===")
|
|
|
|
backup_manager = PVCBackupManager()
|
|
timestamp = backup_manager.get_pst_date()
|
|
snapshot_name = backup_manager.generate_snapshot_name("test-pvc", timestamp)
|
|
print(f"Generated snapshot name: {snapshot_name}")
|
|
assert "test-pvc-snapshot-" in snapshot_name
|
|
|
|
def test_snapshot_yaml_creation():
|
|
"""Test snapshot YAML creation"""
|
|
print("=== Testing Snapshot YAML Creation ===")
|
|
|
|
backup_manager = PVCBackupManager()
|
|
snapshot_yaml = backup_manager.create_snapshot_yaml("test-pvc", "test-snapshot")
|
|
print(f"Snapshot YAML: {snapshot_yaml}")
|
|
|
|
assert snapshot_yaml["apiVersion"] == "snapshot.storage.k8s.io/v1"
|
|
assert snapshot_yaml["kind"] == "VolumeSnapshot"
|
|
assert snapshot_yaml["metadata"]["name"] == "test-snapshot"
|
|
assert snapshot_yaml["spec"]["source"]["persistentVolumeClaimName"] == "test-pvc"
|
|
|
|
def test_full_backup_process():
|
|
"""Test full backup process with mocked APIs"""
|
|
print("=== Testing Full Backup Process ===")
|
|
|
|
with patch('backup_script.client.CustomObjectsApi') as mock_snapshot_api, \
|
|
patch('backup_script.client.CoreV1Api') as mock_core_api:
|
|
|
|
# Mock API instances
|
|
mock_snapshot_instance = Mock()
|
|
mock_core_instance = Mock()
|
|
mock_snapshot_api.return_value = mock_snapshot_instance
|
|
mock_core_api.return_value = mock_core_instance
|
|
|
|
# Mock PVC verification
|
|
mock_core_instance.read_namespaced_persistent_volume_claim.return_value = Mock()
|
|
|
|
# Mock snapshot creation
|
|
mock_snapshot_instance.create_namespaced_custom_object.return_value = {
|
|
"metadata": {"name": "test-snapshot"}
|
|
}
|
|
|
|
# Mock snapshot ready status
|
|
mock_snapshot_instance.get_namespaced_custom_object.return_value = {
|
|
"status": {"readyToUse": True}
|
|
}
|
|
|
|
backup_manager = PVCBackupManager()
|
|
# Override PVCs for testing
|
|
backup_manager.pvcs_to_backup = ["test-pvc"]
|
|
backup_manager.timeout = 10 # Short timeout for testing
|
|
|
|
result = backup_manager.run_backup()
|
|
print(f"Backup process result: {result}")
|
|
assert result == True
|
|
|
|
def test_environment_variables():
|
|
"""Test environment variable configuration"""
|
|
print("=== Testing Environment Variables ===")
|
|
|
|
# Set test environment variables
|
|
os.environ["BACKUP_NAMESPACE"] = "test-namespace"
|
|
os.environ["SNAPSHOT_CLASS"] = "test-snapshot-class"
|
|
os.environ["TIMEOUT"] = "600"
|
|
|
|
backup_manager = PVCBackupManager()
|
|
print(f"Namespace: {backup_manager.namespace}")
|
|
print(f"Snapshot Class: {backup_manager.snapshot_class}")
|
|
print(f"Timeout: {backup_manager.timeout}")
|
|
|
|
assert backup_manager.namespace == "test-namespace"
|
|
assert backup_manager.snapshot_class == "test-snapshot-class"
|
|
assert backup_manager.timeout == 600
|
|
|
|
def run_integration_test():
|
|
"""Run integration test with real cluster (optional)"""
|
|
print("=== Integration Test (Real Cluster) ===")
|
|
print("This test requires access to a real Kubernetes cluster")
|
|
print("Make sure you have kubeconfig configured")
|
|
|
|
try:
|
|
backup_manager = PVCBackupManager()
|
|
print("Successfully initialized backup manager")
|
|
|
|
# Test with a small timeout to avoid long waits
|
|
backup_manager.timeout = 30
|
|
print("Ready to run backup (will timeout after 30 seconds)")
|
|
|
|
# Uncomment the following line to run actual backup
|
|
# result = backup_manager.run_backup()
|
|
# print(f"Integration test result: {result}")
|
|
|
|
except Exception as e:
|
|
print(f"Integration test failed: {e}")
|
|
|
|
def main():
|
|
"""Run all tests"""
|
|
print("Starting backup script tests...")
|
|
|
|
try:
|
|
test_pvc_verification()
|
|
test_snapshot_name_generation()
|
|
test_snapshot_yaml_creation()
|
|
test_full_backup_process()
|
|
test_environment_variables()
|
|
|
|
print("\n=== All Unit Tests Passed ===")
|
|
|
|
# Ask user if they want to run integration test
|
|
response = input("\nDo you want to run integration test with real cluster? (y/N): ")
|
|
if response.lower() == 'y':
|
|
run_integration_test()
|
|
|
|
except Exception as e:
|
|
print(f"Test failed: {e}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main() |