""" EU-Utility Integration Test - Home Assistant ============================================= Tests Home Assistant integration via: - REST API (webhook triggers) - MQTT publishing - WebSocket API - State updates Author: Integration Tester Version: 1.0.0 """ import json import time import uuid from datetime import datetime from typing import Dict, Any, Optional, List from dataclasses import dataclass, asdict from plugins.base_plugin import BasePlugin from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QTextEdit, QPushButton, QComboBox, QSpinBox, QCheckBox, QGroupBox, QTabWidget, QTableWidget, QTableWidgetItem, QHeaderView, QProgressBar, QMessageBox ) from PyQt6.QtCore import Qt, QThread, pyqtSignal @dataclass class HATestCase: """Home Assistant test case.""" name: str service: str # 'rest', 'mqtt', 'websocket' payload: Dict[str, Any] expected_result: str description: str class HomeAssistantTester(BasePlugin): """Plugin for testing Home Assistant integration.""" name = "Home Assistant Tester" version = "1.0.0" author = "Integration Tester" description = "Test Home Assistant integration via REST, MQTT, and WebSocket" # Test cases for different HA integration methods TEST_CASES = [ HATestCase( name="REST Webhook - Loot", service="rest", payload={ "event_type": "eu_utility_loot", "event_data": { "mob": "Atrox Young", "total_value": 50.25, "items": [{"name": "Animal Oil", "value": 0.05}], "timestamp": datetime.now().isoformat() } }, expected_result="200/202", description="Send loot event via REST webhook" ), HATestCase( name="REST Webhook - Skill", service="rest", payload={ "event_type": "eu_utility_skill", "event_data": { "skill": "Rifle", "gain": 0.25, "new_total": 4500.75 } }, expected_result="200/202", description="Send skill gain event via REST webhook" ), HATestCase( name="REST Sensor Update", service="rest", payload={ "state": "hunting", "attributes": { "current_mob": "Atrox", "session_ped": 150.50, "dpp": 2.85, "weapon": "ArMatrix LP-35" } }, expected_result="200", description="Update sensor state via REST API" ), HATestCase( name="MQTT - State Publish", service="mqtt", payload={ "topic": "eu_utility/player/state", "payload": json.dumps({ "status": "active", "activity": "hunting", "location": "Calypso" }) }, expected_result="published", description="Publish state to MQTT topic" ), HATestCase( name="MQTT - Loot Topic", service="mqtt", payload={ "topic": "eu_utility/events/loot", "payload": json.dumps({ "event": "loot_drop", "value": 45.50, "mob": "Atrox" }) }, expected_result="published", description="Publish loot event to MQTT" ), HATestCase( name="WebSocket - Subscribe Events", service="websocket", payload={ "type": "subscribe_events", "event_type": "eu_utility_*" }, expected_result="subscribed", description="Subscribe to EU-Utility events via WebSocket" ), ] def initialize(self): """Initialize the tester.""" self.log_info("Home Assistant Tester initialized") self._ha_url = "" self._ha_token = "" self._mqtt_broker = "" self._mqtt_port = 1883 self._test_results: List[Dict] = [] self._mqtt_client = None self._ws_client = None def get_ui(self) -> QWidget: """Create the plugin UI.""" widget = QWidget() layout = QVBoxLayout(widget) # Title title = QLabel("Home Assistant Integration Tester") title.setStyleSheet("font-size: 16px; font-weight: bold;") layout.addWidget(title) # Tabs tabs = QTabWidget() # Configuration tab tabs.addTab(self._create_config_tab(), "Configuration") # REST API tab tabs.addTab(self._create_rest_tab(), "REST API") # MQTT tab tabs.addTab(self._create_mqtt_tab(), "MQTT") # WebSocket tab tabs.addTab(self._create_websocket_tab(), "WebSocket") # Test Cases tab tabs.addTab(self._create_tests_tab(), "Test Cases") # Results tab tabs.addTab(self._create_results_tab(), "Results") layout.addWidget(tabs) return widget def _create_config_tab(self) -> QWidget: """Create the configuration tab.""" widget = QWidget() layout = QVBoxLayout(widget) # REST API Configuration rest_group = QGroupBox("Home Assistant REST API") rest_layout = QVBoxLayout(rest_group) # URL url_row = QHBoxLayout() url_row.addWidget(QLabel("HA URL:")) self.ha_url_input = QLineEdit() self.ha_url_input.setPlaceholderText("http://homeassistant.local:8123") self.ha_url_input.textChanged.connect(self._on_config_changed) url_row.addWidget(self.ha_url_input) rest_layout.addLayout(url_row) # Token token_row = QHBoxLayout() token_row.addWidget(QLabel("Token:")) self.ha_token_input = QLineEdit() self.ha_token_input.setPlaceholderText("Long-Lived Access Token") self.ha_token_input.setEchoMode(QLineEdit.EchoMode.Password) self.ha_token_input.textChanged.connect(self._on_config_changed) token_row.addWidget(self.ha_token_input) rest_layout.addLayout(token_row) # Test REST connection test_rest_btn = QPushButton("Test REST Connection") test_rest_btn.clicked.connect(self._test_rest_connection) rest_layout.addWidget(test_rest_btn) layout.addWidget(rest_group) # MQTT Configuration mqtt_group = QGroupBox("MQTT Broker (Optional)") mqtt_layout = QVBoxLayout(mqtt_group) mqtt_row = QHBoxLayout() mqtt_row.addWidget(QLabel("Broker:")) self.mqtt_broker_input = QLineEdit() self.mqtt_broker_input.setPlaceholderText("mqtt.homeassistant.local") mqtt_row.addWidget(self.mqtt_broker_input) mqtt_row.addWidget(QLabel("Port:")) self.mqtt_port_input = QSpinBox() self.mqtt_port_input.setRange(1, 65535) self.mqtt_port_input.setValue(1883) mqtt_row.addWidget(self.mqtt_port_input) mqtt_layout.addLayout(mqtt_row) test_mqtt_btn = QPushButton("Test MQTT Connection") test_mqtt_btn.clicked.connect(self._test_mqtt_connection) mqtt_layout.addWidget(test_mqtt_btn) layout.addWidget(mqtt_group) # Instructions instructions = QTextEdit() instructions.setReadOnly(True) instructions.setHtml("""

