Testing Backups

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338

#!/usr/bin/env python3
"""
Comprehensive Backup Testing Framework
Implements multi-tier backup validation with logging and alerting
"""
 
import hashlib
import subprocess
import logging
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Optional, List, Dict
import json
 
class TestLevel(Enum):
    COMPLETION = 1     # Basic job completion check
    INTEGRITY = 2      # Checksum and consistency validation
    CATALOG = 3        # Metadata synchronization verification
    PARTIAL_RESTORE = 4  # Selective object restoration
    FULL_RESTORE = 5   # Complete database restoration
    APPLICATION = 6    # Full application functionality validation
 
@dataclass
class BackupTestResult:
    """Result of a backup verification test"""
    backup_id: str
    test_level: TestLevel
    test_time: datetime
    success: bool
    duration_seconds: float
    details: Dict[str, any]
    error_message: Optional[str] = None
 
class BackupTestFramework:
    """
    Multi-tier backup testing framework
    
    Implements the testing pyramid for database backups:
    - Level 1-3: Automated, run after every backup
    - Level 4: Weekly automated partial restores
    - Level 5-6: Monthly full validation
    """
    
    def __init__(self, config_path: str):
        self.config = self._load_config(config_path)
        self.logger = self._setup_logging()
        self.results: List[BackupTestResult] = []
    
    def _load_config(self, path: str) -> Dict:
        """Load test framework configuration"""
        with open(path) as f:
            return json.load(f)
    
    def _setup_logging(self) -> logging.Logger:
        """Configure structured logging for audit trail"""
        logger = logging.getLogger('backup_testing')
        logger.setLevel(logging.INFO)
        handler = logging.FileHandler(
            f"backup_test_{datetime.now().strftime('%Y%m%d')}.log"
        )
        handler.setFormatter(logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        ))
        logger.addHandler(handler)
        return logger
    
    def test_completion(self, backup_id: str, job_log_path: str) -> BackupTestResult:
        """
        Level 1: Verify backup job completed successfully
        
        Checks:
        - Job exit code was zero
        - No ERROR entries in job log
        - Backup file exists with non-zero size
        - Backup completed within expected time window
        """
        start_time = datetime.now()
        details = {}
        
        try:
            # Parse job log for completion status
            with open(job_log_path) as f:
                log_content = f.read()
            
            details['has_errors'] = 'ERROR' in log_content
            details['has_completion_marker'] = 'Backup completed successfully' in log_content
            
            # Verify backup file exists
            backup_path = self._get_backup_path(backup_id)
            details['file_exists'] = backup_path.exists()
            details['file_size_bytes'] = backup_path.stat().st_size if backup_path.exists() else 0
            
            # Determine success
            success = (
                not details['has_errors'] and 
                details['has_completion_marker'] and 
                details['file_exists'] and 
                details['file_size_bytes'] > 0
            )
            
            return BackupTestResult(
                backup_id=backup_id,
                test_level=TestLevel.COMPLETION,
                test_time=start_time,
                success=success,
                duration_seconds=(datetime.now() - start_time).total_seconds(),
                details=details,
                error_message=None if success else "Completion verification failed"
            )
            
        except Exception as e:
            return BackupTestResult(
                backup_id=backup_id,
                test_level=TestLevel.COMPLETION,
                test_time=start_time,
                success=False,
                duration_seconds=(datetime.now() - start_time).total_seconds(),
                details=details,
                error_message=str(e)
            )
    
    def test_integrity(self, backup_id: str) -> BackupTestResult:
        """
        Level 2: Validate backup file integrity
        
        Implements:
        - SHA-256 checksum verification against stored hash
        - Internal consistency check using native tools
        - File structure validation
        """
        start_time = datetime.now()
        details = {}
        
        try:
            backup_path = self._get_backup_path(backup_id)
            
            # Calculate and verify checksum
            with open(backup_path, 'rb') as f:
                calculated_hash = hashlib.sha256(f.read()).hexdigest()
            
            stored_hash = self._get_stored_checksum(backup_id)
            details['calculated_checksum'] = calculated_hash
            details['stored_checksum'] = stored_hash
            details['checksum_match'] = calculated_hash == stored_hash
            
            # Run native validation tool (example: pg_restore --list for PostgreSQL)
            validation_result = self._run_native_validation(backup_path)
            details['native_validation'] = validation_result
            
            success = details['checksum_match'] and validation_result['valid']
            
            return BackupTestResult(
                backup_id=backup_id,
                test_level=TestLevel.INTEGRITY,
                test_time=start_time,
                success=success,
                duration_seconds=(datetime.now() - start_time).total_seconds(),
                details=details
            )
            
        except Exception as e:
            return BackupTestResult(
                backup_id=backup_id,
                test_level=TestLevel.INTEGRITY,
                test_time=start_time,
                success=False,
                duration_seconds=(datetime.now() - start_time).total_seconds(),
                details=details,
                error_message=str(e)
            )
    
    def test_full_restore(
        self, 
        backup_id: str, 
        target_instance: str,
        validation_queries: List[str]
    ) -> BackupTestResult:
        """
        Level 5: Complete database restoration and validation
        
        Process:
        1. Provision isolated test environment
        2. Restore complete backup to test instance
        3. Verify database is online and accessible
        4. Run validation queries to confirm data integrity
        5. Measure and record restore duration
        6. Clean up test environment
        """
        start_time = datetime.now()
        details = {
            'target_instance': target_instance,
            'phases': {}
        }
        
        try:
            # Phase 1: Restore the backup
            self.logger.info(f"Starting full restore test for {backup_id}")
            restore_start = datetime.now()
            
            restore_result = self._execute_restore(
                backup_id=backup_id,
                target_instance=target_instance
            )
            
            details['phases']['restore'] = {
                'duration_seconds': (datetime.now() - restore_start).total_seconds(),
                'success': restore_result['success'],
                'output': restore_result.get('output', '')[:1000]  # Truncate for logging
            }
            
            if not restore_result['success']:
                return BackupTestResult(
                    backup_id=backup_id,
                    test_level=TestLevel.FULL_RESTORE,
                    test_time=start_time,
                    success=False,
                    duration_seconds=(datetime.now() - start_time).total_seconds(),
                    details=details,
                    error_message="Restore phase failed"
                )
            
            # Phase 2: Validate database accessibility
            accessibility_result = self._test_database_accessibility(target_instance)
            details['phases']['accessibility'] = accessibility_result
            
            # Phase 3: Run validation queries
            query_results = []
            for query in validation_queries:
                result = self._execute_validation_query(target_instance, query)
                query_results.append(result)
            
            details['phases']['validation_queries'] = {
                'total': len(validation_queries),
                'passed': sum(1 for r in query_results if r['success']),
                'results': query_results
            }
            
            # Determine overall success
            all_queries_passed = all(r['success'] for r in query_results)
            success = restore_result['success'] and accessibility_result['success'] and all_queries_passed
            
            # Phase 4: Calculate RTO metrics
            details['rto_metrics'] = {
                'total_restore_seconds': (datetime.now() - start_time).total_seconds(),
                'meets_rto': (datetime.now() - start_time).total_seconds() < self.config['rto_seconds']
            }
            
            return BackupTestResult(
                backup_id=backup_id,
                test_level=TestLevel.FULL_RESTORE,
                test_time=start_time,
                success=success,
                duration_seconds=(datetime.now() - start_time).total_seconds(),
                details=details
            )
            
        except Exception as e:
            self.logger.error(f"Full restore test failed: {str(e)}")
            return BackupTestResult(
                backup_id=backup_id,
                test_level=TestLevel.FULL_RESTORE,
                test_time=start_time,
                success=False,
                duration_seconds=(datetime.now() - start_time).total_seconds(),
                details=details,
                error_message=str(e)
            )
        finally:
            # Always clean up test environment
            self._cleanup_test_environment(target_instance)
    
    def _get_backup_path(self, backup_id: str) -> Path:
        """Resolve backup file path from backup ID"""
        return Path(self.config['backup_dir']) / f"{backup_id}.backup"
    
    def _get_stored_checksum(self, backup_id: str) -> str:
        """Retrieve stored checksum for backup"""
        checksum_file = Path(self.config['checksum_dir']) / f"{backup_id}.sha256"
        return checksum_file.read_text().strip()
    
    def _run_native_validation(self, backup_path: Path) -> Dict:
        """Run database-native backup validation"""
        # PostgreSQL example
        result = subprocess.run(
            ['pg_restore', '--list', str(backup_path)],
            capture_output=True,
            text=True
        )
        return {
            'valid': result.returncode == 0,
            'object_count': len(result.stdout.splitlines()),
            'errors': result.stderr if result.returncode != 0 else None
        }
    
    def _execute_restore(self, backup_id: str, target_instance: str) -> Dict:
        """Execute backup restoration to target instance"""
        # Implementation depends on database system
        pass
    
    def _test_database_accessibility(self, instance: str) -> Dict:
        """Verify database is online and accepting connections"""
        pass
    
    def _execute_validation_query(self, instance: str, query: str) -> Dict:
        """Execute a validation query and return results"""
        pass
    
    def _cleanup_test_environment(self, instance: str) -> None:
        """Clean up test instance after validation"""
        pass
    
    def generate_test_report(self) -> Dict:
        """Generate comprehensive test report"""
        return {
            'generated_at': datetime.now().isoformat(),
            'total_tests': len(self.results),
            'passed': sum(1 for r in self.results if r.success),
            'failed': sum(1 for r in self.results if not r.success),
            'by_level': {
                level.name: {
                    'total': sum(1 for r in self.results if r.test_level == level),
                    'passed': sum(1 for r in self.results if r.test_level == level and r.success)
                }
                for level in TestLevel
            },
            'results': [
                {
                    'backup_id': r.backup_id,
                    'level': r.test_level.name,
                    'success': r.success,
                    'duration': r.duration_seconds,
                    'error': r.error_message
                }
                for r in self.results
            ]
        }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181

