test_server.py

#!/usr/bin/env python3
"""
Test Server - Mock telco troubleshooting server for local testing
Simulates server.py API endpoints for agent testing
"""

import json
import time
import random
from flask import Flask, request, jsonify
from typing import Dict, Any
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Flask(__name__)

# Mock data store
nodes = {
    "Node-A": {"status": "online", "cpu_usage": 45.2, "memory_usage": 67.8, "uptime": 86400},
    "Node-B": {"status": "online", "cpu_usage": 23.1, "memory_usage": 45.6, "uptime": 86400},
    "Node-C": {"status": "degraded", "cpu_usage": 89.7, "memory_usage": 92.3, "uptime": 7200},
    "Node-D": {"status": "offline", "cpu_usage": 0.0, "memory_usage": 0.0, "uptime": 0},
    "Router-01": {"status": "online", "cpu_usage": 12.3, "memory_usage": 34.5, "uptime": 172800},
    "Switch-01": {"status": "online", "cpu_usage": 8.7, "memory_usage": 23.4, "uptime": 172800}
}

services = {
    "Node-A": ["networking", "web-server", "database"],
    "Node-B": ["networking", "cache"],
    "Node-C": ["web-server", "application"],
    "Node-D": ["networking", "web-server"],
    "Router-01": ["routing", "firewall"],
    "Switch-01": ["switching"]
}

@app.route('/api/nodes/<node_id>/status', methods=['GET'])
def get_node_status(node_id):
    """Get status of a specific network node"""
    time.sleep(random.uniform(0.1, 0.3))  # Simulate network delay
    
    if node_id not in nodes:
        return jsonify({
            "status": "error",
            "error": f"Node {node_id} not found"
        }), 404
    
    node_info = nodes[node_id].copy()
    node_info["node_id"] = node_id
    node_info["timestamp"] = time.time()
    
    return jsonify({
        "status": "success",
        "node_status": node_info.pop("status"),  # Rename to avoid conflict
        **node_info
    })

@app.route('/api/commands/execute', methods=['POST'])
def execute_command():
    """Execute CLI commands on network devices"""
    time.sleep(random.uniform(0.2, 0.5))  # Simulate command execution
    
    data = request.get_json()
    node_id = data.get("node_id")
    command = data.get("command")
    
    if not node_id or not command:
        return jsonify({
            "status": "error",
            "error": "node_id and command are required"
        }), 400
    
    if node_id not in nodes:
        return jsonify({
            "status": "error",
            "error": f"Node {node_id} not found"
        }), 404
    
    # Simulate command execution
    if "ping" in command.lower():
        output = f"PING 8.8.8.8 (8.8.8.8): 56 data bytes\n64 bytes from 8.8.8.8: icmp_seq=1 ttl=116 time={random.uniform(10, 50):.1f} ms"
    elif "top" in command.lower():
        node = nodes[node_id]
        output = f"top - 12:34:56 up {node['uptime']//3600} days, 1 user, load average: 0.15, 0.12, 0.08\nCPU: {node['cpu_usage']:.1f}% us, Memory: {node['memory_usage']:.1f}% used"
    else:
        output = f"Command '{command}' executed successfully on {node_id}"
    
    return jsonify({
        "status": "success",
        "node_id": node_id,
        "command": command,
        "output": output,
        "execution_time": random.uniform(0.2, 0.5),
        "timestamp": time.time()
    })

@app.route('/api/connectivity/test', methods=['POST'])
def check_connectivity():
    """Test connectivity between nodes"""
    time.sleep(random.uniform(0.1, 0.4))  # Simulate connectivity test
    
    data = request.get_json()
    source_node = data.get("source_node")
    destination_node = data.get("destination_node")
    test_type = data.get("test_type", "ping")
    
    if not source_node or not destination_node:
        return jsonify({
            "status": "error",
            "error": "source_node and destination_node are required"
        }), 400
    
    # Check if nodes exist
    if source_node not in nodes or destination_node not in nodes:
        return jsonify({
            "status": "error",
            "error": "One or both nodes not found"
        }), 404
    
    # Simulate connectivity test
    source_status = nodes[source_node]["status"]
    dest_status = nodes[destination_node]["status"]
    
    if source_status == "offline" or dest_status == "offline":
        connectivity = "failed"
        latency_ms = None
        error = "Connection failed: node offline"
    elif source_status == "degraded" or dest_status == "degraded":
        connectivity = "poor"
        latency_ms = random.uniform(100, 300)
        error = None
    else:
        connectivity = "good"
        latency_ms = random.uniform(5, 25)
        error = None
    
    result = {
        "status": "success",
        "source_node": source_node,
        "destination_node": destination_node,
        "test_type": test_type,
        "connectivity": connectivity,
        "timestamp": time.time()
    }
    
    if latency_ms:
        result["latency_ms"] = latency_ms
    
    if error:
        result["error"] = error
    
    return jsonify(result)

