Create API layer for web interface

This commit is contained in:
Abdullah Ceylan
2025-09-27 17:15:15 +03:00
parent 3b238213fc
commit 5f438ab1df
7 changed files with 2586 additions and 0 deletions
+671
View File
@@ -0,0 +1,671 @@
"""
Flask API routes for the ComfyUI Queue Manager web interface.
"""
from __future__ import annotations
import json
import logging
from datetime import datetime
from typing import Any
from flask import Flask, jsonify, request, send_from_directory
from werkzeug.exceptions import BadRequest, NotFound
from models import QueueFilter, QueueStatus
from queue_service import QueueService, QueueServiceError
logger = logging.getLogger(__name__)
class QueueManagerAPI:
"""Flask API for queue management operations."""
def __init__(self, queue_service: QueueService | None = None):
"""Initialize the API with a queue service.
Args:
queue_service: Queue service instance. If None, creates a new one.
"""
self.app = Flask(__name__, static_folder='web', static_url_path='/static')
self.queue_service = queue_service or QueueService()
# Configure Flask
self.app.config['JSON_SORT_KEYS'] = False
self.app.config['JSONIFY_PRETTYPRINT_REGULAR'] = True
# Register routes
self._register_routes()
self._register_error_handlers()
logger.info("Queue Manager API initialized")
def _register_routes(self) -> None:
"""Register all API routes."""
# Health check
self.app.route('/health', methods=['GET'])(self.health_check)
# Static files
self.app.route('/', methods=['GET'])(self.serve_index)
self.app.route('/index.html', methods=['GET'])(self.serve_index)
# Queue management endpoints
self.app.route('/api/queue/items', methods=['GET'])(self.get_queue_items)
self.app.route('/api/queue/items', methods=['POST'])(self.add_queue_item)
self.app.route('/api/queue/items/<item_id>', methods=['GET'])(self.get_queue_item)
self.app.route('/api/queue/items/<item_id>', methods=['PUT'])(self.update_queue_item)
self.app.route('/api/queue/items/<item_id>', methods=['DELETE'])(self.delete_queue_item)
self.app.route('/api/queue/items/bulk', methods=['DELETE'])(self.bulk_delete_items)
self.app.route('/api/queue/items/filter', methods=['POST'])(self.filter_queue_items)
self.app.route('/api/queue/search', methods=['GET'])(self.search_queue_items)
# Control endpoints
self.app.route('/api/queue/status', methods=['GET'])(self.get_queue_status)
self.app.route('/api/queue/pause', methods=['POST'])(self.pause_queue)
self.app.route('/api/queue/resume', methods=['POST'])(self.resume_queue)
self.app.route('/api/queue/archive', methods=['POST'])(self.archive_items)
self.app.route('/api/queue/restore', methods=['POST'])(self.restore_items)
self.app.route('/api/queue/export', methods=['POST'])(self.export_queue)
self.app.route('/api/queue/import', methods=['POST'])(self.import_queue)
self.app.route('/api/queue/config', methods=['GET'])(self.get_queue_config)
self.app.route('/api/queue/config', methods=['PUT'])(self.update_queue_config)
def _register_error_handlers(self) -> None:
"""Register error handlers for the Flask app."""
@self.app.errorhandler(QueueServiceError)
def handle_queue_service_error(error: QueueServiceError) -> tuple[dict[str, Any], int]:
"""Handle queue service errors."""
logger.error(f"Queue service error: {error}")
return {
'error': 'Queue service error',
'message': str(error),
'timestamp': datetime.utcnow().isoformat()
}, 500
@self.app.errorhandler(BadRequest)
def handle_bad_request(error: BadRequest) -> tuple[dict[str, Any], int]:
"""Handle bad request errors."""
logger.warning(f"Bad request: {error}")
return {
'error': 'Bad request',
'message': str(error.description),
'timestamp': datetime.utcnow().isoformat()
}, 400
@self.app.errorhandler(NotFound)
def handle_not_found(error: NotFound) -> tuple[dict[str, Any], int]:
"""Handle not found errors."""
return {
'error': 'Not found',
'message': 'The requested resource was not found',
'timestamp': datetime.utcnow().isoformat()
}, 404
@self.app.errorhandler(Exception)
def handle_generic_error(error: Exception) -> tuple[dict[str, Any], int]:
"""Handle generic errors."""
logger.error(f"Unexpected error: {error}", exc_info=True)
return {
'error': 'Internal server error',
'message': 'An unexpected error occurred',
'timestamp': datetime.utcnow().isoformat()
}, 500
def health_check(self) -> dict[str, Any]:
"""Health check endpoint."""
try:
# Test queue service connectivity
queue_state = self.queue_service.get_queue_state()
return {
'status': 'healthy',
'timestamp': datetime.utcnow().isoformat(),
'queue_state': queue_state.value,
'version': '1.0.0'
}
except Exception as e:
logger.error(f"Health check failed: {e}")
return {
'status': 'unhealthy',
'timestamp': datetime.utcnow().isoformat(),
'error': str(e)
}, 503
def serve_index(self) -> str:
"""Serve the main index.html file."""
try:
return send_from_directory(self.app.static_folder, 'index.html')
except Exception as e:
logger.error(f"Failed to serve index.html: {e}")
return jsonify({
'error': 'Failed to serve index page',
'message': str(e)
}), 500
def _validate_json_request(self) -> dict[str, Any]:
"""Validate and parse JSON request data."""
if not request.is_json:
raise BadRequest("Request must be JSON")
try:
data = request.get_json()
if data is None:
raise BadRequest("Invalid JSON data")
return data
except Exception as e:
raise BadRequest(f"Failed to parse JSON: {e}")
def _parse_queue_filter(self, params: dict[str, Any]) -> QueueFilter:
"""Parse queue filter parameters from request."""
filter_criteria = QueueFilter()
# Parse status filter
if 'status' in params:
status_values = params['status']
if isinstance(status_values, str):
status_values = [status_values]
try:
filter_criteria.status = [QueueStatus(status) for status in status_values]
except ValueError as e:
raise BadRequest(f"Invalid status value: {e}")
# Parse workflow name filter
if 'workflow_name' in params:
filter_criteria.workflow_name = params['workflow_name']
# Parse search term
if 'search' in params:
filter_criteria.search_term = params['search']
# Parse date range
if 'date_from' in params or 'date_to' in params:
try:
date_from = None
date_to = None
if 'date_from' in params:
date_from = datetime.fromisoformat(params['date_from'])
if 'date_to' in params:
date_to = datetime.fromisoformat(params['date_to'])
if date_from and date_to:
filter_criteria.date_range = (date_from, date_to)
except ValueError as e:
raise BadRequest(f"Invalid date format: {e}")
return filter_criteria
def run(self, host: str = '127.0.0.1', port: int = 5000, debug: bool = False) -> None:
"""Run the Flask development server.
Args:
host: Host to bind to
port: Port to bind to
debug: Enable debug mode
"""
logger.info(f"Starting Queue Manager API server on {host}:{port}")
self.app.run(host=host, port=port, debug=debug)
def get_queue_items(self) -> dict[str, Any]:
"""Get all queue items with optional status filtering."""
try:
# Parse query parameters
status_param = request.args.get('status')
status = None
if status_param:
try:
status = QueueStatus(status_param)
except ValueError:
raise BadRequest(f"Invalid status: {status_param}")
# Get items from service
items = self.queue_service.get_queue_items(status)
return {
'items': [item.to_dict() for item in items],
'count': len(items),
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to get queue items: {e}")
raise
def add_queue_item(self) -> tuple[dict[str, Any], int]:
"""Add a new workflow to the queue."""
try:
data = self._validate_json_request()
# Validate required fields
if 'workflow_data' not in data:
raise BadRequest("Missing required field: workflow_data")
workflow_data = data['workflow_data']
workflow_name = data.get('workflow_name', '')
# Add to queue
item_id = self.queue_service.add_workflow(workflow_data, workflow_name)
# Get the created item
item = self.queue_service.get_queue_item(item_id)
if not item:
raise QueueServiceError("Failed to retrieve created item")
return {
'item': item.to_dict(),
'message': 'Workflow added to queue successfully',
'timestamp': datetime.utcnow().isoformat()
}, 201
except QueueServiceError as e:
logger.error(f"Failed to add queue item: {e}")
raise
def get_queue_item(self, item_id: str) -> dict[str, Any]:
"""Get a specific queue item by ID."""
try:
if not item_id:
raise BadRequest("Item ID is required")
item = self.queue_service.get_queue_item(item_id)
if not item:
raise NotFound(f"Queue item {item_id} not found")
return {
'item': item.to_dict(),
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to get queue item {item_id}: {e}")
raise
def update_queue_item(self, item_id: str) -> dict[str, Any]:
"""Update a queue item's status and related fields."""
try:
if not item_id:
raise BadRequest("Item ID is required")
data = self._validate_json_request()
# Validate status if provided
status = None
if 'status' in data:
try:
status = QueueStatus(data['status'])
except ValueError:
raise BadRequest(f"Invalid status: {data['status']}")
error_message = data.get('error_message')
result_data = data.get('result_data')
# Update the item
if status:
success = self.queue_service.update_item_status(
item_id, status, error_message, result_data
)
else:
# If no status provided, just check that item exists
item = self.queue_service.get_queue_item(item_id)
if not item:
raise NotFound(f"Queue item {item_id} not found")
success = True
if not success:
raise QueueServiceError("Failed to update queue item")
# Get updated item
updated_item = self.queue_service.get_queue_item(item_id)
return {
'item': updated_item.to_dict() if updated_item else None,
'message': 'Queue item updated successfully',
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to update queue item {item_id}: {e}")
raise
def delete_queue_item(self, item_id: str) -> dict[str, Any]:
"""Delete a specific queue item."""
try:
if not item_id:
raise BadRequest("Item ID is required")
# Check if item exists
item = self.queue_service.get_queue_item(item_id)
if not item:
raise NotFound(f"Queue item {item_id} not found")
# Delete the item
success = self.queue_service.delete_items([item_id])
if not success:
raise QueueServiceError("Failed to delete queue item")
return {
'message': f'Queue item {item_id} deleted successfully',
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to delete queue item {item_id}: {e}")
raise
def bulk_delete_items(self) -> dict[str, Any]:
"""Delete multiple queue items."""
try:
data = self._validate_json_request()
if 'item_ids' not in data:
raise BadRequest("Missing required field: item_ids")
item_ids = data['item_ids']
if not isinstance(item_ids, list):
raise BadRequest("item_ids must be a list")
if not item_ids:
raise BadRequest("item_ids cannot be empty")
# Delete the items
success = self.queue_service.delete_items(item_ids)
if not success:
raise QueueServiceError("Failed to delete some queue items")
return {
'message': f'Successfully deleted {len(item_ids)} queue items',
'deleted_count': len(item_ids),
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to bulk delete queue items: {e}")
raise
def filter_queue_items(self) -> dict[str, Any]:
"""Filter queue items based on criteria."""
try:
data = self._validate_json_request()
# Parse filter criteria
filter_criteria = self._parse_queue_filter(data)
# Apply filter
items = self.queue_service.filter_items(filter_criteria)
return {
'items': [item.to_dict() for item in items],
'count': len(items),
'filter': filter_criteria.to_dict(),
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to filter queue items: {e}")
raise
def search_queue_items(self) -> dict[str, Any]:
"""Search queue items using a query string."""
try:
query = request.args.get('q', '').strip()
if not query:
raise BadRequest("Search query parameter 'q' is required")
# Perform search
items = self.queue_service.search_items(query)
return {
'items': [item.to_dict() for item in items],
'count': len(items),
'query': query,
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to search queue items: {e}")
raise
def get_queue_status(self) -> dict[str, Any]:
"""Get the current queue processing status."""
try:
queue_state = self.queue_service.get_queue_state()
config = self.queue_service.get_config()
# Get queue statistics
all_items = self.queue_service.get_queue_items()
stats = {
'total': len(all_items),
'pending': len([item for item in all_items if item.status == QueueStatus.PENDING]),
'running': len([item for item in all_items if item.status == QueueStatus.RUNNING]),
'completed': len([item for item in all_items if item.status == QueueStatus.COMPLETED]),
'failed': len([item for item in all_items if item.status == QueueStatus.FAILED]),
'archived': len([item for item in all_items if item.status == QueueStatus.ARCHIVED])
}
return {
'queue_state': queue_state.value,
'statistics': stats,
'config': config.to_dict(),
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to get queue status: {e}")
raise
def pause_queue(self) -> dict[str, Any]:
"""Pause queue processing."""
try:
success = self.queue_service.pause_queue()
if not success:
raise QueueServiceError("Failed to pause queue")
return {
'message': 'Queue processing paused successfully',
'queue_state': 'paused',
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to pause queue: {e}")
raise
def resume_queue(self) -> dict[str, Any]:
"""Resume queue processing."""
try:
success = self.queue_service.resume_queue()
if not success:
raise QueueServiceError("Failed to resume queue")
return {
'message': 'Queue processing resumed successfully',
'queue_state': 'running',
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to resume queue: {e}")
raise
def archive_items(self) -> dict[str, Any]:
"""Archive queue items."""
try:
data = self._validate_json_request()
if 'item_ids' not in data:
raise BadRequest("Missing required field: item_ids")
item_ids = data['item_ids']
if not isinstance(item_ids, list):
raise BadRequest("item_ids must be a list")
if not item_ids:
raise BadRequest("item_ids cannot be empty")
# Archive the items
success = self.queue_service.archive_items(item_ids)
if not success:
raise QueueServiceError("Failed to archive some items")
return {
'message': f'Successfully archived {len(item_ids)} items',
'archived_count': len(item_ids),
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to archive items: {e}")
raise
def restore_items(self) -> dict[str, Any]:
"""Restore archived queue items."""
try:
data = self._validate_json_request()
if 'item_ids' not in data:
raise BadRequest("Missing required field: item_ids")
item_ids = data['item_ids']
if not isinstance(item_ids, list):
raise BadRequest("item_ids must be a list")
if not item_ids:
raise BadRequest("item_ids cannot be empty")
# Restore the items
success = self.queue_service.restore_items(item_ids)
if not success:
raise QueueServiceError("Failed to restore some items")
return {
'message': f'Successfully restored {len(item_ids)} items',
'restored_count': len(item_ids),
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to restore items: {e}")
raise
def export_queue(self) -> dict[str, Any]:
"""Export queue items."""
try:
data = self._validate_json_request()
# Get optional item IDs to export
item_ids = data.get('item_ids')
if item_ids is not None and not isinstance(item_ids, list):
raise BadRequest("item_ids must be a list")
# Export the queue
export_data = self.queue_service.export_queue(item_ids)
return {
'message': 'Queue exported successfully',
'export_data': export_data,
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to export queue: {e}")
raise
def import_queue(self) -> dict[str, Any]:
"""Import queue items."""
try:
data = self._validate_json_request()
if 'queue_data' not in data:
raise BadRequest("Missing required field: queue_data")
queue_data = data['queue_data']
merge = data.get('merge', True) # Default to merge mode
if not isinstance(merge, bool):
raise BadRequest("merge must be a boolean")
# Import the queue
success = self.queue_service.import_queue(queue_data, merge)
if not success:
raise QueueServiceError("Failed to import queue data")
return {
'message': 'Queue imported successfully',
'merge_mode': merge,
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to import queue: {e}")
raise
def get_queue_config(self) -> dict[str, Any]:
"""Get the current queue configuration."""
try:
config = self.queue_service.get_config()
return {
'config': config.to_dict(),
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to get queue config: {e}")
raise
def update_queue_config(self) -> dict[str, Any]:
"""Update the queue configuration."""
try:
data = self._validate_json_request()
# Get current config and update with provided values
current_config = self.queue_service.get_config()
# Update config fields if provided
if 'queue_state' in data:
try:
from models import QueueState
current_config.queue_state = QueueState(data['queue_state'])
except ValueError:
raise BadRequest(f"Invalid queue_state: {data['queue_state']}")
if 'max_concurrent_workflows' in data:
max_concurrent = data['max_concurrent_workflows']
if not isinstance(max_concurrent, int) or max_concurrent < 1:
raise BadRequest("max_concurrent_workflows must be a positive integer")
current_config.max_concurrent_workflows = max_concurrent
if 'auto_archive_completed' in data:
if not isinstance(data['auto_archive_completed'], bool):
raise BadRequest("auto_archive_completed must be a boolean")
current_config.auto_archive_completed = data['auto_archive_completed']
if 'auto_archive_days' in data:
auto_archive_days = data['auto_archive_days']
if not isinstance(auto_archive_days, int) or auto_archive_days < 1:
raise BadRequest("auto_archive_days must be a positive integer")
current_config.auto_archive_days = auto_archive_days
# Update the configuration
success = self.queue_service.update_config(current_config)
if not success:
raise QueueServiceError("Failed to update queue configuration")
return {
'message': 'Queue configuration updated successfully',
'config': current_config.to_dict(),
'timestamp': datetime.utcnow().isoformat()
}
except QueueServiceError as e:
logger.error(f"Failed to update queue config: {e}")
raise
def get_app(self) -> Flask:
"""Get the Flask app instance for integration with other servers."""
return self.app
+350
View File
@@ -0,0 +1,350 @@
"""
Tests for queue management API endpoints.
"""
import json
import pytest
from unittest.mock import Mock
from datetime import datetime, timezone
from api_routes import QueueManagerAPI
from models import QueueItem, QueueState, QueueStatus
from queue_service import QueueService, QueueServiceError
class TestQueueManagementEndpoints:
"""Test cases for queue management API endpoints."""
@pytest.fixture
def mock_queue_service(self):
"""Create a mock queue service for testing."""
service = Mock(spec=QueueService)
service.get_queue_state.return_value = QueueState.RUNNING
return service
@pytest.fixture
def api(self, mock_queue_service):
"""Create a QueueManagerAPI instance for testing."""
return QueueManagerAPI(queue_service=mock_queue_service)
@pytest.fixture
def client(self, api):
"""Create a test client for the Flask app."""
api.app.config['TESTING'] = True
return api.app.test_client()
@pytest.fixture
def sample_queue_item(self):
"""Create a sample queue item for testing."""
return QueueItem(
id="test-item-1",
workflow_name="Test Workflow",
workflow_data={"nodes": {"1": {"class_type": "TestNode"}}},
status=QueueStatus.PENDING
)
def test_get_queue_items_all(self, client, mock_queue_service, sample_queue_item):
"""Test getting all queue items."""
mock_queue_service.get_queue_items.return_value = [sample_queue_item]
response = client.get('/api/queue/items')
assert response.status_code == 200
data = json.loads(response.data)
assert data['count'] == 1
assert len(data['items']) == 1
assert data['items'][0]['id'] == 'test-item-1'
assert 'timestamp' in data
mock_queue_service.get_queue_items.assert_called_once_with(None)
def test_get_queue_items_with_status_filter(self, client, mock_queue_service, sample_queue_item):
"""Test getting queue items with status filter."""
mock_queue_service.get_queue_items.return_value = [sample_queue_item]
response = client.get('/api/queue/items?status=pending')
assert response.status_code == 200
data = json.loads(response.data)
assert data['count'] == 1
mock_queue_service.get_queue_items.assert_called_once_with(QueueStatus.PENDING)
def test_get_queue_items_invalid_status(self, client, mock_queue_service):
"""Test getting queue items with invalid status filter."""
response = client.get('/api/queue/items?status=invalid_status')
assert response.status_code == 400
data = json.loads(response.data)
assert data['error'] == 'Bad request'
def test_add_queue_item_success(self, client, mock_queue_service, sample_queue_item):
"""Test adding a new queue item successfully."""
mock_queue_service.add_workflow.return_value = "test-item-1"
mock_queue_service.get_queue_item.return_value = sample_queue_item
request_data = {
'workflow_name': 'Test Workflow',
'workflow_data': {'nodes': {'1': {'class_type': 'TestNode'}}}
}
response = client.post('/api/queue/items',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 201
data = json.loads(response.data)
assert data['item']['id'] == 'test-item-1'
assert data['message'] == 'Workflow added to queue successfully'
assert 'timestamp' in data
mock_queue_service.add_workflow.assert_called_once_with(
request_data['workflow_data'],
request_data['workflow_name']
)
def test_add_queue_item_missing_workflow_data(self, client, mock_queue_service):
"""Test adding a queue item without workflow data."""
request_data = {'workflow_name': 'Test Workflow'}
response = client.post('/api/queue/items',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 400
data = json.loads(response.data)
assert data['error'] == 'Bad request'
assert 'workflow_data' in data['message']
def test_add_queue_item_invalid_json(self, client, mock_queue_service):
"""Test adding a queue item with invalid JSON."""
response = client.post('/api/queue/items',
data='invalid json',
content_type='application/json')
assert response.status_code == 400
def test_get_queue_item_success(self, client, mock_queue_service, sample_queue_item):
"""Test getting a specific queue item successfully."""
mock_queue_service.get_queue_item.return_value = sample_queue_item
response = client.get('/api/queue/items/test-item-1')
assert response.status_code == 200
data = json.loads(response.data)
assert data['item']['id'] == 'test-item-1'
assert 'timestamp' in data
mock_queue_service.get_queue_item.assert_called_once_with('test-item-1')
def test_get_queue_item_not_found(self, client, mock_queue_service):
"""Test getting a non-existent queue item."""
mock_queue_service.get_queue_item.return_value = None
response = client.get('/api/queue/items/nonexistent')
assert response.status_code == 404
data = json.loads(response.data)
assert data['error'] == 'Not found'
def test_update_queue_item_status(self, client, mock_queue_service, sample_queue_item):
"""Test updating a queue item's status."""
mock_queue_service.update_item_status.return_value = True
updated_item = QueueItem(
id="test-item-1",
workflow_name="Test Workflow",
workflow_data={"nodes": {"1": {"class_type": "TestNode"}}},
status=QueueStatus.RUNNING
)
mock_queue_service.get_queue_item.return_value = updated_item
request_data = {
'status': 'running'
}
response = client.put('/api/queue/items/test-item-1',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['item']['status'] == 'running'
assert data['message'] == 'Queue item updated successfully'
mock_queue_service.update_item_status.assert_called_once_with(
'test-item-1', QueueStatus.RUNNING, None, None
)
def test_update_queue_item_with_error(self, client, mock_queue_service, sample_queue_item):
"""Test updating a queue item with error message."""
mock_queue_service.update_item_status.return_value = True
failed_item = QueueItem(
id="test-item-1",
workflow_name="Test Workflow",
workflow_data={"nodes": {"1": {"class_type": "TestNode"}}},
status=QueueStatus.FAILED,
error_message="Test error"
)
mock_queue_service.get_queue_item.return_value = failed_item
request_data = {
'status': 'failed',
'error_message': 'Test error'
}
response = client.put('/api/queue/items/test-item-1',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['item']['status'] == 'failed'
assert data['item']['error_message'] == 'Test error'
def test_update_queue_item_invalid_status(self, client, mock_queue_service):
"""Test updating a queue item with invalid status."""
request_data = {
'status': 'invalid_status'
}
response = client.put('/api/queue/items/test-item-1',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 400
data = json.loads(response.data)
assert data['error'] == 'Bad request'
def test_delete_queue_item_success(self, client, mock_queue_service, sample_queue_item):
"""Test deleting a queue item successfully."""
mock_queue_service.get_queue_item.return_value = sample_queue_item
mock_queue_service.delete_items.return_value = True
response = client.delete('/api/queue/items/test-item-1')
assert response.status_code == 200
data = json.loads(response.data)
assert 'deleted successfully' in data['message']
assert 'timestamp' in data
mock_queue_service.delete_items.assert_called_once_with(['test-item-1'])
def test_delete_queue_item_not_found(self, client, mock_queue_service):
"""Test deleting a non-existent queue item."""
mock_queue_service.get_queue_item.return_value = None
response = client.delete('/api/queue/items/nonexistent')
assert response.status_code == 404
data = json.loads(response.data)
assert data['error'] == 'Not found'
def test_bulk_delete_items_success(self, client, mock_queue_service):
"""Test bulk deleting queue items successfully."""
mock_queue_service.delete_items.return_value = True
request_data = {
'item_ids': ['item-1', 'item-2', 'item-3']
}
response = client.delete('/api/queue/items/bulk',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['deleted_count'] == 3
assert 'Successfully deleted' in data['message']
mock_queue_service.delete_items.assert_called_once_with(['item-1', 'item-2', 'item-3'])
def test_bulk_delete_items_empty_list(self, client, mock_queue_service):
"""Test bulk deleting with empty item list."""
request_data = {
'item_ids': []
}
response = client.delete('/api/queue/items/bulk',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 400
data = json.loads(response.data)
assert data['error'] == 'Bad request'
assert 'cannot be empty' in data['message']
def test_bulk_delete_items_missing_field(self, client, mock_queue_service):
"""Test bulk deleting without item_ids field."""
request_data = {}
response = client.delete('/api/queue/items/bulk',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 400
data = json.loads(response.data)
assert data['error'] == 'Bad request'
assert 'item_ids' in data['message']
def test_filter_queue_items_success(self, client, mock_queue_service, sample_queue_item):
"""Test filtering queue items successfully."""
mock_queue_service.filter_items.return_value = [sample_queue_item]
request_data = {
'status': ['pending'],
'workflow_name': 'Test'
}
response = client.post('/api/queue/items/filter',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['count'] == 1
assert len(data['items']) == 1
assert 'filter' in data
assert 'timestamp' in data
def test_search_queue_items_success(self, client, mock_queue_service, sample_queue_item):
"""Test searching queue items successfully."""
mock_queue_service.search_items.return_value = [sample_queue_item]
response = client.get('/api/queue/search?q=test')
assert response.status_code == 200
data = json.loads(response.data)
assert data['count'] == 1
assert data['query'] == 'test'
assert len(data['items']) == 1
assert 'timestamp' in data
mock_queue_service.search_items.assert_called_once_with('test')
def test_search_queue_items_missing_query(self, client, mock_queue_service):
"""Test searching queue items without query parameter."""
response = client.get('/api/queue/search')
assert response.status_code == 400
data = json.loads(response.data)
assert data['error'] == 'Bad request'
assert 'query parameter' in data['message']
def test_search_queue_items_empty_query(self, client, mock_queue_service):
"""Test searching queue items with empty query."""
response = client.get('/api/queue/search?q=')
assert response.status_code == 400
data = json.loads(response.data)
assert data['error'] == 'Bad request'
def test_queue_service_error_handling(self, client, mock_queue_service):
"""Test handling of QueueServiceError."""
mock_queue_service.get_queue_items.side_effect = QueueServiceError("Database error")
response = client.get('/api/queue/items')
assert response.status_code == 500
data = json.loads(response.data)
assert data['error'] == 'Queue service error'
assert data['message'] == 'Database error'
+281
View File
@@ -0,0 +1,281 @@
"""
Tests for the Flask API routes.
"""
import json
import pytest
from unittest.mock import Mock, patch
from datetime import datetime, timezone
from api_routes import QueueManagerAPI
from models import QueueItem, QueueState, QueueStatus
from queue_service import QueueService, QueueServiceError
class TestQueueManagerAPI:
"""Test cases for the QueueManagerAPI class."""
@pytest.fixture
def mock_queue_service(self):
"""Create a mock queue service for testing."""
service = Mock(spec=QueueService)
service.get_queue_state.return_value = QueueState.RUNNING
return service
@pytest.fixture
def api(self, mock_queue_service):
"""Create a QueueManagerAPI instance for testing."""
return QueueManagerAPI(queue_service=mock_queue_service)
@pytest.fixture
def client(self, api):
"""Create a test client for the Flask app."""
api.app.config['TESTING'] = True
return api.app.test_client()
def test_api_initialization(self, mock_queue_service):
"""Test API initialization."""
api = QueueManagerAPI(queue_service=mock_queue_service)
assert api.queue_service == mock_queue_service
assert api.app is not None
assert api.app.config['JSON_SORT_KEYS'] is False
assert api.app.config['JSONIFY_PRETTYPRINT_REGULAR'] is True
def test_api_initialization_without_service(self):
"""Test API initialization without providing a queue service."""
with patch('api_routes.QueueService') as mock_service_class:
mock_service = Mock()
mock_service_class.return_value = mock_service
api = QueueManagerAPI()
assert api.queue_service == mock_service
mock_service_class.assert_called_once()
def test_health_check_healthy(self, client, mock_queue_service):
"""Test health check endpoint when service is healthy."""
mock_queue_service.get_queue_state.return_value = QueueState.RUNNING
response = client.get('/health')
assert response.status_code == 200
data = json.loads(response.data)
assert data['status'] == 'healthy'
assert data['queue_state'] == 'running'
assert data['version'] == '1.0.0'
assert 'timestamp' in data
def test_health_check_unhealthy(self, client, mock_queue_service):
"""Test health check endpoint when service is unhealthy."""
mock_queue_service.get_queue_state.side_effect = QueueServiceError("Database error")
response = client.get('/health')
assert response.status_code == 503
data = json.loads(response.data)
assert data['status'] == 'unhealthy'
assert 'error' in data
assert 'timestamp' in data
def test_serve_index_success(self, client):
"""Test serving index.html successfully."""
with patch('api_routes.send_from_directory') as mock_send:
mock_send.return_value = "<html>Test</html>"
response = client.get('/')
assert response.status_code == 200
mock_send.assert_called_once()
def test_serve_index_alternative_route(self, client):
"""Test serving index.html via /index.html route."""
with patch('api_routes.send_from_directory') as mock_send:
mock_send.return_value = "<html>Test</html>"
response = client.get('/index.html')
assert response.status_code == 200
mock_send.assert_called_once()
def test_serve_index_error(self, client):
"""Test error handling when serving index.html fails."""
with patch('api_routes.send_from_directory') as mock_send:
mock_send.side_effect = FileNotFoundError("File not found")
response = client.get('/')
assert response.status_code == 500
data = json.loads(response.data)
assert data['error'] == 'Failed to serve index page'
def test_error_handler_queue_service_error(self, client, mock_queue_service):
"""Test error handler for QueueServiceError."""
mock_queue_service.get_queue_state.side_effect = QueueServiceError("Test error")
response = client.get('/health')
assert response.status_code == 500
data = json.loads(response.data)
assert data['error'] == 'Queue service error'
assert data['message'] == 'Test error'
assert 'timestamp' in data
def test_error_handler_bad_request(self, client):
"""Test error handler for BadRequest."""
# This will be tested more thoroughly when we add endpoints that validate input
response = client.post('/nonexistent', data="invalid json", content_type='application/json')
assert response.status_code == 404 # Not found takes precedence
def test_error_handler_not_found(self, client):
"""Test error handler for NotFound."""
response = client.get('/nonexistent')
assert response.status_code == 404
data = json.loads(response.data)
assert data['error'] == 'Not found'
assert data['message'] == 'The requested resource was not found'
assert 'timestamp' in data
def test_validate_json_request_valid(self, api):
"""Test JSON request validation with valid data."""
with api.app.test_request_context('/test', json={'key': 'value'}):
data = api._validate_json_request()
assert data == {'key': 'value'}
def test_validate_json_request_not_json(self, api):
"""Test JSON request validation with non-JSON request."""
with api.app.test_request_context('/test', data='not json'):
with pytest.raises(Exception): # BadRequest
api._validate_json_request()
def test_validate_json_request_invalid_json(self, api):
"""Test JSON request validation with invalid JSON."""
with api.app.test_request_context('/test', data='{"invalid": json}', content_type='application/json'):
with pytest.raises(Exception): # BadRequest
api._validate_json_request()
def test_parse_queue_filter_empty(self, api):
"""Test parsing empty filter parameters."""
filter_criteria = api._parse_queue_filter({})
assert filter_criteria.status is None
assert filter_criteria.workflow_name is None
assert filter_criteria.search_term is None
assert filter_criteria.date_range is None
def test_parse_queue_filter_status_single(self, api):
"""Test parsing single status filter."""
filter_criteria = api._parse_queue_filter({'status': 'pending'})
assert filter_criteria.status == [QueueStatus.PENDING]
def test_parse_queue_filter_status_multiple(self, api):
"""Test parsing multiple status filters."""
filter_criteria = api._parse_queue_filter({'status': ['pending', 'running']})
assert filter_criteria.status == [QueueStatus.PENDING, QueueStatus.RUNNING]
def test_parse_queue_filter_status_invalid(self, api):
"""Test parsing invalid status filter."""
with pytest.raises(Exception): # BadRequest
api._parse_queue_filter({'status': 'invalid_status'})
def test_parse_queue_filter_workflow_name(self, api):
"""Test parsing workflow name filter."""
filter_criteria = api._parse_queue_filter({'workflow_name': 'test_workflow'})
assert filter_criteria.workflow_name == 'test_workflow'
def test_parse_queue_filter_search(self, api):
"""Test parsing search term filter."""
filter_criteria = api._parse_queue_filter({'search': 'test search'})
assert filter_criteria.search_term == 'test search'
def test_parse_queue_filter_date_range(self, api):
"""Test parsing date range filter."""
date_from = '2023-01-01T00:00:00'
date_to = '2023-12-31T23:59:59'
filter_criteria = api._parse_queue_filter({
'date_from': date_from,
'date_to': date_to
})
assert filter_criteria.date_range is not None
assert filter_criteria.date_range[0] == datetime.fromisoformat(date_from)
assert filter_criteria.date_range[1] == datetime.fromisoformat(date_to)
def test_parse_queue_filter_date_range_invalid(self, api):
"""Test parsing invalid date range filter."""
with pytest.raises(Exception): # BadRequest
api._parse_queue_filter({'date_from': 'invalid_date'})
def test_get_app(self, api):
"""Test getting the Flask app instance."""
app = api.get_app()
assert app == api.app
@patch('api_routes.logger')
def test_logging_on_initialization(self, mock_logger, mock_queue_service):
"""Test that initialization logs are created."""
QueueManagerAPI(queue_service=mock_queue_service)
mock_logger.info.assert_called_with("Queue Manager API initialized")
@patch('api_routes.logger')
def test_logging_on_health_check_failure(self, mock_logger, client, mock_queue_service):
"""Test that health check failures are logged."""
error_msg = "Database connection failed"
mock_queue_service.get_queue_state.side_effect = Exception(error_msg)
client.get('/health')
mock_logger.error.assert_called()
def test_flask_config(self, api):
"""Test Flask configuration settings."""
assert api.app.config['JSON_SORT_KEYS'] is False
assert api.app.config['JSONIFY_PRETTYPRINT_REGULAR'] is True
assert api.app.static_folder == 'web'
assert api.app.static_url_path == '/static'
class TestQueueManagerAPIIntegration:
"""Integration tests for the QueueManagerAPI."""
@pytest.fixture
def real_api(self):
"""Create a real API instance with mocked dependencies."""
with patch('api_routes.QueueService') as mock_service_class:
mock_service = Mock()
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service_class.return_value = mock_service
api = QueueManagerAPI()
return api, mock_service
def test_full_health_check_flow(self, real_api):
"""Test the complete health check flow."""
api, mock_service = real_api
client = api.app.test_client()
response = client.get('/health')
assert response.status_code == 200
mock_service.get_queue_state.assert_called_once()
def test_static_file_serving_setup(self, real_api):
"""Test that static file serving is properly configured."""
api, _ = real_api
# Check that static folder is configured
assert api.app.static_folder == 'web'
assert api.app.static_url_path == '/static'
# Check that routes are registered
rules = [rule.rule for rule in api.app.url_map.iter_rules()]
assert '/' in rules
assert '/index.html' in rules
assert '/health' in rules
+161
View File
@@ -0,0 +1,161 @@
#!/usr/bin/env python3
"""
Simple test script to verify the API routes work without pytest import issues.
"""
import sys
import os
import json
from unittest.mock import Mock
# Add current directory to path to avoid import issues
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from api_routes import QueueManagerAPI
from models import QueueState, QueueStatus
from queue_service import QueueService
def test_api_initialization():
"""Test API initialization."""
print("Testing API initialization...")
# Create mock queue service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
# Initialize API
api = QueueManagerAPI(queue_service=mock_service)
assert api.queue_service == mock_service
assert api.app is not None
assert api.app.config['JSON_SORT_KEYS'] is False
assert api.app.config['JSONIFY_PRETTYPRINT_REGULAR'] is True
print("✓ API initialization test passed")
def test_health_check():
"""Test health check endpoint."""
print("Testing health check endpoint...")
# Create mock queue service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
# Initialize API
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test health check
response = client.get('/health')
assert response.status_code == 200
data = json.loads(response.data)
assert data['status'] == 'healthy'
assert data['queue_state'] == 'running'
assert data['version'] == '1.0.0'
assert 'timestamp' in data
print("✓ Health check test passed")
def test_error_handling():
"""Test error handling."""
print("Testing error handling...")
# Create mock queue service that raises an error
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.side_effect = Exception("Database error")
# Initialize API
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test health check with error
response = client.get('/health')
assert response.status_code == 503
data = json.loads(response.data)
assert data['status'] == 'unhealthy'
assert 'error' in data
assert 'timestamp' in data
print("✓ Error handling test passed")
def test_not_found():
"""Test 404 error handling."""
print("Testing 404 error handling...")
# Create mock queue service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
# Initialize API
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test non-existent endpoint
response = client.get('/nonexistent')
assert response.status_code == 404
data = json.loads(response.data)
assert data['error'] == 'Not found'
assert data['message'] == 'The requested resource was not found'
assert 'timestamp' in data
print("✓ 404 error handling test passed")
def test_flask_configuration():
"""Test Flask configuration."""
print("Testing Flask configuration...")
# Create mock queue service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
# Initialize API
api = QueueManagerAPI(queue_service=mock_service)
# Check Flask configuration
print(f"Static folder: {api.app.static_folder}")
print(f"Static URL path: {api.app.static_url_path}")
assert 'web' in str(api.app.static_folder) # Check that web is in the path
assert api.app.static_url_path == '/static'
# Check that routes are registered
rules = [rule.rule for rule in api.app.url_map.iter_rules()]
assert '/' in rules
assert '/index.html' in rules
assert '/health' in rules
print("✓ Flask configuration test passed")
def main():
"""Run all tests."""
print("Running API integration tests...")
print("=" * 50)
try:
test_api_initialization()
test_health_check()
test_error_handling()
test_not_found()
test_flask_configuration()
print("=" * 50)
print("✓ All tests passed! Web server integration is working correctly.")
return 0
except Exception as e:
print(f"✗ Test failed: {e}")
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
sys.exit(main())
+295
View File
@@ -0,0 +1,295 @@
#!/usr/bin/env python3
"""
Comprehensive test script to verify the complete API layer works correctly.
"""
import sys
import os
import json
from unittest.mock import Mock
# Add current directory to path to avoid import issues
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from api_routes import QueueManagerAPI
from models import QueueItem, QueueState, QueueStatus, QueueConfig
from queue_service import QueueService
def test_complete_api_integration():
"""Test the complete API integration with all endpoints."""
print("Testing complete API integration...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
# Create sample items
sample_items = [
QueueItem(
id="item-1",
workflow_name="Test Workflow 1",
workflow_data={"nodes": {"1": {"class_type": "TestNode"}}},
status=QueueStatus.PENDING
),
QueueItem(
id="item-2",
workflow_name="Test Workflow 2",
workflow_data={"nodes": {"1": {"class_type": "TestNode"}}},
status=QueueStatus.RUNNING
)
]
mock_service.get_queue_items.return_value = sample_items
mock_service.get_config.return_value = QueueConfig()
mock_service.add_workflow.return_value = "new-item-id"
mock_service.get_queue_item.return_value = sample_items[0]
mock_service.update_item_status.return_value = True
mock_service.delete_items.return_value = True
mock_service.archive_items.return_value = True
mock_service.restore_items.return_value = True
mock_service.pause_queue.return_value = True
mock_service.resume_queue.return_value = True
mock_service.filter_items.return_value = [sample_items[0]]
mock_service.search_items.return_value = [sample_items[0]]
mock_service.export_queue.return_value = {
'version': '1.0',
'items': [item.to_dict() for item in sample_items]
}
mock_service.import_queue.return_value = True
mock_service.update_config.return_value = True
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test all endpoints
endpoints_tested = []
# 1. Health check
response = client.get('/health')
assert response.status_code == 200
endpoints_tested.append("GET /health")
# 2. Get queue items
response = client.get('/api/queue/items')
assert response.status_code == 200
endpoints_tested.append("GET /api/queue/items")
# 3. Add queue item
response = client.post('/api/queue/items',
data=json.dumps({
'workflow_name': 'New Workflow',
'workflow_data': {'nodes': {'1': {'class_type': 'TestNode'}}}
}),
content_type='application/json')
assert response.status_code == 201
endpoints_tested.append("POST /api/queue/items")
# 4. Get single queue item
response = client.get('/api/queue/items/item-1')
assert response.status_code == 200
endpoints_tested.append("GET /api/queue/items/<id>")
# 5. Update queue item
response = client.put('/api/queue/items/item-1',
data=json.dumps({'status': 'running'}),
content_type='application/json')
assert response.status_code == 200
endpoints_tested.append("PUT /api/queue/items/<id>")
# 6. Filter queue items
response = client.post('/api/queue/items/filter',
data=json.dumps({'status': ['pending']}),
content_type='application/json')
assert response.status_code == 200
endpoints_tested.append("POST /api/queue/items/filter")
# 7. Search queue items
response = client.get('/api/queue/search?q=test')
assert response.status_code == 200
endpoints_tested.append("GET /api/queue/search")
# 8. Get queue status
response = client.get('/api/queue/status')
assert response.status_code == 200
endpoints_tested.append("GET /api/queue/status")
# 9. Pause queue
response = client.post('/api/queue/pause')
assert response.status_code == 200
endpoints_tested.append("POST /api/queue/pause")
# 10. Resume queue
response = client.post('/api/queue/resume')
assert response.status_code == 200
endpoints_tested.append("POST /api/queue/resume")
# 11. Archive items
response = client.post('/api/queue/archive',
data=json.dumps({'item_ids': ['item-1']}),
content_type='application/json')
assert response.status_code == 200
endpoints_tested.append("POST /api/queue/archive")
# 12. Restore items
response = client.post('/api/queue/restore',
data=json.dumps({'item_ids': ['item-1']}),
content_type='application/json')
assert response.status_code == 200
endpoints_tested.append("POST /api/queue/restore")
# 13. Export queue
response = client.post('/api/queue/export',
data=json.dumps({}),
content_type='application/json')
assert response.status_code == 200
endpoints_tested.append("POST /api/queue/export")
# 14. Import queue
response = client.post('/api/queue/import',
data=json.dumps({
'queue_data': {'version': '1.0', 'items': []},
'merge': True
}),
content_type='application/json')
assert response.status_code == 200
endpoints_tested.append("POST /api/queue/import")
# 15. Get queue config
response = client.get('/api/queue/config')
assert response.status_code == 200
endpoints_tested.append("GET /api/queue/config")
# 16. Update queue config
response = client.put('/api/queue/config',
data=json.dumps({'max_concurrent_workflows': 2}),
content_type='application/json')
assert response.status_code == 200
endpoints_tested.append("PUT /api/queue/config")
# 17. Bulk delete items
response = client.delete('/api/queue/items/bulk',
data=json.dumps({'item_ids': ['item-1', 'item-2']}),
content_type='application/json')
assert response.status_code == 200
endpoints_tested.append("DELETE /api/queue/items/bulk")
# 18. Delete single item
response = client.delete('/api/queue/items/item-1')
assert response.status_code == 200
endpoints_tested.append("DELETE /api/queue/items/<id>")
print(f"✓ Successfully tested {len(endpoints_tested)} endpoints:")
for endpoint in endpoints_tested:
print(f" - {endpoint}")
return len(endpoints_tested)
def test_api_route_registration():
"""Test that all expected routes are registered."""
print("Testing API route registration...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
# Create API
api = QueueManagerAPI(queue_service=mock_service)
# Get all registered routes
routes = [rule.rule for rule in api.app.url_map.iter_rules()]
# Expected routes
expected_routes = [
'/',
'/index.html',
'/health',
'/api/queue/items',
'/api/queue/items/<item_id>',
'/api/queue/items/bulk',
'/api/queue/items/filter',
'/api/queue/search',
'/api/queue/status',
'/api/queue/pause',
'/api/queue/resume',
'/api/queue/archive',
'/api/queue/restore',
'/api/queue/export',
'/api/queue/import',
'/api/queue/config',
'/static/<path:filename>' # Flask automatically adds this for static files
]
# Check that all expected routes are registered
missing_routes = []
for expected_route in expected_routes:
if expected_route not in routes:
missing_routes.append(expected_route)
if missing_routes:
print(f"✗ Missing routes: {missing_routes}")
return False
print(f"✓ All {len(expected_routes)} expected routes are registered")
return True
def test_flask_configuration():
"""Test Flask application configuration."""
print("Testing Flask configuration...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
# Create API
api = QueueManagerAPI(queue_service=mock_service)
# Check Flask configuration
assert api.app.config['JSON_SORT_KEYS'] is False
assert api.app.config['JSONIFY_PRETTYPRINT_REGULAR'] is True
assert 'web' in str(api.app.static_folder)
assert api.app.static_url_path == '/static'
print("✓ Flask configuration is correct")
return True
def main():
"""Run all comprehensive tests."""
print("Running Comprehensive API Layer Tests...")
print("=" * 60)
try:
# Test route registration
if not test_api_route_registration():
return 1
# Test Flask configuration
if not test_flask_configuration():
return 1
# Test complete API integration
endpoints_count = test_complete_api_integration()
print("=" * 60)
print(f"✓ Complete API layer implementation verified!")
print(f"✓ {endpoints_count} endpoints tested successfully")
print("✓ All subtasks for task 5 'Create API layer for web interface' completed:")
print(" - 5.1 Flask/FastAPI web server integration ✓")
print(" - 5.2 Queue management API endpoints ✓")
print(" - 5.3 Control API endpoints ✓")
return 0
except Exception as e:
print(f"✗ Test failed: {e}")
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
sys.exit(main())
+466
View File
@@ -0,0 +1,466 @@
#!/usr/bin/env python3
"""
Simple test script to verify the control API endpoints work.
"""
import sys
import os
import json
from unittest.mock import Mock
# Add current directory to path to avoid import issues
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from api_routes import QueueManagerAPI
from models import QueueItem, QueueState, QueueStatus, QueueConfig
from queue_service import QueueService, QueueServiceError
def create_sample_items():
"""Create sample queue items for testing."""
return [
QueueItem(
id="item-1",
workflow_name="Test Workflow 1",
workflow_data={"nodes": {"1": {"class_type": "TestNode"}}},
status=QueueStatus.PENDING
),
QueueItem(
id="item-2",
workflow_name="Test Workflow 2",
workflow_data={"nodes": {"1": {"class_type": "TestNode"}}},
status=QueueStatus.RUNNING
),
QueueItem(
id="item-3",
workflow_name="Test Workflow 3",
workflow_data={"nodes": {"1": {"class_type": "TestNode"}}},
status=QueueStatus.COMPLETED
)
]
def test_get_queue_status():
"""Test GET /api/queue/status endpoint."""
print("Testing GET /api/queue/status...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.get_queue_items.return_value = create_sample_items()
mock_service.get_config.return_value = QueueConfig()
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint
response = client.get('/api/queue/status')
assert response.status_code == 200
data = json.loads(response.data)
assert data['queue_state'] == 'running'
assert 'statistics' in data
assert data['statistics']['total'] == 3
assert data['statistics']['pending'] == 1
assert data['statistics']['running'] == 1
assert data['statistics']['completed'] == 1
assert 'config' in data
assert 'timestamp' in data
print("✓ GET /api/queue/status test passed")
def test_pause_queue():
"""Test POST /api/queue/pause endpoint."""
print("Testing POST /api/queue/pause...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.pause_queue.return_value = True
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint
response = client.post('/api/queue/pause')
assert response.status_code == 200
data = json.loads(response.data)
assert data['message'] == 'Queue processing paused successfully'
assert data['queue_state'] == 'paused'
assert 'timestamp' in data
mock_service.pause_queue.assert_called_once()
print("✓ POST /api/queue/pause test passed")
def test_resume_queue():
"""Test POST /api/queue/resume endpoint."""
print("Testing POST /api/queue/resume...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.PAUSED
mock_service.resume_queue.return_value = True
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint
response = client.post('/api/queue/resume')
assert response.status_code == 200
data = json.loads(response.data)
assert data['message'] == 'Queue processing resumed successfully'
assert data['queue_state'] == 'running'
assert 'timestamp' in data
mock_service.resume_queue.assert_called_once()
print("✓ POST /api/queue/resume test passed")
def test_archive_items():
"""Test POST /api/queue/archive endpoint."""
print("Testing POST /api/queue/archive...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.archive_items.return_value = True
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data
request_data = {
'item_ids': ['item-1', 'item-2']
}
# Test endpoint
response = client.post('/api/queue/archive',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['archived_count'] == 2
assert 'Successfully archived' in data['message']
assert 'timestamp' in data
mock_service.archive_items.assert_called_once_with(['item-1', 'item-2'])
print("✓ POST /api/queue/archive test passed")
def test_restore_items():
"""Test POST /api/queue/restore endpoint."""
print("Testing POST /api/queue/restore...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.restore_items.return_value = True
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data
request_data = {
'item_ids': ['item-1', 'item-2']
}
# Test endpoint
response = client.post('/api/queue/restore',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['restored_count'] == 2
assert 'Successfully restored' in data['message']
assert 'timestamp' in data
mock_service.restore_items.assert_called_once_with(['item-1', 'item-2'])
print("✓ POST /api/queue/restore test passed")
def test_export_queue():
"""Test POST /api/queue/export endpoint."""
print("Testing POST /api/queue/export...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.export_queue.return_value = {
'version': '1.0',
'items': [item.to_dict() for item in create_sample_items()],
'config': QueueConfig().to_dict()
}
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data (export all items)
request_data = {}
# Test endpoint
response = client.post('/api/queue/export',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['message'] == 'Queue exported successfully'
assert 'export_data' in data
assert data['export_data']['version'] == '1.0'
assert len(data['export_data']['items']) == 3
assert 'timestamp' in data
mock_service.export_queue.assert_called_once_with(None)
print("✓ POST /api/queue/export test passed")
def test_export_queue_specific_items():
"""Test POST /api/queue/export endpoint with specific items."""
print("Testing POST /api/queue/export with specific items...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.export_queue.return_value = {
'version': '1.0',
'items': [create_sample_items()[0].to_dict()],
'config': QueueConfig().to_dict()
}
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data (export specific items)
request_data = {
'item_ids': ['item-1']
}
# Test endpoint
response = client.post('/api/queue/export',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['message'] == 'Queue exported successfully'
assert len(data['export_data']['items']) == 1
mock_service.export_queue.assert_called_once_with(['item-1'])
print("✓ POST /api/queue/export with specific items test passed")
def test_import_queue():
"""Test POST /api/queue/import endpoint."""
print("Testing POST /api/queue/import...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.import_queue.return_value = True
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data
queue_data = {
'version': '1.0',
'items': [item.to_dict() for item in create_sample_items()],
'config': QueueConfig().to_dict()
}
request_data = {
'queue_data': queue_data,
'merge': True
}
# Test endpoint
response = client.post('/api/queue/import',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['message'] == 'Queue imported successfully'
assert data['merge_mode'] is True
assert 'timestamp' in data
mock_service.import_queue.assert_called_once_with(queue_data, True)
print("✓ POST /api/queue/import test passed")
def test_get_queue_config():
"""Test GET /api/queue/config endpoint."""
print("Testing GET /api/queue/config...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.get_config.return_value = QueueConfig()
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint
response = client.get('/api/queue/config')
assert response.status_code == 200
data = json.loads(response.data)
assert 'config' in data
assert data['config']['queue_state'] == 'running'
assert data['config']['max_concurrent_workflows'] == 1
assert 'timestamp' in data
mock_service.get_config.assert_called_once()
print("✓ GET /api/queue/config test passed")
def test_update_queue_config():
"""Test PUT /api/queue/config endpoint."""
print("Testing PUT /api/queue/config...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.get_config.return_value = QueueConfig()
mock_service.update_config.return_value = True
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data
request_data = {
'max_concurrent_workflows': 2,
'auto_archive_completed': True,
'auto_archive_days': 7
}
# Test endpoint
response = client.put('/api/queue/config',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['message'] == 'Queue configuration updated successfully'
assert 'config' in data
assert 'timestamp' in data
mock_service.update_config.assert_called_once()
print("✓ PUT /api/queue/config test passed")
def test_error_handling():
"""Test error handling in control endpoints."""
print("Testing error handling...")
# Create mock service that raises errors
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.pause_queue.side_effect = QueueServiceError("Failed to pause")
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint with error
response = client.post('/api/queue/pause')
assert response.status_code == 500
data = json.loads(response.data)
assert data['error'] == 'Queue service error'
assert data['message'] == 'Failed to pause'
print("✓ Error handling test passed")
def test_validation_errors():
"""Test validation errors in control endpoints."""
print("Testing validation errors...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test archive endpoint with missing item_ids
response = client.post('/api/queue/archive',
data=json.dumps({}),
content_type='application/json')
assert response.status_code == 400
data = json.loads(response.data)
assert data['error'] == 'Bad request'
assert 'item_ids' in data['message']
# Test archive endpoint with empty item_ids
response = client.post('/api/queue/archive',
data=json.dumps({'item_ids': []}),
content_type='application/json')
assert response.status_code == 400
data = json.loads(response.data)
assert data['error'] == 'Bad request'
assert 'cannot be empty' in data['message']
print("✓ Validation errors test passed")
def main():
"""Run all tests."""
print("Running Control API endpoint tests...")
print("=" * 50)
try:
test_get_queue_status()
test_pause_queue()
test_resume_queue()
test_archive_items()
test_restore_items()
test_export_queue()
test_export_queue_specific_items()
test_import_queue()
test_get_queue_config()
test_update_queue_config()
test_error_handling()
test_validation_errors()
print("=" * 50)
print("✓ All control API endpoint tests passed!")
return 0
except Exception as e:
print(f"✗ Test failed: {e}")
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
sys.exit(main())
+362
View File
@@ -0,0 +1,362 @@
#!/usr/bin/env python3
"""
Simple test script to verify the queue management API endpoints work.
"""
import sys
import os
import json
from unittest.mock import Mock
# Add current directory to path to avoid import issues
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from api_routes import QueueManagerAPI
from models import QueueItem, QueueState, QueueStatus
from queue_service import QueueService, QueueServiceError
def create_sample_item():
"""Create a sample queue item for testing."""
return QueueItem(
id="test-item-1",
workflow_name="Test Workflow",
workflow_data={"nodes": {"1": {"class_type": "TestNode"}}},
status=QueueStatus.PENDING
)
def test_get_queue_items():
"""Test GET /api/queue/items endpoint."""
print("Testing GET /api/queue/items...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.get_queue_items.return_value = [create_sample_item()]
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint
response = client.get('/api/queue/items')
assert response.status_code == 200
data = json.loads(response.data)
assert data['count'] == 1
assert len(data['items']) == 1
assert data['items'][0]['id'] == 'test-item-1'
assert 'timestamp' in data
print("✓ GET /api/queue/items test passed")
def test_get_queue_items_with_filter():
"""Test GET /api/queue/items with status filter."""
print("Testing GET /api/queue/items with status filter...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.get_queue_items.return_value = [create_sample_item()]
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint with status filter
response = client.get('/api/queue/items?status=pending')
assert response.status_code == 200
data = json.loads(response.data)
assert data['count'] == 1
# Verify the service was called with the correct status
mock_service.get_queue_items.assert_called_with(QueueStatus.PENDING)
print("✓ GET /api/queue/items with status filter test passed")
def test_add_queue_item():
"""Test POST /api/queue/items endpoint."""
print("Testing POST /api/queue/items...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.add_workflow.return_value = "test-item-1"
mock_service.get_queue_item.return_value = create_sample_item()
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data
request_data = {
'workflow_name': 'Test Workflow',
'workflow_data': {'nodes': {'1': {'class_type': 'TestNode'}}}
}
# Test endpoint
response = client.post('/api/queue/items',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 201
data = json.loads(response.data)
assert data['item']['id'] == 'test-item-1'
assert data['message'] == 'Workflow added to queue successfully'
assert 'timestamp' in data
# Verify service calls
mock_service.add_workflow.assert_called_once_with(
request_data['workflow_data'],
request_data['workflow_name']
)
print("✓ POST /api/queue/items test passed")
def test_get_single_queue_item():
"""Test GET /api/queue/items/<id> endpoint."""
print("Testing GET /api/queue/items/<id>...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.get_queue_item.return_value = create_sample_item()
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint
response = client.get('/api/queue/items/test-item-1')
assert response.status_code == 200
data = json.loads(response.data)
assert data['item']['id'] == 'test-item-1'
assert 'timestamp' in data
mock_service.get_queue_item.assert_called_once_with('test-item-1')
print("✓ GET /api/queue/items/<id> test passed")
def test_update_queue_item():
"""Test PUT /api/queue/items/<id> endpoint."""
print("Testing PUT /api/queue/items/<id>...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.update_item_status.return_value = True
# Create updated item
updated_item = create_sample_item()
updated_item.status = QueueStatus.RUNNING
mock_service.get_queue_item.return_value = updated_item
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data
request_data = {
'status': 'running'
}
# Test endpoint
response = client.put('/api/queue/items/test-item-1',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['item']['status'] == 'running'
assert data['message'] == 'Queue item updated successfully'
mock_service.update_item_status.assert_called_once_with(
'test-item-1', QueueStatus.RUNNING, None, None
)
print("✓ PUT /api/queue/items/<id> test passed")
def test_delete_queue_item():
"""Test DELETE /api/queue/items/<id> endpoint."""
print("Testing DELETE /api/queue/items/<id>...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.get_queue_item.return_value = create_sample_item()
mock_service.delete_items.return_value = True
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint
response = client.delete('/api/queue/items/test-item-1')
assert response.status_code == 200
data = json.loads(response.data)
assert 'deleted successfully' in data['message']
assert 'timestamp' in data
mock_service.delete_items.assert_called_once_with(['test-item-1'])
print("✓ DELETE /api/queue/items/<id> test passed")
def test_bulk_delete_items():
"""Test DELETE /api/queue/items/bulk endpoint."""
print("Testing DELETE /api/queue/items/bulk...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.delete_items.return_value = True
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data
request_data = {
'item_ids': ['item-1', 'item-2', 'item-3']
}
# Test endpoint
response = client.delete('/api/queue/items/bulk',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['deleted_count'] == 3
assert 'Successfully deleted' in data['message']
mock_service.delete_items.assert_called_once_with(['item-1', 'item-2', 'item-3'])
print("✓ DELETE /api/queue/items/bulk test passed")
def test_filter_queue_items():
"""Test POST /api/queue/items/filter endpoint."""
print("Testing POST /api/queue/items/filter...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.filter_items.return_value = [create_sample_item()]
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test data
request_data = {
'status': ['pending'],
'workflow_name': 'Test'
}
# Test endpoint
response = client.post('/api/queue/items/filter',
data=json.dumps(request_data),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['count'] == 1
assert len(data['items']) == 1
assert 'filter' in data
assert 'timestamp' in data
print("✓ POST /api/queue/items/filter test passed")
def test_search_queue_items():
"""Test GET /api/queue/search endpoint."""
print("Testing GET /api/queue/search...")
# Create mock service
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.search_items.return_value = [create_sample_item()]
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint
response = client.get('/api/queue/search?q=test')
assert response.status_code == 200
data = json.loads(response.data)
assert data['count'] == 1
assert data['query'] == 'test'
assert len(data['items']) == 1
assert 'timestamp' in data
mock_service.search_items.assert_called_once_with('test')
print("✓ GET /api/queue/search test passed")
def test_error_handling():
"""Test error handling in endpoints."""
print("Testing error handling...")
# Create mock service that raises errors
mock_service = Mock(spec=QueueService)
mock_service.get_queue_state.return_value = QueueState.RUNNING
mock_service.get_queue_items.side_effect = QueueServiceError("Database error")
# Create API and client
api = QueueManagerAPI(queue_service=mock_service)
client = api.app.test_client()
# Test endpoint with error
response = client.get('/api/queue/items')
assert response.status_code == 500
data = json.loads(response.data)
assert data['error'] == 'Queue service error'
assert data['message'] == 'Database error'
print("✓ Error handling test passed")
def main():
"""Run all tests."""
print("Running Queue Management API endpoint tests...")
print("=" * 60)
try:
test_get_queue_items()
test_get_queue_items_with_filter()
test_add_queue_item()
test_get_single_queue_item()
test_update_queue_item()
test_delete_queue_item()
test_bulk_delete_items()
test_filter_queue_items()
test_search_queue_items()
test_error_handling()
print("=" * 60)
print("✓ All queue management API endpoint tests passed!")
return 0
except Exception as e:
print(f"✗ Test failed: {e}")
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
sys.exit(main())