-- =====================================================
-- Backup Validation Query Suite
-- Execute after restoration to verify backup integrity
-- =====================================================
 
-- 1. ROW COUNT VALIDATION
-- Compare against known baseline (typically from backup manifest)
-- Expected values should be stored in backup metadata
 
SELECT 'customers' as table_name, 
       COUNT(*) as actual_count,
       1500000 as expected_count,  -- From backup manifest
       CASE 
           WHEN COUNT(*) = 1500000 THEN 'PASS'
           WHEN COUNT(*) > 1500000 THEN 'EXTRA_ROWS'
           ELSE 'MISSING_ROWS'
       END as status
FROM customers
 
UNION ALL
 
SELECT 'orders' as table_name,
       COUNT(*) as actual_count,
       8750000 as expected_count,
       CASE 
           WHEN COUNT(*) = 8750000 THEN 'PASS'
           WHEN COUNT(*) > 8750000 THEN 'EXTRA_ROWS'
           ELSE 'MISSING_ROWS'
       END as status
FROM orders
 
UNION ALL
 
SELECT 'transactions' as table_name,
       COUNT(*) as actual_count,
       45000000 as expected_count,
       CASE 
           WHEN COUNT(*) = 45000000 THEN 'PASS'
           WHEN COUNT(*) > 45000000 THEN 'EXTRA_ROWS'
           ELSE 'MISSING_ROWS'
       END as status
