|
|
@@ -1,437 +1,307 @@
|
|
|
#!/usr/bin/env python3
|
|
|
|
|
|
-"""
|
|
|
-Test script for Emergency Access Server
|
|
|
-Validates system configuration, file access, notifications, and endpoints
|
|
|
-"""
|
|
|
-
|
|
|
import os
|
|
|
import sys
|
|
|
import json
|
|
|
-import time
|
|
|
-import yaml
|
|
|
import tempfile
|
|
|
-import subprocess
|
|
|
-from typing import Dict, List, Tuple, Any
|
|
|
-from config import Config
|
|
|
-
|
|
|
-class EmergencyAccessTester:
|
|
|
- def __init__(self, config_path: str = None):
|
|
|
- """Initialize tester with configuration"""
|
|
|
- self.config = Config(config_path)
|
|
|
- self.test_results = []
|
|
|
- self.server_process = None
|
|
|
-
|
|
|
- def log_test(self, test_name: str, success: bool, message: str = ""):
|
|
|
- """Log test result"""
|
|
|
- status = "PASS" if success else "FAIL"
|
|
|
- result = {
|
|
|
- 'test': test_name,
|
|
|
- 'status': status,
|
|
|
- 'message': message,
|
|
|
- 'timestamp': time.time()
|
|
|
+import shutil
|
|
|
+from pathlib import Path
|
|
|
+
|
|
|
+# Add the current directory to Python path to import our modules
|
|
|
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
+
|
|
|
+from config import Config, KeyConfig
|
|
|
+
|
|
|
+def test_multikey_config():
|
|
|
+ """Test multi-key configuration loading"""
|
|
|
+ print("Testing multi-key configuration...")
|
|
|
+
|
|
|
+ # Create temporary config file
|
|
|
+ config_data = {
|
|
|
+ "server": {
|
|
|
+ "host": "127.0.0.1",
|
|
|
+ "port": 1127
|
|
|
+ },
|
|
|
+ "routes": {
|
|
|
+ "health_route": "/health-check-test"
|
|
|
+ },
|
|
|
+ "files": {
|
|
|
+ "dummy_file": "/tmp/dummy.txt"
|
|
|
+ },
|
|
|
+ "keys": {
|
|
|
+ "backup_key": {
|
|
|
+ "route": "/emergency-key-backup-a7f9d2e1b4c7f3e6d9a2b5c8e1f4d7a0b3c6e9f2a5d8b1c4e7f0a3d6b9c2e5f8a1d4b7c0e3f6d9a2b5c8e1f4a7b0d3",
|
|
|
+ "file": "/tmp/backup-key.txt",
|
|
|
+ "backends": ["test_backend1", "test_backend2"],
|
|
|
+ "message": "Backup key accessed"
|
|
|
+ },
|
|
|
+ "master_key": {
|
|
|
+ "route": "/emergency-key-master-x3k8m9n2p5q7r1t4u6v9w2y5z8b1c4d7e0f3g6h9j2k5l8m1n4o7p0q3r6s9t2u5v8w1x4y7z0a3b6c9d2e5f8",
|
|
|
+ "file": "/tmp/master-key.txt",
|
|
|
+ "backends": ["test_backend1", "test_backend3"],
|
|
|
+ "message": "Master key accessed"
|
|
|
+ }
|
|
|
+ },
|
|
|
+ "notifications": {
|
|
|
+ "health_backends": ["health_backend"],
|
|
|
+ "config_path": "/tmp/ntfy.yml",
|
|
|
+ "health_message": "Health check",
|
|
|
+ "log_level": "WARNING",
|
|
|
+ "send_all_logs": true
|
|
|
}
|
|
|
- self.test_results.append(result)
|
|
|
+ }
|
|
|
|
|
|
- color = '\033[92m' if success else '\033[91m' # Green or Red
|
|
|
- reset = '\033[0m'
|
|
|
- print(f"{color}[{status}]{reset} {test_name}: {message}")
|
|
|
+ # Write config to temporary file
|
|
|
+ with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
|
|
|
+ json.dump(config_data, f)
|
|
|
+ config_file = f.name
|
|
|
|
|
|
- return success
|
|
|
+ try:
|
|
|
+ # Load config
|
|
|
+ config = Config(config_file)
|
|
|
+
|
|
|
+ # Test basic properties
|
|
|
+ assert config.server_host == "127.0.0.1"
|
|
|
+ assert config.server_port == 1127
|
|
|
+ assert config.health_route == "/health-check-test"
|
|
|
+
|
|
|
+ # Test keys
|
|
|
+ keys = config.keys
|
|
|
+ assert len(keys) == 2
|
|
|
+ assert "backup_key" in keys
|
|
|
+ assert "master_key" in keys
|
|
|
+
|
|
|
+ # Test backup key
|
|
|
+ backup_key = keys["backup_key"]
|
|
|
+ assert backup_key.route == "/emergency-key-backup-a7f9d2e1b4c7f3e6d9a2b5c8e1f4d7a0b3c6e9f2a5d8b1c4e7f0a3d6b9c2e5f8a1d4b7c0e3f6d9a2b5c8e1f4a7b0d3"
|
|
|
+ assert backup_key.file_path == "/tmp/backup-key.txt"
|
|
|
+ assert backup_key.backends == ["test_backend1", "test_backend2"]
|
|
|
+ assert backup_key.message == "Backup key accessed"
|
|
|
+
|
|
|
+ # Test master key
|
|
|
+ master_key = keys["master_key"]
|
|
|
+ assert master_key.route == "/emergency-key-master-x3k8m9n2p5q7r1t4u6v9w2y5z8b1c4d7e0f3g6h9j2k5l8m1n4o7p0q3r6s9t2u5v8w1x4y7z0a3b6c9d2e5f8"
|
|
|
+ assert master_key.file_path == "/tmp/master-key.txt"
|
|
|
+ assert master_key.backends == ["test_backend1", "test_backend3"]
|
|
|
+ assert master_key.message == "Master key accessed"
|
|
|
+
|
|
|
+ # Test key lookup methods
|
|
|
+ found_key = config.get_key_by_route("/emergency-key-backup-a7f9d2e1b4c7f3e6d9a2b5c8e1f4d7a0b3c6e9f2a5d8b1c4e7f0a3d6b9c2e5f8a1d4b7c0e3f6d9a2b5c8e1f4a7b0d3")
|
|
|
+ assert found_key is not None
|
|
|
+ assert found_key.key_id == "backup_key"
|
|
|
+
|
|
|
+ found_key = config.get_key_by_id("master_key")
|
|
|
+ assert found_key is not None
|
|
|
+ assert found_key.route == "/emergency-key-master-x3k8m9n2p5q7r1t4u6v9w2y5z8b1c4d7e0f3g6h9j2k5l8m1n4o7p0q3r6s9t2u5v8w1x4y7z0a3b6c9d2e5f8"
|
|
|
+
|
|
|
+ print("✅ Multi-key configuration test passed")
|
|
|
+
|
|
|
+ finally:
|
|
|
+ os.unlink(config_file)
|
|
|
+
|
|
|
+def test_invalid_config():
|
|
|
+ """Test invalid configuration handling"""
|
|
|
+ print("Testing invalid configuration handling...")
|
|
|
+
|
|
|
+ # Test config without keys section
|
|
|
+ config_data = {
|
|
|
+ "server": {
|
|
|
+ "host": "127.0.0.1",
|
|
|
+ "port": 1127
|
|
|
+ },
|
|
|
+ "routes": {
|
|
|
+ "health_route": "/health-check-test"
|
|
|
+ },
|
|
|
+ "files": {
|
|
|
+ "dummy_file": "/tmp/dummy.txt"
|
|
|
+ },
|
|
|
+ "notifications": {
|
|
|
+ "health_backends": ["health_backend"],
|
|
|
+ "config_path": "/tmp/ntfy.yml",
|
|
|
+ "health_message": "Health check",
|
|
|
+ "log_level": "WARNING",
|
|
|
+ "send_all_logs": true
|
|
|
+ }
|
|
|
+ }
|
|
|
|
|
|
- def test_config_loading(self) -> bool:
|
|
|
- """Test configuration file loading"""
|
|
|
+ # Write config to temporary file
|
|
|
+ with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
|
|
|
+ json.dump(config_data, f)
|
|
|
+ config_file = f.name
|
|
|
+
|
|
|
+ try:
|
|
|
+ # Should raise exception for missing keys
|
|
|
try:
|
|
|
- # Test all required config properties
|
|
|
- self.config.server_host
|
|
|
- self.config.server_port
|
|
|
- self.config.key_route
|
|
|
- self.config.health_route
|
|
|
- self.config.key_file_path
|
|
|
- self.config.dummy_file_path
|
|
|
- self.config.ntfy_backends_key
|
|
|
- self.config.ntfy_backends_health
|
|
|
-
|
|
|
- return self.log_test("Config Loading", True, "All config properties accessible")
|
|
|
+ config = Config(config_file)
|
|
|
+ assert False, "Should have raised exception for missing keys section"
|
|
|
except Exception as e:
|
|
|
- return self.log_test("Config Loading", False, f"Error: {str(e)}")
|
|
|
+ assert "No keys configured" in str(e)
|
|
|
|
|
|
- def test_file_access(self) -> bool:
|
|
|
- """Test access to key and dummy files"""
|
|
|
- success = True
|
|
|
+ print("✅ Invalid configuration test passed")
|
|
|
|
|
|
- # Test key file
|
|
|
- try:
|
|
|
- if not os.path.exists(self.config.key_file_path):
|
|
|
- self.log_test("Key File Exists", False, f"File not found: {self.config.key_file_path}")
|
|
|
- success = False
|
|
|
- else:
|
|
|
- with open(self.config.key_file_path, 'r') as f:
|
|
|
- content = f.read().strip()
|
|
|
- if not content:
|
|
|
- self.log_test("Key File Content", False, "Key file is empty")
|
|
|
- success = False
|
|
|
- else:
|
|
|
- self.log_test("Key File Access", True, f"Key file readable, length: {len(content)}")
|
|
|
- except Exception as e:
|
|
|
- self.log_test("Key File Access", False, f"Error: {str(e)}")
|
|
|
- success = False
|
|
|
+ finally:
|
|
|
+ os.unlink(config_file)
|
|
|
|
|
|
- # Test dummy file
|
|
|
- try:
|
|
|
- if not os.path.exists(self.config.dummy_file_path):
|
|
|
- self.log_test("Dummy File Exists", False, f"File not found: {self.config.dummy_file_path}")
|
|
|
- success = False
|
|
|
- else:
|
|
|
- with open(self.config.dummy_file_path, 'r') as f:
|
|
|
- content = f.read().strip()
|
|
|
- if not content:
|
|
|
- self.log_test("Dummy File Content", False, "Dummy file is empty")
|
|
|
- success = False
|
|
|
- else:
|
|
|
- self.log_test("Dummy File Access", True, f"Dummy file readable, length: {len(content)}")
|
|
|
- except Exception as e:
|
|
|
- self.log_test("Dummy File Access", False, f"Error: {str(e)}")
|
|
|
- success = False
|
|
|
+def test_key_config_validation():
|
|
|
+ """Test key configuration validation"""
|
|
|
+ print("Testing key configuration validation...")
|
|
|
|
|
|
- return success
|
|
|
+ # Test missing route
|
|
|
+ try:
|
|
|
+ KeyConfig("test", {"file": "/tmp/test.txt"}, {})
|
|
|
+ assert False, "Should have raised exception for missing route"
|
|
|
+ except Exception as e:
|
|
|
+ assert "Route not configured" in str(e)
|
|
|
|
|
|
- def test_ntfy_connectivity(self) -> bool:
|
|
|
- """Test ntfy backend connectivity"""
|
|
|
- success = True
|
|
|
+ # Test missing file
|
|
|
+ try:
|
|
|
+ KeyConfig("test", {"route": "/test"}, {})
|
|
|
+ assert False, "Should have raised exception for missing file"
|
|
|
+ except Exception as e:
|
|
|
+ assert "File path not configured" in str(e)
|
|
|
|
|
|
- # Test importing ntfy and loading config
|
|
|
- try:
|
|
|
- import ntfy
|
|
|
- self.log_test("ntfy Import", True, "dschep/ntfy package available")
|
|
|
-
|
|
|
- # Test loading the ntfy config file
|
|
|
- ntfy_config = ntfy.load_config(self.config.ntfy_config_path)
|
|
|
- self.log_test("ntfy Config Load", True, f"Config loaded from {self.config.ntfy_config_path}")
|
|
|
- except ImportError:
|
|
|
- self.log_test("ntfy Import", False, "dschep/ntfy package not installed")
|
|
|
- return False
|
|
|
- except Exception as e:
|
|
|
- self.log_test("ntfy Config Load", False, f"Failed to load config: {str(e)}")
|
|
|
- return False
|
|
|
-
|
|
|
- # Test key notification backends
|
|
|
- for backend in self.config.ntfy_backends_key:
|
|
|
- try:
|
|
|
- # Just verify the backend name is valid (assumes global ntfy config is set up)
|
|
|
- # We don't actually send notifications during testing
|
|
|
- if backend and isinstance(backend, str) and len(backend.strip()) > 0:
|
|
|
- self.log_test(f"Key Backend: {backend}", True, "Backend name valid (using global ntfy config)")
|
|
|
- else:
|
|
|
- self.log_test(f"Key Backend: {backend}", False, "Invalid backend name")
|
|
|
- success = False
|
|
|
-
|
|
|
- except Exception as e:
|
|
|
- self.log_test(f"Key Backend: {backend}", False, f"Error: {str(e)}")
|
|
|
- success = False
|
|
|
-
|
|
|
- # Test health notification backends
|
|
|
- for backend in self.config.ntfy_backends_health:
|
|
|
- try:
|
|
|
- # Similar validation for health backends
|
|
|
- if backend and isinstance(backend, str) and len(backend.strip()) > 0:
|
|
|
- self.log_test(f"Health Backend: {backend}", True, "Backend name valid (using global ntfy config)")
|
|
|
- else:
|
|
|
- self.log_test(f"Health Backend: {backend}", False, "Invalid backend name")
|
|
|
- success = False
|
|
|
-
|
|
|
- except Exception as e:
|
|
|
- self.log_test(f"Health Backend: {backend}", False, f"Error: {str(e)}")
|
|
|
- success = False
|
|
|
-
|
|
|
- return success
|
|
|
-
|
|
|
- def start_test_server(self) -> bool:
|
|
|
- """Start the server for endpoint testing"""
|
|
|
- try:
|
|
|
- import subprocess
|
|
|
- import time
|
|
|
-
|
|
|
- # Start server in background
|
|
|
- cmd = [sys.executable, "main.py"]
|
|
|
- env = os.environ.copy()
|
|
|
- env['EMERGENCY_CONFIG'] = self.config.config_path
|
|
|
-
|
|
|
- self.server_process = subprocess.Popen(
|
|
|
- cmd,
|
|
|
- stdout=subprocess.PIPE,
|
|
|
- stderr=subprocess.PIPE,
|
|
|
- env=env
|
|
|
- )
|
|
|
-
|
|
|
- # Wait for server to start
|
|
|
- time.sleep(3)
|
|
|
-
|
|
|
- # Check if server is running
|
|
|
- if self.server_process.poll() is None:
|
|
|
- return self.log_test("Server Start", True, "Server started successfully")
|
|
|
- else:
|
|
|
- stdout, stderr = self.server_process.communicate()
|
|
|
- error_msg = stderr.decode() if stderr else "Unknown error"
|
|
|
- return self.log_test("Server Start", False, f"Server failed to start: {error_msg}")
|
|
|
+ # Test missing backends
|
|
|
+ try:
|
|
|
+ KeyConfig("test", {"route": "/test", "file": "/tmp/test.txt"}, {})
|
|
|
+ assert False, "Should have raised exception for missing backends"
|
|
|
+ except Exception as e:
|
|
|
+ assert "No notification backends configured" in str(e)
|
|
|
+
|
|
|
+ # Test valid config
|
|
|
+ key_config = KeyConfig(
|
|
|
+ "test",
|
|
|
+ {
|
|
|
+ "route": "/test",
|
|
|
+ "file": "/tmp/test.txt",
|
|
|
+ "backends": ["backend1"],
|
|
|
+ "message": "Test message"
|
|
|
+ },
|
|
|
+ {}
|
|
|
+ )
|
|
|
+
|
|
|
+ assert key_config.key_id == "test"
|
|
|
+ assert key_config.route == "/test"
|
|
|
+ assert key_config.file_path == "/tmp/test.txt"
|
|
|
+ assert key_config.backends == ["backend1"]
|
|
|
+ assert key_config.message == "Test message"
|
|
|
+
|
|
|
+ print("✅ Key configuration validation test passed")
|
|
|
+
|
|
|
+def test_app_integration():
|
|
|
+ """Test Flask app integration with multiple keys"""
|
|
|
+ print("Testing Flask app integration...")
|
|
|
+
|
|
|
+ # Create test files
|
|
|
+ temp_dir = tempfile.mkdtemp()
|
|
|
+ try:
|
|
|
+ # Create test key files
|
|
|
+ backup_key_file = os.path.join(temp_dir, "backup-key.txt")
|
|
|
+ master_key_file = os.path.join(temp_dir, "master-key.txt")
|
|
|
+ dummy_file = os.path.join(temp_dir, "dummy.txt")
|
|
|
+
|
|
|
+ with open(backup_key_file, 'w') as f:
|
|
|
+ f.write("backup_key_content_123")
|
|
|
+ with open(master_key_file, 'w') as f:
|
|
|
+ f.write("master_key_content_456")
|
|
|
+ with open(dummy_file, 'w') as f:
|
|
|
+ f.write("system_healthy")
|
|
|
+
|
|
|
+ # Create config
|
|
|
+ config_data = {
|
|
|
+ "server": {
|
|
|
+ "host": "127.0.0.1",
|
|
|
+ "port": 1127
|
|
|
+ },
|
|
|
+ "routes": {
|
|
|
+ "health_route": "/health-check-test"
|
|
|
+ },
|
|
|
+ "files": {
|
|
|
+ "dummy_file": dummy_file
|
|
|
+ },
|
|
|
+ "keys": {
|
|
|
+ "backup_key": {
|
|
|
+ "route": "/emergency-key-backup-a7f9d2e1b4c7f3e6d9a2b5c8e1f4d7a0b3c6e9f2a5d8b1c4e7f0a3d6b9c2e5f8a1d4b7c0e3f6d9a2b5c8e1f4a7b0d3",
|
|
|
+ "file": backup_key_file,
|
|
|
+ "backends": ["test_backend1"],
|
|
|
+ "message": "Backup key accessed"
|
|
|
+ },
|
|
|
+ "master_key": {
|
|
|
+ "route": "/emergency-key-master-x3k8m9n2p5q7r1t4u6v9w2y5z8b1c4d7e0f3g6h9j2k5l8m1n4o7p0q3r6s9t2u5v8w1x4y7z0a3b6c9d2e5f8",
|
|
|
+ "file": master_key_file,
|
|
|
+ "backends": ["test_backend2"],
|
|
|
+ "message": "Master key accessed"
|
|
|
+ }
|
|
|
+ },
|
|
|
+ "notifications": {
|
|
|
+ "health_backends": ["health_backend"],
|
|
|
+ "config_path": "/tmp/ntfy.yml",
|
|
|
+ "health_message": "Health check",
|
|
|
+ "log_level": "WARNING",
|
|
|
+ "send_all_logs": false
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ # Write config
|
|
|
+ with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
|
|
|
+ json.dump(config_data, f)
|
|
|
+ config_file = f.name
|
|
|
|
|
|
- except Exception as e:
|
|
|
- return self.log_test("Server Start", False, f"Error: {str(e)}")
|
|
|
-
|
|
|
- def stop_test_server(self):
|
|
|
- """Stop the test server"""
|
|
|
- if self.server_process and self.server_process.poll() is None:
|
|
|
- self.server_process.terminate()
|
|
|
- try:
|
|
|
- self.server_process.wait(timeout=5)
|
|
|
- except subprocess.TimeoutExpired:
|
|
|
- self.server_process.kill()
|
|
|
- self.server_process.wait()
|
|
|
- self.log_test("Server Stop", True, "Server stopped successfully")
|
|
|
-
|
|
|
- def test_endpoints(self) -> bool:
|
|
|
- """Test server endpoints"""
|
|
|
- success = True
|
|
|
- base_url = f"http://{self.config.server_host}:{self.config.server_port}"
|
|
|
-
|
|
|
- # Test health endpoint
|
|
|
try:
|
|
|
- import requests
|
|
|
- response = requests.get(
|
|
|
- f"{base_url}{self.config.health_route}",
|
|
|
- timeout=30
|
|
|
- )
|
|
|
-
|
|
|
- if response.status_code == 200:
|
|
|
- data = response.json()
|
|
|
- if data.get('status') == 'ok':
|
|
|
- self.log_test("Health Endpoint", True, f"Response: {response.status_code}")
|
|
|
- else:
|
|
|
- self.log_test("Health Endpoint", False, f"Invalid response: {data}")
|
|
|
- success = False
|
|
|
- else:
|
|
|
- self.log_test("Health Endpoint", False, f"HTTP {response.status_code}")
|
|
|
- success = False
|
|
|
+ # Set environment variable
|
|
|
+ os.environ['EMERGENCY_CONFIG'] = config_file
|
|
|
|
|
|
- except Exception as e:
|
|
|
- self.log_test("Health Endpoint", False, f"Error: {str(e)}")
|
|
|
- success = False
|
|
|
+ # Import main module (this tests configuration loading)
|
|
|
+ import main
|
|
|
|
|
|
- # Test key endpoint
|
|
|
- try:
|
|
|
- import requests
|
|
|
- response = requests.get(
|
|
|
- f"{base_url}{self.config.key_route}",
|
|
|
- timeout=30
|
|
|
- )
|
|
|
-
|
|
|
- if response.status_code == 200:
|
|
|
- data = response.json()
|
|
|
- if data.get('success') and 'key_part' in data:
|
|
|
- self.log_test("Key Endpoint", True, f"Key retrieved successfully")
|
|
|
- else:
|
|
|
- self.log_test("Key Endpoint", False, f"Invalid response: {data}")
|
|
|
- success = False
|
|
|
- else:
|
|
|
- self.log_test("Key Endpoint", False, f"HTTP {response.status_code}")
|
|
|
- success = False
|
|
|
+ # Test config loading
|
|
|
+ main.config = Config(config_file)
|
|
|
|
|
|
- except Exception as e:
|
|
|
- self.log_test("Key Endpoint", False, f"Error: {str(e)}")
|
|
|
- success = False
|
|
|
+ # Test file reading
|
|
|
+ success, content = main.read_file_safely(backup_key_file)
|
|
|
+ assert success == True
|
|
|
+ assert content == "backup_key_content_123"
|
|
|
|
|
|
- # Test 404 handling
|
|
|
- try:
|
|
|
- import requests
|
|
|
- response = requests.get(f"{base_url}/nonexistent-path", timeout=10)
|
|
|
- if response.status_code == 404:
|
|
|
- self.log_test("404 Handling", True, "Correctly returns 404 for invalid paths")
|
|
|
- else:
|
|
|
- self.log_test("404 Handling", False, f"Expected 404, got {response.status_code}")
|
|
|
- success = False
|
|
|
- except Exception as e:
|
|
|
- self.log_test("404 Handling", False, f"Error: {str(e)}")
|
|
|
- success = False
|
|
|
+ success, content = main.read_file_safely(master_key_file)
|
|
|
+ assert success == True
|
|
|
+ assert content == "master_key_content_456"
|
|
|
|
|
|
- return success
|
|
|
+ # Test key handler creation
|
|
|
+ backup_key_config = main.config.get_key_by_id("backup_key")
|
|
|
+ handler = main.create_key_handler(backup_key_config)
|
|
|
+ assert handler is not None
|
|
|
|
|
|
- def test_fail_safe_behavior(self) -> bool:
|
|
|
- """Test fail-safe behavior with invalid backends"""
|
|
|
- success = True
|
|
|
+ print("✅ Flask app integration test passed")
|
|
|
|
|
|
- # Create temporary config with invalid backends
|
|
|
- try:
|
|
|
- with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
|
|
|
- invalid_config = {
|
|
|
- "server": {
|
|
|
- "host": self.config.server_host,
|
|
|
- "port": self.config.server_port + 1 # Different port
|
|
|
- },
|
|
|
- "routes": {
|
|
|
- "key_route": "/test-key",
|
|
|
- "health_route": "/test-health"
|
|
|
- },
|
|
|
- "files": {
|
|
|
- "key_file": self.config.key_file_path,
|
|
|
- "dummy_file": self.config.dummy_file_path
|
|
|
- },
|
|
|
- "notifications": {
|
|
|
- "key_backends": ["invalid_backend_test"],
|
|
|
- "health_backends": ["invalid_backend_test"],
|
|
|
- "key_message": "Test message",
|
|
|
- "health_message": "Test health message"
|
|
|
- }
|
|
|
- }
|
|
|
- json.dump(invalid_config, f)
|
|
|
- temp_config_path = f.name
|
|
|
-
|
|
|
- # Test with invalid config
|
|
|
- test_config = Config(temp_config_path)
|
|
|
-
|
|
|
- # Start server with invalid config
|
|
|
- cmd = [sys.executable, "main.py"]
|
|
|
- env = os.environ.copy()
|
|
|
- env['EMERGENCY_CONFIG'] = temp_config_path
|
|
|
-
|
|
|
- test_process = subprocess.Popen(
|
|
|
- cmd,
|
|
|
- stdout=subprocess.PIPE,
|
|
|
- stderr=subprocess.PIPE,
|
|
|
- env=env
|
|
|
- )
|
|
|
-
|
|
|
- time.sleep(3)
|
|
|
-
|
|
|
- if test_process.poll() is None:
|
|
|
- # Try to access endpoints - should fail due to notification failures
|
|
|
- base_url = f"http://{test_config.server_host}:{test_config.server_port}"
|
|
|
-
|
|
|
- try:
|
|
|
- import requests
|
|
|
- response = requests.get(f"{base_url}/test-key", timeout=15)
|
|
|
- if response.status_code == 500:
|
|
|
- self.log_test("Fail-Safe Key", True, "Correctly blocks access when notifications fail")
|
|
|
- else:
|
|
|
- self.log_test("Fail-Safe Key", False, f"Expected 500, got {response.status_code}")
|
|
|
- success = False
|
|
|
- except Exception as e:
|
|
|
- self.log_test("Fail-Safe Key", False, f"Error testing fail-safe: {str(e)}")
|
|
|
- success = False
|
|
|
-
|
|
|
- # Clean up
|
|
|
- test_process.terminate()
|
|
|
- try:
|
|
|
- test_process.wait(timeout=5)
|
|
|
- except subprocess.TimeoutExpired:
|
|
|
- test_process.kill()
|
|
|
- else:
|
|
|
- self.log_test("Fail-Safe Test", False, "Test server failed to start")
|
|
|
- success = False
|
|
|
-
|
|
|
- # Clean up temp file
|
|
|
- os.unlink(temp_config_path)
|
|
|
+ finally:
|
|
|
+ os.unlink(config_file)
|
|
|
+ if 'EMERGENCY_CONFIG' in os.environ:
|
|
|
+ del os.environ['EMERGENCY_CONFIG']
|
|
|
|
|
|
- except Exception as e:
|
|
|
- self.log_test("Fail-Safe Test", False, f"Error: {str(e)}")
|
|
|
- success = False
|
|
|
-
|
|
|
- return success
|
|
|
-
|
|
|
- def run_all_tests(self) -> bool:
|
|
|
- """Run all tests and return overall success"""
|
|
|
- print("=" * 60)
|
|
|
- print("Emergency Access Server Test Suite")
|
|
|
- print("=" * 60)
|
|
|
-
|
|
|
- overall_success = True
|
|
|
-
|
|
|
- # Configuration tests
|
|
|
- print("\n--- Configuration Tests ---")
|
|
|
- overall_success &= self.test_config_loading()
|
|
|
-
|
|
|
- # File system tests
|
|
|
- print("\n--- File System Tests ---")
|
|
|
- overall_success &= self.test_file_access()
|
|
|
-
|
|
|
- # Network tests
|
|
|
- print("\n--- Network Tests ---")
|
|
|
- overall_success &= self.test_ntfy_connectivity()
|
|
|
-
|
|
|
- # Server tests
|
|
|
- print("\n--- Server Tests ---")
|
|
|
- if self.start_test_server():
|
|
|
- time.sleep(2) # Give server time to fully start
|
|
|
- overall_success &= self.test_endpoints()
|
|
|
- self.stop_test_server()
|
|
|
- else:
|
|
|
- overall_success = False
|
|
|
-
|
|
|
- # Fail-safe tests
|
|
|
- print("\n--- Fail-Safe Tests ---")
|
|
|
- overall_success &= self.test_fail_safe_behavior()
|
|
|
-
|
|
|
- # Print summary
|
|
|
- self.print_summary()
|
|
|
-
|
|
|
- return overall_success
|
|
|
-
|
|
|
- def print_summary(self):
|
|
|
- """Print test summary"""
|
|
|
- print("\n" + "=" * 60)
|
|
|
- print("TEST SUMMARY")
|
|
|
- print("=" * 60)
|
|
|
-
|
|
|
- passed = sum(1 for r in self.test_results if r['status'] == 'PASS')
|
|
|
- failed = sum(1 for r in self.test_results if r['status'] == 'FAIL')
|
|
|
- total = len(self.test_results)
|
|
|
-
|
|
|
- print(f"Total Tests: {total}")
|
|
|
- print(f"Passed: {passed}")
|
|
|
- print(f"Failed: {failed}")
|
|
|
-
|
|
|
- if failed > 0:
|
|
|
- print("\nFAILED TESTS:")
|
|
|
- for result in self.test_results:
|
|
|
- if result['status'] == 'FAIL':
|
|
|
- print(f" - {result['test']}: {result['message']}")
|
|
|
-
|
|
|
- overall_status = "PASS" if failed == 0 else "FAIL"
|
|
|
- color = '\033[92m' if failed == 0 else '\033[91m'
|
|
|
- reset = '\033[0m'
|
|
|
- print(f"\n{color}Overall Status: {overall_status}{reset}")
|
|
|
+ finally:
|
|
|
+ shutil.rmtree(temp_dir)
|
|
|
|
|
|
def main():
|
|
|
- """Main function"""
|
|
|
- import argparse
|
|
|
+ """Run all tests"""
|
|
|
+ print("Running multi-key functionality tests...\n")
|
|
|
|
|
|
- parser = argparse.ArgumentParser(description='Test Emergency Access Server')
|
|
|
- parser.add_argument('--config', help='Configuration file path')
|
|
|
- parser.add_argument('--quick', action='store_true', help='Run quick tests only (skip server startup)')
|
|
|
+ try:
|
|
|
+ test_multikey_config()
|
|
|
+ test_invalid_config()
|
|
|
+ test_key_config_validation()
|
|
|
+ test_app_integration()
|
|
|
|
|
|
- args = parser.parse_args()
|
|
|
+ print("\n🎉 All tests passed! Multi-key functionality is working correctly.")
|
|
|
+ return True
|
|
|
|
|
|
- try:
|
|
|
- tester = EmergencyAccessTester(args.config)
|
|
|
-
|
|
|
- if args.quick:
|
|
|
- # Quick tests only
|
|
|
- success = True
|
|
|
- success &= tester.test_config_loading()
|
|
|
- success &= tester.test_file_access()
|
|
|
- success &= tester.test_ntfy_connectivity()
|
|
|
- tester.print_summary()
|
|
|
- else:
|
|
|
- # Full test suite
|
|
|
- success = tester.run_all_tests()
|
|
|
-
|
|
|
- sys.exit(0 if success else 1)
|
|
|
-
|
|
|
- except KeyboardInterrupt:
|
|
|
- print("\nTest interrupted by user")
|
|
|
- sys.exit(1)
|
|
|
+ except AssertionError as e:
|
|
|
+ print(f"\n❌ Test failed: {e}")
|
|
|
+ return False
|
|
|
except Exception as e:
|
|
|
- print(f"Test suite failed: {str(e)}")
|
|
|
- sys.exit(1)
|
|
|
-
|
|
|
-if __name__ == '__main__':
|
|
|
- main()
|
|
|
+ print(f"\n❌ Unexpected error: {e}")
|
|
|
+ import traceback
|
|
|
+ traceback.print_exc()
|
|
|
+ return False
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ success = main()
|
|
|
+ sys.exit(0 if success else 1)
|
zehe