home-automation/tools/test_config_loader.py

#!/usr/bin/env python3
"""Test configuration loader with different YAML formats."""

import sys
import tempfile
from pathlib import Path

# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))

from apps.abstraction.main import load_config, validate_devices


def test_device_id_format():
    """Test YAML with 'device_id' key."""
    yaml_content = """version: 1

mqtt:
  broker: "172.16.2.16"
  port: 1883
  client_id: "test"

redis:
  url: "redis://localhost:6379/8"
  channel: "ui:updates"

devices:
  - device_id: test_lampe
    type: light
    cap_version: "light@1.2.0"
    technology: zigbee2mqtt
    features:
      power: true
    topics:
      set: "vendor/test_lampe/set"
      state: "vendor/test_lampe/state"
"""
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
        f.write(yaml_content)
        temp_path = Path(f.name)
    
    try:
        config = load_config(temp_path)
        devices = config.get("devices", [])
        validate_devices(devices)
        print("✓ Test 1 passed: YAML with 'device_id' loaded successfully")
        print(f"  Device ID: {devices[0]['device_id']}")
        return True
    except Exception as e:
        print(f"✗ Test 1 failed: {e}")
        return False
    finally:
        temp_path.unlink()


def test_id_format():
    """Test YAML with 'id' key (legacy format)."""
    yaml_content = """version: 1

mqtt:
  broker: "172.16.2.16"
  port: 1883
  client_id: "test"

redis:
  url: "redis://localhost:6379/8"
  channel: "ui:updates"

devices:
  - id: test_lampe
    type: light
    cap_version: "light@1.2.0"
    technology: zigbee2mqtt
    features:
      power: true
    topics:
      set: "vendor/test_lampe/set"
      state: "vendor/test_lampe/state"
"""
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
        f.write(yaml_content)
        temp_path = Path(f.name)
    
    try:
        config = load_config(temp_path)
        devices = config.get("devices", [])
        validate_devices(devices)
        print("✓ Test 2 passed: YAML with 'id' loaded successfully")
        print(f"  Device ID: {devices[0]['device_id']}")
        return True
    except Exception as e:
        print(f"✗ Test 2 failed: {e}")
        return False
    finally:
        temp_path.unlink()


def test_missing_id():
    """Test YAML without 'id' or 'device_id' key."""
    yaml_content = """version: 1

mqtt:
  broker: "172.16.2.16"
  port: 1883

devices:
  - type: light
    cap_version: "light@1.2.0"
    technology: zigbee2mqtt
    topics:
      set: "vendor/test_lampe/set"
      state: "vendor/test_lampe/state"
"""
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
        f.write(yaml_content)
        temp_path = Path(f.name)
    
    try:
        config = load_config(temp_path)
        devices = config.get("devices", [])
        validate_devices(devices)
        print("✗ Test 3 failed: Should have raised ValueError for missing device_id")
        return False
    except ValueError as e:
        if "requires 'id' or 'device_id'" in str(e):
            print(f"✓ Test 3 passed: Correct error for missing device_id")
            print(f"  Error: {e}")
            return True
        else:
            print(f"✗ Test 3 failed: Wrong error message: {e}")
            return False
    except Exception as e:
        print(f"✗ Test 3 failed: Unexpected error: {e}")
        return False
    finally:
        temp_path.unlink()


def test_missing_topics_set():
    """Test YAML without 'topics.set'."""
    yaml_content = """version: 1

mqtt:
  broker: "172.16.2.16"
  port: 1883

devices:
  - device_id: test_lampe
    type: light
    cap_version: "light@1.2.0"
    technology: zigbee2mqtt
    topics:
      state: "vendor/test_lampe/state"
"""
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
        f.write(yaml_content)
        temp_path = Path(f.name)
    
    try:
        config = load_config(temp_path)
        devices = config.get("devices", [])
        validate_devices(devices)
        print("✗ Test 4 failed: Should have raised ValueError for missing topics.set")
        return False
    except ValueError as e:
        if "missing 'topics.set'" in str(e):
            print(f"✓ Test 4 passed: Correct error for missing topics.set")
            print(f"  Error: {e}")
            return True
        else:
            print(f"✗ Test 4 failed: Wrong error message: {e}")
            return False
    except Exception as e:
        print(f"✗ Test 4 failed: Unexpected error: {e}")
        return False
    finally:
        temp_path.unlink()


def test_missing_topics_state():
    """Test YAML without 'topics.state'."""
    yaml_content = """version: 1

mqtt:
  broker: "172.16.2.16"
  port: 1883

devices:
  - device_id: test_lampe
    type: light
    cap_version: "light@1.2.0"
    technology: zigbee2mqtt
    topics:
      set: "vendor/test_lampe/set"
"""
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
        f.write(yaml_content)
        temp_path = Path(f.name)
    
    try:
        config = load_config(temp_path)
        devices = config.get("devices", [])
        validate_devices(devices)
        print("✗ Test 5 failed: Should have raised ValueError for missing topics.state")
        return False
    except ValueError as e:
        if "missing 'topics.state'" in str(e):
            print(f"✓ Test 5 passed: Correct error for missing topics.state")
            print(f"  Error: {e}")
            return True
        else:
            print(f"✗ Test 5 failed: Wrong error message: {e}")
            return False
    except Exception as e:
        print(f"✗ Test 5 failed: Unexpected error: {e}")
        return False
    finally:
        temp_path.unlink()


def main():
    """Run all tests."""
    print("Testing Configuration Loader")
    print("=" * 60)
    
    tests = [
        test_device_id_format,
        test_id_format,
        test_missing_id,
        test_missing_topics_set,
        test_missing_topics_state,
    ]
    
    results = []
    for test in tests:
        results.append(test())
        print()
    
    print("=" * 60)
    passed = sum(results)
    total = len(results)
    print(f"Results: {passed}/{total} tests passed")
    
    return 0 if all(results) else 1


if __name__ == "__main__":
    sys.exit(main())