""" 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'])