Setup Instructions

REST API:

  1. In Home Assistant, go to your Profile → Long-Lived Access Tokens
  2. Create a new token and copy it
  3. Paste the token above

MQTT (optional):

  1. Install MQTT integration in HA
  2. Configure your MQTT broker
  3. Enter broker details above
""") layout.addWidget(instructions) layout.addStretch() return widget def _create_rest_tab(self) -> QWidget: """Create the REST API testing tab.""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("REST API Test Panel")) # Endpoint selection endpoint_layout = QHBoxLayout() endpoint_layout.addWidget(QLabel("Endpoint:")) self.rest_endpoint = QComboBox() self.rest_endpoint.addItems([ "/api/webhook/eu_utility", "/api/states/sensor.eu_utility_status", "/api/events/eu_utility_loot", "/api/services/input_text/set_value", "/api/config" ]) self.rest_endpoint.setEditable(True) endpoint_layout.addWidget(self.rest_endpoint) layout.addLayout(endpoint_layout) # Method method_layout = QHBoxLayout() method_layout.addWidget(QLabel("Method:")) self.rest_method = QComboBox() self.rest_method.addItems(["POST", "GET", "PUT", "PATCH"]) method_layout.addWidget(self.rest_method) layout.addLayout(method_layout) # Payload layout.addWidget(QLabel("Payload (JSON):")) self.rest_payload = QTextEdit() self.rest_payload.setPlaceholderText('{"state": "active", "attributes": {}}') self.rest_payload.setMaximumHeight(150) layout.addWidget(self.rest_payload) # Action buttons btn_layout = QHBoxLayout() send_btn = QPushButton("Send Request") send_btn.clicked.connect(self._send_rest_request) btn_layout.addWidget(send_btn) preset_btn = QPushButton("Load Preset") preset_btn.clicked.connect(self._load_rest_preset) btn_layout.addWidget(preset_btn) layout.addLayout(btn_layout) # Response layout.addWidget(QLabel("Response:")) self.rest_response = QTextEdit() self.rest_response.setReadOnly(True) layout.addWidget(self.rest_response) return widget def _create_mqtt_tab(self) -> QWidget: """Create the MQTT testing tab.""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("MQTT Test Panel")) # Topic topic_layout = QHBoxLayout() topic_layout.addWidget(QLabel("Topic:")) self.mqtt_topic = QLineEdit() self.mqtt_topic.setText("eu_utility/test") topic_layout.addWidget(self.mqtt_topic) layout.addLayout(topic_layout) # QoS qos_layout = QHBoxLayout() qos_layout.addWidget(QLabel("QoS:")) self.mqtt_qos = QComboBox() self.mqtt_qos.addItems(["0 (At most once)", "1 (At least once)", "2 (Exactly once)"]) qos_layout.addWidget(self.mqtt_qos) layout.addLayout(qos_layout) # Retain self.mqtt_retain = QCheckBox("Retain message") layout.addWidget(self.mqtt_retain) # Payload layout.addWidget(QLabel("Payload (JSON):")) self.mqtt_payload = QTextEdit() self.mqtt_payload.setPlaceholderText('{"status": "test", "value": 123}') self.mqtt_payload.setMaximumHeight(100) layout.addWidget(self.mqtt_payload) # Action buttons btn_layout = QHBoxLayout() publish_btn = QPushButton("Publish") publish_btn.clicked.connect(self._publish_mqtt) btn_layout.addWidget(publish_btn) subscribe_btn = QPushButton("Subscribe") subscribe_btn.clicked.connect(self._subscribe_mqtt) btn_layout.addWidget(subscribe_btn) layout.addLayout(btn_layout) # Messages layout.addWidget(QLabel("Messages:")) self.mqtt_messages = QTextEdit() self.mqtt_messages.setReadOnly(True) layout.addWidget(self.mqtt_messages) return widget def _create_websocket_tab(self) -> QWidget: """Create the WebSocket testing tab.""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("WebSocket API Test Panel")) # Connection status self.ws_status = QLabel("Status: Disconnected") layout.addWidget(self.ws_status) # Action buttons btn_layout = QHBoxLayout() connect_btn = QPushButton("Connect") connect_btn.clicked.connect(self._connect_websocket) btn_layout.addWidget(connect_btn) disconnect_btn = QPushButton("Disconnect") disconnect_btn.clicked.connect(self._disconnect_websocket) btn_layout.addWidget(disconnect_btn) subscribe_btn = QPushButton("Subscribe to Events") subscribe_btn.clicked.connect(self._ws_subscribe) btn_layout.addWidget(subscribe_btn) layout.addLayout(btn_layout) # Message type msg_layout = QHBoxLayout() msg_layout.addWidget(QLabel("Message Type:")) self.ws_msg_type = QComboBox() self.ws_msg_type.addItems([ "subscribe_events", "unsubscribe_events", "call_service", "get_states", "get_config" ]) msg_layout.addWidget(self.ws_msg_type) layout.addLayout(msg_layout) # Message payload layout.addWidget(QLabel("Message (JSON):")) self.ws_message = QTextEdit() self.ws_message.setPlaceholderText('{"type": "subscribe_events", "event_type": "state_changed"}') self.ws_message.setMaximumHeight(100) layout.addWidget(self.ws_message) send_btn = QPushButton("Send Message") send_btn.clicked.connect(self._send_ws_message) layout.addWidget(send_btn) # Messages log layout.addWidget(QLabel("WebSocket Log:")) self.ws_log = QTextEdit() self.ws_log.setReadOnly(True) layout.addWidget(self.ws_log) return widget def _create_tests_tab(self) -> QWidget: """Create the test cases tab.""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("Automated Test Cases:")) # Test table self.test_table = QTableWidget() self.test_table.setColumnCount(4) self.test_table.setHorizontalHeaderLabels(["Test", "Service", "Description", "Status"]) self.test_table.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeMode.Stretch) self.test_table.setRowCount(len(self.TEST_CASES)) for i, test in enumerate(self.TEST_CASES): self.test_table.setItem(i, 0, QTableWidgetItem(test.name)) self.test_table.setItem(i, 1, QTableWidgetItem(test.service.upper())) self.test_table.setItem(i, 2, QTableWidgetItem(test.description)) self.test_table.setItem(i, 3, QTableWidgetItem("⏳ Pending")) layout.addWidget(self.test_table) # Buttons btn_layout = QHBoxLayout() run_all_btn = QPushButton("Run All Tests") run_all_btn.clicked.connect(self._run_all_tests) btn_layout.addWidget(run_all_btn) run_sel_btn = QPushButton("Run Selected") run_sel_btn.clicked.connect(self._run_selected_test) btn_layout.addWidget(run_sel_btn) layout.addLayout(btn_layout) self.ha_progress = QProgressBar() layout.addWidget(self.ha_progress) return widget def _create_results_tab(self) -> QWidget: """Create the results tab.""" widget = QWidget() layout = QVBoxLayout(widget) self.ha_results_summary = QLabel("No tests run yet") layout.addWidget(self.ha_results_summary) layout.addWidget(QLabel("Detailed Results:")) self.ha_results_text = QTextEdit() self.ha_results_text.setReadOnly(True) layout.addWidget(self.ha_results_text) export_btn = QPushButton("Export Results") export_btn.clicked.connect(self._export_results) layout.addWidget(export_btn) return widget def _on_config_changed(self): """Handle configuration changes.""" self._ha_url = self.ha_url_input.text().strip() self._ha_token = self.ha_token_input.text().strip() self._mqtt_broker = self.mqtt_broker_input.text().strip() self._mqtt_port = self.mqtt_port_input.value() def _test_rest_connection(self): """Test REST API connection.""" if not self._ha_url: self.notify_error("Missing URL", "Please enter Home Assistant URL") return try: import requests headers = {"Authorization": f"Bearer {self._ha_token}"} if self._ha_token else {} response = requests.get( f"{self._ha_url}/api/", headers=headers, timeout=10 ) if response.status_code == 200: data = response.json() message = f"Connected! HA Version: {data.get('version', 'unknown')}" self.notify_success("Connection Successful", message) else: self.notify_error("Connection Failed", f"Status: {response.status_code}") except Exception as e: self.notify_error("Connection Error", str(e)) def _test_mqtt_connection(self): """Test MQTT connection.""" if not self._mqtt_broker: self.notify_error("Missing Broker", "Please enter MQTT broker address") return try: import paho.mqtt.client as mqtt client = mqtt.Client() result = client.connect(self._mqtt_broker, self._mqtt_port, 5) if result == 0: client.disconnect() self.notify_success("MQTT Connected", f"Connected to {self._mqtt_broker}") else: self.notify_error("MQTT Failed", f"Connection result code: {result}") except ImportError: self.notify_error("Missing Library", "Install paho-mqtt: pip install paho-mqtt") except Exception as e: self.notify_error("MQTT Error", str(e)) def _send_rest_request(self): """Send a REST API request.""" if not self._ha_url: self.notify_error("Missing URL", "Please configure Home Assistant URL") return try: import requests endpoint = self.rest_endpoint.currentText() url = f"{self._ha_url}{endpoint}" method = self.rest_method.currentText() headers = { "Authorization": f"Bearer {self._ha_token}", "Content-Type": "application/json" } # Parse payload payload_text = self.rest_payload.toPlainText() payload = json.loads(payload_text) if payload_text else None response = requests.request( method=method, url=url, headers=headers, json=payload, timeout=10 ) self.rest_response.setText( f"Status: {response.status_code}\n" f"Headers: {dict(response.headers)}\n\n" f"Body: {response.text[:2000]}" ) except Exception as e: self.rest_response.setText(f"Error: {str(e)}") def _load_rest_preset(self): """Load a REST payload preset.""" presets = { "/api/webhook/eu_utility": { "event_type": "loot", "data": {"value": 50.25, "mob": "Atrox"} }, "/api/states/sensor.eu_utility_status": { "state": "hunting", "attributes": {"dpp": 2.85, "weapon": "ArMatrix LP-35"} }, "/api/services/input_text/set_value": { "entity_id": "input_text.eu_utility_status", "value": "Active hunting session" } } endpoint = self.rest_endpoint.currentText() if endpoint in presets: self.rest_payload.setText(json.dumps(presets[endpoint], indent=2)) def _publish_mqtt(self): """Publish an MQTT message.""" if not self._mqtt_broker: self.notify_error("Missing Broker", "Please configure MQTT broker") return try: import paho.mqtt.client as mqtt client = mqtt.Client() client.connect(self._mqtt_broker, self._mqtt_port, 5) topic = self.mqtt_topic.text() payload = self.mqtt_payload.toPlainText() qos = self.mqtt_qos.currentIndex() retain = self.mqtt_retain.isChecked() result = client.publish(topic, payload, qos, retain) client.disconnect() if result.rc == 0: self.mqtt_messages.append(f"Published to {topic}") self.notify_success("Published", f"Message sent to {topic}") else: self.notify_error("Publish Failed", f"Result code: {result.rc}") except Exception as e: self.notify_error("MQTT Error", str(e)) def _subscribe_mqtt(self): """Subscribe to an MQTT topic.""" self.notify_info("Not Implemented", "MQTT subscription not yet implemented") def _connect_websocket(self): """Connect to WebSocket API.""" self.notify_info("Not Implemented", "WebSocket connection not yet implemented") def _disconnect_websocket(self): """Disconnect WebSocket.""" self.notify_info("Not Implemented", "WebSocket not connected") def _ws_subscribe(self): """Subscribe to WebSocket events.""" self.notify_info("Not Implemented", "WebSocket subscription not yet implemented") def _send_ws_message(self): """Send WebSocket message.""" self.notify_info("Not Implemented", "WebSocket messaging not yet implemented") def _run_all_tests(self): """Run all test cases.""" self._test_results.clear() self.ha_progress.setMaximum(len(self.TEST_CASES)) for i, test in enumerate(self.TEST_CASES): self.ha_progress.setValue(i) self._run_test(test, i) self.ha_progress.setValue(len(self.TEST_CASES)) self._update_results_display() def _run_selected_test(self): """Run selected test case.""" row = self.test_table.currentRow() if row < 0: self.notify_warning("No Selection", "Please select a test case") return self._run_test(self.TEST_CASES[row], row) self._update_results_display() def _run_test(self, test: HATestCase, index: int): """Run a single test case.""" self.log_info(f"Running test: {test.name}") start_time = time.time() if test.service == "rest": success, response = self._test_rest_webhook(test.payload) elif test.service == "mqtt": success, response = self._test_mqtt_publish(test.payload) else: success, response = False, "Not implemented" elapsed = (time.time() - start_time) * 1000 result = { "test_name": test.name, "service": test.service, "success": success, "response": response, "elapsed_ms": round(elapsed, 2), "timestamp": datetime.now().isoformat() } self._test_results.append(result) status = "✅ PASS" if success else "❌ FAIL" self.test_table.setItem(index, 3, QTableWidgetItem(status)) def _test_rest_webhook(self, payload: Dict) -> tuple[bool, Any]: """Test REST webhook.""" if not self._ha_url: return False, "HA URL not configured" try: import requests headers = { "Authorization": f"Bearer {self._ha_token}", "Content-Type": "application/json" } response = requests.post( f"{self._ha_url}/api/webhook/eu_utility", headers=headers, json=payload, timeout=10 ) success = response.status_code in [200, 202, 204] return success, {"status": response.status_code} except Exception as e: return False, {"error": str(e)} def _test_mqtt_publish(self, payload: Dict) -> tuple[bool, Any]: """Test MQTT publish.""" if not self._mqtt_broker: return False, "MQTT broker not configured" try: import paho.mqtt.client as mqtt client = mqtt.Client() client.connect(self._mqtt_broker, self._mqtt_port, 5) result = client.publish( payload.get("topic", "test"), payload.get("payload", ""), qos=0 ) client.disconnect() success = result.rc == 0 return success, {"result_code": result.rc} except Exception as e: return False, {"error": str(e)} def _update_results_display(self): """Update the results display.""" if not self._test_results: return passed = sum(1 for r in self._test_results if r["success"]) total = len(self._test_results) self.ha_results_summary.setText(f"Results: {passed}/{total} tests passed") text = [] for result in self._test_results: status = "✅ PASS" if result["success"] else "❌ FAIL" text.append(f"[{status}] {result['test_name']}") text.append(f" Service: {result['service']}") text.append(f" Elapsed: {result['elapsed_ms']}ms") text.append(f" Response: {result['response']}") text.append("") self.ha_results_text.setText("\n".join(text)) def _export_results(self): """Export test results.""" if not self._test_results: self.notify_warning("No Results", "No test results to export") return filename = f"ha_test_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(filename, 'w') as f: json.dump({ "timestamp": datetime.now().isoformat(), "results": self._test_results }, f, indent=2) self.notify_success("Exported", f"Results saved to {filename}") plugin_class = HomeAssistantTester