Mike Swanson
06f7617718
Reorganized project structure for better maintainability and reduced disk usage by 95.9% (11 GB -> 451 MB). Directory Reorganization (85% reduction in root files): - Created docs/ with subdirectories (deployment, testing, database, etc.) - Created infrastructure/vpn-configs/ for VPN scripts - Moved 90+ files from root to organized locations - Archived obsolete documentation (context system, offline mode, zombie debugging) - Moved all test files to tests/ directory - Root directory: 119 files -> 18 files Disk Cleanup (10.55 GB recovered): - Deleted Rust build artifacts: 9.6 GB (target/ directories) - Deleted Python virtual environments: 161 MB (venv/ directories) - Deleted Python cache: 50 KB (__pycache__/) New Structure: - docs/ - All documentation organized by category - docs/archives/ - Obsolete but preserved documentation - infrastructure/ - VPN configs and SSH setup - tests/ - All test files consolidated - logs/ - Ready for future logs Benefits: - Cleaner root directory (18 vs 119 files) - Logical organization of documentation - 95.9% disk space reduction - Faster navigation and discovery - Better portability (build artifacts excluded) Build artifacts can be regenerated: - Rust: cargo build --release (5-15 min per project) - Python: pip install -r requirements.txt (2-3 min) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
822 lines
30 KiB
Python
822 lines
30 KiB
Python
"""
|
|
Comprehensive API Endpoint Tests for ClaudeTools FastAPI Application
|
|
|
|
This test suite validates all 5 core API endpoints:
|
|
- Machines
|
|
- Clients
|
|
- Projects
|
|
- Sessions
|
|
- Tags
|
|
|
|
Tests include:
|
|
- API startup and health checks
|
|
- CRUD operations for all entities
|
|
- Authentication (with and without JWT tokens)
|
|
- Pagination parameters
|
|
- Error handling (404, 409, 422 responses)
|
|
"""
|
|
|
|
import sys
|
|
from datetime import timedelta
|
|
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
# Import the FastAPI app and auth utilities
|
|
from api.main import app
|
|
from api.middleware.auth import create_access_token
|
|
|
|
# Create test client
|
|
client = TestClient(app)
|
|
|
|
# Test counters
|
|
tests_passed = 0
|
|
tests_failed = 0
|
|
test_results = []
|
|
|
|
|
|
def log_test(test_name: str, passed: bool, error_msg: str = ""):
|
|
"""Log test result and update counters."""
|
|
global tests_passed, tests_failed
|
|
if passed:
|
|
tests_passed += 1
|
|
status = "PASS"
|
|
symbol = "[+]"
|
|
else:
|
|
tests_failed += 1
|
|
status = "FAIL"
|
|
symbol = "[-]"
|
|
|
|
result = f"{symbol} {status}: {test_name}"
|
|
if error_msg:
|
|
result += f"\n Error: {error_msg}"
|
|
|
|
test_results.append((test_name, passed, error_msg))
|
|
print(result)
|
|
|
|
|
|
def create_test_token():
|
|
"""Create a test JWT token for authentication."""
|
|
token_data = {
|
|
"sub": "test_user@claudetools.com",
|
|
"scopes": ["msp:read", "msp:write", "msp:admin"]
|
|
}
|
|
return create_access_token(token_data, expires_delta=timedelta(hours=1))
|
|
|
|
|
|
def get_auth_headers():
|
|
"""Get authorization headers with test token."""
|
|
token = create_test_token()
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
# ============================================================================
|
|
# SECTION 1: API Health and Startup Tests
|
|
# ============================================================================
|
|
|
|
print("\n" + "="*70)
|
|
print("SECTION 1: API Health and Startup Tests")
|
|
print("="*70 + "\n")
|
|
|
|
def test_root_endpoint():
|
|
"""Test root endpoint returns API status."""
|
|
try:
|
|
response = client.get("/")
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["status"] == "online", f"Expected status 'online', got {data.get('status')}"
|
|
assert "service" in data, "Response missing 'service' field"
|
|
assert "version" in data, "Response missing 'version' field"
|
|
log_test("Root endpoint (/)", True)
|
|
except Exception as e:
|
|
log_test("Root endpoint (/)", False, str(e))
|
|
|
|
def test_health_endpoint():
|
|
"""Test health check endpoint."""
|
|
try:
|
|
response = client.get("/health")
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["status"] == "healthy", f"Expected status 'healthy', got {data.get('status')}"
|
|
log_test("Health check endpoint (/health)", True)
|
|
except Exception as e:
|
|
log_test("Health check endpoint (/health)", False, str(e))
|
|
|
|
def test_jwt_token_creation():
|
|
"""Test JWT token creation."""
|
|
try:
|
|
token = create_test_token()
|
|
assert token is not None, "Token creation returned None"
|
|
assert len(token) > 20, "Token seems too short"
|
|
log_test("JWT token creation", True)
|
|
except Exception as e:
|
|
log_test("JWT token creation", False, str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# SECTION 2: Authentication Tests
|
|
# ============================================================================
|
|
|
|
print("\n" + "="*70)
|
|
print("SECTION 2: Authentication Tests")
|
|
print("="*70 + "\n")
|
|
|
|
def test_unauthenticated_access():
|
|
"""Test that protected endpoints reject requests without auth."""
|
|
try:
|
|
response = client.get("/api/machines")
|
|
# Can be 401 (Unauthorized) or 403 (Forbidden) depending on implementation
|
|
assert response.status_code in [401, 403], f"Expected 401 or 403, got {response.status_code}"
|
|
log_test("Unauthenticated access rejected", True)
|
|
except Exception as e:
|
|
log_test("Unauthenticated access rejected", False, str(e))
|
|
|
|
def test_authenticated_access():
|
|
"""Test that protected endpoints accept valid JWT tokens."""
|
|
try:
|
|
headers = get_auth_headers()
|
|
response = client.get("/api/machines", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
log_test("Authenticated access accepted", True)
|
|
except Exception as e:
|
|
log_test("Authenticated access accepted", False, str(e))
|
|
|
|
def test_invalid_token():
|
|
"""Test that invalid tokens are rejected."""
|
|
try:
|
|
headers = {"Authorization": "Bearer invalid_token_string"}
|
|
response = client.get("/api/machines", headers=headers)
|
|
assert response.status_code == 401, f"Expected 401, got {response.status_code}"
|
|
log_test("Invalid token rejected", True)
|
|
except Exception as e:
|
|
log_test("Invalid token rejected", False, str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# SECTION 3: Machine CRUD Operations
|
|
# ============================================================================
|
|
|
|
print("\n" + "="*70)
|
|
print("SECTION 3: Machine CRUD Operations")
|
|
print("="*70 + "\n")
|
|
|
|
machine_id = None
|
|
|
|
def test_create_machine():
|
|
"""Test creating a new machine."""
|
|
global machine_id
|
|
try:
|
|
headers = get_auth_headers()
|
|
machine_data = {
|
|
"hostname": f"test-machine-{uuid4().hex[:8]}",
|
|
"friendly_name": "Test Machine",
|
|
"machine_type": "laptop",
|
|
"platform": "win32",
|
|
"is_active": True
|
|
}
|
|
response = client.post("/api/machines", json=machine_data, headers=headers)
|
|
assert response.status_code == 201, f"Expected 201, got {response.status_code}. Response: {response.text}"
|
|
data = response.json()
|
|
assert "id" in data, f"Response missing 'id' field. Data: {data}"
|
|
machine_id = data["id"]
|
|
print(f" Created machine with ID: {machine_id}")
|
|
log_test("Create machine", True)
|
|
except Exception as e:
|
|
log_test("Create machine", False, str(e))
|
|
|
|
def test_list_machines():
|
|
"""Test listing machines with pagination."""
|
|
try:
|
|
headers = get_auth_headers()
|
|
response = client.get("/api/machines?skip=0&limit=10", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert "total" in data, "Response missing 'total' field"
|
|
assert "machines" in data, "Response missing 'machines' field"
|
|
assert isinstance(data["machines"], list), "machines field is not a list"
|
|
log_test("List machines", True)
|
|
except Exception as e:
|
|
log_test("List machines", False, str(e))
|
|
|
|
def test_get_machine():
|
|
"""Test retrieving a specific machine by ID."""
|
|
try:
|
|
if machine_id is None:
|
|
raise Exception("No machine_id available (create test may have failed)")
|
|
headers = get_auth_headers()
|
|
print(f" Fetching machine with ID: {machine_id} (type: {type(machine_id)})")
|
|
|
|
# List all machines to check if our machine exists
|
|
list_response = client.get("/api/machines", headers=headers)
|
|
all_machines = list_response.json().get("machines", [])
|
|
print(f" Total machines in DB: {len(all_machines)}")
|
|
if all_machines:
|
|
print(f" First machine ID: {all_machines[0].get('id')} (type: {type(all_machines[0].get('id'))})")
|
|
|
|
response = client.get(f"/api/machines/{machine_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}. Response: {response.text}"
|
|
data = response.json()
|
|
assert str(data["id"]) == str(machine_id), f"Expected ID {machine_id}, got {data.get('id')}"
|
|
log_test("Get machine by ID", True)
|
|
except Exception as e:
|
|
log_test("Get machine by ID", False, str(e))
|
|
|
|
def test_update_machine():
|
|
"""Test updating a machine."""
|
|
try:
|
|
if machine_id is None:
|
|
raise Exception("No machine_id available (create test may have failed)")
|
|
headers = get_auth_headers()
|
|
update_data = {
|
|
"friendly_name": "Updated Test Machine",
|
|
"notes": "Updated during testing"
|
|
}
|
|
response = client.put(f"/api/machines/{machine_id}", json=update_data, headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["friendly_name"] == "Updated Test Machine", "Update not reflected"
|
|
log_test("Update machine", True)
|
|
except Exception as e:
|
|
log_test("Update machine", False, str(e))
|
|
|
|
def test_machine_not_found():
|
|
"""Test getting non-existent machine returns 404."""
|
|
try:
|
|
headers = get_auth_headers()
|
|
fake_id = str(uuid4())
|
|
response = client.get(f"/api/machines/{fake_id}", headers=headers)
|
|
assert response.status_code == 404, f"Expected 404, got {response.status_code}"
|
|
log_test("Machine not found (404)", True)
|
|
except Exception as e:
|
|
log_test("Machine not found (404)", False, str(e))
|
|
|
|
def test_delete_machine():
|
|
"""Test deleting a machine."""
|
|
try:
|
|
if machine_id is None:
|
|
raise Exception("No machine_id available (create test may have failed)")
|
|
headers = get_auth_headers()
|
|
response = client.delete(f"/api/machines/{machine_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
log_test("Delete machine", True)
|
|
except Exception as e:
|
|
log_test("Delete machine", False, str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# SECTION 4: Client CRUD Operations
|
|
# ============================================================================
|
|
|
|
print("\n" + "="*70)
|
|
print("SECTION 4: Client CRUD Operations")
|
|
print("="*70 + "\n")
|
|
|
|
client_id = None
|
|
|
|
def test_create_client():
|
|
"""Test creating a new client."""
|
|
global client_id
|
|
try:
|
|
headers = get_auth_headers()
|
|
client_data = {
|
|
"name": f"Test Client {uuid4().hex[:8]}",
|
|
"type": "msp_client",
|
|
"primary_contact": "John Doe",
|
|
"is_active": True
|
|
}
|
|
response = client.post("/api/clients", json=client_data, headers=headers)
|
|
assert response.status_code == 201, f"Expected 201, got {response.status_code}. Response: {response.text}"
|
|
data = response.json()
|
|
assert "id" in data, f"Response missing 'id' field. Data: {data}"
|
|
client_id = data["id"]
|
|
print(f" Created client with ID: {client_id}")
|
|
log_test("Create client", True)
|
|
except Exception as e:
|
|
log_test("Create client", False, str(e))
|
|
|
|
def test_list_clients():
|
|
"""Test listing clients with pagination."""
|
|
try:
|
|
headers = get_auth_headers()
|
|
response = client.get("/api/clients?skip=0&limit=10", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert "total" in data, "Response missing 'total' field"
|
|
assert "clients" in data, "Response missing 'clients' field"
|
|
log_test("List clients", True)
|
|
except Exception as e:
|
|
log_test("List clients", False, str(e))
|
|
|
|
def test_get_client():
|
|
"""Test retrieving a specific client by ID."""
|
|
try:
|
|
if client_id is None:
|
|
raise Exception("No client_id available")
|
|
headers = get_auth_headers()
|
|
response = client.get(f"/api/clients/{client_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["id"] == client_id, f"Expected ID {client_id}, got {data.get('id')}"
|
|
log_test("Get client by ID", True)
|
|
except Exception as e:
|
|
log_test("Get client by ID", False, str(e))
|
|
|
|
def test_update_client():
|
|
"""Test updating a client."""
|
|
try:
|
|
if client_id is None:
|
|
raise Exception("No client_id available")
|
|
headers = get_auth_headers()
|
|
update_data = {
|
|
"primary_contact": "Jane Doe",
|
|
"notes": "Updated contact"
|
|
}
|
|
response = client.put(f"/api/clients/{client_id}", json=update_data, headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["primary_contact"] == "Jane Doe", "Update not reflected"
|
|
log_test("Update client", True)
|
|
except Exception as e:
|
|
log_test("Update client", False, str(e))
|
|
|
|
def test_delete_client():
|
|
"""Test deleting a client."""
|
|
try:
|
|
if client_id is None:
|
|
raise Exception("No client_id available")
|
|
headers = get_auth_headers()
|
|
response = client.delete(f"/api/clients/{client_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
log_test("Delete client", True)
|
|
except Exception as e:
|
|
log_test("Delete client", False, str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# SECTION 5: Project CRUD Operations
|
|
# ============================================================================
|
|
|
|
print("\n" + "="*70)
|
|
print("SECTION 5: Project CRUD Operations")
|
|
print("="*70 + "\n")
|
|
|
|
project_id = None
|
|
project_client_id = None
|
|
|
|
def test_create_project():
|
|
"""Test creating a new project."""
|
|
global project_id, project_client_id
|
|
try:
|
|
headers = get_auth_headers()
|
|
|
|
# First create a client for the project
|
|
client_data = {
|
|
"name": f"Project Test Client {uuid4().hex[:8]}",
|
|
"type": "msp_client",
|
|
"is_active": True
|
|
}
|
|
client_response = client.post("/api/clients", json=client_data, headers=headers)
|
|
assert client_response.status_code == 201, f"Failed to create test client: {client_response.text}"
|
|
project_client_id = client_response.json()["id"]
|
|
|
|
# Now create the project
|
|
project_data = {
|
|
"name": f"Test Project {uuid4().hex[:8]}",
|
|
"client_id": project_client_id,
|
|
"status": "active"
|
|
}
|
|
response = client.post("/api/projects", json=project_data, headers=headers)
|
|
assert response.status_code == 201, f"Expected 201, got {response.status_code}. Response: {response.text}"
|
|
data = response.json()
|
|
assert "id" in data, f"Response missing 'id' field. Data: {data}"
|
|
project_id = data["id"]
|
|
print(f" Created project with ID: {project_id}")
|
|
log_test("Create project", True)
|
|
except Exception as e:
|
|
log_test("Create project", False, str(e))
|
|
|
|
def test_list_projects():
|
|
"""Test listing projects with pagination."""
|
|
try:
|
|
headers = get_auth_headers()
|
|
response = client.get("/api/projects?skip=0&limit=10", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert "total" in data, "Response missing 'total' field"
|
|
assert "projects" in data, "Response missing 'projects' field"
|
|
log_test("List projects", True)
|
|
except Exception as e:
|
|
log_test("List projects", False, str(e))
|
|
|
|
def test_get_project():
|
|
"""Test retrieving a specific project by ID."""
|
|
try:
|
|
if project_id is None:
|
|
raise Exception("No project_id available")
|
|
headers = get_auth_headers()
|
|
response = client.get(f"/api/projects/{project_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["id"] == project_id, f"Expected ID {project_id}, got {data.get('id')}"
|
|
log_test("Get project by ID", True)
|
|
except Exception as e:
|
|
log_test("Get project by ID", False, str(e))
|
|
|
|
def test_update_project():
|
|
"""Test updating a project."""
|
|
try:
|
|
if project_id is None:
|
|
raise Exception("No project_id available")
|
|
headers = get_auth_headers()
|
|
update_data = {
|
|
"status": "completed",
|
|
"notes": "Project completed during testing"
|
|
}
|
|
response = client.put(f"/api/projects/{project_id}", json=update_data, headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["status"] == "completed", "Update not reflected"
|
|
log_test("Update project", True)
|
|
except Exception as e:
|
|
log_test("Update project", False, str(e))
|
|
|
|
def test_delete_project():
|
|
"""Test deleting a project."""
|
|
try:
|
|
if project_id is None:
|
|
raise Exception("No project_id available")
|
|
headers = get_auth_headers()
|
|
response = client.delete(f"/api/projects/{project_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
|
|
# Clean up test client
|
|
if project_client_id:
|
|
client.delete(f"/api/clients/{project_client_id}", headers=headers)
|
|
|
|
log_test("Delete project", True)
|
|
except Exception as e:
|
|
log_test("Delete project", False, str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# SECTION 6: Session CRUD Operations
|
|
# ============================================================================
|
|
|
|
print("\n" + "="*70)
|
|
print("SECTION 6: Session CRUD Operations")
|
|
print("="*70 + "\n")
|
|
|
|
session_id = None
|
|
session_client_id = None
|
|
session_project_id = None
|
|
|
|
def test_create_session():
|
|
"""Test creating a new session."""
|
|
global session_id, session_client_id, session_project_id
|
|
try:
|
|
headers = get_auth_headers()
|
|
|
|
# Create client for session
|
|
client_data = {
|
|
"name": f"Session Test Client {uuid4().hex[:8]}",
|
|
"type": "msp_client",
|
|
"is_active": True
|
|
}
|
|
client_response = client.post("/api/clients", json=client_data, headers=headers)
|
|
assert client_response.status_code == 201, f"Failed to create test client: {client_response.text}"
|
|
session_client_id = client_response.json()["id"]
|
|
|
|
# Create project for session
|
|
project_data = {
|
|
"name": f"Session Test Project {uuid4().hex[:8]}",
|
|
"client_id": session_client_id,
|
|
"status": "active"
|
|
}
|
|
project_response = client.post("/api/projects", json=project_data, headers=headers)
|
|
assert project_response.status_code == 201, f"Failed to create test project: {project_response.text}"
|
|
session_project_id = project_response.json()["id"]
|
|
|
|
# Create session
|
|
from datetime import date
|
|
session_data = {
|
|
"session_title": f"Test Session {uuid4().hex[:8]}",
|
|
"session_date": str(date.today()),
|
|
"client_id": session_client_id,
|
|
"project_id": session_project_id,
|
|
"status": "completed"
|
|
}
|
|
response = client.post("/api/sessions", json=session_data, headers=headers)
|
|
assert response.status_code == 201, f"Expected 201, got {response.status_code}. Response: {response.text}"
|
|
data = response.json()
|
|
assert "id" in data, f"Response missing 'id' field. Data: {data}"
|
|
session_id = data["id"]
|
|
print(f" Created session with ID: {session_id}")
|
|
log_test("Create session", True)
|
|
except Exception as e:
|
|
log_test("Create session", False, str(e))
|
|
|
|
def test_list_sessions():
|
|
"""Test listing sessions with pagination."""
|
|
try:
|
|
headers = get_auth_headers()
|
|
response = client.get("/api/sessions?skip=0&limit=10", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert "total" in data, "Response missing 'total' field"
|
|
assert "sessions" in data, "Response missing 'sessions' field"
|
|
log_test("List sessions", True)
|
|
except Exception as e:
|
|
log_test("List sessions", False, str(e))
|
|
|
|
def test_get_session():
|
|
"""Test retrieving a specific session by ID."""
|
|
try:
|
|
if session_id is None:
|
|
raise Exception("No session_id available")
|
|
headers = get_auth_headers()
|
|
response = client.get(f"/api/sessions/{session_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["id"] == session_id, f"Expected ID {session_id}, got {data.get('id')}"
|
|
log_test("Get session by ID", True)
|
|
except Exception as e:
|
|
log_test("Get session by ID", False, str(e))
|
|
|
|
def test_update_session():
|
|
"""Test updating a session."""
|
|
try:
|
|
if session_id is None:
|
|
raise Exception("No session_id available")
|
|
headers = get_auth_headers()
|
|
update_data = {
|
|
"status": "completed",
|
|
"summary": "Test session completed"
|
|
}
|
|
response = client.put(f"/api/sessions/{session_id}", json=update_data, headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["status"] == "completed", "Update not reflected"
|
|
log_test("Update session", True)
|
|
except Exception as e:
|
|
log_test("Update session", False, str(e))
|
|
|
|
def test_delete_session():
|
|
"""Test deleting a session."""
|
|
try:
|
|
if session_id is None:
|
|
raise Exception("No session_id available")
|
|
headers = get_auth_headers()
|
|
response = client.delete(f"/api/sessions/{session_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
|
|
# Clean up test data
|
|
if session_project_id:
|
|
client.delete(f"/api/projects/{session_project_id}", headers=headers)
|
|
if session_client_id:
|
|
client.delete(f"/api/clients/{session_client_id}", headers=headers)
|
|
|
|
log_test("Delete session", True)
|
|
except Exception as e:
|
|
log_test("Delete session", False, str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# SECTION 7: Tag CRUD Operations
|
|
# ============================================================================
|
|
|
|
print("\n" + "="*70)
|
|
print("SECTION 7: Tag CRUD Operations")
|
|
print("="*70 + "\n")
|
|
|
|
tag_id = None
|
|
|
|
def test_create_tag():
|
|
"""Test creating a new tag."""
|
|
global tag_id
|
|
try:
|
|
headers = get_auth_headers()
|
|
tag_data = {
|
|
"name": f"test-tag-{uuid4().hex[:8]}",
|
|
"category": "technology",
|
|
"color": "#FF5733"
|
|
}
|
|
response = client.post("/api/tags", json=tag_data, headers=headers)
|
|
assert response.status_code == 201, f"Expected 201, got {response.status_code}"
|
|
data = response.json()
|
|
assert "id" in data, "Response missing 'id' field"
|
|
tag_id = data["id"]
|
|
log_test("Create tag", True)
|
|
except Exception as e:
|
|
log_test("Create tag", False, str(e))
|
|
|
|
def test_list_tags():
|
|
"""Test listing tags with pagination."""
|
|
try:
|
|
headers = get_auth_headers()
|
|
response = client.get("/api/tags?skip=0&limit=10", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert "total" in data, "Response missing 'total' field"
|
|
assert "tags" in data, "Response missing 'tags' field"
|
|
log_test("List tags", True)
|
|
except Exception as e:
|
|
log_test("List tags", False, str(e))
|
|
|
|
def test_get_tag():
|
|
"""Test retrieving a specific tag by ID."""
|
|
try:
|
|
if tag_id is None:
|
|
raise Exception("No tag_id available")
|
|
headers = get_auth_headers()
|
|
response = client.get(f"/api/tags/{tag_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["id"] == tag_id, f"Expected ID {tag_id}, got {data.get('id')}"
|
|
log_test("Get tag by ID", True)
|
|
except Exception as e:
|
|
log_test("Get tag by ID", False, str(e))
|
|
|
|
def test_update_tag():
|
|
"""Test updating a tag."""
|
|
try:
|
|
if tag_id is None:
|
|
raise Exception("No tag_id available")
|
|
headers = get_auth_headers()
|
|
update_data = {
|
|
"color": "#00FF00",
|
|
"description": "Updated test tag"
|
|
}
|
|
response = client.put(f"/api/tags/{tag_id}", json=update_data, headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["color"] == "#00FF00", "Update not reflected"
|
|
log_test("Update tag", True)
|
|
except Exception as e:
|
|
log_test("Update tag", False, str(e))
|
|
|
|
def test_tag_duplicate_name():
|
|
"""Test creating tag with duplicate name returns 409."""
|
|
try:
|
|
if tag_id is None:
|
|
raise Exception("No tag_id available")
|
|
headers = get_auth_headers()
|
|
|
|
# Get existing tag name
|
|
existing_response = client.get(f"/api/tags/{tag_id}", headers=headers)
|
|
existing_name = existing_response.json()["name"]
|
|
|
|
# Try to create duplicate
|
|
duplicate_data = {
|
|
"name": existing_name,
|
|
"category": "test"
|
|
}
|
|
response = client.post("/api/tags", json=duplicate_data, headers=headers)
|
|
assert response.status_code == 409, f"Expected 409, got {response.status_code}"
|
|
log_test("Tag duplicate name (409)", True)
|
|
except Exception as e:
|
|
log_test("Tag duplicate name (409)", False, str(e))
|
|
|
|
def test_delete_tag():
|
|
"""Test deleting a tag."""
|
|
try:
|
|
if tag_id is None:
|
|
raise Exception("No tag_id available")
|
|
headers = get_auth_headers()
|
|
response = client.delete(f"/api/tags/{tag_id}", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
log_test("Delete tag", True)
|
|
except Exception as e:
|
|
log_test("Delete tag", False, str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# SECTION 8: Pagination Tests
|
|
# ============================================================================
|
|
|
|
print("\n" + "="*70)
|
|
print("SECTION 8: Pagination Tests")
|
|
print("="*70 + "\n")
|
|
|
|
def test_pagination_skip_limit():
|
|
"""Test pagination with skip and limit parameters."""
|
|
try:
|
|
headers = get_auth_headers()
|
|
response = client.get("/api/machines?skip=0&limit=5", headers=headers)
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
|
data = response.json()
|
|
assert data["skip"] == 0, f"Expected skip=0, got {data.get('skip')}"
|
|
assert data["limit"] == 5, f"Expected limit=5, got {data.get('limit')}"
|
|
log_test("Pagination skip/limit parameters", True)
|
|
except Exception as e:
|
|
log_test("Pagination skip/limit parameters", False, str(e))
|
|
|
|
def test_pagination_max_limit():
|
|
"""Test that pagination enforces maximum limit."""
|
|
try:
|
|
headers = get_auth_headers()
|
|
# Try to request more than max (1000)
|
|
response = client.get("/api/machines?limit=2000", headers=headers)
|
|
# Should either return 422 or clamp to max
|
|
assert response.status_code in [200, 422], f"Unexpected status {response.status_code}"
|
|
log_test("Pagination max limit enforcement", True)
|
|
except Exception as e:
|
|
log_test("Pagination max limit enforcement", False, str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# Run All Tests
|
|
# ============================================================================
|
|
|
|
def run_all_tests():
|
|
"""Run all test functions."""
|
|
print("\n" + "="*70)
|
|
print("CLAUDETOOLS API ENDPOINT TESTS")
|
|
print("="*70)
|
|
|
|
# Section 1: Health
|
|
test_root_endpoint()
|
|
test_health_endpoint()
|
|
test_jwt_token_creation()
|
|
|
|
# Section 2: Auth
|
|
test_unauthenticated_access()
|
|
test_authenticated_access()
|
|
test_invalid_token()
|
|
|
|
# Section 3: Machines
|
|
test_create_machine()
|
|
test_list_machines()
|
|
test_get_machine()
|
|
test_update_machine()
|
|
test_machine_not_found()
|
|
test_delete_machine()
|
|
|
|
# Section 4: Clients
|
|
test_create_client()
|
|
test_list_clients()
|
|
test_get_client()
|
|
test_update_client()
|
|
test_delete_client()
|
|
|
|
# Section 5: Projects
|
|
test_create_project()
|
|
test_list_projects()
|
|
test_get_project()
|
|
test_update_project()
|
|
test_delete_project()
|
|
|
|
# Section 6: Sessions
|
|
test_create_session()
|
|
test_list_sessions()
|
|
test_get_session()
|
|
test_update_session()
|
|
test_delete_session()
|
|
|
|
# Section 7: Tags
|
|
test_create_tag()
|
|
test_list_tags()
|
|
test_get_tag()
|
|
test_update_tag()
|
|
test_tag_duplicate_name()
|
|
test_delete_tag()
|
|
|
|
# Section 8: Pagination
|
|
test_pagination_skip_limit()
|
|
test_pagination_max_limit()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("\n>> Starting ClaudeTools API Test Suite...")
|
|
|
|
try:
|
|
run_all_tests()
|
|
|
|
# Print summary
|
|
print("\n" + "="*70)
|
|
print("TEST SUMMARY")
|
|
print("="*70)
|
|
print(f"\nTotal Tests: {tests_passed + tests_failed}")
|
|
print(f"Passed: {tests_passed}")
|
|
print(f"Failed: {tests_failed}")
|
|
|
|
if tests_failed > 0:
|
|
print("\nFAILED TESTS:")
|
|
for name, passed, error in test_results:
|
|
if not passed:
|
|
print(f" - {name}")
|
|
if error:
|
|
print(f" Error: {error}")
|
|
|
|
if tests_failed == 0:
|
|
print("\n>> All tests passed!")
|
|
sys.exit(0)
|
|
else:
|
|
print(f"\n>> {tests_failed} test(s) failed")
|
|
sys.exit(1)
|
|
|
|
except Exception as e:
|
|
print(f"\n>> Fatal error running tests: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|