test_api.py•7.37 kB
"""
Test script for the Sectional MCP Panel API endpoints.
"""
import asyncio
import sys
import os
import json
import requests
from pathlib import Path
# Add the project root to the Python path
project_root = Path(__file__).parent.parent
sys.path.append(str(project_root))
# API base URL - assumes the server is running locally
API_BASE_URL = "http://localhost:8000/api/v1"
def test_api_endpoints():
"""Test the API endpoints functionality."""
print("Testing API endpoints...")
# Test panel endpoints
print("\nTesting panel endpoints:")
try:
# Get panel configuration
response = requests.get(f"{API_BASE_URL}/panel")
if response.status_code == 200:
panel = response.json()
print(f"Retrieved panel configuration: {panel['name']}, version {panel['version']}")
else:
print(f"Failed to get panel configuration: {response.status_code}")
return
# Update panel configuration
panel_update = {
"name": "Updated Test Panel",
"global_settings": {
"settings": {
"environmentVars": {
"GLOBAL_ENV": "updated"
}
}
}
}
response = requests.put(f"{API_BASE_URL}/panel", json=panel_update)
if response.status_code == 200:
updated_panel = response.json()
print(f"Updated panel configuration: {updated_panel['name']}")
else:
print(f"Failed to update panel configuration: {response.status_code}")
except Exception as e:
print(f"Error testing panel endpoints: {e}")
# Test section endpoints
print("\nTesting section endpoints:")
try:
# Create a section
section_data = {
"name": "API Test Section",
"description": "A test section created via API",
"settings": {
"environmentVars": {
"SECTION_ENV": "api_test"
}
}
}
response = requests.post(f"{API_BASE_URL}/sections", json=section_data)
if response.status_code == 201:
section = response.json()
section_id = section["id"]
print(f"Created section: {section['name']} (ID: {section_id})")
else:
print(f"Failed to create section: {response.status_code}")
return
# Get all sections
response = requests.get(f"{API_BASE_URL}/sections")
if response.status_code == 200:
sections = response.json()
print(f"Retrieved {len(sections)} sections")
else:
print(f"Failed to get sections: {response.status_code}")
# Get section by ID
response = requests.get(f"{API_BASE_URL}/sections/{section_id}")
if response.status_code == 200:
section = response.json()
print(f"Retrieved section: {section['name']}")
else:
print(f"Failed to get section: {response.status_code}")
# Update section
section_update = {
"description": "Updated test section description"
}
response = requests.put(f"{API_BASE_URL}/sections/{section_id}", json=section_update)
if response.status_code == 200:
updated_section = response.json()
print(f"Updated section: {updated_section['name']}, description: {updated_section['description']}")
else:
print(f"Failed to update section: {response.status_code}")
except Exception as e:
print(f"Error testing section endpoints: {e}")
section_id = None
# Test server endpoints
print("\nTesting server endpoints:")
try:
if not section_id:
print("Skipping server tests because section creation failed")
return
# Create a server
server_data = {
"name": "API Test Server",
"section_id": section_id,
"description": "A test server created via API",
"runtime_definition": {
"type": "docker_image",
"command": "nginx:latest",
"args": [],
"ports": [
{
"containerPort": 80,
"protocol": "TCP"
}
]
},
"settings": {
"environmentVars": {
"SERVER_ENV": "api_test"
}
}
}
response = requests.post(f"{API_BASE_URL}/servers", json=server_data)
if response.status_code == 201:
server = response.json()
server_id = server["id"]
print(f"Created server: {server['name']} (ID: {server_id})")
else:
print(f"Failed to create server: {response.status_code}")
return
# Get all servers
response = requests.get(f"{API_BASE_URL}/servers")
if response.status_code == 200:
servers = response.json()
print(f"Retrieved {len(servers)} servers")
else:
print(f"Failed to get servers: {response.status_code}")
# Get server by ID
response = requests.get(f"{API_BASE_URL}/servers/{server_id}")
if response.status_code == 200:
server = response.json()
print(f"Retrieved server: {server['name']}, status: {server['status']}")
else:
print(f"Failed to get server: {response.status_code}")
# Update server
server_update = {
"description": "Updated test server description"
}
response = requests.put(f"{API_BASE_URL}/servers/{server_id}", json=server_update)
if response.status_code == 200:
updated_server = response.json()
print(f"Updated server: {updated_server['name']}, description: {updated_server['description']}")
else:
print(f"Failed to update server: {response.status_code}")
# Note: We're not testing server start/stop/restart here as they require Docker
# Clean up - delete server
response = requests.delete(f"{API_BASE_URL}/servers/{server_id}")
if response.status_code == 204:
print(f"Deleted server: {server_id}")
else:
print(f"Failed to delete server: {response.status_code}")
except Exception as e:
print(f"Error testing server endpoints: {e}")
# Clean up - delete section
try:
if section_id:
response = requests.delete(f"{API_BASE_URL}/sections/{section_id}")
if response.status_code == 204:
print(f"Deleted section: {section_id}")
else:
print(f"Failed to delete section: {response.status_code}")
except Exception as e:
print(f"Error deleting section: {e}")
print("\nAPI endpoints test completed")
if __name__ == "__main__":
print("Note: This test requires the API server to be running.")
print("Start the server with: python -m src.main")
print("Press Enter to continue or Ctrl+C to cancel...")
input()
test_api_endpoints()