TasmotaManager/tests/test_fulltopic_approaches.py

#!/usr/bin/env python3
"""
Test script to try different approaches for setting the FullTopic parameter
to find a solution that avoids the extra '=' being added to the beginning of the value.
"""

import sys
import logging
import requests
import json
import argparse
import time
import urllib.parse

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger("FullTopicApproachesTest")

def load_config():
    """Load the network configuration."""
    try:
        with open('network_configuration.json', 'r') as f:
            return json.load(f)
    except Exception as e:
        logger.error(f"Error loading configuration: {str(e)}")
        sys.exit(1)

def get_current_fulltopic(ip_address):
    """Get the current FullTopic value from the device."""
    try:
        status_url = f"http://{ip_address}/cm?cmnd=FullTopic"
        response = requests.get(status_url, timeout=5)
        if response.status_code == 200:
            try:
                # Try to parse as JSON
                data = response.json()
                if isinstance(data, dict) and "FullTopic" in data:
                    current_value = data["FullTopic"]
                else:
                    current_value = response.text
            except:
                # If not JSON, use the raw text
                current_value = response.text
                
            logger.info(f"Current FullTopic value: {current_value}")
            return current_value
        else:
            logger.error(f"Failed to get current FullTopic value: {response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        logger.error(f"Error connecting to device: {str(e)}")
        return None

def test_approach(ip_address, approach_name, url):
    """Test a specific approach for setting the FullTopic parameter."""
    logger.info(f"Testing approach: {approach_name}")
    logger.info(f"URL: {url}")
    
    try:
        response = requests.get(url, timeout=5)
        
        # Log the raw response for debugging
        logger.info(f"Raw response: {response.text}")
        
        if response.status_code == 200:
            try:
                # Try to parse as JSON
                data = response.json()
                logger.info(f"Response JSON: {data}")
            except:
                logger.info(f"Response is not JSON: {response.text}")
        else:
            logger.error(f"Failed to set FullTopic: {response.status_code}")
            return False
    except requests.exceptions.RequestException as e:
        logger.error(f"Error setting FullTopic: {str(e)}")
        return False
    
    # Wait a moment for the change to take effect
    time.sleep(1)
    
    # Verify the FullTopic was set correctly
    new_value = get_current_fulltopic(ip_address)
    if new_value is None:
        return False
    
    # Check if the value has an extra '=' at the beginning
    if new_value.startswith('='):
        logger.error(f"ISSUE DETECTED: FullTopic still has an extra '=' at the beginning: {new_value}")
        return False
    else:
        logger.info(f"SUCCESS: FullTopic set correctly without an extra '=': {new_value}")
        return True

def main():
    """Main function to test different approaches for setting the FullTopic parameter."""
    parser = argparse.ArgumentParser(description='Test different approaches for setting the FullTopic parameter')
    parser.add_argument('ip_address', help='IP address of the Tasmota device to test')
    args = parser.parse_args()
    
    if not args.ip_address:
        print("Usage: python test_fulltopic_approaches.py <ip_address>")
        sys.exit(1)
    
    # Load configuration
    config = load_config()
    mqtt_config = config.get('mqtt', {})
    
    if not mqtt_config:
        logger.error("No MQTT configuration found")
        sys.exit(1)
    
    # Get the FullTopic value from configuration
    full_topic = mqtt_config.get('FullTopic', '%prefix%/%topic%/')
    logger.info(f"FullTopic from configuration: {full_topic}")
    
    # Get the current FullTopic value
    current_value = get_current_fulltopic(args.ip_address)
    if current_value is None:
        sys.exit(1)
    
    # Try different approaches
    approaches = [
        # Current approach in TasmotaManager.py
        {
            "name": "Current approach (setting=value)",
            "url": f"http://{args.ip_address}/cm?cmnd=FullTopic={full_topic}"
        },
        # Try with URL encoding the value
        {
            "name": "URL encoded value",
            "url": f"http://{args.ip_address}/cm?cmnd=FullTopic={urllib.parse.quote(full_topic)}"
        },
        # Try with a space (%20) instead of equals
        {
            "name": "Using space (%20) instead of equals",
            "url": f"http://{args.ip_address}/cm?cmnd=FullTopic%20{full_topic}"
        },
        # Try with backslash before equals
        {
            "name": "Backslash before equals",
            "url": f"http://{args.ip_address}/cm?cmnd=FullTopic\\={full_topic}"
        },
        # Try with double equals
        {
            "name": "Double equals",
            "url": f"http://{args.ip_address}/cm?cmnd=FullTopic=={full_topic}"
        },
        # Try with no separator (direct value)
        {
            "name": "No separator (direct value)",
            "url": f"http://{args.ip_address}/cm?cmnd=FullTopic{full_topic}"
        }
    ]
    
    # Test each approach
    successful_approaches = []
    for approach in approaches:
        success = test_approach(args.ip_address, approach["name"], approach["url"])
        if success:
            successful_approaches.append(approach["name"])
    
    # Print summary
    print("\n=== SUMMARY ===")
    if successful_approaches:
        print(f"Successful approaches: {len(successful_approaches)}/{len(approaches)}")
        for i, approach in enumerate(successful_approaches, 1):
            print(f"{i}. {approach}")
    else:
        print("No successful approaches found.")
    
    # Exit with success if at least one approach worked
    sys.exit(0 if successful_approaches else 1)

if __name__ == "__main__":
    main()