diff --git a/.woodpecker/predeploy.yml b/.woodpecker/predeploy.yml index 2f0ce6d..b43af7e 100644 --- a/.woodpecker/predeploy.yml +++ b/.woodpecker/predeploy.yml @@ -33,6 +33,7 @@ steps: --namespace=$NAMESPACE --dry-run=client -o yaml | kubectl apply -f - - kubectl apply -f deployment/configmap.yaml -n $NAMESPACE + - kubectl apply -f deployment/mtls-config.yaml -n $NAMESPACE when: event: [tag] diff --git a/deployment/api-deployment.yaml b/deployment/api-deployment.yaml index c6b9756..c433592 100644 --- a/deployment/api-deployment.yaml +++ b/deployment/api-deployment.yaml @@ -107,6 +107,9 @@ metadata: name: api-ingress annotations: cert-manager.io/cluster-issuer: letsencrypt-production-http + # Traefik 2 mTLS Configuration + traefik.ingress.kubernetes.io/router.tls.options: homea2-mtls@kubernetescrd + traefik.ingress.kubernetes.io/router.middlewares: homea2-mtls-auth@kubernetescrd spec: tls: - hosts: diff --git a/deployment/mtls-config.yaml b/deployment/mtls-config.yaml new file mode 100644 index 0000000..bf66781 --- /dev/null +++ b/deployment/mtls-config.yaml @@ -0,0 +1,45 @@ +apiVersion: traefik.containo.us/v1alpha1 +kind: TLSOption +metadata: + name: homea2-mtls +spec: + clientAuth: + secretNames: + - mtls-ca-cert + clientAuthType: RequireAndVerifyClientCert + minVersion: "VersionTLS12" + cipherSuites: + - "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" + - "TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305" + - "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256" + - "TLS_RSA_WITH_AES_256_GCM_SHA384" + - "TLS_RSA_WITH_AES_128_GCM_SHA256" +--- +apiVersion: traefik.containo.us/v1alpha1 +kind: Middleware +metadata: + name: mtls-auth +spec: + headers: + customRequestHeaders: + X-Client-Cert: "" + customResponseHeaders: + X-mTLS-Verified: "true" + # Optional: Add IP whitelist for additional security + # ipWhiteList: + # sourceRange: + # - "10.0.0.0/8" + # - "192.168.0.0/16" +--- +apiVersion: traefik.containo.us/v1alpha1 +kind: Middleware +metadata: + name: security-headers +spec: + headers: + customResponseHeaders: + X-Frame-Options: "SAMEORIGIN" + X-Content-Type-Options: "nosniff" + X-XSS-Protection: "1; mode=block" + Strict-Transport-Security: "max-age=31536000; includeSubDomains; preload" + contentSecurityPolicy: "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'" \ No newline at end of file diff --git a/deployment/ui-deployment.yaml b/deployment/ui-deployment.yaml index 4854976..5d915e9 100644 --- a/deployment/ui-deployment.yaml +++ b/deployment/ui-deployment.yaml @@ -84,6 +84,9 @@ metadata: name: ui-ingress annotations: cert-manager.io/cluster-issuer: letsencrypt-production-http + # Traefik 2 mTLS Configuration + traefik.ingress.kubernetes.io/router.tls.options: homea2-mtls@kubernetescrd + traefik.ingress.kubernetes.io/router.middlewares: homea2-mtls-auth@kubernetescrd spec: tls: - hosts: diff --git a/tools/README_device_simulator.md b/tools/README_device_simulator.md deleted file mode 100644 index 15967db..0000000 --- a/tools/README_device_simulator.md +++ /dev/null @@ -1,190 +0,0 @@ -# Device Simulator - -Unified MQTT device simulator für das Home Automation System. - -## Übersicht - -Dieser Simulator ersetzt die einzelnen Simulatoren (`sim_test_lampe.py`, `sim_thermo.py`) und vereint alle Device-Typen in einer einzigen Anwendung. - -## Unterstützte Geräte - -### Lampen (3 Geräte) -- `test_lampe_1` - Mit Power und Brightness -- `test_lampe_2` - Mit Power und Brightness -- `test_lampe_3` - Mit Power und Brightness - -**Features:** -- `power`: "on" oder "off" -- `brightness`: 0-100 - -### Thermostaten (1 Gerät) -- `test_thermo_1` - Vollständiger Thermostat mit Temperatur-Simulation - -**Features:** -- `mode`: "off", "heat", oder "auto" -- `target`: Soll-Temperatur (5.0-30.0°C) -- `current`: Ist-Temperatur (wird simuliert) -- `battery`: Batteriestand (90%) -- `window_open`: Fensterstatus (false) - -**Temperatur-Simulation:** -- Alle 5 Sekunden wird die Ist-Temperatur angepasst -- **HEAT/AUTO Mode**: Drift zu `target` (+0.2°C pro Intervall) -- **OFF Mode**: Drift zu Ambient-Temperatur 18°C (-0.2°C pro Intervall) - -## MQTT-Konfiguration - -- **Broker**: 172.16.2.16:1883 (konfigurierbar via ENV) -- **QoS**: 1 für alle Publishes -- **Retained**: Ja für alle State-Messages -- **Client ID**: device_simulator - -### Topics - -Für jedes Gerät: -- Subscribe: `vendor/{device_id}/set` (QoS 1) -- Publish: `vendor/{device_id}/state` (QoS 1, retained) - -## Verwendung - -### Starten - -```bash -poetry run python tools/device_simulator.py -``` - -Oder im Hintergrund: - -```bash -poetry run python tools/device_simulator.py > /tmp/simulator.log 2>&1 & -``` - -### Umgebungsvariablen - -```bash -export MQTT_BROKER="172.16.2.16" # MQTT Broker Host -export MQTT_PORT="1883" # MQTT Broker Port -``` - -## Testen - -Ein umfassendes Test-Skript ist verfügbar: - -```bash -./tools/test_device_simulator.sh -``` - -Das Test-Skript: -1. Stoppt alle laufenden Services -2. Startet Abstraction Layer, API und Simulator -3. Testet alle Lampen-Operationen -4. Testet alle Thermostat-Operationen -5. Verifiziert MQTT State Messages -6. Zeigt Simulator-Logs - -## Beispiele - -### Lampe einschalten - -```bash -curl -X POST http://localhost:8001/devices/test_lampe_1/set \ - -H "Content-Type: application/json" \ - -d '{"type":"light","payload":{"power":"on"}}' -``` - -### Helligkeit setzen - -```bash -curl -X POST http://localhost:8001/devices/test_lampe_1/set \ - -H "Content-Type: application/json" \ - -d '{"type":"light","payload":{"brightness":75}}' -``` - -### Thermostat Mode setzen - -```bash -curl -X POST http://localhost:8001/devices/test_thermo_1/set \ - -H "Content-Type: application/json" \ - -d '{"type":"thermostat","payload":{"mode":"heat","target":22.5}}' -``` - -### State abfragen via MQTT - -```bash -# Lampe -mosquitto_sub -h 172.16.2.16 -t 'vendor/test_lampe_1/state' -C 1 - -# Thermostat -mosquitto_sub -h 172.16.2.16 -t 'vendor/test_thermo_1/state' -C 1 -``` - -## Architektur - -``` -Browser/API - ↓ POST /devices/{id}/set -API Server (Port 8001) - ↓ MQTT: home/{type}/{id}/set -Abstraction Layer - ↓ MQTT: vendor/{id}/set -Device Simulator - ↓ MQTT: vendor/{id}/state (retained) -Abstraction Layer - ↓ MQTT: home/{type}/{id}/state (retained) - ↓ Redis Pub/Sub: ui:updates -UI / Dashboard -``` - -## Logs - -Der Simulator loggt alle Aktivitäten: -- Startup und MQTT-Verbindung -- Empfangene SET-Commands -- State-Änderungen -- Temperature-Drift (Thermostaten) -- Publizierte State-Messages - -Log-Level: INFO - -## Troubleshooting - -### Simulator startet nicht - -```bash -# Prüfe ob Port bereits belegt -lsof -ti:1883 - -# Prüfe MQTT Broker -mosquitto_sub -h 172.16.2.16 -t '#' -C 1 -``` - -### Keine State-Updates - -```bash -# Prüfe Simulator-Log -tail -f /tmp/simulator.log - -# Prüfe MQTT Topics -mosquitto_sub -h 172.16.2.16 -t 'vendor/#' -v -``` - -### API antwortet nicht - -```bash -# Prüfe ob API läuft -curl http://localhost:8001/devices - -# Prüfe API-Log -tail -f /tmp/api.log -``` - -## Integration - -Der Simulator integriert sich nahtlos in das Home Automation System: - -1. **Abstraction Layer** empfängt Commands und sendet sie an Simulator -2. **Simulator** reagiert und publiziert neuen State -3. **Abstraction Layer** empfängt State und publiziert zu Redis -4. **UI** empfängt Updates via SSE und aktualisiert Dashboard - -Alle Komponenten arbeiten vollständig asynchron über MQTT. diff --git a/tools/create-client-cert.sh b/tools/create-client-cert.sh new file mode 100755 index 0000000..4d64631 --- /dev/null +++ b/tools/create-client-cert.sh @@ -0,0 +1,73 @@ +#!/bin/bash +set -e + +# Check if client name is provided +if [ $# -eq 0 ]; then + echo "Usage: $0 " + echo "Example: $0 john.doe" + exit 1 +fi + +CLIENT_NAME="$1" + +# Check if CA exists +if [ ! -f "ca/ca.crt" ] || [ ! -f "ca/ca.key" ]; then + echo "Error: CA not found. Please run setup-ca.sh first." + exit 1 +fi + +echo "=== Creating Client Certificate ===" +echo "Client Name: $CLIENT_NAME" + +# Create client directory +mkdir -p clients/$CLIENT_NAME + +# Generate client private key +echo "Generating client private key..." +openssl genrsa -out clients/$CLIENT_NAME/$CLIENT_NAME.key 2048 + +# Generate client certificate signing request +echo "Generating client certificate signing request..." +openssl req -new -key clients/$CLIENT_NAME/$CLIENT_NAME.key \ + -out clients/$CLIENT_NAME/$CLIENT_NAME.csr \ + -subj "/DC=de/DC=hottis/DC=homea2/CN=$CLIENT_NAME" + +# Sign the client certificate +echo "Signing client certificate..." +openssl x509 -req -in clients/$CLIENT_NAME/$CLIENT_NAME.csr \ + -CA ca/ca.crt -CAkey ca/ca.key -CAcreateserial \ + -out clients/$CLIENT_NAME/$CLIENT_NAME.crt \ + -days 365 -sha256 + +# Create PKCS#12 bundle +echo "Creating PKCS#12 bundle..." +openssl pkcs12 -export \ + -out clients/$CLIENT_NAME/$CLIENT_NAME.p12 \ + -inkey clients/$CLIENT_NAME/$CLIENT_NAME.key \ + -in clients/$CLIENT_NAME/$CLIENT_NAME.crt \ + -certfile ca/ca.crt \ + -name "$CLIENT_NAME Home Automation Client" \ + -passout pass: + +# Set appropriate permissions +chmod 400 clients/$CLIENT_NAME/$CLIENT_NAME.key +chmod 644 clients/$CLIENT_NAME/$CLIENT_NAME.crt +chmod 644 clients/$CLIENT_NAME/$CLIENT_NAME.p12 + +# Verify client certificate +echo "Verifying client certificate..." +openssl x509 -noout -text -in clients/$CLIENT_NAME/$CLIENT_NAME.crt + +echo "" +echo "=== Client Certificate Created ===" +echo "Client Certificate: clients/$CLIENT_NAME/$CLIENT_NAME.crt" +echo "Client Private Key: clients/$CLIENT_NAME/$CLIENT_NAME.key" +echo "PKCS#12 Bundle: clients/$CLIENT_NAME/$CLIENT_NAME.p12" +echo "" +echo "Installation Instructions:" +echo "1. Import the PKCS#12 file into your browser/application" +echo "2. The bundle contains both the client certificate and CA certificate" +echo "3. No password is set for the PKCS#12 file (you can add one by modifying the -passout parameter)" +echo "" +echo "For testing with curl:" +echo "curl --cert clients/$CLIENT_NAME/$CLIENT_NAME.crt --key clients/$CLIENT_NAME/$CLIENT_NAME.key --cacert ca/ca.crt https://homea2.hottis.de/" \ No newline at end of file diff --git a/tools/device_simulator.py b/tools/device_simulator.py deleted file mode 100755 index 01c0543..0000000 --- a/tools/device_simulator.py +++ /dev/null @@ -1,291 +0,0 @@ -#!/usr/bin/env python3 -""" -Unified Device Simulator for Home Automation. - -Simulates multiple device types: -- Lights (test_lampe_1, test_lampe_2, test_lampe_3) -- Thermostats (test_thermo_1) - -Each device: -- Subscribes to vendor/{device_id}/set -- Maintains local state -- Publishes state changes to vendor/{device_id}/state (retained, QoS 1) -- Thermostats simulate temperature drift every 5 seconds -""" - -import asyncio -import json -import logging -import os -import signal -import sys -import uuid -from datetime import datetime -from typing import Dict, Any - -from aiomqtt import Client, MqttError - -# Configure logging -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" -) -logger = logging.getLogger(__name__) - -# Configuration -BROKER_HOST = os.getenv("MQTT_BROKER", "172.16.2.16") -BROKER_PORT = int(os.getenv("MQTT_PORT", "1883")) -DRIFT_INTERVAL = 5 # seconds for thermostat temperature drift - -# Device configurations -LIGHT_DEVICES = ["test_lampe_1", "test_lampe_2", "test_lampe_3"] -THERMOSTAT_DEVICES = ["test_thermo_1"] - - -class DeviceSimulator: - """Unified simulator for lights and thermostats.""" - - def __init__(self): - # Light states - self.light_states: Dict[str, Dict[str, Any]] = { - "test_lampe_1": {"power": "off", "brightness": 50}, - "test_lampe_2": {"power": "off", "brightness": 50}, - "test_lampe_3": {"power": "off", "brightness": 50} - } - - # Thermostat states - self.thermostat_states: Dict[str, Dict[str, Any]] = { - "test_thermo_1": { - "mode": "auto", - "target": 21.0, - "current": 20.5, - "battery": 90, - "window_open": False - } - } - - self.client = None - self.running = True - self.drift_task = None - - async def publish_state(self, device_id: str, device_type: str): - """Publish device state to MQTT (retained, QoS 1).""" - if not self.client: - return - - if device_type == "light": - state = self.light_states.get(device_id) - elif device_type == "thermostat": - state = self.thermostat_states.get(device_id) - else: - logger.warning(f"Unknown device type: {device_type}") - return - - if not state: - logger.warning(f"Unknown device: {device_id}") - return - - state_topic = f"vendor/{device_id}/state" - payload = json.dumps(state) - - await self.client.publish( - state_topic, - payload=payload, - qos=1, - retain=True - ) - logger.info(f"[{device_id}] Published state: {payload}") - - async def handle_light_set(self, device_id: str, payload: dict): - """Handle SET command for light device.""" - if device_id not in self.light_states: - logger.warning(f"Unknown light device: {device_id}") - return - - state = self.light_states[device_id] - updated = False - - if "power" in payload: - old_power = state["power"] - state["power"] = payload["power"] - if old_power != state["power"]: - updated = True - logger.info(f"[{device_id}] Power: {old_power} -> {state['power']}") - - if "brightness" in payload: - old_brightness = state["brightness"] - state["brightness"] = int(payload["brightness"]) - if old_brightness != state["brightness"]: - updated = True - logger.info(f"[{device_id}] Brightness: {old_brightness} -> {state['brightness']}") - - if updated: - await self.publish_state(device_id, "light") - - async def handle_thermostat_set(self, device_id: str, payload: dict): - """Handle SET command for thermostat device.""" - if device_id not in self.thermostat_states: - logger.warning(f"Unknown thermostat device: {device_id}") - return - - state = self.thermostat_states[device_id] - updated = False - - if "mode" in payload: - new_mode = payload["mode"] - if new_mode in ["off", "heat", "auto"]: - old_mode = state["mode"] - state["mode"] = new_mode - if old_mode != new_mode: - updated = True - logger.info(f"[{device_id}] Mode: {old_mode} -> {new_mode}") - else: - logger.warning(f"[{device_id}] Invalid mode: {new_mode}") - - if "target" in payload: - try: - new_target = float(payload["target"]) - if 5.0 <= new_target <= 30.0: - old_target = state["target"] - state["target"] = new_target - if old_target != new_target: - updated = True - logger.info(f"[{device_id}] Target: {old_target}°C -> {new_target}°C") - else: - logger.warning(f"[{device_id}] Target out of range: {new_target}") - except (ValueError, TypeError): - logger.warning(f"[{device_id}] Invalid target value: {payload['target']}") - - if updated: - await self.publish_state(device_id, "thermostat") - - def apply_temperature_drift(self, device_id: str): - """ - Simulate temperature drift for thermostat. - Max change: ±0.2°C per interval. - """ - if device_id not in self.thermostat_states: - return - - state = self.thermostat_states[device_id] - - if state["mode"] == "off": - # Drift towards ambient (18°C) - ambient = 18.0 - diff = ambient - state["current"] - else: - # Drift towards target - diff = state["target"] - state["current"] - - # Apply max ±0.2°C drift - if abs(diff) < 0.1: - state["current"] = round(state["current"] + diff, 1) - elif diff > 0: - state["current"] = round(state["current"] + 0.2, 1) - else: - state["current"] = round(state["current"] - 0.2, 1) - - logger.info(f"[{device_id}] Temperature drift: current={state['current']}°C (target={state['target']}°C, mode={state['mode']})") - - async def thermostat_drift_loop(self): - """Background loop for thermostat temperature drift.""" - while self.running: - await asyncio.sleep(DRIFT_INTERVAL) - - for device_id in THERMOSTAT_DEVICES: - self.apply_temperature_drift(device_id) - await self.publish_state(device_id, "thermostat") - - async def handle_message(self, message): - """Handle incoming MQTT message.""" - try: - # Extract device_id from topic (vendor/{device_id}/set) - topic_parts = message.topic.value.split('/') - if len(topic_parts) != 3 or topic_parts[0] != "vendor" or topic_parts[2] != "set": - logger.warning(f"Unexpected topic format: {message.topic}") - return - - device_id = topic_parts[1] - payload = json.loads(message.payload.decode()) - - logger.info(f"[{device_id}] Received SET: {payload}") - - # Determine device type and handle accordingly - if device_id in self.light_states: - await self.handle_light_set(device_id, payload) - elif device_id in self.thermostat_states: - await self.handle_thermostat_set(device_id, payload) - else: - logger.warning(f"Unknown device: {device_id}") - - except json.JSONDecodeError as e: - logger.error(f"Invalid JSON: {e}") - except Exception as e: - logger.error(f"Error handling message: {e}") - - async def run(self): - """Main simulator loop.""" - # Generate unique client ID to avoid collisions - base_client_id = "device_simulator" - client_suffix = os.environ.get("MQTT_CLIENT_ID_SUFFIX") or uuid.uuid4().hex[:6] - unique_client_id = f"{base_client_id}-{client_suffix}" - - try: - async with Client( - hostname=BROKER_HOST, - port=BROKER_PORT, - identifier=unique_client_id - ) as client: - self.client = client - logger.info(f"✅ Connected to MQTT broker {BROKER_HOST}:{BROKER_PORT}") - - # Publish initial states - for device_id in LIGHT_DEVICES: - await self.publish_state(device_id, "light") - logger.info(f"💡 Light simulator started: {device_id}") - - for device_id in THERMOSTAT_DEVICES: - await self.publish_state(device_id, "thermostat") - logger.info(f"🌡️ Thermostat simulator started: {device_id}") - - # Subscribe to all SET topics - all_devices = LIGHT_DEVICES + THERMOSTAT_DEVICES - for device_id in all_devices: - set_topic = f"vendor/{device_id}/set" - await client.subscribe(set_topic, qos=1) - logger.info(f"👂 Subscribed to {set_topic}") - - # Start thermostat drift loop - self.drift_task = asyncio.create_task(self.thermostat_drift_loop()) - - # Listen for messages - async for message in client.messages: - await self.handle_message(message) - - # Cancel drift loop on disconnect - if self.drift_task: - self.drift_task.cancel() - - except MqttError as e: - logger.error(f"❌ MQTT Error: {e}") - except KeyboardInterrupt: - logger.info("⚠️ Interrupted by user") - finally: - self.running = False - if self.drift_task: - self.drift_task.cancel() - logger.info("👋 Simulator stopped") - - -async def main(): - """Entry point.""" - simulator = DeviceSimulator() - await simulator.run() - - -if __name__ == "__main__": - try: - asyncio.run(main()) - except KeyboardInterrupt: - print("\n👋 Simulator terminated") - sys.exit(0) diff --git a/tools/setup-ca.sh b/tools/setup-ca.sh new file mode 100755 index 0000000..d94929e --- /dev/null +++ b/tools/setup-ca.sh @@ -0,0 +1,24 @@ +#!/bin/bash +set -e + +echo "=== mTLS CA Setup ===" + +# Create CA directory +mkdir -p ca + +# Generate CA private key +echo "Generating CA private key..." +openssl genrsa -out ca/ca.key 2048 + +# Generate CA certificate +echo "Generating CA certificate..." +openssl req -new -x509 -days 3650 -key ca/ca.key -out ca/ca.crt \ + -subj "/DC=de/DC=hottis/DC=homea2/CN=Home Automation CA" + +echo "" +echo "=== CA Setup Complete ===" +echo "CA Certificate: ca/ca.crt" +echo "CA Private Key: ca/ca.key" +echo "" +echo "Deploy to Kubernetes:" +echo "kubectl create secret generic mtls-ca-cert --from-file=ca.crt=ca/ca.crt -n homea2" \ No newline at end of file diff --git a/tools/test_api_client.py b/tools/test_api_client.py deleted file mode 100644 index 87a9816..0000000 --- a/tools/test_api_client.py +++ /dev/null @@ -1,169 +0,0 @@ -""" -Acceptance tests for API Client (fetch_devices). - -Tests: -1. API running: fetch_devices() returns list with test_lampe_1, test_lampe_2 -2. API down: fetch_devices() returns empty list without crash -""" -import sys -import time -import subprocess -from apps.ui.api_client import fetch_devices - - -def test_api_running(): - """Test 1: API is running -> fetch_devices() returns devices.""" - print("Test 1: API running") - print("-" * 60) - - devices = fetch_devices("http://localhost:8001") - - if not devices: - print(" ✗ FAILED: Expected devices, got empty list") - return False - - print(f" ✓ Received {len(devices)} devices") - - # Check structure - for device in devices: - device_id = device.get("device_id") - device_type = device.get("type") - name = device.get("name") - - if not device_id: - print(f" ✗ FAILED: Device missing 'device_id': {device}") - return False - - if not device_type: - print(f" ✗ FAILED: Device missing 'type': {device}") - return False - - if not name: - print(f" ✗ FAILED: Device missing 'name': {device}") - return False - - print(f" • {device_id} (type={device_type}, name={name})") - - # Check for expected devices - device_ids = {d["device_id"] for d in devices} - expected = {"test_lampe_1", "test_lampe_2"} - - if not expected.issubset(device_ids): - missing = expected - device_ids - print(f" ✗ FAILED: Missing devices: {missing}") - return False - - print(" ✓ All expected devices present") - print() - return True - - -def test_api_down(): - """Test 2: API is down -> fetch_devices() returns empty list without crash.""" - print("Test 2: API down (invalid port)") - print("-" * 60) - - try: - devices = fetch_devices("http://localhost:9999") - - if devices != []: - print(f" ✗ FAILED: Expected empty list, got {len(devices)} devices") - return False - - print(" ✓ Returns empty list without crash") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: Exception raised: {e}") - print() - return False - - -def test_api_timeout(): - """Test 3: API timeout -> fetch_devices() returns empty list.""" - print("Test 3: API timeout") - print("-" * 60) - - # Use httpbin.org delay endpoint to simulate slow API - try: - start = time.time() - devices = fetch_devices("https://httpbin.org/delay/5") # 5s delay, but 3s timeout - elapsed = time.time() - start - - if devices != []: - print(f" ✗ FAILED: Expected empty list, got {len(devices)} devices") - return False - - if elapsed >= 4.0: - print(f" ✗ FAILED: Timeout not enforced (took {elapsed:.1f}s)") - return False - - print(f" ✓ Returns empty list after timeout ({elapsed:.1f}s)") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: Exception raised: {e}") - print() - return False - - -def main(): - """Run all acceptance tests.""" - print("=" * 60) - print("Testing API Client (fetch_devices)") - print("=" * 60) - print() - - # Check if API is running - print("Prerequisites: Checking if API is running on port 8001...") - try: - devices = fetch_devices("http://localhost:8001") - if devices: - print(f"✓ API is running ({len(devices)} devices found)") - print() - else: - print("⚠ API might not be running (no devices returned)") - print(" Start API with: poetry run uvicorn apps.api.main:app --port 8001") - print() - except Exception as e: - print(f"✗ Cannot reach API: {e}") - print(" Start API with: poetry run uvicorn apps.api.main:app --port 8001") - print() - sys.exit(1) - - results = [] - - # Test 1: API running - results.append(("API running", test_api_running())) - - # Test 2: API down - results.append(("API down", test_api_down())) - - # Test 3: Timeout - results.append(("API timeout", test_api_timeout())) - - # Summary - print("=" * 60) - passed = sum(1 for _, result in results if result) - total = len(results) - print(f"Results: {passed}/{total} tests passed") - print("=" * 60) - - for name, result in results: - status = "✓" if result else "✗" - print(f" {status} {name}") - - if passed == total: - print() - print("All tests passed!") - sys.exit(0) - else: - print() - print("Some tests failed.") - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/tools/test_config_loader.py b/tools/test_config_loader.py deleted file mode 100644 index 429c014..0000000 --- a/tools/test_config_loader.py +++ /dev/null @@ -1,253 +0,0 @@ -#!/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()) diff --git a/tools/test_dashboard.py b/tools/test_dashboard.py deleted file mode 100644 index 93c3f22..0000000 --- a/tools/test_dashboard.py +++ /dev/null @@ -1,210 +0,0 @@ -""" -Acceptance tests for Dashboard Route. - -Tests: -1. GET /dashboard loads without errors -2. Rooms are shown in layout.yaml order -3. Devices within each room are sorted by rank (ascending) -4. GET / redirects to dashboard -""" -import sys -import httpx - - -def test_dashboard_loads(): - """Test 1: Dashboard loads without errors.""" - print("Test 1: Dashboard loads") - print("-" * 60) - - try: - response = httpx.get("http://localhost:8002/dashboard", timeout=5.0) - response.raise_for_status() - - html = response.text - - # Check for essential elements - if "Home Automation" not in html: - print(" ✗ FAILED: Missing title 'Home Automation'") - return False - - if "Wohnzimmer" not in html: - print(" ✗ FAILED: Missing room 'Wohnzimmer'") - return False - - if "Schlafzimmer" not in html: - print(" ✗ FAILED: Missing room 'Schlafzimmer'") - return False - - print(" ✓ Dashboard loads successfully") - print(" ✓ Contains expected rooms") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: {e}") - print() - return False - - -def test_room_order(): - """Test 2: Rooms appear in layout.yaml order.""" - print("Test 2: Room order matches layout.yaml") - print("-" * 60) - - try: - response = httpx.get("http://localhost:8002/dashboard", timeout=5.0) - response.raise_for_status() - - html = response.text - - # Find positions of room titles - wohnzimmer_pos = html.find('class="room-title">Wohnzimmer<') - schlafzimmer_pos = html.find('class="room-title">Schlafzimmer<') - - if wohnzimmer_pos == -1: - print(" ✗ FAILED: Room 'Wohnzimmer' not found") - return False - - if schlafzimmer_pos == -1: - print(" ✗ FAILED: Room 'Schlafzimmer' not found") - return False - - # Wohnzimmer should appear before Schlafzimmer - if wohnzimmer_pos > schlafzimmer_pos: - print(" ✗ FAILED: Room order incorrect") - print(f" Wohnzimmer at position {wohnzimmer_pos}") - print(f" Schlafzimmer at position {schlafzimmer_pos}") - return False - - print(" ✓ Rooms appear in correct order:") - print(" 1. Wohnzimmer") - print(" 2. Schlafzimmer") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: {e}") - print() - return False - - -def test_device_rank_sorting(): - """Test 3: Devices are sorted by rank (ascending).""" - print("Test 3: Devices sorted by rank") - print("-" * 60) - - try: - response = httpx.get("http://localhost:8002/dashboard", timeout=5.0) - response.raise_for_status() - - html = response.text - - # In Wohnzimmer: Deckenlampe (rank=5) should come before Stehlampe (rank=10) - deckenlampe_pos = html.find('device-title">Deckenlampe<') - stehlampe_pos = html.find('device-title">Stehlampe<') - - if deckenlampe_pos == -1: - print(" ✗ FAILED: Device 'Deckenlampe' not found") - return False - - if stehlampe_pos == -1: - print(" ✗ FAILED: Device 'Stehlampe' not found") - return False - - if deckenlampe_pos > stehlampe_pos: - print(" ✗ FAILED: Devices not sorted by rank") - print(f" Deckenlampe (rank=5) at position {deckenlampe_pos}") - print(f" Stehlampe (rank=10) at position {stehlampe_pos}") - return False - - print(" ✓ Devices sorted by rank (ascending):") - print(" Wohnzimmer: Deckenlampe (rank=5) → Stehlampe (rank=10)") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: {e}") - print() - return False - - -def test_root_redirect(): - """Test 4: GET / shows dashboard.""" - print("Test 4: Root path (/) shows dashboard") - print("-" * 60) - - try: - response = httpx.get("http://localhost:8002/", timeout=5.0, follow_redirects=True) - response.raise_for_status() - - html = response.text - - # Should contain dashboard elements - if "Home Automation" not in html: - print(" ✗ FAILED: Root path doesn't show dashboard") - return False - - if "Wohnzimmer" not in html: - print(" ✗ FAILED: Root path missing rooms") - return False - - print(" ✓ Root path shows dashboard") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: {e}") - print() - return False - - -def main(): - """Run all acceptance tests.""" - print("=" * 60) - print("Testing Dashboard Route") - print("=" * 60) - print() - - # Check if UI is running - print("Prerequisites: Checking if UI is running on port 8002...") - try: - response = httpx.get("http://localhost:8002/dashboard", timeout=3.0) - print("✓ UI is running") - print() - except Exception as e: - print(f"✗ Cannot reach UI: {e}") - print(" Start UI with: poetry run uvicorn apps.ui.main:app --port 8002") - print() - sys.exit(1) - - results = [] - - # Run tests - results.append(("Dashboard loads", test_dashboard_loads())) - results.append(("Room order", test_room_order())) - results.append(("Device rank sorting", test_device_rank_sorting())) - results.append(("Root redirect", test_root_redirect())) - - # Summary - print("=" * 60) - passed = sum(1 for _, result in results if result) - total = len(results) - print(f"Results: {passed}/{total} tests passed") - print("=" * 60) - - for name, result in results: - status = "✓" if result else "✗" - print(f" {status} {name}") - - if passed == total: - print() - print("All tests passed!") - sys.exit(0) - else: - print() - print("Some tests failed.") - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/tools/test_device_simulator.sh b/tools/test_device_simulator.sh deleted file mode 100755 index 6455422..0000000 --- a/tools/test_device_simulator.sh +++ /dev/null @@ -1,154 +0,0 @@ -#!/bin/bash -# Test script for device_simulator.py - -set -e # Exit on error - -echo "=== Device Simulator Test Suite ===" -echo "" - -# 1. Stop all running services -echo "1. Stoppe alle laufenden Services..." -pkill -f "device_simulator" 2>/dev/null || true -pkill -f "uvicorn apps" 2>/dev/null || true -pkill -f "apps.abstraction" 2>/dev/null || true -sleep 2 -echo " ✓ Services gestoppt" -echo "" - -# 2. Start services -echo "2. Starte Services..." -poetry run python -m apps.abstraction.main > /tmp/abstraction.log 2>&1 & -ABSTRACTION_PID=$! -echo " Abstraction Layer gestartet (PID: $ABSTRACTION_PID)" -sleep 2 - -poetry run uvicorn apps.api.main:app --host 0.0.0.0 --port 8001 > /tmp/api.log 2>&1 & -API_PID=$! -echo " API Server gestartet (PID: $API_PID)" -sleep 2 - -poetry run python tools/device_simulator.py > /tmp/simulator.log 2>&1 & -SIM_PID=$! -echo " Device Simulator gestartet (PID: $SIM_PID)" -sleep 2 - -echo " ✓ Alle Services laufen" -echo "" - -# 3. Test API reachability -echo "3. Teste API Erreichbarkeit..." -if timeout 3 curl -s http://localhost:8001/devices > /dev/null; then - echo " ✓ API antwortet" -else - echo " ✗ API antwortet nicht!" - echo " API Log:" - tail -10 /tmp/api.log - exit 1 -fi -echo "" - -# 4. Test Light Operations -echo "4. Teste Lampen-Operationen..." - -# 4.1 Power On -echo " 4.1 Lampe einschalten (test_lampe_1)..." -RESPONSE=$(timeout 3 curl -s -X POST http://localhost:8001/devices/test_lampe_1/set \ - -H "Content-Type: application/json" \ - -d '{"type":"light","payload":{"power":"on"}}') -echo " Response: $RESPONSE" -sleep 1 - -# 4.2 Check state via MQTT -echo " 4.2 Prüfe State via MQTT..." -STATE=$(timeout 2 mosquitto_sub -h 172.16.2.16 -t 'vendor/test_lampe_1/state' -C 1) -echo " State: $STATE" -if echo "$STATE" | grep -q '"power": "on"'; then - echo " ✓ Power ist ON" -else - echo " ✗ Power nicht ON!" -fi - -# 4.3 Brightness -echo " 4.3 Helligkeit setzen (75%)..." -RESPONSE=$(timeout 3 curl -s -X POST http://localhost:8001/devices/test_lampe_1/set \ - -H "Content-Type: application/json" \ - -d '{"type":"light","payload":{"brightness":75}}') -echo " Response: $RESPONSE" -sleep 1 - -STATE=$(timeout 2 mosquitto_sub -h 172.16.2.16 -t 'vendor/test_lampe_1/state' -C 1) -echo " State: $STATE" -if echo "$STATE" | grep -q '"brightness": 75'; then - echo " ✓ Brightness ist 75" -else - echo " ✗ Brightness nicht 75!" -fi -echo "" - -# 5. Test Thermostat Operations -echo "5. Teste Thermostat-Operationen..." - -# 5.1 Set mode and target -echo " 5.1 Setze Mode HEAT und Target 22.5°C..." -RESPONSE=$(timeout 3 curl -s -X POST http://localhost:8001/devices/test_thermo_1/set \ - -H "Content-Type: application/json" \ - -d '{"type":"thermostat","payload":{"mode":"heat","target":22.5}}') -echo " Response: $RESPONSE" -sleep 1 - -STATE=$(timeout 2 mosquitto_sub -h 172.16.2.16 -t 'vendor/test_thermo_1/state' -C 1) -echo " State: $STATE" -if echo "$STATE" | grep -q '"mode": "heat"' && echo "$STATE" | grep -q '"target": 22.5'; then - echo " ✓ Mode ist HEAT, Target ist 22.5" -else - echo " ✗ Mode oder Target nicht korrekt!" -fi - -# 5.2 Wait for temperature drift -echo " 5.2 Warte 6 Sekunden auf Temperature Drift..." -sleep 6 - -STATE=$(timeout 2 mosquitto_sub -h 172.16.2.16 -t 'vendor/test_thermo_1/state' -C 1) -echo " State: $STATE" -CURRENT=$(echo "$STATE" | grep -o '"current": [0-9.]*' | grep -o '[0-9.]*$') -echo " Current Temperature: ${CURRENT}°C" -if [ -n "$CURRENT" ]; then - echo " ✓ Temperature drift funktioniert" -else - echo " ✗ Temperature drift nicht sichtbar!" -fi - -# 5.3 Set mode OFF -echo " 5.3 Setze Mode OFF..." -RESPONSE=$(timeout 3 curl -s -X POST http://localhost:8001/devices/test_thermo_1/set \ - -H "Content-Type: application/json" \ - -d '{"type":"thermostat","payload":{"mode":"off","target":22.5}}') -echo " Response: $RESPONSE" -sleep 1 - -STATE=$(timeout 2 mosquitto_sub -h 172.16.2.16 -t 'vendor/test_thermo_1/state' -C 1) -if echo "$STATE" | grep -q '"mode": "off"'; then - echo " ✓ Mode ist OFF" -else - echo " ✗ Mode nicht OFF!" -fi -echo "" - -# 6. Check simulator log -echo "6. Simulator Log (letzte 20 Zeilen)..." -tail -20 /tmp/simulator.log -echo "" - -# 7. Summary -echo "=== Test Summary ===" -echo "✓ Alle Tests abgeschlossen" -echo "" -echo "Laufende Prozesse:" -echo " Abstraction: PID $ABSTRACTION_PID" -echo " API: PID $API_PID" -echo " Simulator: PID $SIM_PID" -echo "" -echo "Logs verfügbar in:" -echo " /tmp/abstraction.log" -echo " /tmp/api.log" -echo " /tmp/simulator.log" diff --git a/tools/test_layout_loader.py b/tools/test_layout_loader.py deleted file mode 100644 index baa1832..0000000 --- a/tools/test_layout_loader.py +++ /dev/null @@ -1,288 +0,0 @@ -#!/usr/bin/env python3 -"""Test layout loader and models.""" - -import sys -import tempfile -from pathlib import Path - -# Add parent directory to path -sys.path.insert(0, str(Path(__file__).parent.parent)) - -from packages.home_capabilities.layout import DeviceTile, Room, UiLayout, load_layout - - -def test_device_tile_creation(): - """Test DeviceTile model creation.""" - print("Test 1: DeviceTile Creation") - - tile = DeviceTile( - device_id="test_lamp", - title="Test Lamp", - icon="💡", - rank=10 - ) - - assert tile.device_id == "test_lamp" - assert tile.title == "Test Lamp" - assert tile.icon == "💡" - assert tile.rank == 10 - - print(" ✓ DeviceTile created successfully") - return True - - -def test_room_creation(): - """Test Room model creation.""" - print("\nTest 2: Room Creation") - - tiles = [ - DeviceTile(device_id="lamp1", title="Lamp 1", icon="💡", rank=1), - DeviceTile(device_id="lamp2", title="Lamp 2", icon="💡", rank=2) - ] - - room = Room(name="Living Room", devices=tiles) - - assert room.name == "Living Room" - assert len(room.devices) == 2 - assert room.devices[0].rank == 1 - - print(" ✓ Room with 2 devices created") - return True - - -def test_ui_layout_creation(): - """Test UiLayout model creation.""" - print("\nTest 3: UiLayout Creation") - - rooms = [ - Room( - name="Room 1", - devices=[DeviceTile(device_id="d1", title="D1", icon="💡", rank=1)] - ), - Room( - name="Room 2", - devices=[DeviceTile(device_id="d2", title="D2", icon="💡", rank=1)] - ) - ] - - layout = UiLayout(rooms=rooms) - - assert len(layout.rooms) == 2 - assert layout.total_devices() == 2 - - print(" ✓ UiLayout with 2 rooms created") - print(f" ✓ Total devices: {layout.total_devices()}") - return True - - -def test_layout_validation_empty_rooms(): - """Test that layout validation rejects empty rooms list.""" - print("\nTest 4: Validation - Empty Rooms") - - try: - UiLayout(rooms=[]) - print(" ✗ Should have raised ValueError") - return False - except ValueError as e: - if "at least one room" in str(e): - print(f" ✓ Correct validation error: {e}") - return True - else: - print(f" ✗ Wrong error message: {e}") - return False - - -def test_load_layout_from_file(): - """Test loading layout from YAML file.""" - print("\nTest 5: Load Layout from YAML") - - yaml_content = """ -rooms: - - name: Wohnzimmer - devices: - - device_id: test_lampe_1 - title: Stehlampe - icon: "💡" - rank: 10 - - device_id: test_lampe_2 - title: Tischlampe - icon: "💡" - rank: 20 - - - name: Schlafzimmer - devices: - - device_id: test_lampe_3 - title: Nachttischlampe - icon: "🛏️" - rank: 5 -""" - - with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f: - f.write(yaml_content) - temp_path = Path(f.name) - - try: - layout = load_layout(str(temp_path)) - - assert len(layout.rooms) == 2 - assert layout.rooms[0].name == "Wohnzimmer" - assert len(layout.rooms[0].devices) == 2 - assert layout.total_devices() == 3 - - # Check device sorting by rank - wohnzimmer_devices = layout.rooms[0].devices - assert wohnzimmer_devices[0].rank == 10 - assert wohnzimmer_devices[1].rank == 20 - - print(" ✓ Layout loaded from YAML") - print(f" ✓ Rooms: {len(layout.rooms)}") - print(f" ✓ Total devices: {layout.total_devices()}") - print(f" ✓ Room order preserved: {[r.name for r in layout.rooms]}") - - return True - - finally: - temp_path.unlink() - - -def test_load_layout_missing_file(): - """Test error handling for missing file.""" - print("\nTest 6: Missing File Error") - - try: - load_layout("/nonexistent/path/layout.yaml") - print(" ✗ Should have raised FileNotFoundError") - return False - except FileNotFoundError as e: - if "not found" in str(e): - print(f" ✓ Correct error for missing file") - return True - else: - print(f" ✗ Wrong error message: {e}") - return False - - -def test_load_layout_invalid_yaml(): - """Test error handling for invalid YAML.""" - print("\nTest 7: Invalid YAML Error") - - yaml_content = """ -rooms: - - name: Room1 - devices: [invalid yaml structure -""" - - with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f: - f.write(yaml_content) - temp_path = Path(f.name) - - try: - load_layout(str(temp_path)) - print(" ✗ Should have raised YAMLError") - temp_path.unlink() - return False - except Exception as e: - temp_path.unlink() - if "YAML" in str(type(e).__name__) or "parse" in str(e).lower(): - print(f" ✓ Correct YAML parsing error") - return True - else: - print(f" ✗ Unexpected error: {type(e).__name__}: {e}") - return False - - -def test_load_layout_missing_required_fields(): - """Test validation for missing required fields.""" - print("\nTest 8: Missing Required Fields") - - yaml_content = """ -rooms: - - name: Room1 - devices: - - device_id: lamp1 - title: Lamp - # Missing: icon, rank -""" - - with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f: - f.write(yaml_content) - temp_path = Path(f.name) - - try: - load_layout(str(temp_path)) - print(" ✗ Should have raised ValueError") - temp_path.unlink() - return False - except ValueError as e: - temp_path.unlink() - if "icon" in str(e) or "rank" in str(e) or "required" in str(e).lower(): - print(f" ✓ Validation error for missing fields") - return True - else: - print(f" ✗ Wrong error: {e}") - return False - - -def test_load_default_layout(): - """Test loading default layout from workspace.""" - print("\nTest 9: Load Default Layout (config/layout.yaml)") - - try: - layout = load_layout() - - print(f" ✓ Default layout loaded") - print(f" ✓ Rooms: {len(layout.rooms)} ({', '.join(r.name for r in layout.rooms)})") - print(f" ✓ Total devices: {layout.total_devices()}") - - for room in layout.rooms: - print(f" - {room.name}: {len(room.devices)} devices") - for device in room.devices: - print(f" • {device.title} (id={device.device_id}, rank={device.rank})") - - return True - - except FileNotFoundError: - print(" ⚠ Default layout.yaml not found (expected at config/layout.yaml)") - return True # Not a failure if file doesn't exist yet - except Exception as e: - print(f" ✗ Error loading default layout: {e}") - return False - - -def main(): - """Run all tests.""" - print("=" * 60) - print("Testing Layout Loader and Models") - print("=" * 60) - - tests = [ - test_device_tile_creation, - test_room_creation, - test_ui_layout_creation, - test_layout_validation_empty_rooms, - test_load_layout_from_file, - test_load_layout_missing_file, - test_load_layout_invalid_yaml, - test_load_layout_missing_required_fields, - test_load_default_layout, - ] - - results = [] - for test in tests: - try: - results.append(test()) - except Exception as e: - print(f" ✗ Unexpected exception: {e}") - results.append(False) - - print("\n" + "=" * 60) - passed = sum(results) - total = len(results) - print(f"Results: {passed}/{total} tests passed") - print("=" * 60) - - return 0 if all(results) else 1 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/tools/test_responsive_dashboard.py b/tools/test_responsive_dashboard.py deleted file mode 100644 index 919d2e6..0000000 --- a/tools/test_responsive_dashboard.py +++ /dev/null @@ -1,296 +0,0 @@ -""" -Acceptance tests for Responsive Dashboard. - -Tests: -1. Desktop: 4 columns grid -2. Tablet: 2 columns grid -3. Mobile: 1 column grid -4. POST requests work (via API check) -5. No horizontal scrolling on any viewport -""" -import sys -import httpx -from bs4 import BeautifulSoup - - -def test_dashboard_html_structure(): - """Test 1: Dashboard has correct HTML structure.""" - print("Test 1: Dashboard HTML Structure") - print("-" * 60) - - try: - response = httpx.get("http://localhost:8002/dashboard", timeout=5.0) - response.raise_for_status() - - html = response.text - soup = BeautifulSoup(html, 'html.parser') - - # Check for grid container - grid = soup.find('div', class_='devices-grid') - if not grid: - print(" ✗ FAILED: Missing .devices-grid container") - return False - - # Check for device tiles - tiles = soup.find_all('div', class_='device-tile') - if len(tiles) < 2: - print(f" ✗ FAILED: Expected at least 2 device tiles, found {len(tiles)}") - return False - - # Check for state spans - state_spans = soup.find_all('span', id=lambda x: x and x.startswith('state-')) - if len(state_spans) < 2: - print(f" ✗ FAILED: Expected state spans, found {len(state_spans)}") - return False - - # Check for ON/OFF buttons - btn_on = soup.find_all('button', class_='btn-on') - btn_off = soup.find_all('button', class_='btn-off') - - if not btn_on or not btn_off: - print(" ✗ FAILED: Missing ON/OFF buttons") - return False - - print(f" ✓ Found {len(tiles)} device tiles") - print(f" ✓ Found {len(state_spans)} state indicators") - print(f" ✓ Found {len(btn_on)} ON buttons and {len(btn_off)} OFF buttons") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: {e}") - print() - return False - - -def test_responsive_css(): - """Test 2: CSS has responsive grid rules.""" - print("Test 2: Responsive CSS") - print("-" * 60) - - try: - response = httpx.get("http://localhost:8002/static/style.css", timeout=5.0) - response.raise_for_status() - - css = response.text - - # Check for desktop 4 columns - if 'grid-template-columns: repeat(4, 1fr)' not in css: - print(" ✗ FAILED: Missing desktop grid (4 columns)") - return False - - # Check for tablet media query (2 columns) - if 'max-width: 1024px' not in css or 'repeat(2, 1fr)' not in css: - print(" ✗ FAILED: Missing tablet media query (2 columns)") - return False - - # Check for mobile media query (1 column) - if 'max-width: 640px' not in css: - print(" ✗ FAILED: Missing mobile media query") - return False - - print(" ✓ Desktop: 4 columns (grid-template-columns: repeat(4, 1fr))") - print(" ✓ Tablet: 2 columns (@media max-width: 1024px)") - print(" ✓ Mobile: 1 column (@media max-width: 640px)") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: {e}") - print() - return False - - -def test_javascript_functions(): - """Test 3: JavaScript POST function exists.""" - print("Test 3: JavaScript POST Function") - print("-" * 60) - - try: - response = httpx.get("http://localhost:8002/dashboard", timeout=5.0) - response.raise_for_status() - - html = response.text - - # Check for setDeviceState function - if 'function setDeviceState' not in html and 'async function setDeviceState' not in html: - print(" ✗ FAILED: Missing setDeviceState function") - return False - - # Check for fetch POST call - if 'fetch(' not in html or 'method: \'POST\'' not in html: - print(" ✗ FAILED: Missing fetch POST call") - return False - - # Check for correct API endpoint pattern - if '/devices/${deviceId}/set' not in html and '/devices/' not in html: - print(" ✗ FAILED: Missing correct API endpoint") - return False - - # Check for JSON payload - if 'type: \'light\'' not in html and '"type":"light"' not in html: - print(" ✗ FAILED: Missing correct JSON payload") - return False - - print(" ✓ setDeviceState function defined") - print(" ✓ Uses fetch with POST method") - print(" ✓ Correct endpoint: /devices/{deviceId}/set") - print(" ✓ Correct payload: {type:'light', payload:{power:...}}") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: {e}") - print() - return False - - -def test_device_controls(): - """Test 4: Devices have correct controls.""" - print("Test 4: Device Controls") - print("-" * 60) - - try: - response = httpx.get("http://localhost:8002/dashboard", timeout=5.0) - response.raise_for_status() - - html = response.text - soup = BeautifulSoup(html, 'html.parser') - - # Find device tiles - tiles = soup.find_all('div', class_='device-tile') - - for tile in tiles: - device_id = tile.get('data-device-id') - - # Check for device header - header = tile.find('div', class_='device-header') - if not header: - print(f" ✗ FAILED: Device {device_id} missing header") - return False - - # Check for icon and title - icon = tile.find('div', class_='device-icon') - title = tile.find('h3', class_='device-title') - device_id_elem = tile.find('p', class_='device-id') - - if not icon or not title or not device_id_elem: - print(f" ✗ FAILED: Device {device_id} missing icon/title/id") - return False - - # Check for state indicator - state_span = tile.find('span', id=f'state-{device_id}') - if not state_span: - print(f" ✗ FAILED: Device {device_id} missing state indicator") - return False - - print(f" ✓ All {len(tiles)} devices have:") - print(" • Icon, title, and device_id") - print(" • State indicator (span#state-{{device_id}})") - print(" • ON/OFF buttons (for lights with power feature)") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: {e}") - print() - return False - - -def test_rank_sorting(): - """Test 5: Devices sorted by rank.""" - print("Test 5: Device Rank Sorting") - print("-" * 60) - - try: - response = httpx.get("http://localhost:8002/dashboard", timeout=5.0) - response.raise_for_status() - - html = response.text - - # In Wohnzimmer: Deckenlampe (rank=5) should come before Stehlampe (rank=10) - deckenlampe_pos = html.find('device-title">Deckenlampe<') - stehlampe_pos = html.find('device-title">Stehlampe<') - - if deckenlampe_pos == -1 or stehlampe_pos == -1: - print(" ℹ INFO: Test devices not found (expected for test)") - print() - return True - - if deckenlampe_pos > stehlampe_pos: - print(" ✗ FAILED: Devices not sorted by rank") - return False - - print(" ✓ Devices sorted by rank (Deckenlampe before Stehlampe)") - print() - return True - - except Exception as e: - print(f" ✗ FAILED: {e}") - print() - return False - - -def main(): - """Run all acceptance tests.""" - print("=" * 60) - print("Testing Responsive Dashboard") - print("=" * 60) - print() - - # Check if UI is running - print("Prerequisites: Checking if UI is running on port 8002...") - try: - response = httpx.get("http://localhost:8002/dashboard", timeout=3.0) - print("✓ UI is running") - print() - except Exception as e: - print(f"✗ Cannot reach UI: {e}") - print(" Start UI with: poetry run uvicorn apps.ui.main:app --port 8002") - print() - sys.exit(1) - - results = [] - - # Run tests - results.append(("HTML Structure", test_dashboard_html_structure())) - results.append(("Responsive CSS", test_responsive_css())) - results.append(("JavaScript POST", test_javascript_functions())) - results.append(("Device Controls", test_device_controls())) - results.append(("Rank Sorting", test_rank_sorting())) - - # Summary - print("=" * 60) - passed = sum(1 for _, result in results if result) - total = len(results) - print(f"Results: {passed}/{total} tests passed") - print("=" * 60) - - for name, result in results: - status = "✓" if result else "✗" - print(f" {status} {name}") - - print() - print("Manual Tests Required:") - print(" 1. Open http://localhost:8002/dashboard in browser") - print(" 2. Resize browser window to test responsive breakpoints:") - print(" - Desktop (>1024px): Should show 4 columns") - print(" - Tablet (640-1024px): Should show 2 columns") - print(" - Mobile (<640px): Should show 1 column") - print(" 3. Click ON/OFF buttons and check Network tab in DevTools") - print(" - Should see POST to http://localhost:8001/devices/.../set") - print(" - No JavaScript errors in Console") - print(" 4. Verify no horizontal scrolling at any viewport size") - - if passed == total: - print() - print("All automated tests passed!") - sys.exit(0) - else: - print() - print("Some tests failed.") - sys.exit(1) - - -if __name__ == "__main__": - main()