@app.route('/api/logs', methods=['GET'])
def get_logs():
    """Retrieve system logs"""
    time.sleep(random.uniform(0.1, 0.3))  # Simulate log retrieval
    
    node_id = request.args.get("node_id")
    log_type = request.args.get("log_type", "syslog")
    lines = int(request.args.get("lines", 50))
    
    if not node_id:
        return jsonify({
            "status": "error",
            "error": "node_id is required"
        }), 400
    
    if node_id not in nodes:
        return jsonify({
            "status": "error",
            "error": f"Node {node_id} not found"
        }), 404
    
    # Generate mock logs
    mock_logs = []
    for i in range(min(lines, 20)):  # Limit to 20 lines for testing
        timestamp = time.time() - (i * 60)  # 1 minute intervals
        
        if log_type == "error":
            log_entries = [
                f"[ERROR] Network interface down on {node_id}",
                f"[ERROR] Connection timeout to external service",
                f"[ERROR] Memory threshold exceeded on {node_id}"
            ]
        elif log_type == "access":
            log_entries = [
                f"[ACCESS] User login to {node_id} from 192.168.1.100",
                f"[ACCESS] API request processed successfully",
                f"[ACCESS] Configuration updated on {node_id}"
            ]
        else:  # syslog
            log_entries = [
                f"[INFO] System health check completed on {node_id}",
                f"[INFO] Network interface status: active",
                f"[WARNING] High CPU usage detected on {node_id}",
                f"[INFO] Service restart completed successfully"
            ]
        
        log_entry = random.choice(log_entries)
        mock_logs.append({
            "timestamp": timestamp,
            "level": log_type.upper(),
            "message": log_entry,
            "node_id": node_id
        })
    
    return jsonify({
        "status": "success",
        "node_id": node_id,
        "log_type": log_type,
        "lines": len(mock_logs),
        "logs": mock_logs
    })

@app.route('/api/metrics/analyze', methods=['POST'])
def analyze_metrics():
    """Analyze network performance metrics"""
    time.sleep(random.uniform(0.2, 0.4))  # Simulate metric analysis
    
    data = request.get_json()
    metric_type = data.get("metric_type")
    time_range = data.get("time_range", "1h")
    nodes_list = data.get("nodes", [])
    
    if not metric_type:
        return jsonify({
            "status": "error",
            "error": "metric_type is required"
        }), 400
    
    # Generate mock metrics
    metrics = []
    for node_id in nodes_list if nodes_list else list(nodes.keys())[:3]:
        if node_id not in nodes:
            continue
        
        node_info = nodes[node_id]
        
        if metric_type == "latency":
            value = random.uniform(5, 100) if node_info["status"] == "online" else random.uniform(100, 500)
            unit = "ms"
        elif metric_type == "bandwidth":
            value = random.uniform(100, 1000) if node_info["status"] == "online" else random.uniform(10, 100)
            unit = "Mbps"
        elif metric_type == "errors":
            value = random.uniform(0, 10) if node_info["status"] == "online" else random.uniform(50, 200)
            unit = "count"
        else:
            value = random.uniform(0, 100)
            unit = "percentage"
        
        metrics.append({
            "node_id": node_id,
            "metric_type": metric_type,
            "value": value,
            "unit": unit,
            "timestamp": time.time() - random.uniform(0, 3600),
            "status": node_info["status"]
        })
    
    return jsonify({
        "status": "success",
        "metric_type": metric_type,
        "time_range": time_range,
        "metrics": metrics,
        "summary": {
            "total_nodes": len(metrics),
            "average_value": sum(m["value"] for m in metrics) / len(metrics) if metrics else 0,
            "unit": unit
        }
    })

