799 lines
27 KiB
Python
799 lines
27 KiB
Python
"""
|
|
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("""
|
|
<h3>Setup Instructions</h3>
|
|
<p><b>REST API:</b></p>
|
|
<ol>
|
|
<li>In Home Assistant, go to your Profile → Long-Lived Access Tokens</li>
|
|
<li>Create a new token and copy it</li>
|
|
<li>Paste the token above</li>
|
|
</ol>
|
|
<p><b>MQTT (optional):</b></p>
|
|
<ol>
|
|
<li>Install MQTT integration in HA</li>
|
|
<li>Configure your MQTT broker</li>
|
|
<li>Enter broker details above</li>
|
|
</ol>
|
|
""")
|
|
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 |