268 lines
9.4 KiB
Python
268 lines
9.4 KiB
Python
"""
|
|
Tests for error handling and logging functionality.
|
|
|
|
Tests error handlers, request/response logging, database rollback on errors,
|
|
and proper error responses (JSON vs HTML).
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import pytest
|
|
from flask import Flask
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from web.app import create_app
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
"""Create test Flask app."""
|
|
test_config = {
|
|
'TESTING': True,
|
|
'SQLALCHEMY_DATABASE_URI': 'sqlite:///:memory:',
|
|
'SECRET_KEY': 'test-secret-key',
|
|
'WTF_CSRF_ENABLED': False
|
|
}
|
|
app = create_app(test_config)
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
"""Create test client."""
|
|
return app.test_client()
|
|
|
|
|
|
class TestErrorHandlers:
|
|
"""Test error handler functionality."""
|
|
|
|
def test_404_json_response(self, client):
|
|
"""Test 404 error returns JSON for API requests."""
|
|
response = client.get('/api/nonexistent')
|
|
assert response.status_code == 404
|
|
assert response.content_type == 'application/json'
|
|
|
|
data = json.loads(response.data)
|
|
assert 'error' in data
|
|
assert data['error'] == 'Not Found'
|
|
assert 'message' in data
|
|
|
|
def test_404_html_response(self, client):
|
|
"""Test 404 error returns HTML for web requests."""
|
|
response = client.get('/nonexistent')
|
|
assert response.status_code == 404
|
|
assert 'text/html' in response.content_type
|
|
assert b'404' in response.data
|
|
|
|
def test_400_json_response(self, client):
|
|
"""Test 400 error returns JSON for API requests."""
|
|
# Trigger 400 by sending invalid JSON
|
|
response = client.post(
|
|
'/api/scans',
|
|
data='invalid json',
|
|
content_type='application/json'
|
|
)
|
|
assert response.status_code in [400, 401] # 401 if auth required
|
|
|
|
def test_405_method_not_allowed(self, client):
|
|
"""Test 405 error for method not allowed."""
|
|
# Try POST to health check (only GET allowed)
|
|
response = client.post('/api/scans/health')
|
|
assert response.status_code == 405
|
|
|
|
data = json.loads(response.data)
|
|
assert 'error' in data
|
|
assert data['error'] == 'Method Not Allowed'
|
|
|
|
def test_json_accept_header(self, client):
|
|
"""Test JSON response when Accept header specifies JSON."""
|
|
response = client.get(
|
|
'/nonexistent',
|
|
headers={'Accept': 'application/json'}
|
|
)
|
|
assert response.status_code == 404
|
|
assert response.content_type == 'application/json'
|
|
|
|
|
|
class TestLogging:
|
|
"""Test logging functionality."""
|
|
|
|
def test_request_logging(self, client, caplog):
|
|
"""Test that requests are logged."""
|
|
with caplog.at_level(logging.INFO):
|
|
response = client.get('/api/scans/health')
|
|
|
|
# Check log messages
|
|
log_messages = [record.message for record in caplog.records]
|
|
# Should log incoming request and response
|
|
assert any('GET /api/scans/health' in msg for msg in log_messages)
|
|
|
|
def test_error_logging(self, client, caplog):
|
|
"""Test that errors are logged with full context."""
|
|
with caplog.at_level(logging.INFO):
|
|
client.get('/api/nonexistent')
|
|
|
|
# Check that 404 was logged
|
|
log_messages = [record.message for record in caplog.records]
|
|
assert any('not found' in msg.lower() or '404' in msg for msg in log_messages)
|
|
|
|
def test_request_id_in_logs(self, client, caplog):
|
|
"""Test that request ID is included in log records."""
|
|
with caplog.at_level(logging.INFO):
|
|
client.get('/api/scans/health')
|
|
|
|
# Check that log records have request_id attribute
|
|
for record in caplog.records:
|
|
assert hasattr(record, 'request_id')
|
|
assert record.request_id # Should not be empty
|
|
|
|
|
|
class TestRequestResponseHandlers:
|
|
"""Test request and response handler middleware."""
|
|
|
|
def test_request_id_header(self, client):
|
|
"""Test that response includes X-Request-ID header for API requests."""
|
|
response = client.get('/api/scans/health')
|
|
assert 'X-Request-ID' in response.headers
|
|
|
|
def test_request_duration_header(self, client):
|
|
"""Test that response includes X-Request-Duration-Ms header."""
|
|
response = client.get('/api/scans/health')
|
|
assert 'X-Request-Duration-Ms' in response.headers
|
|
|
|
duration = float(response.headers['X-Request-Duration-Ms'])
|
|
assert duration >= 0 # Should be non-negative
|
|
|
|
def test_security_headers(self, client):
|
|
"""Test that security headers are added to API responses."""
|
|
response = client.get('/api/scans/health')
|
|
|
|
# Check security headers
|
|
assert response.headers.get('X-Content-Type-Options') == 'nosniff'
|
|
assert response.headers.get('X-Frame-Options') == 'DENY'
|
|
assert response.headers.get('X-XSS-Protection') == '1; mode=block'
|
|
|
|
def test_request_timing(self, client):
|
|
"""Test that request timing is calculated correctly."""
|
|
response = client.get('/api/scans/health')
|
|
|
|
duration_header = response.headers.get('X-Request-Duration-Ms')
|
|
assert duration_header is not None
|
|
|
|
duration = float(duration_header)
|
|
# Should complete in reasonable time (less than 5 seconds)
|
|
assert duration < 5000
|
|
|
|
|
|
class TestDatabaseErrorHandling:
|
|
"""Test database error handling and rollback."""
|
|
|
|
def test_database_rollback_on_error(self, app):
|
|
"""Test that database session is rolled back on error."""
|
|
# This test would require triggering a database error
|
|
# For now, just verify the error handler is registered
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
# Check that SQLAlchemyError handler is registered
|
|
assert SQLAlchemyError in app.error_handler_spec[None]
|
|
|
|
|
|
class TestLogRotation:
|
|
"""Test log rotation configuration."""
|
|
|
|
def test_log_files_created(self, app, tmp_path):
|
|
"""Test that log files are created in logs directory."""
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# Check that logs directory exists
|
|
log_dir = Path('logs')
|
|
# Note: In test environment, logs may not be created immediately
|
|
# Just verify the configuration is set up
|
|
|
|
# Verify app logger has handlers
|
|
assert len(app.logger.handlers) > 0
|
|
|
|
# Verify at least one handler is a RotatingFileHandler
|
|
from logging.handlers import RotatingFileHandler
|
|
has_rotating_handler = any(
|
|
isinstance(h, RotatingFileHandler)
|
|
for h in app.logger.handlers
|
|
)
|
|
assert has_rotating_handler, "Should have RotatingFileHandler configured"
|
|
|
|
def test_log_handler_configuration(self, app):
|
|
"""Test that log handlers are configured correctly."""
|
|
from logging.handlers import RotatingFileHandler
|
|
|
|
# Find RotatingFileHandler
|
|
rotating_handlers = [
|
|
h for h in app.logger.handlers
|
|
if isinstance(h, RotatingFileHandler)
|
|
]
|
|
|
|
assert len(rotating_handlers) > 0, "Should have rotating file handlers"
|
|
|
|
# Check handler configuration
|
|
for handler in rotating_handlers:
|
|
# Should have max size configured
|
|
assert handler.maxBytes > 0
|
|
# Should have backup count configured
|
|
assert handler.backupCount > 0
|
|
|
|
|
|
class TestStructuredLogging:
|
|
"""Test structured logging features."""
|
|
|
|
def test_log_format_includes_request_id(self, client, caplog):
|
|
"""Test that log format includes request ID."""
|
|
with caplog.at_level(logging.INFO):
|
|
client.get('/api/scans/health')
|
|
|
|
# Verify log records have request_id
|
|
for record in caplog.records:
|
|
assert hasattr(record, 'request_id')
|
|
|
|
def test_error_log_includes_traceback(self, app, caplog):
|
|
"""Test that errors are logged with traceback."""
|
|
with app.test_request_context('/api/test'):
|
|
with caplog.at_level(logging.ERROR):
|
|
try:
|
|
raise ValueError("Test error")
|
|
except ValueError as e:
|
|
app.logger.error("Test error occurred", exc_info=True)
|
|
|
|
# Check that traceback is in logs
|
|
log_output = caplog.text
|
|
assert 'Test error' in log_output
|
|
assert 'Traceback' in log_output or 'ValueError' in log_output
|
|
|
|
|
|
class TestErrorTemplates:
|
|
"""Test error template rendering."""
|
|
|
|
def test_404_template_exists(self, client):
|
|
"""Test that 404 error template is rendered."""
|
|
response = client.get('/nonexistent')
|
|
assert response.status_code == 404
|
|
assert b'404' in response.data
|
|
assert b'Page Not Found' in response.data or b'Not Found' in response.data
|
|
|
|
def test_500_template_exists(self, app):
|
|
"""Test that 500 error template can be rendered."""
|
|
# We can't easily trigger a 500 without breaking the app
|
|
# Just verify the template file exists
|
|
from pathlib import Path
|
|
template_path = Path('web/templates/errors/500.html')
|
|
assert template_path.exists(), "500 error template should exist"
|
|
|
|
def test_error_template_styling(self, client):
|
|
"""Test that error templates include styling."""
|
|
response = client.get('/nonexistent')
|
|
# Should include CSS styling
|
|
assert b'style' in response.data or b'css' in response.data.lower()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v'])
|