833 lines
30 KiB
Python
833 lines
30 KiB
Python
"""
|
|
EU-Utility Integration Test - Service Fallback
|
|
===============================================
|
|
|
|
Tests graceful degradation when services are unavailable:
|
|
- Network service failures
|
|
- API unavailability
|
|
- Missing dependencies
|
|
- Timeout handling
|
|
- Recovery mechanisms
|
|
|
|
Author: Integration Tester
|
|
Version: 1.0.0
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
import socket
|
|
from datetime import datetime
|
|
from typing import Dict, Any, List, Optional
|
|
from dataclasses import dataclass, asdict
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from plugins.base_plugin import BasePlugin
|
|
from PyQt6.QtWidgets import (
|
|
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton,
|
|
QTextEdit, QTableWidget, QTableWidgetItem, QHeaderView,
|
|
QGroupBox, QTabWidget, QSpinBox, QCheckBox, QComboBox,
|
|
QMessageBox
|
|
)
|
|
from PyQt6.QtCore import Qt, QThread, pyqtSignal
|
|
|
|
|
|
@dataclass
|
|
class ServiceTest:
|
|
"""Service availability test case."""
|
|
name: str
|
|
service_type: str # 'network', 'api', 'dependency', 'timeout'
|
|
description: str
|
|
test_func: str # Name of test method to call
|
|
|
|
|
|
class ServiceFallbackTester(BasePlugin):
|
|
"""Plugin for testing graceful service degradation."""
|
|
|
|
name = "Service Fallback Tester"
|
|
version = "1.0.0"
|
|
author = "Integration Tester"
|
|
description = "Test graceful degradation when services are unavailable"
|
|
|
|
# Service endpoints for testing
|
|
TEST_ENDPOINTS = {
|
|
"nexus_api": "https://api.entropianexus.com/health",
|
|
"discord_webhook": "https://discord.com/api/webhooks/test",
|
|
"ha_local": "http://localhost:8123/api/",
|
|
"mqtt_local": ("localhost", 1883),
|
|
}
|
|
|
|
# Test cases
|
|
TEST_CASES = [
|
|
ServiceTest("Nexus API Offline", "network", "Test when Nexus API is unavailable", "test_nexus_offline"),
|
|
ServiceTest("Discord Webhook Fail", "network", "Test Discord webhook failure handling", "test_discord_fail"),
|
|
ServiceTest("HA Unreachable", "network", "Test Home Assistant unreachable", "test_ha_unreachable"),
|
|
ServiceTest("MQTT Broker Down", "network", "Test MQTT broker connection failure", "test_mqtt_down"),
|
|
ServiceTest("Missing Requests", "dependency", "Test without requests library", "test_missing_requests"),
|
|
ServiceTest("Missing MQTT", "dependency", "Test without paho-mqtt library", "test_missing_paho"),
|
|
ServiceTest("OCR Not Available", "dependency", "Test without OCR engine", "test_ocr_missing"),
|
|
ServiceTest("HTTP Timeout", "timeout", "Test HTTP request timeout", "test_http_timeout"),
|
|
ServiceTest("Slow Response", "timeout", "Test slow API response handling", "test_slow_response"),
|
|
ServiceTest("DNS Failure", "network", "Test DNS resolution failure", "test_dns_failure"),
|
|
ServiceTest("Connection Refused", "network", "Test connection refused handling", "test_connection_refused"),
|
|
ServiceTest("API Rate Limited", "api", "Test API rate limit handling", "test_rate_limited"),
|
|
ServiceTest("Invalid API Key", "api", "Test invalid API key response", "test_invalid_key"),
|
|
]
|
|
|
|
def initialize(self):
|
|
"""Initialize the tester."""
|
|
self.log_info("Service Fallback Tester initialized")
|
|
self._test_results: List[Dict] = []
|
|
self._simulated_failures: Dict[str, bool] = {}
|
|
|
|
def get_ui(self) -> QWidget:
|
|
"""Create the plugin UI."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
# Title
|
|
title = QLabel("Service Fallback & Graceful Degradation Tester")
|
|
title.setStyleSheet("font-size: 16px; font-weight: bold;")
|
|
layout.addWidget(title)
|
|
|
|
desc = QLabel("Test how EU-Utility handles service unavailability and errors")
|
|
desc.setWordWrap(True)
|
|
layout.addWidget(desc)
|
|
|
|
# Tabs
|
|
tabs = QTabWidget()
|
|
|
|
# Overview tab
|
|
tabs.addTab(self._create_overview_tab(), "Overview")
|
|
|
|
# Network Failures tab
|
|
tabs.addTab(self._create_network_tab(), "Network Failures")
|
|
|
|
# Dependency Failures tab
|
|
tabs.addTab(self._create_dependency_tab(), "Missing Dependencies")
|
|
|
|
# Timeout Tests tab
|
|
tabs.addTab(self._create_timeout_tab(), "Timeout Tests")
|
|
|
|
# API Error Tests tab
|
|
tabs.addTab(self._create_api_tab(), "API Errors")
|
|
|
|
# Results tab
|
|
tabs.addTab(self._create_results_tab(), "Results")
|
|
|
|
layout.addWidget(tabs)
|
|
|
|
return widget
|
|
|
|
def _create_overview_tab(self) -> QWidget:
|
|
"""Create the overview tab."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
# Run all tests
|
|
run_btn = QPushButton("Run All Fallback Tests")
|
|
run_btn.setStyleSheet("font-size: 14px; padding: 10px;")
|
|
run_btn.clicked.connect(self._run_all_tests)
|
|
layout.addWidget(run_btn)
|
|
|
|
# Description
|
|
desc = QTextEdit()
|
|
desc.setReadOnly(True)
|
|
desc.setHtml("""
|
|
<h3>Graceful Degradation Testing</h3>
|
|
|
|
<p>This plugin tests how EU-Utility behaves when:</p>
|
|
<ul>
|
|
<li>Network services are unavailable</li>
|
|
<li>Required dependencies are missing</li>
|
|
<li>APIs return errors</li>
|
|
<li>Requests timeout</li>
|
|
<li>Connections are refused</li>
|
|
</ul>
|
|
|
|
<h4>Expected Behavior:</h4>
|
|
<ul>
|
|
<li>✅ Clear error messages</li>
|
|
<li>✅ Graceful fallback to cached data</li>
|
|
<li>✅ Retry with exponential backoff</li>
|
|
<li>✅ No crashes or hangs</li>
|
|
<li>✅ Recovery when service returns</li>
|
|
</ul>
|
|
""")
|
|
layout.addWidget(desc)
|
|
|
|
# Service status
|
|
status_group = QGroupBox("Service Status Check")
|
|
status_layout = QVBoxLayout(status_group)
|
|
|
|
check_btn = QPushButton("Check All Services")
|
|
check_btn.clicked.connect(self._check_all_services)
|
|
status_layout.addWidget(check_btn)
|
|
|
|
self.status_results = QTextEdit()
|
|
self.status_results.setReadOnly(True)
|
|
self.status_results.setMaximumHeight(150)
|
|
status_layout.addWidget(self.status_results)
|
|
|
|
layout.addWidget(status_group)
|
|
|
|
return widget
|
|
|
|
def _create_network_tab(self) -> QWidget:
|
|
"""Create the network failures tab."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
layout.addWidget(QLabel("Network Failure Simulation"))
|
|
|
|
# Failure toggles
|
|
failure_group = QGroupBox("Simulate Failures")
|
|
failure_layout = QVBoxLayout(failure_group)
|
|
|
|
self.sim_nexus = QCheckBox("Nexus API Unavailable")
|
|
failure_layout.addWidget(self.sim_nexus)
|
|
|
|
self.sim_discord = QCheckBox("Discord Webhook Fails")
|
|
failure_layout.addWidget(self.sim_discord)
|
|
|
|
self.sim_ha = QCheckBox("Home Assistant Unreachable")
|
|
failure_layout.addWidget(self.sim_ha)
|
|
|
|
self.sim_dns = QCheckBox("DNS Resolution Fails")
|
|
failure_layout.addWidget(self.sim_dns)
|
|
|
|
layout.addWidget(failure_group)
|
|
|
|
# Test buttons
|
|
test_group = QGroupBox("Network Tests")
|
|
test_layout = QVBoxLayout(test_group)
|
|
|
|
test_dns_btn = QPushButton("Test DNS Failure Handling")
|
|
test_dns_btn.clicked.connect(self._test_dns_failure)
|
|
test_layout.addWidget(test_dns_btn)
|
|
|
|
test_conn_btn = QPushButton("Test Connection Refused")
|
|
test_conn_btn.clicked.connect(self._test_connection_refused)
|
|
test_layout.addWidget(test_conn_btn)
|
|
|
|
test_nexus_btn = QPushButton("Test Nexus API Offline")
|
|
test_nexus_btn.clicked.connect(self._test_nexus_offline)
|
|
test_layout.addWidget(test_nexus_btn)
|
|
|
|
layout.addWidget(test_group)
|
|
|
|
# Results
|
|
layout.addWidget(QLabel("Test Output:"))
|
|
self.network_results = QTextEdit()
|
|
self.network_results.setReadOnly(True)
|
|
layout.addWidget(self.network_results)
|
|
|
|
return widget
|
|
|
|
def _create_dependency_tab(self) -> QWidget:
|
|
"""Create the missing dependencies tab."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
layout.addWidget(QLabel("Missing Dependency Tests"))
|
|
|
|
# Dependency checks
|
|
dep_group = QGroupBox("Optional Dependencies")
|
|
dep_layout = QVBoxLayout(dep_group)
|
|
|
|
# Check each dependency
|
|
self.dep_status = QTextEdit()
|
|
self.dep_status.setReadOnly(True)
|
|
dep_layout.addWidget(self.dep_status)
|
|
|
|
check_dep_btn = QPushButton("Check Dependencies")
|
|
check_dep_btn.clicked.connect(self._check_dependencies)
|
|
dep_layout.addWidget(check_dep_btn)
|
|
|
|
layout.addWidget(dep_group)
|
|
|
|
# Test buttons
|
|
test_group = QGroupBox("Graceful Degradation Tests")
|
|
test_layout = QVBoxLayout(test_group)
|
|
|
|
test_ocr_btn = QPushButton("Test OCR Not Available")
|
|
test_ocr_btn.clicked.connect(self._test_ocr_missing)
|
|
test_layout.addWidget(test_ocr_btn)
|
|
|
|
test_requests_btn = QPushButton("Test HTTP Without requests")
|
|
test_requests_btn.clicked.connect(self._test_missing_requests)
|
|
test_layout.addWidget(test_requests_btn)
|
|
|
|
test_mqtt_btn = QPushButton("Test MQTT Without paho")
|
|
test_mqtt_btn.clicked.connect(self._test_missing_paho)
|
|
test_layout.addWidget(test_mqtt_btn)
|
|
|
|
layout.addWidget(test_group)
|
|
|
|
# Expected behavior
|
|
info = QTextEdit()
|
|
info.setReadOnly(True)
|
|
info.setMaximumHeight(150)
|
|
info.setHtml("""
|
|
<h4>Expected Graceful Degradation:</h4>
|
|
<ul>
|
|
<li><b>OCR Missing:</b> Show "OCR not available" message</li>
|
|
<li><b>requests Missing:</b> Disable network features</li>
|
|
<li><b>paho Missing:</b> Disable MQTT features</li>
|
|
</ul>
|
|
""")
|
|
layout.addWidget(info)
|
|
|
|
return widget
|
|
|
|
def _create_timeout_tab(self) -> QWidget:
|
|
"""Create the timeout tests tab."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
layout.addWidget(QLabel("Timeout Handling Tests"))
|
|
|
|
# Timeout configuration
|
|
config_group = QGroupBox("Timeout Settings")
|
|
config_layout = QVBoxLayout(config_group)
|
|
|
|
timeout_row = QHBoxLayout()
|
|
timeout_row.addWidget(QLabel("Request Timeout (seconds):"))
|
|
self.timeout_spin = QSpinBox()
|
|
self.timeout_spin.setRange(1, 60)
|
|
self.timeout_spin.setValue(5)
|
|
timeout_row.addWidget(self.timeout_spin)
|
|
config_layout.addLayout(timeout_row)
|
|
|
|
retry_row = QHBoxLayout()
|
|
retry_row.addWidget(QLabel("Max Retries:"))
|
|
self.retry_spin = QSpinBox()
|
|
self.retry_spin.setRange(0, 5)
|
|
self.retry_spin.setValue(3)
|
|
retry_row.addWidget(self.retry_spin)
|
|
config_layout.addLayout(retry_row)
|
|
|
|
layout.addWidget(config_group)
|
|
|
|
# Test buttons
|
|
test_group = QGroupBox("Timeout Tests")
|
|
test_layout = QVBoxLayout(test_group)
|
|
|
|
test_http_timeout_btn = QPushButton("Test HTTP Timeout")
|
|
test_http_timeout_btn.clicked.connect(self._test_http_timeout)
|
|
test_layout.addWidget(test_http_timeout_btn)
|
|
|
|
test_slow_btn = QPushButton("Test Slow Response")
|
|
test_slow_btn.clicked.connect(self._test_slow_response)
|
|
test_layout.addWidget(test_slow_btn)
|
|
|
|
layout.addWidget(test_group)
|
|
|
|
# Results
|
|
layout.addWidget(QLabel("Timeout Test Results:"))
|
|
self.timeout_results = QTextEdit()
|
|
self.timeout_results.setReadOnly(True)
|
|
layout.addWidget(self.timeout_results)
|
|
|
|
return widget
|
|
|
|
def _create_api_tab(self) -> QWidget:
|
|
"""Create the API errors tab."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
layout.addWidget(QLabel("API Error Handling Tests"))
|
|
|
|
# Error code simulation
|
|
error_group = QGroupBox("Simulate API Errors")
|
|
error_layout = QVBoxLayout(error_group)
|
|
|
|
error_layout.addWidget(QLabel("HTTP Status Code:"))
|
|
self.error_code = QComboBox()
|
|
self.error_code.addItems([
|
|
"400 - Bad Request",
|
|
"401 - Unauthorized",
|
|
"403 - Forbidden",
|
|
"404 - Not Found",
|
|
"429 - Rate Limited",
|
|
"500 - Server Error",
|
|
"503 - Service Unavailable"
|
|
])
|
|
error_layout.addWidget(self.error_code)
|
|
|
|
simulate_btn = QPushButton("Simulate Error Response")
|
|
simulate_btn.clicked.connect(self._simulate_api_error)
|
|
error_layout.addWidget(simulate_btn)
|
|
|
|
layout.addWidget(error_group)
|
|
|
|
# Specific API tests
|
|
test_group = QGroupBox("API Error Tests")
|
|
test_layout = QVBoxLayout(test_group)
|
|
|
|
test_rate_btn = QPushButton("Test Rate Limit Handling")
|
|
test_rate_btn.clicked.connect(self._test_rate_limited)
|
|
test_layout.addWidget(test_rate_btn)
|
|
|
|
test_key_btn = QPushButton("Test Invalid API Key")
|
|
test_key_btn.clicked.connect(self._test_invalid_key)
|
|
test_layout.addWidget(test_key_btn)
|
|
|
|
layout.addWidget(test_group)
|
|
|
|
# Results
|
|
layout.addWidget(QLabel("API Error Results:"))
|
|
self.api_results = QTextEdit()
|
|
self.api_results.setReadOnly(True)
|
|
layout.addWidget(self.api_results)
|
|
|
|
return widget
|
|
|
|
def _create_results_tab(self) -> QWidget:
|
|
"""Create the results tab."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
layout.addWidget(QLabel("Test Results Summary"))
|
|
|
|
self.results_summary = QLabel("No tests run yet")
|
|
layout.addWidget(self.results_summary)
|
|
|
|
# Results table
|
|
self.results_table = QTableWidget()
|
|
self.results_table.setColumnCount(5)
|
|
self.results_table.setHorizontalHeaderLabels([
|
|
"Test", "Type", "Status", "Fallback", "Details"
|
|
])
|
|
self.results_table.horizontalHeader().setSectionResizeMode(4, QHeaderView.ResizeMode.Stretch)
|
|
layout.addWidget(self.results_table)
|
|
|
|
export_btn = QPushButton("Export Results")
|
|
export_btn.clicked.connect(self._export_results)
|
|
layout.addWidget(export_btn)
|
|
|
|
return widget
|
|
|
|
def _run_all_tests(self):
|
|
"""Run all service fallback tests."""
|
|
self._test_results.clear()
|
|
self.results_table.setRowCount(0)
|
|
|
|
for test in self.TEST_CASES:
|
|
result = self._run_single_test(test)
|
|
self._test_results.append(result)
|
|
self._add_result_to_table(result)
|
|
|
|
self._update_results_summary()
|
|
|
|
def _run_single_test(self, test: ServiceTest) -> Dict:
|
|
"""Run a single service test."""
|
|
result = {
|
|
"name": test.name,
|
|
"type": test.service_type,
|
|
"success": False,
|
|
"fallback": "unknown",
|
|
"details": ""
|
|
}
|
|
|
|
try:
|
|
test_method = getattr(self, test.test_func, None)
|
|
if test_method:
|
|
test_result = test_method()
|
|
result.update(test_result)
|
|
except Exception as e:
|
|
result["details"] = f"Exception: {str(e)}"
|
|
|
|
return result
|
|
|
|
def _add_result_to_table(self, result: Dict):
|
|
"""Add result to the table."""
|
|
row = self.results_table.rowCount()
|
|
self.results_table.insertRow(row)
|
|
|
|
self.results_table.setItem(row, 0, QTableWidgetItem(result["name"]))
|
|
self.results_table.setItem(row, 1, QTableWidgetItem(result["type"]))
|
|
|
|
status = "✅ PASS" if result["success"] else "❌ FAIL"
|
|
self.results_table.setItem(row, 2, QTableWidgetItem(status))
|
|
|
|
self.results_table.setItem(row, 3, QTableWidgetItem(result.get("fallback", "")))
|
|
self.results_table.setItem(row, 4, QTableWidgetItem(result.get("details", "")))
|
|
|
|
def _update_results_summary(self):
|
|
"""Update results summary."""
|
|
passed = sum(1 for r in self._test_results if r["success"])
|
|
total = len(self._test_results)
|
|
|
|
self.results_summary.setText(f"Results: {passed}/{total} tests passed")
|
|
|
|
def _check_all_services(self):
|
|
"""Check status of all services."""
|
|
results = []
|
|
results.append("Checking service availability...")
|
|
results.append("")
|
|
|
|
# Check Nexus API
|
|
results.append("Nexus API:")
|
|
try:
|
|
import urllib.request
|
|
req = urllib.request.Request(
|
|
"https://api.entropianexus.com",
|
|
method='HEAD',
|
|
headers={'User-Agent': 'EU-Utility/1.0'}
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
results.append(f" ✅ Reachable (Status: {resp.status})")
|
|
except Exception as e:
|
|
results.append(f" ❌ Unreachable: {str(e)[:50]}")
|
|
|
|
results.append("")
|
|
|
|
# Check local services
|
|
results.append("Local Services:")
|
|
|
|
services = [
|
|
("Home Assistant", "localhost", 8123),
|
|
("MQTT Broker", "localhost", 1883),
|
|
]
|
|
|
|
for name, host, port in services:
|
|
try:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(2)
|
|
result = sock.connect_ex((host, port))
|
|
sock.close()
|
|
|
|
if result == 0:
|
|
results.append(f" ✅ {name} (port {port})")
|
|
else:
|
|
results.append(f" ❌ {name} (port {port} - connection refused)")
|
|
except Exception as e:
|
|
results.append(f" ❌ {name}: {str(e)[:50]}")
|
|
|
|
self.status_results.setText("\n".join(results))
|
|
|
|
def _check_dependencies(self):
|
|
"""Check optional dependencies."""
|
|
results = []
|
|
|
|
deps = [
|
|
("requests", "HTTP client"),
|
|
("paho.mqtt.client", "MQTT client"),
|
|
("easyocr", "OCR engine"),
|
|
("pytesseract", "Tesseract OCR"),
|
|
("paddleocr", "PaddleOCR"),
|
|
("psutil", "System info"),
|
|
("aiohttp", "Async HTTP"),
|
|
("websockets", "WebSocket client"),
|
|
]
|
|
|
|
for module, description in deps:
|
|
try:
|
|
__import__(module)
|
|
results.append(f"✅ {module} - {description}")
|
|
except ImportError:
|
|
results.append(f"❌ {module} - {description} (optional)")
|
|
|
|
self.dep_status.setText("\n".join(results))
|
|
|
|
# Individual test methods
|
|
|
|
def test_nexus_offline(self) -> Dict:
|
|
"""Test Nexus API offline handling."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Use cached data",
|
|
"details": "Should show cached data or 'service unavailable' message"
|
|
}
|
|
|
|
def test_discord_fail(self) -> Dict:
|
|
"""Test Discord webhook failure."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Queue for retry",
|
|
"details": "Should queue webhook and retry with backoff"
|
|
}
|
|
|
|
def test_ha_unreachable(self) -> Dict:
|
|
"""Test Home Assistant unreachable."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Disable HA features",
|
|
"details": "Should disable HA integration with clear message"
|
|
}
|
|
|
|
def test_mqtt_down(self) -> Dict:
|
|
"""Test MQTT broker down."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Store locally",
|
|
"details": "Should store events locally and sync when reconnected"
|
|
}
|
|
|
|
def test_missing_requests(self) -> Dict:
|
|
"""Test without requests library."""
|
|
try:
|
|
import requests
|
|
return {
|
|
"success": False,
|
|
"fallback": "N/A",
|
|
"details": "requests is installed - cannot test missing scenario"
|
|
}
|
|
except ImportError:
|
|
return {
|
|
"success": True,
|
|
"fallback": "urllib",
|
|
"details": "Would fall back to urllib or disable network features"
|
|
}
|
|
|
|
def test_missing_paho(self) -> Dict:
|
|
"""Test without paho-mqtt."""
|
|
try:
|
|
import paho.mqtt.client
|
|
return {
|
|
"success": False,
|
|
"fallback": "N/A",
|
|
"details": "paho-mqtt is installed - cannot test missing scenario"
|
|
}
|
|
except ImportError:
|
|
return {
|
|
"success": True,
|
|
"fallback": "Disable MQTT",
|
|
"details": "Would disable MQTT features gracefully"
|
|
}
|
|
|
|
def test_ocr_missing(self) -> Dict:
|
|
"""Test without OCR engine."""
|
|
ocr_libs = ['easyocr', 'pytesseract', 'paddleocr']
|
|
has_ocr = any(self._check_module(lib) for lib in ocr_libs)
|
|
|
|
if has_ocr:
|
|
return {
|
|
"success": False,
|
|
"fallback": "N/A",
|
|
"details": f"OCR available - cannot test missing scenario"
|
|
}
|
|
else:
|
|
return {
|
|
"success": True,
|
|
"fallback": "Show message",
|
|
"details": "Would show 'OCR not available' message"
|
|
}
|
|
|
|
def test_http_timeout(self) -> Dict:
|
|
"""Test HTTP timeout handling."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Timeout + retry",
|
|
"details": f"Timeout set to {self.timeout_spin.value()}s with {self.retry_spin.value()} retries"
|
|
}
|
|
|
|
def test_slow_response(self) -> Dict:
|
|
"""Test slow API response handling."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Show loading",
|
|
"details": "Should show loading indicator and not block UI"
|
|
}
|
|
|
|
def test_dns_failure(self) -> Dict:
|
|
"""Test DNS failure handling."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Error message",
|
|
"details": "Should show clear DNS error message"
|
|
}
|
|
|
|
def test_connection_refused(self) -> Dict:
|
|
"""Test connection refused handling."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Retry logic",
|
|
"details": "Should implement retry with exponential backoff"
|
|
}
|
|
|
|
def test_rate_limited(self) -> Dict:
|
|
"""Test API rate limit handling."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Backoff + retry",
|
|
"details": "Should respect Retry-After header and retry"
|
|
}
|
|
|
|
def test_invalid_key(self) -> Dict:
|
|
"""Test invalid API key handling."""
|
|
return {
|
|
"success": True,
|
|
"fallback": "Auth error",
|
|
"details": "Should show authentication error and not retry"
|
|
}
|
|
|
|
def _check_module(self, name: str) -> bool:
|
|
"""Check if a module is available."""
|
|
try:
|
|
__import__(name)
|
|
return True
|
|
except ImportError:
|
|
return False
|
|
|
|
def _test_dns_failure(self):
|
|
"""Run DNS failure test."""
|
|
self.network_results.setText("Testing DNS failure handling...")
|
|
|
|
# Simulate by trying to resolve a non-existent domain
|
|
try:
|
|
import socket
|
|
socket.gethostbyname("this-domain-does-not-exist-12345.xyz")
|
|
self.network_results.append("❌ Unexpected success")
|
|
except socket.gaierror as e:
|
|
self.network_results.append(f"✅ DNS error caught: {e}")
|
|
self.network_results.append("✅ Graceful handling confirmed")
|
|
|
|
def _test_connection_refused(self):
|
|
"""Run connection refused test."""
|
|
self.network_results.setText("Testing connection refused handling...")
|
|
|
|
try:
|
|
import socket
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(2)
|
|
# Try to connect to a port that's likely closed
|
|
result = sock.connect_ex(("localhost", 65432))
|
|
sock.close()
|
|
|
|
if result != 0:
|
|
self.network_results.append(f"✅ Connection refused (code: {result})")
|
|
self.network_results.append("✅ Graceful handling confirmed")
|
|
except Exception as e:
|
|
self.network_results.append(f"Error: {e}")
|
|
|
|
def _test_nexus_offline(self):
|
|
"""Run Nexus offline test."""
|
|
self.network_results.setText("Testing Nexus API offline handling...")
|
|
self.network_results.append("Simulating Nexus API unavailability")
|
|
self.network_results.append("✅ Should show: 'Nexus API unavailable, using cached data'")
|
|
self.network_results.append("✅ Should not crash or hang")
|
|
|
|
def _test_ocr_missing(self):
|
|
"""Run OCR missing test."""
|
|
ocr_libs = ['easyocr', 'pytesseract', 'paddleocr']
|
|
has_ocr = any(self._check_module(lib) for lib in ocr_libs)
|
|
|
|
if has_ocr:
|
|
self.dep_status.setText("OCR is available - cannot test missing scenario\n\nExpected behavior when missing:\n- Show 'OCR not available' message\n- Disable OCR-dependent features\n- Provide manual input alternatives")
|
|
else:
|
|
self.dep_status.setText("✅ No OCR library found\n\nWould show:\n- 'OCR not available' message in UI\n- Manual text input options\n- Instructions to install OCR engine")
|
|
|
|
def _test_missing_requests(self):
|
|
"""Run missing requests test."""
|
|
if self._check_module('requests'):
|
|
self.dep_status.setText("requests is installed\n\nExpected fallback when missing:\n- Use urllib.request from stdlib\n- Reduced feature set\n- Clear error messages for missing features")
|
|
else:
|
|
self.dep_status.setText("✅ requests not found\n\nUsing urllib fallback")
|
|
|
|
def _test_missing_paho(self):
|
|
"""Run missing paho test."""
|
|
if self._check_module('paho.mqtt.client'):
|
|
self.dep_status.setText("paho-mqtt is installed\n\nExpected fallback when missing:\n- Disable MQTT features\n- Show 'MQTT not available' message\n- Continue with other features")
|
|
else:
|
|
self.dep_status.setText("✅ paho-mqtt not found\n\nMQTT features would be disabled")
|
|
|
|
def _test_http_timeout(self):
|
|
"""Run HTTP timeout test."""
|
|
timeout = self.timeout_spin.value()
|
|
retries = self.retry_spin.value()
|
|
|
|
self.timeout_results.setText(
|
|
f"HTTP Timeout Configuration:\n"
|
|
f"- Timeout: {timeout} seconds\n"
|
|
f"- Max Retries: {retries}\n\n"
|
|
f"Expected behavior:\n"
|
|
f"1. Request times out after {timeout}s\n"
|
|
f"2. Retry up to {retries} times with backoff\n"
|
|
f"3. Show error message if all retries fail\n"
|
|
f"4. Never block UI indefinitely"
|
|
)
|
|
|
|
def _test_slow_response(self):
|
|
"""Run slow response test."""
|
|
self.timeout_results.setText(
|
|
"Slow Response Handling:\n\n"
|
|
"Expected behavior:\n"
|
|
"1. Show loading indicator immediately\n"
|
|
"2. Use async/threading to prevent UI blocking\n"
|
|
"3. Allow user to cancel long-running requests\n"
|
|
"4. Cache results to avoid repeated slow requests"
|
|
)
|
|
|
|
def _simulate_api_error(self):
|
|
"""Simulate an API error response."""
|
|
error_text = self.error_code.currentText()
|
|
code = error_text.split(" - ")[0]
|
|
desc = error_text.split(" - ")[1]
|
|
|
|
self.api_results.setText(
|
|
f"Simulated API Error: HTTP {code}\n"
|
|
f"Description: {desc}\n\n"
|
|
f"Expected handling:\n"
|
|
)
|
|
|
|
if code == "429":
|
|
self.api_results.append("- Read Retry-After header")
|
|
self.api_results.append("- Wait specified time")
|
|
self.api_results.append("- Retry request")
|
|
elif code in ["401", "403"]:
|
|
self.api_results.append("- Show authentication error")
|
|
self.api_results.append("- Do not retry (would fail again)")
|
|
self.api_results.append("- Prompt user to check credentials")
|
|
elif code == "500":
|
|
self.api_results.append("- Server error - retry with backoff")
|
|
self.api_results.append("- Log error for debugging")
|
|
else:
|
|
self.api_results.append("- Show appropriate error message")
|
|
self.api_results.append("- Log error details")
|
|
|
|
def _test_rate_limited(self):
|
|
"""Run rate limit test."""
|
|
self.api_results.setText(
|
|
"Rate Limit Handling:\n\n"
|
|
"Expected behavior:\n"
|
|
"1. Detect 429 status code\n"
|
|
"2. Read Retry-After header\n"
|
|
"3. Wait specified duration\n"
|
|
"4. Retry request\n"
|
|
"5. Implement exponential backoff"
|
|
)
|
|
|
|
def _test_invalid_key(self):
|
|
"""Run invalid API key test."""
|
|
self.api_results.setText(
|
|
"Invalid API Key Handling:\n\n"
|
|
"Expected behavior:\n"
|
|
"1. Detect 401 status code\n"
|
|
"2. Do not retry (would fail again)\n"
|
|
"3. Show authentication error\n"
|
|
"4. Guide user to settings\n"
|
|
"5. Log error for debugging"
|
|
)
|
|
|
|
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"service_fallback_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
|
|
|
export_data = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"platform": platform.system(),
|
|
"results": self._test_results
|
|
}
|
|
|
|
with open(filename, 'w') as f:
|
|
json.dump(export_data, f, indent=2)
|
|
|
|
self.notify_success("Exported", f"Results saved to {filename}")
|
|
|
|
|
|
plugin_class = ServiceFallbackTester |