@app.route('/api/topology', methods=['GET'])
def get_topology():
    """Get network topology information"""
    time.sleep(random.uniform(0.1, 0.2))  # Simulate topology retrieval
    
    scope = request.args.get("scope", "full")
    
    # Mock topology data
    topology = {
        "nodes": [
            {"id": node_id, "type": "server" if node_id.startswith("Node") else "network", "status": info["status"]}
            for node_id, info in nodes.items()
        ],
        "connections": [
            {"source": "Node-A", "target": "Router-01", "status": "active"},
            {"source": "Node-B", "target": "Router-01", "status": "active"},
            {"source": "Node-C", "target": "Switch-01", "status": "degraded"},
            {"source": "Node-D", "target": "Switch-01", "status": "inactive"},
            {"source": "Router-01", "target": "Switch-01", "status": "active"}
        ]
    }
    
    return jsonify({
        "status": "success",
        "scope": scope,
        "topology": topology,
        "timestamp": time.time()
    })

@app.route('/api/services/restart', methods=['POST'])
def restart_service():
    """Restart a network service"""
    time.sleep(random.uniform(0.5, 1.0))  # Simulate service restart
    
    data = request.get_json()
    node_id = data.get("node_id")
    service_name = data.get("service_name")
    
    if not node_id or not service_name:
        return jsonify({
            "status": "error",
            "error": "node_id and service_name are required"
        }), 400
    
    if node_id not in nodes:
        return jsonify({
            "status": "error",
            "error": f"Node {node_id} not found"
        }), 404
    
    if node_id not in services or service_name not in services[node_id]:
        return jsonify({
            "status": "error",
            "error": f"Service {service_name} not found on {node_id}"
        }), 404
    
    # Simulate service restart
    success = random.random() > 0.1  # 90% success rate
    
    if success:
        # Update node status if it was offline/degraded
        if nodes[node_id]["status"] in ["offline", "degraded"]:
            nodes[node_id]["status"] = "online"
            nodes[node_id]["cpu_usage"] = random.uniform(20, 60)
            nodes[node_id]["memory_usage"] = random.uniform(30, 70)
        
        return jsonify({
            "status": "success",
            "node_id": node_id,
            "service_name": service_name,
            "message": f"Service {service_name} restarted successfully",
            "execution_time": random.uniform(0.5, 1.0),
            "timestamp": time.time()
        })
    else:
        return jsonify({
            "status": "error",
            "node_id": node_id,
            "service_name": service_name,
            "error": f"Failed to restart service {service_name}",
            "timestamp": time.time()
        })

@app.route('/api/ping', methods=['POST'])
def ping_test():
    """Perform ping test"""
    time.sleep(random.uniform(0.1, 0.3))  # Simulate ping
    
    data = request.get_json()
    target = data.get("target")
    count = data.get("count", 4)
    
    if not target:
        return jsonify({
            "status": "error",
            "error": "target is required"
        }), 400
    
    # Simulate ping results
    results = []
    for i in range(count):
        success = random.random() > 0.1  # 90% success rate
        if success:
            results.append({
                "sequence": i + 1,
                "bytes": 64,
                "time_ms": random.uniform(5, 50),
                "ttl": random.randint(50, 120)
            })
        else:
            results.append({
                "sequence": i + 1,
                "status": "timeout"
            })
    
    successful_pings = [r for r in results if "status" not in r]
    
    return jsonify({
        "status": "success",
        "target": target,
        "count": count,
        "successful": len(successful_pings),
        "failed": count - len(successful_pings),
        "results": results,
        "timestamp": time.time()
    })

@app.route('/health', methods=['GET'])
def health_check():
    """Health check endpoint"""
    return jsonify({
        "status": "healthy",
        "timestamp": time.time(),
        "version": "1.0.0"
    })

if __name__ == "__main__":
    print("Starting Telco Troubleshooting Test Server...")
    print("Available endpoints:")
    print("  GET  /api/nodes/<node_id>/status")
    print("  POST /api/commands/execute")
    print("  POST /api/connectivity/test")
    print("  GET  /api/logs")
    print("  POST /api/metrics/analyze")
    print("  GET  /api/topology")
    print("  POST /api/services/restart")
    print("  POST /api/ping")
    print("  GET  /health")
    print("\nServer starting on http://localhost:8000")
    
    app.run(host="0.0.0.0", port=8000, debug=True)