FROM transactions;
 
-- 2. REFERENTIAL INTEGRITY VALIDATION
-- Verify foreign key relationships are intact
 
-- Orders without valid customers (should be 0)
SELECT 'orphaned_orders' as check_name,
       COUNT(*) as violation_count,
       CASE WHEN COUNT(*) = 0 THEN 'PASS' ELSE 'FAIL' END as status
FROM orders o
WHERE NOT EXISTS (
    SELECT 1 FROM customers c WHERE c.customer_id = o.customer_id
);
 
-- Order items without valid orders (should be 0)
SELECT 'orphaned_order_items' as check_name,
       COUNT(*) as violation_count,
       CASE WHEN COUNT(*) = 0 THEN 'PASS' ELSE 'FAIL' END as status
FROM order_items oi
WHERE NOT EXISTS (
    SELECT 1 FROM orders o WHERE o.order_id = oi.order_id
);
 
-- 3. AGGREGATE VALIDATION
-- Critical for financial data integrity
 
SELECT 'total_revenue' as metric,
       SUM(total_amount) as actual_value,
       985746382.45 as expected_value,  -- Known value from source
       CASE 
           WHEN ABS(SUM(total_amount) - 985746382.45) < 0.01 THEN 'PASS'
           ELSE 'FAIL'
       END as status
