emergency-access/test.py

#!/usr/bin/env python3

import os
import sys
import json
import tempfile
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
        }
    }

    # 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:
        # 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
        }
    }

    # 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:
            config = Config(config_file)
            assert False, "Should have raised exception for missing keys section"
        except Exception as e:
            assert "No keys configured" in str(e)

        print("✅ Invalid configuration test passed")

    finally:
        os.unlink(config_file)

def test_key_config_validation():
    """Test key configuration validation"""
    print("Testing key configuration validation...")

    # 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)

    # 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 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

        try:
            # Set environment variable
            os.environ['EMERGENCY_CONFIG'] = config_file

            # Import main module (this tests configuration loading)
            import main

            # Test config loading
            main.config = Config(config_file)

            # Test file reading
            success, content = main.read_file_safely(backup_key_file)
            assert success == True
            assert content == "backup_key_content_123"

            success, content = main.read_file_safely(master_key_file)
            assert success == True
            assert content == "master_key_content_456"

            # 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

            print("✅ Flask app integration test passed")

        finally:
            os.unlink(config_file)
            if 'EMERGENCY_CONFIG' in os.environ:
                del os.environ['EMERGENCY_CONFIG']

    finally:
        shutil.rmtree(temp_dir)

def main():
    """Run all tests"""
    print("Running multi-key functionality tests...\n")

    try:
        test_multikey_config()
        test_invalid_config()
        test_key_config_validation()
        test_app_integration()

        print("\n🎉 All tests passed! Multi-key functionality is working correctly.")
        return True

    except AssertionError as e:
        print(f"\n❌ Test failed: {e}")
        return False
    except Exception as e:
        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)