test_database.py•4.84 kB
"""
Test script for the Sectional MCP Panel database operations.
"""
import asyncio
import sys
import os
import json
from pathlib import Path
# Add the project root to the Python path
project_root = Path(__file__).parent.parent
sys.path.append(str(project_root))
from src.database.database import init_db, get_db
from src.database import crud
from src.config.models import SectionCreate, ServerCreate
async def test_database_operations():
"""Test basic database CRUD operations."""
print("Testing database operations...")
# Initialize the database
await init_db()
print("Database initialized")
# Create a test panel configuration
async with get_db() as db:
panel_config = {
"name": "Test Panel",
"version": "0.1.0",
"config_schema_version": "1.0.0",
"global_settings": {
"settings": {
"environmentVars": {
"ENV": "test"
},
"resourceLimits": {
"cpuRequest": 0.1,
"cpuLimit": 0.5,
"memoryRequestMB": 64,
"memoryLimitMB": 128
},
"runtimeOptions": {
"restartPolicy": "on_failure",
"maxRestarts": 3,
"terminationGracePeriodSeconds": 30
}
}
}
}
panel = await crud.update_panel_config(db, panel_config)
print(f"Created panel configuration: {panel.name}, version {panel.version}")
# Create a test section
section_data = {
"name": "Test Section",
"description": "A test section for database operations",
"settings": {
"environmentVars": {
"SECTION_ENV": "test_section"
}
}
}
section = await crud.create_section(db, section_data)
print(f"Created section: {section.name} (ID: {section.id})")
# Create a test server
server_data = {
"name": "Test Server",
"section_id": section.id,
"description": "A test server for database operations",
"runtime_definition": {
"type": "docker_image",
"command": "nginx:latest",
"args": [],
"ports": [
{
"containerPort": 80,
"protocol": "TCP"
}
]
},
"settings": {
"environmentVars": {
"SERVER_ENV": "test_server"
}
}
}
server = await crud.create_server(db, server_data)
print(f"Created server: {server.name} (ID: {server.id})")
# Create a test task
task = await crud.create_task(db, "test_task", {"test": "data"})
print(f"Created task: {task.task_id}")
# Update task status
await crud.update_task_status(db, task.task_id, "Completed", {"result": "success"})
print(f"Updated task status to Completed")
# Add audit log
await crud.add_audit_log(db, "test", "section", str(section.id), "test_user", {"action": "test"})
print("Added audit log")
# Retrieve data
retrieved_panel = await crud.get_panel_config(db)
print(f"Retrieved panel: {retrieved_panel.name}")
retrieved_section = await crud.get_section_by_id(db, section.id)
print(f"Retrieved section: {retrieved_section.name}")
retrieved_server = await crud.get_server_by_id(db, server.id)
print(f"Retrieved server: {retrieved_server.name}")
retrieved_task = await crud.get_task(db, task.task_id)
print(f"Retrieved task: {retrieved_task.task_id}, status: {retrieved_task.status}")
# Update server status
await crud.update_server_status(db, server.id, "Running", "test-process-id")
print(f"Updated server status to Running")
updated_server = await crud.get_server_by_id(db, server.id)
print(f"Server status: {updated_server.status}, process ID: {updated_server.process_id}")
# Clean up
print("\nCleaning up test data...")
await crud.delete_server(db, server.id)
print(f"Deleted server: {server.name}")
await crud.delete_section(db, section.id)
print(f"Deleted section: {section.name}")
print("Database operations test completed successfully")
if __name__ == "__main__":
asyncio.run(test_database_operations())