FROM orders
WHERE order_date < '2024-01-01';
 
SELECT 'inventory_balance' as metric,
       SUM(quantity_on_hand * unit_cost) as actual_value,
       12847563.78 as expected_value,
       CASE 
           WHEN ABS(SUM(quantity_on_hand * unit_cost) - 12847563.78) < 0.01 THEN 'PASS'
           ELSE 'FAIL'
       END as status
FROM inventory;
 
-- 4. BOUNDARY VALIDATION
-- Verify no data truncation at time boundaries
 
SELECT 'max_order_date' as boundary,
       MAX(order_date) as actual_value,
       '2024-12-31 23:59:47'::timestamp as expected_value,
       CASE 
           WHEN MAX(order_date) = '2024-12-31 23:59:47'::timestamp THEN 'PASS'
           ELSE 'FAIL'
       END as status
FROM orders;
 
SELECT 'max_transaction_id' as boundary,
       MAX(transaction_id) as actual_value,
       45847362 as expected_value,
       CASE 
           WHEN MAX(transaction_id) = 45847362 THEN 'PASS'
           ELSE 'FAIL'
       END as status
FROM transactions;
 
-- 5. SAMPLE RECORD VALIDATION
-- Verify specific known records are intact
 
SELECT 'known_customer_check' as validation,
       CASE 
           WHEN customer_name = 'Acme Corporation'
            AND customer_email = 'orders@acme.example.com'
            AND credit_limit = 50000.00
           THEN 'PASS'
           ELSE 'FAIL'
       END as status,
       customer_name,
       customer_email,
       credit_limit
FROM customers
WHERE customer_id = 'CUST-00001';
 
-- 6. CONSTRAINT VALIDATION
-- Verify constraints are enforced (critical after restore)
 
-- Check for NULL violation in NOT NULL columns
SELECT 'null_constraint_check' as validation,
       COUNT(*) as violations,
       CASE WHEN COUNT(*) = 0 THEN 'PASS' ELSE 'FAIL' END as status
FROM orders
WHERE customer_id IS NULL 
   OR order_date IS NULL 
   OR total_amount IS NULL;
 
-- Check for unique constraint violations
SELECT 'unique_constraint_check' as validation,
       COUNT(*) as duplicate_count,
       CASE WHEN COUNT(*) = 0 THEN 'PASS' ELSE 'FAIL' END as status
FROM (
    SELECT order_number, COUNT(*) as cnt
    FROM orders
    GROUP BY order_number
    HAVING COUNT(*) > 1
) duplicates;
 
-- 7. CHECKSUM VALIDATION
-- Row-level data integrity using checksums
 
SELECT 'checksum_validation' as validation,
       COUNT(*) as mismatch_count,
       CASE WHEN COUNT(*) = 0 THEN 'PASS' ELSE 'FAIL' END as status
FROM customers c
JOIN customer_checksums cs ON c.customer_id = cs.customer_id
WHERE MD5(CONCAT(c.customer_name, c.customer_email, c.credit_limit::text)) 
      != cs.row_checksum;
 
-- 8. STORED PROCEDURE VALIDATION
-- Verify critical stored procedures exist and are executable
 
DO $$
DECLARE
    proc_count INTEGER;
BEGIN
    SELECT COUNT(*) INTO proc_count
    FROM information_schema.routines
    WHERE routine_schema = 'public'
      AND routine_name IN (
          'calculate_order_total',
          'process_payment',
          'generate_invoice',
          'update_inventory'
      );
    
    IF proc_count = 4 THEN
        RAISE NOTICE 'PASS: All required stored procedures present';
    ELSE
        RAISE EXCEPTION 'FAIL: Missing stored procedures. Found % of 4', proc_count;
    END IF;
END $$;