236 lines
8.7 KiB
Python
236 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify that templates are properly activated after being set.
|
|
|
|
This script:
|
|
1. Gets a test device from current.json
|
|
2. Sets a template on the device
|
|
3. Verifies that the template was properly activated
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import requests
|
|
import time
|
|
import sys
|
|
import os
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Import TasmotaManager class
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
from TasmotaManager import TasmotaDiscovery
|
|
|
|
def get_test_device():
|
|
"""Get a test device from current.json"""
|
|
try:
|
|
with open('current.json', 'r') as f:
|
|
data = json.load(f)
|
|
devices = data.get('tasmota', {}).get('devices', [])
|
|
if devices:
|
|
return devices[0] # Use the first device
|
|
else:
|
|
logger.error("No devices found in current.json")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error reading current.json: {e}")
|
|
return None
|
|
|
|
def get_template_from_config():
|
|
"""Get a template from network_configuration.json"""
|
|
try:
|
|
with open('network_configuration.json', 'r') as f:
|
|
config = json.load(f)
|
|
templates = config.get('mqtt', {}).get('config_other', {})
|
|
if templates:
|
|
# Get the first template
|
|
template_key = next(iter(templates))
|
|
template_value = templates[template_key]
|
|
return template_key, template_value
|
|
else:
|
|
logger.error("No templates found in network_configuration.json")
|
|
return None, None
|
|
except Exception as e:
|
|
logger.error(f"Error reading network_configuration.json: {e}")
|
|
return None, None
|
|
|
|
def check_device_module(ip):
|
|
"""Check the current module of the device"""
|
|
try:
|
|
url = f"http://{ip}/cm?cmnd=Module"
|
|
response = requests.get(url, timeout=5)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
logger.info(f"Module response: {data}")
|
|
|
|
# Extract module information
|
|
if "Module" in data:
|
|
module = data["Module"]
|
|
return module
|
|
else:
|
|
logger.error(f"Unexpected response format: {data}")
|
|
return None
|
|
else:
|
|
logger.error(f"Failed to get module: HTTP {response.status_code}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error checking module: {e}")
|
|
return None
|
|
|
|
def check_template_on_device(ip):
|
|
"""Check the current template on the device"""
|
|
try:
|
|
url = f"http://{ip}/cm?cmnd=Template"
|
|
response = requests.get(url, timeout=5)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
logger.info(f"Template response: {data}")
|
|
|
|
# Extract template information
|
|
template = None
|
|
if "Template" in data:
|
|
template = data["Template"]
|
|
elif isinstance(data, dict) and len(data) > 0:
|
|
# If there's no "Template" key but we have a dict, try to get the first value
|
|
first_key = next(iter(data))
|
|
if isinstance(data[first_key], str) and "{" in data[first_key]:
|
|
template = data[first_key]
|
|
# Handle the case where the template is returned as a dict with NAME, GPIO, FLAG, BASE keys
|
|
elif all(key in data for key in ['NAME', 'GPIO', 'FLAG', 'BASE']):
|
|
import json
|
|
template = json.dumps(data)
|
|
|
|
return template
|
|
else:
|
|
logger.error(f"Failed to get template: HTTP {response.status_code}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error checking template: {e}")
|
|
return None
|
|
|
|
def set_template_on_device(ip, template_value):
|
|
"""Set a template on the device and activate it"""
|
|
try:
|
|
# URL encode the template value
|
|
import urllib.parse
|
|
encoded_value = urllib.parse.quote(template_value)
|
|
url = f"http://{ip}/cm?cmnd=Template%20{encoded_value}"
|
|
|
|
logger.info(f"Setting template: {url}")
|
|
response = requests.get(url, timeout=5)
|
|
if response.status_code == 200:
|
|
logger.info(f"Template set response: {response.text}")
|
|
|
|
# Set module to 0 to activate the template
|
|
logger.info("Setting module to 0 to activate template")
|
|
module_url = f"http://{ip}/cm?cmnd=Module%200"
|
|
module_response = requests.get(module_url, timeout=5)
|
|
|
|
if module_response.status_code == 200:
|
|
logger.info(f"Module set response: {module_response.text}")
|
|
|
|
# Restart the device to apply the template
|
|
logger.info("Restarting device to apply template")
|
|
restart_url = f"http://{ip}/cm?cmnd=Restart%201"
|
|
restart_response = requests.get(restart_url, timeout=5)
|
|
|
|
if restart_response.status_code == 200:
|
|
logger.info("Device restart initiated successfully")
|
|
return True
|
|
else:
|
|
logger.error(f"Failed to restart device: HTTP {restart_response.status_code}")
|
|
else:
|
|
logger.error(f"Failed to set module: HTTP {module_response.status_code}")
|
|
else:
|
|
logger.error(f"Failed to set template: HTTP {response.status_code}")
|
|
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Error setting template: {e}")
|
|
return False
|
|
|
|
def main():
|
|
"""Main test function"""
|
|
# Get a test device
|
|
device = get_test_device()
|
|
if not device:
|
|
logger.error("No test device available. Run discovery first.")
|
|
return 1
|
|
|
|
device_name = device.get('name')
|
|
device_ip = device.get('ip')
|
|
|
|
logger.info(f"Testing with device: {device_name} (IP: {device_ip})")
|
|
|
|
# Get a template from the configuration
|
|
template_key, template_value = get_template_from_config()
|
|
if not template_key or not template_value:
|
|
logger.error("No template available in configuration.")
|
|
return 1
|
|
|
|
logger.info(f"Using template: {template_key} = {template_value}")
|
|
|
|
# Check current module and template
|
|
logger.info("Checking current module and template")
|
|
current_module = check_device_module(device_ip)
|
|
current_template = check_template_on_device(device_ip)
|
|
|
|
logger.info(f"Current module: {current_module}")
|
|
logger.info(f"Current template: {current_template}")
|
|
|
|
# Set the template on the device
|
|
logger.info("Setting and activating template")
|
|
success = set_template_on_device(device_ip, template_value)
|
|
if not success:
|
|
logger.error("Failed to set and activate template")
|
|
return 1
|
|
|
|
# Wait for the device to restart
|
|
logger.info("Waiting for device to restart...")
|
|
time.sleep(10)
|
|
|
|
# Check module and template after restart
|
|
logger.info("Checking module and template after restart")
|
|
after_module = check_device_module(device_ip)
|
|
after_template = check_template_on_device(device_ip)
|
|
|
|
logger.info(f"Module after restart: {after_module}")
|
|
logger.info(f"Template after restart: {after_template}")
|
|
|
|
# Verify that the template was activated
|
|
if after_module == 0:
|
|
logger.info("SUCCESS: Module is set to 0 (Template module)")
|
|
else:
|
|
logger.error(f"FAILURE: Module is not set to 0, got {after_module}")
|
|
return 1
|
|
|
|
# Compare templates (this is approximate since formatting might differ)
|
|
import json
|
|
try:
|
|
# Try to parse both as JSON for comparison
|
|
template_json = json.loads(template_value)
|
|
after_json = json.loads(after_template) if after_template else None
|
|
|
|
if after_json and all(key in after_json for key in ['NAME', 'GPIO', 'FLAG', 'BASE']):
|
|
logger.info("SUCCESS: Template appears to be correctly set and activated")
|
|
return 0
|
|
else:
|
|
logger.error("FAILURE: Template does not appear to be correctly set")
|
|
return 1
|
|
except json.JSONDecodeError:
|
|
# If JSON parsing fails, do a simple string comparison
|
|
if template_value == after_template:
|
|
logger.info("SUCCESS: Template appears to be correctly set and activated")
|
|
return 0
|
|
else:
|
|
logger.error("FAILURE: Template does not appear to be correctly set")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |