zwei Lampen und Test-Werkzeug
This commit is contained in:
253
tools/test_config_loader.py
Normal file
253
tools/test_config_loader.py
Normal file
@@ -0,0 +1,253 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user