mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2025-12-20 17:31:09 -05:00
- Implemented extensive test cases for the Parks API, covering endpoints for listing, retrieving, creating, updating, and deleting parks. - Added tests for filtering, searching, and ordering parks in the API. - Created tests for error handling in the API, including malformed JSON and unsupported methods. - Developed model tests for Park, ParkArea, Company, and ParkReview models, ensuring validation and constraints are enforced. - Introduced utility mixins for API and model testing to streamline assertions and enhance test readability. - Included integration tests to validate complete workflows involving park creation, retrieval, updating, and deletion.
234 lines
6.8 KiB
Python
234 lines
6.8 KiB
Python
"""
|
|
Centralized logging configuration for ThrillWiki.
|
|
Provides structured logging with proper formatting and context.
|
|
"""
|
|
|
|
import logging
|
|
import sys
|
|
from typing import Dict, Any, Optional
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
|
|
|
|
class ThrillWikiFormatter(logging.Formatter):
|
|
"""Custom formatter for ThrillWiki logs with structured output."""
|
|
|
|
def format(self, record):
|
|
# Add timestamp if not present
|
|
if not hasattr(record, 'timestamp'):
|
|
record.timestamp = timezone.now().isoformat()
|
|
|
|
# Add request context if available
|
|
if hasattr(record, 'request'):
|
|
record.request_id = getattr(record.request, 'id', 'unknown')
|
|
record.user_id = getattr(record.request.user, 'id', 'anonymous') if hasattr(record.request, 'user') else 'unknown'
|
|
record.path = getattr(record.request, 'path', 'unknown')
|
|
record.method = getattr(record.request, 'method', 'unknown')
|
|
|
|
# Structure the log message
|
|
if hasattr(record, 'extra_data'):
|
|
record.structured_data = record.extra_data
|
|
|
|
return super().format(record)
|
|
|
|
|
|
def get_logger(name: str) -> logging.Logger:
|
|
"""
|
|
Get a configured logger for ThrillWiki components.
|
|
|
|
Args:
|
|
name: Logger name (usually __name__)
|
|
|
|
Returns:
|
|
Configured logger instance
|
|
"""
|
|
logger = logging.getLogger(name)
|
|
|
|
# Only configure if not already configured
|
|
if not logger.handlers:
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
formatter = ThrillWikiFormatter(
|
|
fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
handler.setFormatter(formatter)
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logging.INFO if settings.DEBUG else logging.WARNING)
|
|
|
|
return logger
|
|
|
|
|
|
def log_exception(
|
|
logger: logging.Logger,
|
|
exception: Exception,
|
|
*,
|
|
context: Optional[Dict[str, Any]] = None,
|
|
request=None,
|
|
level: int = logging.ERROR
|
|
) -> None:
|
|
"""
|
|
Log an exception with structured context.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
exception: Exception to log
|
|
context: Additional context data
|
|
request: Django request object
|
|
level: Log level
|
|
"""
|
|
log_data = {
|
|
'exception_type': exception.__class__.__name__,
|
|
'exception_message': str(exception),
|
|
'context': context or {}
|
|
}
|
|
|
|
if request:
|
|
log_data.update({
|
|
'request_path': getattr(request, 'path', 'unknown'),
|
|
'request_method': getattr(request, 'method', 'unknown'),
|
|
'user_id': getattr(request.user, 'id', 'anonymous') if hasattr(request, 'user') else 'unknown'
|
|
})
|
|
|
|
logger.log(level, f"Exception occurred: {exception}", extra={'extra_data': log_data}, exc_info=True)
|
|
|
|
|
|
def log_business_event(
|
|
logger: logging.Logger,
|
|
event_type: str,
|
|
*,
|
|
message: str,
|
|
context: Optional[Dict[str, Any]] = None,
|
|
request=None,
|
|
level: int = logging.INFO
|
|
) -> None:
|
|
"""
|
|
Log a business event with structured context.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
event_type: Type of business event
|
|
message: Event message
|
|
context: Additional context data
|
|
request: Django request object
|
|
level: Log level
|
|
"""
|
|
log_data = {
|
|
'event_type': event_type,
|
|
'context': context or {}
|
|
}
|
|
|
|
if request:
|
|
log_data.update({
|
|
'request_path': getattr(request, 'path', 'unknown'),
|
|
'request_method': getattr(request, 'method', 'unknown'),
|
|
'user_id': getattr(request.user, 'id', 'anonymous') if hasattr(request, 'user') else 'unknown'
|
|
})
|
|
|
|
logger.log(level, message, extra={'extra_data': log_data})
|
|
|
|
|
|
def log_performance_metric(
|
|
logger: logging.Logger,
|
|
operation: str,
|
|
*,
|
|
duration_ms: float,
|
|
context: Optional[Dict[str, Any]] = None,
|
|
level: int = logging.INFO
|
|
) -> None:
|
|
"""
|
|
Log a performance metric.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
operation: Operation name
|
|
duration_ms: Duration in milliseconds
|
|
context: Additional context data
|
|
level: Log level
|
|
"""
|
|
log_data = {
|
|
'metric_type': 'performance',
|
|
'operation': operation,
|
|
'duration_ms': duration_ms,
|
|
'context': context or {}
|
|
}
|
|
|
|
message = f"Performance: {operation} took {duration_ms:.2f}ms"
|
|
logger.log(level, message, extra={'extra_data': log_data})
|
|
|
|
|
|
def log_api_request(
|
|
logger: logging.Logger,
|
|
request,
|
|
*,
|
|
response_status: Optional[int] = None,
|
|
duration_ms: Optional[float] = None,
|
|
level: int = logging.INFO
|
|
) -> None:
|
|
"""
|
|
Log an API request with context.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
request: Django request object
|
|
response_status: HTTP response status code
|
|
duration_ms: Request duration in milliseconds
|
|
level: Log level
|
|
"""
|
|
log_data = {
|
|
'request_type': 'api',
|
|
'path': getattr(request, 'path', 'unknown'),
|
|
'method': getattr(request, 'method', 'unknown'),
|
|
'user_id': getattr(request.user, 'id', 'anonymous') if hasattr(request, 'user') else 'unknown',
|
|
'response_status': response_status,
|
|
'duration_ms': duration_ms
|
|
}
|
|
|
|
message = f"API Request: {request.method} {request.path}"
|
|
if response_status:
|
|
message += f" -> {response_status}"
|
|
if duration_ms:
|
|
message += f" ({duration_ms:.2f}ms)"
|
|
|
|
logger.log(level, message, extra={'extra_data': log_data})
|
|
|
|
|
|
def log_security_event(
|
|
logger: logging.Logger,
|
|
event_type: str,
|
|
*,
|
|
message: str,
|
|
severity: str = 'medium',
|
|
context: Optional[Dict[str, Any]] = None,
|
|
request=None
|
|
) -> None:
|
|
"""
|
|
Log a security-related event.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
event_type: Type of security event
|
|
message: Event message
|
|
severity: Event severity (low, medium, high, critical)
|
|
context: Additional context data
|
|
request: Django request object
|
|
"""
|
|
log_data = {
|
|
'security_event': True,
|
|
'event_type': event_type,
|
|
'severity': severity,
|
|
'context': context or {}
|
|
}
|
|
|
|
if request:
|
|
log_data.update({
|
|
'request_path': getattr(request, 'path', 'unknown'),
|
|
'request_method': getattr(request, 'method', 'unknown'),
|
|
'user_id': getattr(request.user, 'id', 'anonymous') if hasattr(request, 'user') else 'unknown',
|
|
'remote_addr': request.META.get('REMOTE_ADDR', 'unknown'),
|
|
'user_agent': request.META.get('HTTP_USER_AGENT', 'unknown')
|
|
})
|
|
|
|
# Use WARNING for medium/high, ERROR for critical
|
|
level = logging.ERROR if severity in ['high', 'critical'] else logging.WARNING
|
|
|
|
logger.log(level, f"SECURITY: {message}", extra={'extra_data': log_data})
|