Add state machine diagrams and code examples for ThrillWiki

- Created a comprehensive documentation file for state machine diagrams, detailing various states and transitions for models such as EditSubmission, ModerationReport, and Park Status.
- Included transition matrices for each state machine to clarify role requirements and guards.
- Developed a new document providing code examples for implementing state machines, including adding new state machines to models, defining custom guards, implementing callbacks, and testing state machines.
- Added examples for document approval workflows, custom guards, email notifications, and cache invalidation callbacks.
- Implemented a test suite for document workflows, covering various scenarios including approval, rejection, and transition logging.
This commit is contained in:
pacnpal
2025-12-21 20:21:54 -05:00
parent 8f6acbdc23
commit b508434574
24 changed files with 9979 additions and 360 deletions

View File

@@ -1,4 +1,65 @@
"""StateTransitionBuilder - Reads RichChoice metadata and generates FSM configurations."""
"""
StateTransitionBuilder - Reads RichChoice metadata and generates FSM configurations.
This module provides utilities for building FSM transition configurations
from rich choice metadata, enabling declarative state machine definitions
where transitions are defined in the choice registry rather than code.
Key Features:
- Extract valid transitions from choice metadata
- Build complete transition graphs
- Extract permission requirements for transitions
- Identify terminal and actionable states
- Generate consistent transition method names
Example Usage:
Create a builder for a choice group::
from apps.core.state_machine.builder import StateTransitionBuilder
builder = StateTransitionBuilder(
choice_group='submission_status',
domain='moderation'
)
# Get valid transitions from a state
targets = builder.extract_valid_transitions('PENDING')
# Returns: ['APPROVED', 'REJECTED', 'ESCALATED']
# Build complete transition graph
graph = builder.build_transition_graph()
# Returns: {
# 'PENDING': ['APPROVED', 'REJECTED', 'ESCALATED'],
# 'ESCALATED': ['APPROVED', 'REJECTED'],
# 'APPROVED': [], # Terminal state
# 'REJECTED': [], # Terminal state
# }
# Check state properties
builder.is_terminal_state('APPROVED') # True
builder.is_actionable_state('PENDING') # True
Generate transition method names::
from apps.core.state_machine.builder import determine_method_name_for_transition
method = determine_method_name_for_transition('PENDING', 'APPROVED')
# Returns: 'transition_to_approved'
Rich Choice Metadata Keys:
The builder reads these metadata keys from RichChoice definitions:
- can_transition_to (List[str]): Valid target states from this state
- is_final (bool): Whether this is a terminal state
- is_actionable (bool): Whether this state requires action
- requires_moderator (bool): Whether moderator role is required
- requires_admin_approval (bool): Whether admin role is required
See Also:
- apps.core.choices.base.RichChoice: Choice definition with metadata
- apps.core.choices.registry: Central choice registry
- apps.core.state_machine.guards: Guard extraction from metadata
"""
from typing import Dict, List, Optional, Any
from django.core.exceptions import ImproperlyConfigured
@@ -7,7 +68,47 @@ from apps.core.choices.base import RichChoice
class StateTransitionBuilder:
"""Reads RichChoice metadata and generates FSM transition configurations."""
"""
Reads RichChoice metadata and generates FSM transition configurations.
This class provides a bridge between the rich choice registry and FSM
configuration, extracting transition rules, permissions, and state
properties from choice metadata.
Attributes:
choice_group (str): Name of the choice group in the registry
domain (str): Domain namespace for the choice group
choices: List of RichChoice objects for this group
Example:
Basic usage::
builder = StateTransitionBuilder('ride_status', domain='core')
# Get all states
states = builder.get_all_states()
# ['OPERATING', 'CLOSED_TEMP', 'SBNO', 'CLOSED_PERM', ...]
# Get metadata for a state
metadata = builder.get_choice_metadata('SBNO')
# {'can_transition_to': ['OPERATING', 'CLOSED_PERM'], ...}
# Check state properties
builder.is_terminal_state('DEMOLISHED') # True
builder.is_terminal_state('SBNO') # False
Building transition decorators programmatically::
builder = StateTransitionBuilder('park_status')
graph = builder.build_transition_graph()
for source_state, targets in graph.items():
for target in targets:
method_name = determine_method_name_for_transition(
source_state, target
)
# Create transition method dynamically...
"""
def __init__(self, choice_group: str, domain: str = "core"):
"""

View File

@@ -2,7 +2,68 @@
Callback system infrastructure for FSM state transitions.
This module provides the core classes and registry for managing callbacks
that execute during state machine transitions.
that execute during state machine transitions. Callbacks enable side effects
like notifications, cache invalidation, and related model updates.
Key Components:
- CallbackStage: Enum defining when callbacks execute (pre/post/error)
- TransitionContext: Data class with all transition information
- BaseTransitionCallback: Abstract base class for all callbacks
- TransitionCallbackRegistry: Singleton registry for callback management
Callback Lifecycle:
1. PRE callbacks execute before the state change (can abort transition)
2. State transition occurs
3. POST callbacks execute after successful transition
4. ERROR callbacks execute if transition fails
Example Usage:
Define a custom callback::
from apps.core.state_machine.callbacks import (
PostTransitionCallback,
TransitionContext,
register_post_callback
)
class AuditLogCallback(PostTransitionCallback):
name = "AuditLogCallback"
def execute(self, context: TransitionContext) -> bool:
log_entry = AuditLog.objects.create(
model_name=context.model_name,
object_id=context.instance.pk,
from_state=context.source_state,
to_state=context.target_state,
user=context.user,
timestamp=context.timestamp,
)
return True
# Register the callback
register_post_callback(
model_class=EditSubmission,
field_name='status',
source='*', # Any source state
target='APPROVED', # Only for approvals
callback=AuditLogCallback()
)
Conditional callback execution::
class HighPriorityNotification(PostTransitionCallback):
def should_execute(self, context: TransitionContext) -> bool:
# Only execute for high priority items
return getattr(context.instance, 'priority', None) == 'HIGH'
def execute(self, context: TransitionContext) -> bool:
# Send high priority notification
...
See Also:
- apps.core.state_machine.callbacks.notifications: Notification callbacks
- apps.core.state_machine.callbacks.cache: Cache invalidation callbacks
- apps.core.state_machine.callbacks.related_updates: Related model callbacks
"""
from abc import ABC, abstractmethod
@@ -19,7 +80,42 @@ logger = logging.getLogger(__name__)
class CallbackStage(Enum):
"""Stages at which callbacks can be executed during a transition."""
"""
Stages at which callbacks can be executed during a transition.
Attributes:
PRE: Execute before the state change. Can prevent transition by returning False.
POST: Execute after successful state change. Cannot prevent transition.
ERROR: Execute when transition fails. Used for cleanup and error logging.
Example:
Register callbacks at different stages::
from apps.core.state_machine.callbacks import (
CallbackStage,
callback_registry
)
# PRE callback - validate before transition
callback_registry.register(
model_class=Ride,
field_name='status',
source='OPERATING',
target='SBNO',
callback=ValidationCallback(),
stage=CallbackStage.PRE
)
# POST callback - notify after transition
callback_registry.register(
model_class=Ride,
field_name='status',
source='*',
target='CLOSED_PERM',
callback=NotifyCallback(),
stage=CallbackStage.POST
)
"""
PRE = "pre"
POST = "post"
@@ -31,7 +127,40 @@ class TransitionContext:
"""
Context object passed to callbacks containing transition metadata.
Provides all relevant information about the transition being executed.
Provides all relevant information about the transition being executed,
including the model instance, state values, user, and timing.
Attributes:
instance: The model instance undergoing the transition
field_name: Name of the FSM field (e.g., 'status')
source_state: The state before transition (e.g., 'PENDING')
target_state: The state after transition (e.g., 'APPROVED')
user: The user performing the transition (may be None)
timestamp: When the transition occurred
extra_data: Additional data passed to the transition
Properties:
model_class: The model class of the instance
model_name: String name of the model class
Example:
Access context in a callback::
def execute(self, context: TransitionContext) -> bool:
# Access transition details
print(f"Transitioning {context.model_name} #{context.instance.pk}")
print(f"From: {context.source_state} -> To: {context.target_state}")
print(f"By user: {context.user}")
# Access the instance
if hasattr(context.instance, 'notes'):
context.instance.notes = "Processed by callback"
# Use extra data
if context.extra_data.get('urgent'):
self.send_urgent_notification(context)
return True
"""
instance: models.Model

View File

@@ -532,6 +532,35 @@ class ModerationNotificationCallback(NotificationCallback):
return recipient
return None
def _get_notification_title(self, context: TransitionContext, notification_type: str) -> str:
"""Get the notification title based on notification type."""
titles = {
'report_under_review': 'Your report is under review',
'report_resolved': 'Your report has been resolved',
'queue_in_progress': 'Moderation queue item in progress',
'queue_completed': 'Moderation queue item completed',
'bulk_operation_started': 'Bulk operation started',
'bulk_operation_completed': 'Bulk operation completed',
'bulk_operation_failed': 'Bulk operation failed',
}
return titles.get(notification_type, f"{context.model_name} status updated")
def _get_notification_message(self, context: TransitionContext, notification_type: str) -> str:
"""Get the notification message based on notification type."""
messages = {
'report_under_review': 'Your moderation report is now being reviewed by our team.',
'report_resolved': 'Your moderation report has been reviewed and resolved.',
'queue_in_progress': 'A moderation queue item is now being processed.',
'queue_completed': 'A moderation queue item has been completed.',
'bulk_operation_started': 'Your bulk operation has started processing.',
'bulk_operation_completed': 'Your bulk operation has completed successfully.',
'bulk_operation_failed': 'Your bulk operation encountered an error and could not complete.',
}
return messages.get(
notification_type,
f"The {context.model_name} has been updated to {context.target_state}."
)
def execute(self, context: TransitionContext) -> bool:
"""Execute the moderation notification."""
notification_service = self._get_notification_service()
@@ -556,6 +585,8 @@ class ModerationNotificationCallback(NotificationCallback):
notification_service.create_notification(
user=recipient,
notification_type=notification_type,
title=self._get_notification_title(context, notification_type),
message=self._get_notification_message(context, notification_type),
related_object=context.instance,
extra_data=extra_data,
)

View File

@@ -115,6 +115,9 @@ class ParkCountUpdateCallback(RelatedModelUpdateCallback):
return source_affects or target_affects
# Category value for roller coasters (from rides domain choices)
COASTER_CATEGORY = 'RC'
def perform_update(self, context: TransitionContext) -> bool:
"""Update park ride counts."""
instance = context.instance
@@ -142,10 +145,10 @@ class ParkCountUpdateCallback(RelatedModelUpdateCallback):
status__in=active_statuses
).count()
# Count active coasters
# Count active coasters (category='RC' for Roller Coaster)
coaster_count = ride_queryset.filter(
status__in=active_statuses,
ride_type='ROLLER_COASTER'
category=self.COASTER_CATEGORY
).count()
# Update park counts

View File

@@ -1,4 +1,52 @@
"""State machine fields with rich choice integration."""
"""
State machine fields with rich choice integration.
This module provides FSM field implementations that integrate with the rich
choice registry, enabling metadata-driven state machine definitions.
Key Features:
- Automatic choice population from registry
- Deprecated state handling
- Rich choice metadata access on model instances
- Migration support for custom field attributes
Example Usage:
Define a model with a RichFSMField::
from django.db import models
from apps.core.state_machine.fields import RichFSMField
class EditSubmission(models.Model):
status = RichFSMField(
choice_group='submission_status',
domain='moderation',
default='PENDING'
)
Access rich choice metadata on instances::
submission = EditSubmission.objects.first()
rich_choice = submission.get_status_rich_choice()
print(rich_choice.metadata) # {'is_actionable': True, ...}
print(submission.get_status_display()) # "Pending Review"
Define FSM transitions using django-fsm decorators::
from django_fsm import transition
class EditSubmission(models.Model):
status = RichFSMField(...)
@transition(field=status, source='PENDING', target='APPROVED')
def transition_to_approved(self, user=None):
self.handled_by = user
self.handled_at = timezone.now()
See Also:
- apps.core.choices.base.RichChoice: The choice object with metadata
- apps.core.choices.registry: The central choice registry
- apps.core.state_machine.mixins.StateMachineMixin: Convenience helpers
"""
from typing import Any, Optional
from django.core.exceptions import ValidationError
@@ -9,7 +57,54 @@ from apps.core.choices.registry import registry
class RichFSMField(DjangoFSMField):
"""FSMField that uses the rich choice registry for states."""
"""
FSMField that uses the rich choice registry for states.
This field extends django-fsm's FSMField to integrate with the rich choice
registry system, providing metadata-driven state machine definitions with
automatic choice population and validation.
The field automatically:
- Populates choices from the registry based on choice_group and domain
- Validates state values against the registry
- Handles deprecated states appropriately
- Adds convenience methods to the model class for accessing rich choice data
Attributes:
choice_group (str): Name of the choice group in the registry
domain (str): Domain namespace for the choice group (default: "core")
allow_deprecated (bool): Whether to allow deprecated states (default: False)
Auto-generated Model Methods:
- get_{field_name}_rich_choice(): Returns the RichChoice object for current state
- get_{field_name}_display(): Returns the human-readable label
Example:
Basic field definition::
class Ride(models.Model):
status = RichFSMField(
choice_group='ride_status',
domain='core',
default='OPERATING',
max_length=30
)
Using auto-generated methods::
ride = Ride.objects.get(pk=1)
ride.status # 'OPERATING'
ride.get_status_display() # 'Operating'
ride.get_status_rich_choice() # RichChoice(value='OPERATING', ...)
ride.get_status_rich_choice().metadata # {'icon': 'check', ...}
With deprecated states (for historical data)::
status = RichFSMField(
choice_group='legacy_status',
allow_deprecated=True # Include deprecated choices
)
"""
def __init__(
self,

View File

@@ -1,4 +1,43 @@
"""Base mixins for django-fsm state machines."""
"""
Base mixins for django-fsm state machines.
This module provides abstract model mixins that add convenience methods for
working with FSM-enabled models, including state inspection, transition
checking, and display helpers.
Key Features:
- State value and display access
- Transition availability checking
- Rich choice metadata access
- Consistent interface across all FSM models
Example Usage:
Add the mixin to your FSM model::
from django.db import models
from apps.core.state_machine.mixins import StateMachineMixin
from apps.core.state_machine.fields import RichFSMField
class Park(StateMachineMixin, models.Model):
state_field_name = 'status' # Specify your FSM field name
status = RichFSMField(
choice_group='park_status',
default='OPERATING'
)
Use the convenience methods::
park = Park.objects.first()
park.get_state_value() # 'OPERATING'
park.get_state_display_value() # 'Operating'
park.is_in_state('OPERATING') # True
park.can_transition('transition_to_closed_temp') # True
See Also:
- apps.core.state_machine.fields.RichFSMField: The FSM field implementation
- django_fsm.can_proceed: FSM transition checking utility
"""
from typing import Any, Iterable, Optional
from django.db import models
@@ -6,7 +45,56 @@ from django_fsm import can_proceed
class StateMachineMixin(models.Model):
"""Common helpers for models that use django-fsm."""
"""
Common helpers for models that use django-fsm.
This abstract model mixin provides a consistent interface for working with
FSM-enabled models, including methods for state inspection, transition
checking, and display formatting.
Class Attributes:
state_field_name (str): The name of the FSM field on the model.
Override this in subclasses if your field is not named 'state'.
Default: 'state'
Example:
Basic usage with custom field name::
class Ride(StateMachineMixin, models.Model):
state_field_name = 'status'
status = RichFSMField(...)
@transition(field=status, source='OPERATING', target='SBNO')
def transition_to_sbno(self, user=None):
pass
ride = Ride.objects.first()
# State inspection
ride.get_state_value() # 'OPERATING'
ride.is_in_state('OPERATING') # True
ride.is_in_state('SBNO') # False
# Transition checking
ride.can_transition('transition_to_sbno') # True
# Display formatting
ride.get_state_display_value() # 'Operating'
# Rich choice access (when using RichFSMField)
choice = ride.get_state_choice()
choice.metadata # {'icon': 'check', ...}
Multiple FSM fields::
class ComplexModel(StateMachineMixin, models.Model):
status = RichFSMField(...)
approval_status = RichFSMField(...)
# Access non-default field
model.get_state_value('approval_status')
"""
state_field_name: str = "state"

View File

@@ -1 +1,8 @@
"""Test package initialization."""
"""
State machine test package.
This package contains comprehensive tests for the state machine system including:
- Guard tests (test_guards.py)
- Callback tests (test_callbacks.py)
- Test fixtures and helpers (fixtures.py, helpers.py)
"""

View File

@@ -0,0 +1,372 @@
"""
Test fixtures for state machine tests.
This module provides reusable fixtures for creating test data:
- User factories for different roles
- Model instance factories for moderation, parks, rides
- Mock objects for testing guards and callbacks
"""
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from typing import Optional, Any, Dict
User = get_user_model()
class UserFactory:
"""Factory for creating users with different roles."""
_counter = 0
@classmethod
def _get_unique_id(cls) -> int:
"""Get a unique counter for creating unique usernames."""
cls._counter += 1
return cls._counter
@classmethod
def create_user(
cls,
role: str = 'USER',
username: Optional[str] = None,
email: Optional[str] = None,
password: str = 'testpass123',
**kwargs
) -> User:
"""
Create a user with specified role.
Args:
role: User role (USER, MODERATOR, ADMIN, SUPERUSER)
username: Username (auto-generated if not provided)
email: Email (auto-generated if not provided)
password: Password for the user
**kwargs: Additional user fields
Returns:
Created User instance
"""
uid = cls._get_unique_id()
if username is None:
username = f"user_{role.lower()}_{uid}"
if email is None:
email = f"{role.lower()}_{uid}@example.com"
return User.objects.create_user(
username=username,
email=email,
password=password,
role=role,
**kwargs
)
@classmethod
def create_regular_user(cls, **kwargs) -> User:
"""Create a regular user."""
return cls.create_user(role='USER', **kwargs)
@classmethod
def create_moderator(cls, **kwargs) -> User:
"""Create a moderator user."""
return cls.create_user(role='MODERATOR', **kwargs)
@classmethod
def create_admin(cls, **kwargs) -> User:
"""Create an admin user."""
return cls.create_user(role='ADMIN', **kwargs)
@classmethod
def create_superuser(cls, **kwargs) -> User:
"""Create a superuser."""
return cls.create_user(role='SUPERUSER', **kwargs)
class CompanyFactory:
"""Factory for creating company instances."""
_counter = 0
@classmethod
def _get_unique_id(cls) -> int:
cls._counter += 1
return cls._counter
@classmethod
def create_operator(cls, name: Optional[str] = None, **kwargs) -> Any:
"""Create an operator company."""
from apps.parks.models import Company
uid = cls._get_unique_id()
if name is None:
name = f"Test Operator {uid}"
defaults = {
'name': name,
'description': f'Test operator company {uid}',
'roles': ['OPERATOR']
}
defaults.update(kwargs)
return Company.objects.create(**defaults)
@classmethod
def create_manufacturer(cls, name: Optional[str] = None, **kwargs) -> Any:
"""Create a manufacturer company."""
from apps.rides.models import Company
uid = cls._get_unique_id()
if name is None:
name = f"Test Manufacturer {uid}"
defaults = {
'name': name,
'description': f'Test manufacturer company {uid}',
'roles': ['MANUFACTURER']
}
defaults.update(kwargs)
return Company.objects.create(**defaults)
class ParkFactory:
"""Factory for creating park instances."""
_counter = 0
@classmethod
def _get_unique_id(cls) -> int:
cls._counter += 1
return cls._counter
@classmethod
def create_park(
cls,
name: Optional[str] = None,
operator: Optional[Any] = None,
status: str = 'OPERATING',
**kwargs
) -> Any:
"""
Create a park with specified status.
Args:
name: Park name (auto-generated if not provided)
operator: Operator company (auto-created if not provided)
status: Park status
**kwargs: Additional park fields
Returns:
Created Park instance
"""
from apps.parks.models import Park
uid = cls._get_unique_id()
if name is None:
name = f"Test Park {uid}"
if operator is None:
operator = CompanyFactory.create_operator()
defaults = {
'name': name,
'slug': f'test-park-{uid}',
'description': f'A test park {uid}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
class RideFactory:
"""Factory for creating ride instances."""
_counter = 0
@classmethod
def _get_unique_id(cls) -> int:
cls._counter += 1
return cls._counter
@classmethod
def create_ride(
cls,
name: Optional[str] = None,
park: Optional[Any] = None,
manufacturer: Optional[Any] = None,
status: str = 'OPERATING',
**kwargs
) -> Any:
"""
Create a ride with specified status.
Args:
name: Ride name (auto-generated if not provided)
park: Park for the ride (auto-created if not provided)
manufacturer: Manufacturer company (auto-created if not provided)
status: Ride status
**kwargs: Additional ride fields
Returns:
Created Ride instance
"""
from apps.rides.models import Ride
uid = cls._get_unique_id()
if name is None:
name = f"Test Ride {uid}"
if park is None:
park = ParkFactory.create_park()
if manufacturer is None:
manufacturer = CompanyFactory.create_manufacturer()
defaults = {
'name': name,
'slug': f'test-ride-{uid}',
'description': f'A test ride {uid}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
class EditSubmissionFactory:
"""Factory for creating edit submission instances."""
_counter = 0
@classmethod
def _get_unique_id(cls) -> int:
cls._counter += 1
return cls._counter
@classmethod
def create_submission(
cls,
user: Optional[Any] = None,
target_object: Optional[Any] = None,
status: str = 'PENDING',
changes: Optional[Dict[str, Any]] = None,
**kwargs
) -> Any:
"""
Create an edit submission.
Args:
user: User who submitted (auto-created if not provided)
target_object: Object being edited (auto-created if not provided)
status: Submission status
changes: Changes dictionary
**kwargs: Additional fields
Returns:
Created EditSubmission instance
"""
from apps.moderation.models import EditSubmission
from apps.parks.models import Company
uid = cls._get_unique_id()
if user is None:
user = UserFactory.create_regular_user()
if target_object is None:
target_object = Company.objects.create(
name=f'Target Company {uid}',
description='Test company'
)
if changes is None:
changes = {'name': f'Updated Name {uid}'}
content_type = ContentType.objects.get_for_model(target_object)
defaults = {
'user': user,
'content_type': content_type,
'object_id': target_object.id,
'submission_type': 'EDIT',
'changes': changes,
'status': status,
'reason': f'Test reason {uid}'
}
defaults.update(kwargs)
return EditSubmission.objects.create(**defaults)
class ModerationReportFactory:
"""Factory for creating moderation report instances."""
_counter = 0
@classmethod
def _get_unique_id(cls) -> int:
cls._counter += 1
return cls._counter
@classmethod
def create_report(
cls,
reporter: Optional[Any] = None,
target_object: Optional[Any] = None,
status: str = 'PENDING',
**kwargs
) -> Any:
"""
Create a moderation report.
Args:
reporter: User who reported (auto-created if not provided)
target_object: Object being reported (auto-created if not provided)
status: Report status
**kwargs: Additional fields
Returns:
Created ModerationReport instance
"""
from apps.moderation.models import ModerationReport
from apps.parks.models import Company
uid = cls._get_unique_id()
if reporter is None:
reporter = UserFactory.create_regular_user()
if target_object is None:
target_object = Company.objects.create(
name=f'Reported Company {uid}',
description='Test company'
)
content_type = ContentType.objects.get_for_model(target_object)
defaults = {
'report_type': 'CONTENT',
'status': status,
'priority': 'MEDIUM',
'reported_entity_type': target_object._meta.model_name,
'reported_entity_id': target_object.id,
'content_type': content_type,
'reason': f'Test reason {uid}',
'description': f'Test report description {uid}',
'reported_by': reporter
}
defaults.update(kwargs)
return ModerationReport.objects.create(**defaults)
class MockInstance:
"""
Mock instance for testing guards without database.
Example:
instance = MockInstance(
status='PENDING',
created_by=user,
assigned_to=moderator
)
"""
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
def __repr__(self):
attrs = ', '.join(f'{k}={v!r}' for k, v in self.__dict__.items())
return f'MockInstance({attrs})'

View File

@@ -0,0 +1,340 @@
"""
Test helper functions for state machine tests.
This module provides utility functions for testing state machine functionality:
- Transition assertion helpers
- State log verification helpers
- Guard testing utilities
"""
from typing import Any, Optional, List, Callable
from django.contrib.contenttypes.models import ContentType
def assert_transition_allowed(
instance: Any,
method_name: str,
user: Optional[Any] = None
) -> bool:
"""
Assert that a transition is allowed.
Args:
instance: Model instance with FSM field
method_name: Name of the transition method
user: User attempting the transition
Returns:
True if transition is allowed
Raises:
AssertionError: If transition is not allowed
Example:
assert_transition_allowed(submission, 'transition_to_approved', moderator)
"""
from django_fsm import can_proceed
method = getattr(instance, method_name)
result = can_proceed(method)
assert result, f"Transition {method_name} should be allowed but was denied"
return True
def assert_transition_denied(
instance: Any,
method_name: str,
user: Optional[Any] = None
) -> bool:
"""
Assert that a transition is denied.
Args:
instance: Model instance with FSM field
method_name: Name of the transition method
user: User attempting the transition
Returns:
True if transition is denied
Raises:
AssertionError: If transition is allowed
Example:
assert_transition_denied(submission, 'transition_to_approved', regular_user)
"""
from django_fsm import can_proceed
method = getattr(instance, method_name)
result = can_proceed(method)
assert not result, f"Transition {method_name} should be denied but was allowed"
return True
def assert_state_log_created(
instance: Any,
expected_state: str,
user: Optional[Any] = None
) -> Any:
"""
Assert that a StateLog entry was created for a transition.
Args:
instance: Model instance that was transitioned
expected_state: The expected final state in the log
user: Expected user who made the transition (optional)
Returns:
The StateLog entry
Raises:
AssertionError: If StateLog entry not found or doesn't match
Example:
log = assert_state_log_created(submission, 'APPROVED', moderator)
"""
from django_fsm_log.models import StateLog
ct = ContentType.objects.get_for_model(instance)
log = StateLog.objects.filter(
content_type=ct,
object_id=instance.id,
state=expected_state
).first()
assert log is not None, f"StateLog for state '{expected_state}' not found"
if user is not None:
assert log.by == user, f"Expected log.by={user}, got {log.by}"
return log
def assert_state_log_count(instance: Any, expected_count: int) -> List[Any]:
"""
Assert the number of StateLog entries for an instance.
Args:
instance: Model instance to check logs for
expected_count: Expected number of log entries
Returns:
List of StateLog entries
Raises:
AssertionError: If count doesn't match
Example:
logs = assert_state_log_count(submission, 2)
"""
from django_fsm_log.models import StateLog
ct = ContentType.objects.get_for_model(instance)
logs = list(StateLog.objects.filter(
content_type=ct,
object_id=instance.id
).order_by('timestamp'))
actual_count = len(logs)
assert actual_count == expected_count, \
f"Expected {expected_count} StateLog entries, got {actual_count}"
return logs
def assert_state_transition_sequence(
instance: Any,
expected_states: List[str]
) -> List[Any]:
"""
Assert that state transitions occurred in a specific sequence.
Args:
instance: Model instance to check
expected_states: List of expected states in order
Returns:
List of StateLog entries
Raises:
AssertionError: If sequence doesn't match
Example:
assert_state_transition_sequence(submission, ['ESCALATED', 'APPROVED'])
"""
from django_fsm_log.models import StateLog
ct = ContentType.objects.get_for_model(instance)
logs = list(StateLog.objects.filter(
content_type=ct,
object_id=instance.id
).order_by('timestamp'))
actual_states = [log.state for log in logs]
assert actual_states == expected_states, \
f"Expected state sequence {expected_states}, got {actual_states}"
return logs
def assert_guard_passes(
guard: Callable,
instance: Any,
user: Optional[Any] = None,
message: str = ""
) -> bool:
"""
Assert that a guard function passes.
Args:
guard: Guard function or callable
instance: Model instance to check
user: User attempting the action
message: Optional message on failure
Returns:
True if guard passes
Raises:
AssertionError: If guard fails
Example:
assert_guard_passes(permission_guard, instance, moderator)
"""
result = guard(instance, user)
fail_message = message or f"Guard should pass but returned {result}"
assert result is True, fail_message
return True
def assert_guard_fails(
guard: Callable,
instance: Any,
user: Optional[Any] = None,
expected_error_code: Optional[str] = None,
message: str = ""
) -> bool:
"""
Assert that a guard function fails.
Args:
guard: Guard function or callable
instance: Model instance to check
user: User attempting the action
expected_error_code: Expected error code from guard
message: Optional message on failure
Returns:
True if guard fails as expected
Raises:
AssertionError: If guard passes or wrong error code
Example:
assert_guard_fails(permission_guard, instance, regular_user, 'PERMISSION_DENIED')
"""
result = guard(instance, user)
fail_message = message or f"Guard should fail but returned {result}"
assert result is False, fail_message
if expected_error_code and hasattr(guard, 'error_code'):
assert guard.error_code == expected_error_code, \
f"Expected error code {expected_error_code}, got {guard.error_code}"
return True
def transition_and_save(
instance: Any,
transition_method: str,
user: Optional[Any] = None,
**kwargs
) -> Any:
"""
Execute a transition and save the instance.
Args:
instance: Model instance with FSM field
transition_method: Name of the transition method
user: User performing the transition
**kwargs: Additional arguments for the transition
Returns:
The saved instance
Example:
submission = transition_and_save(submission, 'transition_to_approved', moderator)
"""
method = getattr(instance, transition_method)
method(user=user, **kwargs)
instance.save()
instance.refresh_from_db()
return instance
def get_available_transitions(instance: Any) -> List[str]:
"""
Get list of available transitions for an instance.
Args:
instance: Model instance with FSM field
Returns:
List of available transition method names
Example:
transitions = get_available_transitions(submission)
# ['transition_to_approved', 'transition_to_rejected', 'transition_to_escalated']
"""
from django_fsm import get_available_FIELD_transitions
# Get the state field name from the instance
state_field = getattr(instance, 'state_field_name', 'status')
# Build the function name dynamically
func_name = f'get_available_{state_field}_transitions'
if hasattr(instance, func_name):
get_transitions = getattr(instance, func_name)
return [t.name for t in get_transitions()]
# Fallback: look for transition methods
transitions = []
for attr_name in dir(instance):
if attr_name.startswith('transition_to_'):
transitions.append(attr_name)
return transitions
def create_transition_context(
instance: Any,
from_state: str,
to_state: str,
user: Optional[Any] = None,
**extra
) -> dict:
"""
Create a mock transition context dictionary.
Args:
instance: Model instance being transitioned
from_state: Source state
to_state: Target state
user: User performing the transition
**extra: Additional context data
Returns:
Dictionary matching TransitionContext structure
Example:
context = create_transition_context(submission, 'PENDING', 'APPROVED', moderator)
"""
return {
'instance': instance,
'from_state': from_state,
'to_state': to_state,
'user': user,
'model_class': type(instance),
'transition_name': f'transition_to_{to_state.lower()}',
**extra
}

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -1,10 +1,32 @@
"""
Comprehensive tests for the moderation app.
This module contains tests for:
- EditSubmission state machine transitions
- PhotoSubmission state machine transitions
- ModerationReport state machine transitions
- ModerationQueue state machine transitions
- BulkOperation state machine transitions
- FSM transition logging with django-fsm-log
- Mixin functionality tests
"""
from django.test import TestCase, Client
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.contrib.contenttypes.models import ContentType
from django.core.files.uploadedfile import SimpleUploadedFile
from django.http import JsonResponse, HttpRequest
from .models import EditSubmission
from django.utils import timezone
from django_fsm import TransitionNotAllowed
from .models import (
EditSubmission,
PhotoSubmission,
ModerationReport,
ModerationQueue,
BulkOperation,
ModerationAction,
)
from .mixins import (
EditSubmissionMixin,
PhotoSubmissionMixin,
@@ -349,6 +371,480 @@ class ModerationMixinsTests(TestCase):
self.assertEqual(len(context["edit_submissions"]), 1)
# ============================================================================
# EditSubmission FSM Transition Tests
# ============================================================================
class EditSubmissionTransitionTests(TestCase):
"""Comprehensive tests for EditSubmission FSM transitions."""
def setUp(self):
"""Set up test fixtures."""
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123',
role='USER'
)
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
self.admin = User.objects.create_user(
username='admin',
email='admin@example.com',
password='testpass123',
role='ADMIN'
)
self.operator = Operator.objects.create(
name='Test Operator',
description='Test Description'
)
self.content_type = ContentType.objects.get_for_model(Operator)
def _create_submission(self, status='PENDING'):
"""Helper to create an EditSubmission."""
return EditSubmission.objects.create(
user=self.user,
content_type=self.content_type,
object_id=self.operator.id,
submission_type='EDIT',
changes={'name': 'Updated Name'},
status=status,
reason='Test reason'
)
def test_pending_to_approved_transition(self):
"""Test transition from PENDING to APPROVED."""
submission = self._create_submission()
self.assertEqual(submission.status, 'PENDING')
submission.transition_to_approved(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'APPROVED')
self.assertEqual(submission.handled_by, self.moderator)
self.assertIsNotNone(submission.handled_at)
def test_pending_to_rejected_transition(self):
"""Test transition from PENDING to REJECTED."""
submission = self._create_submission()
self.assertEqual(submission.status, 'PENDING')
submission.transition_to_rejected(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.notes = 'Rejected: Insufficient evidence'
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'REJECTED')
self.assertEqual(submission.handled_by, self.moderator)
self.assertIn('Rejected', submission.notes)
def test_pending_to_escalated_transition(self):
"""Test transition from PENDING to ESCALATED."""
submission = self._create_submission()
self.assertEqual(submission.status, 'PENDING')
submission.transition_to_escalated(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.notes = 'Escalated: Needs admin review'
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'ESCALATED')
def test_escalated_to_approved_transition(self):
"""Test transition from ESCALATED to APPROVED."""
submission = self._create_submission(status='ESCALATED')
submission.transition_to_approved(user=self.admin)
submission.handled_by = self.admin
submission.handled_at = timezone.now()
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'APPROVED')
self.assertEqual(submission.handled_by, self.admin)
def test_escalated_to_rejected_transition(self):
"""Test transition from ESCALATED to REJECTED."""
submission = self._create_submission(status='ESCALATED')
submission.transition_to_rejected(user=self.admin)
submission.handled_by = self.admin
submission.handled_at = timezone.now()
submission.notes = 'Rejected by admin'
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'REJECTED')
def test_invalid_transition_from_approved(self):
"""Test that transitions from APPROVED state fail."""
submission = self._create_submission(status='APPROVED')
# Attempting to transition from APPROVED should raise TransitionNotAllowed
with self.assertRaises(TransitionNotAllowed):
submission.transition_to_rejected(user=self.moderator)
def test_invalid_transition_from_rejected(self):
"""Test that transitions from REJECTED state fail."""
submission = self._create_submission(status='REJECTED')
# Attempting to transition from REJECTED should raise TransitionNotAllowed
with self.assertRaises(TransitionNotAllowed):
submission.transition_to_approved(user=self.moderator)
def test_approve_wrapper_method(self):
"""Test the approve() wrapper method."""
submission = self._create_submission()
result = submission.approve(self.moderator)
submission.refresh_from_db()
self.assertEqual(submission.status, 'APPROVED')
self.assertEqual(submission.handled_by, self.moderator)
self.assertIsNotNone(submission.handled_at)
def test_reject_wrapper_method(self):
"""Test the reject() wrapper method."""
submission = self._create_submission()
submission.reject(self.moderator, reason='Not enough evidence')
submission.refresh_from_db()
self.assertEqual(submission.status, 'REJECTED')
self.assertIn('Not enough evidence', submission.notes)
def test_escalate_wrapper_method(self):
"""Test the escalate() wrapper method."""
submission = self._create_submission()
submission.escalate(self.moderator, reason='Needs admin approval')
submission.refresh_from_db()
self.assertEqual(submission.status, 'ESCALATED')
self.assertIn('Needs admin approval', submission.notes)
# ============================================================================
# ModerationReport FSM Transition Tests
# ============================================================================
class ModerationReportTransitionTests(TestCase):
"""Comprehensive tests for ModerationReport FSM transitions."""
def setUp(self):
"""Set up test fixtures."""
self.user = User.objects.create_user(
username='reporter',
email='reporter@example.com',
password='testpass123',
role='USER'
)
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
self.operator = Operator.objects.create(
name='Test Operator',
description='Test Description'
)
self.content_type = ContentType.objects.get_for_model(Operator)
def _create_report(self, status='PENDING'):
"""Helper to create a ModerationReport."""
return ModerationReport.objects.create(
report_type='CONTENT',
status=status,
priority='MEDIUM',
reported_entity_type='company',
reported_entity_id=self.operator.id,
content_type=self.content_type,
reason='Inaccurate information',
description='The company information is incorrect',
reported_by=self.user
)
def test_pending_to_under_review_transition(self):
"""Test transition from PENDING to UNDER_REVIEW."""
report = self._create_report()
self.assertEqual(report.status, 'PENDING')
report.transition_to_under_review(user=self.moderator)
report.assigned_moderator = self.moderator
report.save()
report.refresh_from_db()
self.assertEqual(report.status, 'UNDER_REVIEW')
self.assertEqual(report.assigned_moderator, self.moderator)
def test_under_review_to_resolved_transition(self):
"""Test transition from UNDER_REVIEW to RESOLVED."""
report = self._create_report(status='UNDER_REVIEW')
report.assigned_moderator = self.moderator
report.save()
report.transition_to_resolved(user=self.moderator)
report.resolution_action = 'Content updated'
report.resolution_notes = 'Fixed the incorrect information'
report.resolved_at = timezone.now()
report.save()
report.refresh_from_db()
self.assertEqual(report.status, 'RESOLVED')
self.assertIsNotNone(report.resolved_at)
def test_under_review_to_dismissed_transition(self):
"""Test transition from UNDER_REVIEW to DISMISSED."""
report = self._create_report(status='UNDER_REVIEW')
report.assigned_moderator = self.moderator
report.save()
report.transition_to_dismissed(user=self.moderator)
report.resolution_notes = 'Report is not valid'
report.resolved_at = timezone.now()
report.save()
report.refresh_from_db()
self.assertEqual(report.status, 'DISMISSED')
def test_invalid_transition_from_resolved(self):
"""Test that transitions from RESOLVED state fail."""
report = self._create_report(status='RESOLVED')
with self.assertRaises(TransitionNotAllowed):
report.transition_to_dismissed(user=self.moderator)
def test_invalid_transition_from_dismissed(self):
"""Test that transitions from DISMISSED state fail."""
report = self._create_report(status='DISMISSED')
with self.assertRaises(TransitionNotAllowed):
report.transition_to_resolved(user=self.moderator)
# ============================================================================
# ModerationQueue FSM Transition Tests
# ============================================================================
class ModerationQueueTransitionTests(TestCase):
"""Comprehensive tests for ModerationQueue FSM transitions."""
def setUp(self):
"""Set up test fixtures."""
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_queue_item(self, status='PENDING'):
"""Helper to create a ModerationQueue item."""
return ModerationQueue.objects.create(
item_type='EDIT_SUBMISSION',
status=status,
priority='MEDIUM',
title='Review edit submission',
description='User submitted an edit that needs review',
flagged_by=self.moderator
)
def test_pending_to_in_progress_transition(self):
"""Test transition from PENDING to IN_PROGRESS."""
item = self._create_queue_item()
self.assertEqual(item.status, 'PENDING')
item.transition_to_in_progress(user=self.moderator)
item.assigned_to = self.moderator
item.assigned_at = timezone.now()
item.save()
item.refresh_from_db()
self.assertEqual(item.status, 'IN_PROGRESS')
self.assertEqual(item.assigned_to, self.moderator)
def test_in_progress_to_completed_transition(self):
"""Test transition from IN_PROGRESS to COMPLETED."""
item = self._create_queue_item(status='IN_PROGRESS')
item.assigned_to = self.moderator
item.save()
item.transition_to_completed(user=self.moderator)
item.save()
item.refresh_from_db()
self.assertEqual(item.status, 'COMPLETED')
def test_in_progress_to_cancelled_transition(self):
"""Test transition from IN_PROGRESS to CANCELLED."""
item = self._create_queue_item(status='IN_PROGRESS')
item.assigned_to = self.moderator
item.save()
item.transition_to_cancelled(user=self.moderator)
item.save()
item.refresh_from_db()
self.assertEqual(item.status, 'CANCELLED')
def test_pending_to_cancelled_transition(self):
"""Test transition from PENDING to CANCELLED."""
item = self._create_queue_item()
item.transition_to_cancelled(user=self.moderator)
item.save()
item.refresh_from_db()
self.assertEqual(item.status, 'CANCELLED')
def test_invalid_transition_from_completed(self):
"""Test that transitions from COMPLETED state fail."""
item = self._create_queue_item(status='COMPLETED')
with self.assertRaises(TransitionNotAllowed):
item.transition_to_in_progress(user=self.moderator)
# ============================================================================
# BulkOperation FSM Transition Tests
# ============================================================================
class BulkOperationTransitionTests(TestCase):
"""Comprehensive tests for BulkOperation FSM transitions."""
def setUp(self):
"""Set up test fixtures."""
self.admin = User.objects.create_user(
username='admin',
email='admin@example.com',
password='testpass123',
role='ADMIN'
)
def _create_bulk_operation(self, status='PENDING'):
"""Helper to create a BulkOperation."""
return BulkOperation.objects.create(
operation_type='BULK_UPDATE',
status=status,
priority='MEDIUM',
description='Bulk update park statuses',
parameters={'target': 'parks', 'action': 'update_status'},
created_by=self.admin,
total_items=100
)
def test_pending_to_running_transition(self):
"""Test transition from PENDING to RUNNING."""
operation = self._create_bulk_operation()
self.assertEqual(operation.status, 'PENDING')
operation.transition_to_running(user=self.admin)
operation.started_at = timezone.now()
operation.save()
operation.refresh_from_db()
self.assertEqual(operation.status, 'RUNNING')
self.assertIsNotNone(operation.started_at)
def test_running_to_completed_transition(self):
"""Test transition from RUNNING to COMPLETED."""
operation = self._create_bulk_operation(status='RUNNING')
operation.started_at = timezone.now()
operation.save()
operation.transition_to_completed(user=self.admin)
operation.completed_at = timezone.now()
operation.processed_items = 100
operation.save()
operation.refresh_from_db()
self.assertEqual(operation.status, 'COMPLETED')
self.assertIsNotNone(operation.completed_at)
self.assertEqual(operation.processed_items, 100)
def test_running_to_failed_transition(self):
"""Test transition from RUNNING to FAILED."""
operation = self._create_bulk_operation(status='RUNNING')
operation.started_at = timezone.now()
operation.save()
operation.transition_to_failed(user=self.admin)
operation.completed_at = timezone.now()
operation.results = {'error': 'Database connection failed'}
operation.failed_items = 50
operation.save()
operation.refresh_from_db()
self.assertEqual(operation.status, 'FAILED')
self.assertEqual(operation.failed_items, 50)
def test_pending_to_cancelled_transition(self):
"""Test transition from PENDING to CANCELLED."""
operation = self._create_bulk_operation()
operation.transition_to_cancelled(user=self.admin)
operation.save()
operation.refresh_from_db()
self.assertEqual(operation.status, 'CANCELLED')
def test_running_to_cancelled_transition(self):
"""Test transition from RUNNING to CANCELLED when cancellable."""
operation = self._create_bulk_operation(status='RUNNING')
operation.can_cancel = True
operation.save()
operation.transition_to_cancelled(user=self.admin)
operation.save()
operation.refresh_from_db()
self.assertEqual(operation.status, 'CANCELLED')
def test_invalid_transition_from_completed(self):
"""Test that transitions from COMPLETED state fail."""
operation = self._create_bulk_operation(status='COMPLETED')
with self.assertRaises(TransitionNotAllowed):
operation.transition_to_running(user=self.admin)
def test_invalid_transition_from_failed(self):
"""Test that transitions from FAILED state fail."""
operation = self._create_bulk_operation(status='FAILED')
with self.assertRaises(TransitionNotAllowed):
operation.transition_to_completed(user=self.admin)
def test_progress_percentage_calculation(self):
"""Test progress percentage property."""
operation = self._create_bulk_operation()
operation.total_items = 100
operation.processed_items = 50
self.assertEqual(operation.progress_percentage, 50.0)
operation.processed_items = 0
self.assertEqual(operation.progress_percentage, 0.0)
operation.total_items = 0
self.assertEqual(operation.progress_percentage, 0.0)
# ============================================================================
# FSM Transition Logging Tests
# ============================================================================
@@ -388,7 +884,8 @@ class TransitionLoggingTestCase(TestCase):
object_id=self.operator.id,
submission_type='EDIT',
changes={'name': 'Updated Name'},
status='PENDING'
status='PENDING',
reason='Test reason'
)
# Perform transition
@@ -417,7 +914,8 @@ class TransitionLoggingTestCase(TestCase):
object_id=self.operator.id,
submission_type='EDIT',
changes={'name': 'Updated Name'},
status='PENDING'
status='PENDING',
reason='Test reason'
)
submission_ct = ContentType.objects.get_for_model(submission)
@@ -443,7 +941,6 @@ class TransitionLoggingTestCase(TestCase):
def test_history_endpoint_returns_logs(self):
"""Test history API endpoint returns transition logs."""
from rest_framework.test import APIClient
from django_fsm_log.models import StateLog
api_client = APIClient()
api_client.force_authenticate(user=self.moderator)
@@ -454,7 +951,8 @@ class TransitionLoggingTestCase(TestCase):
object_id=self.operator.id,
submission_type='EDIT',
changes={'name': 'Updated Name'},
status='PENDING'
status='PENDING',
reason='Test reason'
)
# Perform transition to create log
@@ -463,11 +961,9 @@ class TransitionLoggingTestCase(TestCase):
# Note: This assumes EditSubmission has a history endpoint
# Adjust URL pattern based on actual implementation
response = api_client.get(f'/api/moderation/reports/all_history/')
response = api_client.get('/api/moderation/reports/all_history/')
self.assertEqual(response.status_code, 200)
# Response should contain history data
# Actual assertions depend on response format
def test_system_transitions_without_user(self):
"""Test that system transitions work without a user."""
@@ -479,7 +975,8 @@ class TransitionLoggingTestCase(TestCase):
object_id=self.operator.id,
submission_type='EDIT',
changes={'name': 'Updated Name'},
status='PENDING'
status='PENDING',
reason='Test reason'
)
# Perform transition without user
@@ -507,7 +1004,8 @@ class TransitionLoggingTestCase(TestCase):
object_id=self.operator.id,
submission_type='EDIT',
changes={'name': 'Updated Name'},
status='PENDING'
status='PENDING',
reason='Test reason'
)
# Perform transition
@@ -525,3 +1023,367 @@ class TransitionLoggingTestCase(TestCase):
# Description field exists and can be used for audit trails
self.assertTrue(hasattr(log, 'description'))
def test_log_ordering_by_timestamp(self):
"""Test that logs are properly ordered by timestamp."""
from django_fsm_log.models import StateLog
import time
submission = EditSubmission.objects.create(
user=self.user,
content_type=self.content_type,
object_id=self.operator.id,
submission_type='EDIT',
changes={'name': 'Updated Name'},
status='PENDING',
reason='Test reason'
)
submission_ct = ContentType.objects.get_for_model(submission)
# Create multiple transitions
submission.transition_to_escalated(user=self.moderator)
submission.save()
submission.transition_to_approved(user=self.moderator)
submission.save()
# Get logs ordered by timestamp
logs = list(StateLog.objects.filter(
content_type=submission_ct,
object_id=submission.id
).order_by('timestamp'))
# Verify ordering
self.assertEqual(len(logs), 2)
self.assertTrue(logs[0].timestamp <= logs[1].timestamp)
# ============================================================================
# ModerationAction Model Tests
# ============================================================================
class ModerationActionTests(TestCase):
"""Tests for ModerationAction model."""
def setUp(self):
"""Set up test fixtures."""
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
self.target_user = User.objects.create_user(
username='target',
email='target@example.com',
password='testpass123',
role='USER'
)
def test_create_action_with_duration(self):
"""Test creating an action with duration sets expires_at."""
action = ModerationAction.objects.create(
action_type='TEMPORARY_BAN',
reason='Spam',
details='User was spamming the forums',
duration_hours=24,
moderator=self.moderator,
target_user=self.target_user
)
self.assertIsNotNone(action.expires_at)
# expires_at should be approximately 24 hours from now
time_diff = action.expires_at - timezone.now()
self.assertAlmostEqual(time_diff.total_seconds(), 24 * 3600, delta=60)
def test_create_action_without_duration(self):
"""Test creating an action without duration has no expires_at."""
action = ModerationAction.objects.create(
action_type='WARNING',
reason='First offense',
details='Warning issued for minor violation',
moderator=self.moderator,
target_user=self.target_user
)
self.assertIsNone(action.expires_at)
def test_action_is_active_by_default(self):
"""Test that new actions are active by default."""
action = ModerationAction.objects.create(
action_type='WARNING',
reason='Test',
details='Test warning',
moderator=self.moderator,
target_user=self.target_user
)
self.assertTrue(action.is_active)
# ============================================================================
# PhotoSubmission FSM Transition Tests
# ============================================================================
class PhotoSubmissionTransitionTests(TestCase):
"""Comprehensive tests for PhotoSubmission FSM transitions."""
def setUp(self):
"""Set up test fixtures."""
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123',
role='USER'
)
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
self.admin = User.objects.create_user(
username='admin',
email='admin@example.com',
password='testpass123',
role='ADMIN'
)
self.operator = Operator.objects.create(
name='Test Operator',
description='Test Description',
roles=['OPERATOR']
)
self.content_type = ContentType.objects.get_for_model(Operator)
def _create_mock_photo(self):
"""Create a mock CloudflareImage for testing."""
from unittest.mock import Mock
mock_photo = Mock()
mock_photo.pk = 1
mock_photo.id = 1
return mock_photo
def _create_submission(self, status='PENDING'):
"""Helper to create a PhotoSubmission."""
# Create using direct database creation to bypass FK validation
from unittest.mock import patch, Mock
with patch.object(PhotoSubmission, 'photo', Mock()):
submission = PhotoSubmission(
user=self.user,
content_type=self.content_type,
object_id=self.operator.id,
caption='Test Photo',
status=status,
)
# Bypass model save to avoid FK constraint on photo
submission.photo_id = 1
submission.save(update_fields=None)
# Force status after creation for non-PENDING states
if status != 'PENDING':
PhotoSubmission.objects.filter(pk=submission.pk).update(status=status)
submission.refresh_from_db()
return submission
def test_pending_to_approved_transition(self):
"""Test transition from PENDING to APPROVED."""
submission = self._create_submission()
self.assertEqual(submission.status, 'PENDING')
submission.transition_to_approved(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'APPROVED')
self.assertEqual(submission.handled_by, self.moderator)
self.assertIsNotNone(submission.handled_at)
def test_pending_to_rejected_transition(self):
"""Test transition from PENDING to REJECTED."""
submission = self._create_submission()
self.assertEqual(submission.status, 'PENDING')
submission.transition_to_rejected(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.notes = 'Rejected: Image quality too low'
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'REJECTED')
self.assertEqual(submission.handled_by, self.moderator)
self.assertIn('Rejected', submission.notes)
def test_pending_to_escalated_transition(self):
"""Test transition from PENDING to ESCALATED."""
submission = self._create_submission()
self.assertEqual(submission.status, 'PENDING')
submission.transition_to_escalated(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.notes = 'Escalated: Copyright concerns'
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'ESCALATED')
def test_escalated_to_approved_transition(self):
"""Test transition from ESCALATED to APPROVED."""
submission = self._create_submission(status='ESCALATED')
submission.transition_to_approved(user=self.admin)
submission.handled_by = self.admin
submission.handled_at = timezone.now()
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'APPROVED')
self.assertEqual(submission.handled_by, self.admin)
def test_escalated_to_rejected_transition(self):
"""Test transition from ESCALATED to REJECTED."""
submission = self._create_submission(status='ESCALATED')
submission.transition_to_rejected(user=self.admin)
submission.handled_by = self.admin
submission.handled_at = timezone.now()
submission.notes = 'Rejected by admin after review'
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'REJECTED')
def test_invalid_transition_from_approved(self):
"""Test that transitions from APPROVED state fail."""
submission = self._create_submission(status='APPROVED')
with self.assertRaises(TransitionNotAllowed):
submission.transition_to_rejected(user=self.moderator)
def test_invalid_transition_from_rejected(self):
"""Test that transitions from REJECTED state fail."""
submission = self._create_submission(status='REJECTED')
with self.assertRaises(TransitionNotAllowed):
submission.transition_to_approved(user=self.moderator)
def test_reject_wrapper_method(self):
"""Test the reject() wrapper method."""
from unittest.mock import patch
submission = self._create_submission()
# Mock the photo creation part since we don't have actual photos
with patch.object(submission, 'transition_to_rejected'):
submission.reject(self.moderator, notes='Not suitable')
submission.refresh_from_db()
self.assertEqual(submission.status, 'REJECTED')
self.assertIn('Not suitable', submission.notes)
def test_escalate_wrapper_method(self):
"""Test the escalate() wrapper method."""
from unittest.mock import patch
submission = self._create_submission()
with patch.object(submission, 'transition_to_escalated'):
submission.escalate(self.moderator, notes='Needs admin review')
submission.refresh_from_db()
self.assertEqual(submission.status, 'ESCALATED')
self.assertIn('Needs admin review', submission.notes)
def test_transition_creates_state_log(self):
"""Test that transitions create StateLog entries."""
from django_fsm_log.models import StateLog
submission = self._create_submission()
# Perform transition
submission.transition_to_approved(user=self.moderator)
submission.save()
# Check log was created
submission_ct = ContentType.objects.get_for_model(submission)
log = StateLog.objects.filter(
content_type=submission_ct,
object_id=submission.id
).first()
self.assertIsNotNone(log, "StateLog entry should be created")
self.assertEqual(log.state, 'APPROVED')
self.assertEqual(log.by, self.moderator)
def test_multiple_transitions_logged(self):
"""Test that multiple transitions are all logged."""
from django_fsm_log.models import StateLog
submission = self._create_submission()
submission_ct = ContentType.objects.get_for_model(submission)
# First transition: PENDING -> ESCALATED
submission.transition_to_escalated(user=self.moderator)
submission.save()
# Second transition: ESCALATED -> APPROVED
submission.transition_to_approved(user=self.admin)
submission.save()
# Check multiple logs created
logs = StateLog.objects.filter(
content_type=submission_ct,
object_id=submission.id
).order_by('timestamp')
self.assertEqual(logs.count(), 2, "Should have 2 log entries")
self.assertEqual(logs[0].state, 'ESCALATED')
self.assertEqual(logs[1].state, 'APPROVED')
def test_handled_by_and_handled_at_updated(self):
"""Test that handled_by and handled_at are properly updated."""
submission = self._create_submission()
self.assertIsNone(submission.handled_by)
self.assertIsNone(submission.handled_at)
before_time = timezone.now()
submission.transition_to_approved(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.save()
after_time = timezone.now()
submission.refresh_from_db()
self.assertEqual(submission.handled_by, self.moderator)
self.assertIsNotNone(submission.handled_at)
self.assertTrue(before_time <= submission.handled_at <= after_time)
def test_notes_field_updated_on_rejection(self):
"""Test that notes field is updated with rejection reason."""
submission = self._create_submission()
rejection_reason = 'Image contains watermarks'
submission.transition_to_rejected(user=self.moderator)
submission.notes = rejection_reason
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.notes, rejection_reason)
def test_notes_field_updated_on_escalation(self):
"""Test that notes field is updated with escalation reason."""
submission = self._create_submission()
escalation_reason = 'Potentially copyrighted content'
submission.transition_to_escalated(user=self.moderator)
submission.notes = escalation_reason
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.notes, escalation_reason)

View File

@@ -0,0 +1,7 @@
"""
Moderation test package.
This package contains tests for the moderation app including:
- Workflow tests (test_workflows.py)
- Permission tests (test_permissions.py - planned)
"""

View File

@@ -0,0 +1,532 @@
"""
Integration tests for complete moderation workflows.
This module tests end-to-end moderation workflows including:
- Submission approval workflow
- Submission rejection workflow
- Submission escalation workflow
- Report handling workflow
- Bulk operation workflow
"""
from django.test import TestCase
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from django.utils import timezone
from unittest.mock import patch, Mock
User = get_user_model()
class SubmissionApprovalWorkflowTests(TestCase):
"""Tests for the complete submission approval workflow."""
@classmethod
def setUpTestData(cls):
"""Set up test data for all tests."""
cls.regular_user = User.objects.create_user(
username='regular_user',
email='user@example.com',
password='testpass123',
role='USER'
)
cls.moderator = User.objects.create_user(
username='moderator',
email='mod@example.com',
password='testpass123',
role='MODERATOR'
)
cls.admin = User.objects.create_user(
username='admin',
email='admin@example.com',
password='testpass123',
role='ADMIN'
)
def test_edit_submission_approval_workflow(self):
"""
Test complete edit submission approval workflow.
Flow: User submits → Moderator reviews → Moderator approves → Changes applied
"""
from apps.moderation.models import EditSubmission
from apps.parks.models import Company
# Create target object
company = Company.objects.create(
name='Test Company',
description='Original description'
)
# User submits an edit
content_type = ContentType.objects.get_for_model(company)
submission = EditSubmission.objects.create(
user=self.regular_user,
content_type=content_type,
object_id=company.id,
submission_type='EDIT',
changes={'description': 'Updated description'},
status='PENDING',
reason='Fixing typo'
)
self.assertEqual(submission.status, 'PENDING')
self.assertIsNone(submission.handled_by)
self.assertIsNone(submission.handled_at)
# Moderator approves
submission.transition_to_approved(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'APPROVED')
self.assertEqual(submission.handled_by, self.moderator)
self.assertIsNotNone(submission.handled_at)
def test_photo_submission_approval_workflow(self):
"""
Test complete photo submission approval workflow.
Flow: User submits photo → Moderator reviews → Moderator approves → Photo created
"""
from apps.moderation.models import PhotoSubmission
from apps.parks.models import Park, Company
# Create target park
operator = Company.objects.create(
name='Test Operator',
roles=['OPERATOR']
)
park = Park.objects.create(
name='Test Park',
slug='test-park',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
# User submits a photo
content_type = ContentType.objects.get_for_model(park)
submission = PhotoSubmission.objects.create(
user=self.regular_user,
content_type=content_type,
object_id=park.id,
status='PENDING',
photo_type='GENERAL',
description='Beautiful park entrance'
)
self.assertEqual(submission.status, 'PENDING')
# Moderator approves
submission.transition_to_approved(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'APPROVED')
class SubmissionRejectionWorkflowTests(TestCase):
"""Tests for the submission rejection workflow."""
@classmethod
def setUpTestData(cls):
cls.regular_user = User.objects.create_user(
username='user_rej',
email='user_rej@example.com',
password='testpass123',
role='USER'
)
cls.moderator = User.objects.create_user(
username='mod_rej',
email='mod_rej@example.com',
password='testpass123',
role='MODERATOR'
)
def test_edit_submission_rejection_with_reason(self):
"""
Test rejection workflow with reason.
Flow: User submits → Moderator rejects with reason → User notified
"""
from apps.moderation.models import EditSubmission
from apps.parks.models import Company
company = Company.objects.create(
name='Test Company',
description='Original'
)
content_type = ContentType.objects.get_for_model(company)
submission = EditSubmission.objects.create(
user=self.regular_user,
content_type=content_type,
object_id=company.id,
submission_type='EDIT',
changes={'name': 'Spam Content'},
status='PENDING',
reason='Name change request'
)
# Moderator rejects
submission.transition_to_rejected(user=self.moderator)
submission.handled_by = self.moderator
submission.handled_at = timezone.now()
submission.notes = 'Rejected: Content appears to be spam'
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'REJECTED')
self.assertIn('spam', submission.notes.lower())
class SubmissionEscalationWorkflowTests(TestCase):
"""Tests for the submission escalation workflow."""
@classmethod
def setUpTestData(cls):
cls.regular_user = User.objects.create_user(
username='user_esc',
email='user_esc@example.com',
password='testpass123',
role='USER'
)
cls.moderator = User.objects.create_user(
username='mod_esc',
email='mod_esc@example.com',
password='testpass123',
role='MODERATOR'
)
cls.admin = User.objects.create_user(
username='admin_esc',
email='admin_esc@example.com',
password='testpass123',
role='ADMIN'
)
def test_escalation_workflow(self):
"""
Test complete escalation workflow.
Flow: User submits → Moderator escalates → Admin reviews → Admin approves
"""
from apps.moderation.models import EditSubmission
from apps.parks.models import Company
company = Company.objects.create(
name='Sensitive Company',
description='Original'
)
content_type = ContentType.objects.get_for_model(company)
submission = EditSubmission.objects.create(
user=self.regular_user,
content_type=content_type,
object_id=company.id,
submission_type='EDIT',
changes={'name': 'New Sensitive Name'},
status='PENDING',
reason='Major name change'
)
# Moderator escalates
submission.transition_to_escalated(user=self.moderator)
submission.notes = 'Escalated: Major change needs admin review'
submission.save()
self.assertEqual(submission.status, 'ESCALATED')
# Admin approves
submission.transition_to_approved(user=self.admin)
submission.handled_by = self.admin
submission.handled_at = timezone.now()
submission.save()
submission.refresh_from_db()
self.assertEqual(submission.status, 'APPROVED')
self.assertEqual(submission.handled_by, self.admin)
class ReportHandlingWorkflowTests(TestCase):
"""Tests for the moderation report handling workflow."""
@classmethod
def setUpTestData(cls):
cls.reporter = User.objects.create_user(
username='reporter',
email='reporter@example.com',
password='testpass123',
role='USER'
)
cls.moderator = User.objects.create_user(
username='mod_report',
email='mod_report@example.com',
password='testpass123',
role='MODERATOR'
)
def test_report_resolution_workflow(self):
"""
Test complete report resolution workflow.
Flow: User reports → Moderator assigned → Moderator investigates → Resolved
"""
from apps.moderation.models import ModerationReport
from apps.parks.models import Company
reported_company = Company.objects.create(
name='Problematic Company',
description='Some inappropriate content'
)
content_type = ContentType.objects.get_for_model(reported_company)
# User reports content
report = ModerationReport.objects.create(
report_type='CONTENT',
status='PENDING',
priority='HIGH',
reported_entity_type='company',
reported_entity_id=reported_company.id,
content_type=content_type,
reason='INAPPROPRIATE',
description='This content is inappropriate',
reported_by=self.reporter
)
self.assertEqual(report.status, 'PENDING')
# Moderator claims and starts review
report.transition_to_under_review(user=self.moderator)
report.assigned_moderator = self.moderator
report.save()
self.assertEqual(report.status, 'UNDER_REVIEW')
self.assertEqual(report.assigned_moderator, self.moderator)
# Moderator resolves
report.transition_to_resolved(user=self.moderator)
report.resolution_action = 'CONTENT_REMOVED'
report.resolution_notes = 'Content was removed'
report.resolved_at = timezone.now()
report.save()
report.refresh_from_db()
self.assertEqual(report.status, 'RESOLVED')
self.assertIsNotNone(report.resolved_at)
def test_report_dismissal_workflow(self):
"""
Test report dismissal workflow for invalid reports.
Flow: User reports → Moderator reviews → Moderator dismisses
"""
from apps.moderation.models import ModerationReport
from apps.parks.models import Company
company = Company.objects.create(
name='Valid Company',
description='Normal content'
)
content_type = ContentType.objects.get_for_model(company)
report = ModerationReport.objects.create(
report_type='CONTENT',
status='PENDING',
priority='LOW',
reported_entity_type='company',
reported_entity_id=company.id,
content_type=content_type,
reason='OTHER',
description='I just do not like this',
reported_by=self.reporter
)
# Moderator claims
report.transition_to_under_review(user=self.moderator)
report.assigned_moderator = self.moderator
report.save()
# Moderator dismisses as invalid
report.transition_to_dismissed(user=self.moderator)
report.resolution_notes = 'Report does not violate any guidelines'
report.resolved_at = timezone.now()
report.save()
report.refresh_from_db()
self.assertEqual(report.status, 'DISMISSED')
class BulkOperationWorkflowTests(TestCase):
"""Tests for bulk operation workflows."""
@classmethod
def setUpTestData(cls):
cls.admin = User.objects.create_user(
username='admin_bulk',
email='admin_bulk@example.com',
password='testpass123',
role='ADMIN'
)
def test_bulk_operation_success_workflow(self):
"""
Test successful bulk operation workflow.
Flow: Admin creates → Operation runs → Progress tracked → Completed
"""
from apps.moderation.models import BulkOperation
operation = BulkOperation.objects.create(
operation_type='APPROVE_SUBMISSIONS',
status='PENDING',
total_items=10,
processed_items=0,
created_by=self.admin,
parameters={'submission_ids': list(range(1, 11))}
)
self.assertEqual(operation.status, 'PENDING')
# Start operation
operation.transition_to_running(user=self.admin)
operation.started_at = timezone.now()
operation.save()
self.assertEqual(operation.status, 'RUNNING')
# Simulate progress
for i in range(1, 11):
operation.processed_items = i
operation.save()
# Complete operation
operation.transition_to_completed(user=self.admin)
operation.completed_at = timezone.now()
operation.results = {'approved': 10, 'failed': 0}
operation.save()
operation.refresh_from_db()
self.assertEqual(operation.status, 'COMPLETED')
self.assertEqual(operation.processed_items, 10)
def test_bulk_operation_failure_workflow(self):
"""
Test bulk operation failure workflow.
Flow: Admin creates → Operation runs → Error occurs → Failed
"""
from apps.moderation.models import BulkOperation
operation = BulkOperation.objects.create(
operation_type='DELETE_CONTENT',
status='PENDING',
total_items=5,
processed_items=0,
created_by=self.admin,
parameters={'content_ids': list(range(1, 6))}
)
operation.transition_to_running(user=self.admin)
operation.started_at = timezone.now()
operation.save()
# Simulate partial progress then failure
operation.processed_items = 2
operation.failed_items = 3
operation.transition_to_failed(user=self.admin)
operation.completed_at = timezone.now()
operation.results = {'error': 'Database connection lost', 'processed': 2}
operation.save()
operation.refresh_from_db()
self.assertEqual(operation.status, 'FAILED')
self.assertEqual(operation.failed_items, 3)
def test_bulk_operation_cancellation_workflow(self):
"""
Test bulk operation cancellation workflow.
Flow: Admin creates → Operation runs → Admin cancels
"""
from apps.moderation.models import BulkOperation
operation = BulkOperation.objects.create(
operation_type='BATCH_UPDATE',
status='PENDING',
total_items=100,
processed_items=0,
created_by=self.admin,
parameters={'update_field': 'status'},
can_cancel=True
)
operation.transition_to_running(user=self.admin)
operation.save()
# Partial progress
operation.processed_items = 30
operation.save()
# Admin cancels
operation.transition_to_cancelled(user=self.admin)
operation.completed_at = timezone.now()
operation.results = {'cancelled_at': 30, 'reason': 'User requested cancellation'}
operation.save()
operation.refresh_from_db()
self.assertEqual(operation.status, 'CANCELLED')
self.assertEqual(operation.processed_items, 30)
class ModerationQueueWorkflowTests(TestCase):
"""Tests for moderation queue workflows."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='mod_queue',
email='mod_queue@example.com',
password='testpass123',
role='MODERATOR'
)
def test_queue_completion_workflow(self):
"""
Test queue item completion workflow.
Flow: Item created → Moderator claims → Work done → Completed
"""
from apps.moderation.models import ModerationQueue
queue_item = ModerationQueue.objects.create(
queue_type='SUBMISSION_REVIEW',
status='PENDING',
priority='MEDIUM',
item_type='edit_submission',
item_id=123
)
self.assertEqual(queue_item.status, 'PENDING')
# Moderator claims
queue_item.transition_to_in_progress(user=self.moderator)
queue_item.assigned_to = self.moderator
queue_item.assigned_at = timezone.now()
queue_item.save()
self.assertEqual(queue_item.status, 'IN_PROGRESS')
# Work completed
queue_item.transition_to_completed(user=self.moderator)
queue_item.completed_at = timezone.now()
queue_item.save()
queue_item.refresh_from_db()
self.assertEqual(queue_item.status, 'COMPLETED')

View File

@@ -1,117 +1,541 @@
from django.test import TestCase, Client
"""
Comprehensive tests for the parks app state machine.
This module contains tests for:
- Park status FSM transitions
- Park transition wrapper methods
- Transition history logging
- Related model updates during transitions
"""
from django.test import TestCase
from django.contrib.auth import get_user_model
from apps.parks.models import Park, ParkArea, ParkLocation, Company as Operator
from django.contrib.contenttypes.models import ContentType
from django.utils import timezone
from django_fsm import TransitionNotAllowed
from .models import Park, Company
from datetime import date
User = get_user_model()
def create_test_location(park: Park) -> ParkLocation:
"""Helper function to create a test location"""
park_location = ParkLocation.objects.create(
park=park,
street_address="123 Test St",
city="Test City",
state="TS",
country="Test Country",
postal_code="12345",
)
# Set coordinates using the helper method
park_location.set_coordinates(34.0522, -118.2437) # latitude, longitude
park_location.save()
return park_location
# ============================================================================
# Park FSM Transition Tests
# ============================================================================
class ParkModelTests(TestCase):
@classmethod
def setUpTestData(cls) -> None:
# Create test user
cls.user = User.objects.create_user(
username="testuser",
email="test@example.com",
password="testpass123",
)
class ParkTransitionTests(TestCase):
"""Comprehensive tests for Park FSM transitions."""
# Create test company
cls.operator = Operator.objects.create(
name="Test Company", website="http://example.com"
)
# Create test park
cls.park = Park.objects.create(
name="Test Park",
operator=cls.operator,
status="OPERATING",
website="http://testpark.com",
)
# Create test location
cls.location = create_test_location(cls.park)
def test_park_creation(self) -> None:
"""Test park instance creation and field values"""
self.assertEqual(self.park.name, "Test Park")
self.assertEqual(self.park.operator, self.operator)
self.assertEqual(self.park.status, "OPERATING")
self.assertEqual(self.park.website, "http://testpark.com")
self.assertTrue(self.park.slug)
def test_park_str_representation(self) -> None:
"""Test string representation of park"""
self.assertEqual(str(self.park), "Test Park")
def test_park_coordinates(self) -> None:
"""Test park coordinates property"""
coords = self.park.coordinates
self.assertIsNotNone(coords)
if coords:
self.assertAlmostEqual(coords[0], 34.0522, places=4) # latitude
self.assertAlmostEqual(coords[1], -118.2437, places=4) # longitude
def test_park_formatted_location(self) -> None:
"""Test park formatted_location property"""
expected = "123 Test St, Test City, TS, 12345, Test Country"
self.assertEqual(self.park.formatted_location, expected)
class ParkAreaTests(TestCase):
def setUp(self) -> None:
# Create test company
self.operator = Operator.objects.create(
name="Test Company", website="http://example.com"
)
# Create test park
self.park = Park.objects.create(
name="Test Park", operator=self.operator, status="OPERATING"
)
# Create test location
self.location = create_test_location(self.park)
# Create test area
self.area = ParkArea.objects.create(
park=self.park, name="Test Area", description="Test Description"
)
def test_area_creation(self) -> None:
"""Test park area creation"""
self.assertEqual(self.area.name, "Test Area")
self.assertEqual(self.area.park, self.park)
self.assertTrue(self.area.slug)
class ParkViewTests(TestCase):
def setUp(self) -> None:
self.client = Client()
def setUp(self):
"""Set up test fixtures."""
self.user = User.objects.create_user(
username="testuser",
email="test@example.com",
password="testpass123",
username='testuser',
email='test@example.com',
password='testpass123',
role='USER'
)
self.operator = Operator.objects.create(
name="Test Company", website="http://example.com"
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
self.park = Park.objects.create(
name="Test Park", operator=self.operator, status="OPERATING"
self.admin = User.objects.create_user(
username='admin',
email='admin@example.com',
password='testpass123',
role='ADMIN'
)
self.location = create_test_location(self.park)
# Create operator company
self.operator = Company.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
def _create_park(self, status='OPERATING', **kwargs):
"""Helper to create a Park with specified status."""
defaults = {
'name': 'Test Park',
'slug': 'test-park',
'description': 'A test park',
'operator': self.operator,
'timezone': 'America/New_York'
}
defaults.update(kwargs)
return Park.objects.create(status=status, **defaults)
# -------------------------------------------------------------------------
# Operating status transitions
# -------------------------------------------------------------------------
def test_operating_to_closed_temp_transition(self):
"""Test transition from OPERATING to CLOSED_TEMP."""
park = self._create_park(status='OPERATING')
self.assertEqual(park.status, 'OPERATING')
park.transition_to_closed_temp(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_TEMP')
def test_operating_to_closed_perm_transition(self):
"""Test transition from OPERATING to CLOSED_PERM."""
park = self._create_park(status='OPERATING')
park.transition_to_closed_perm(user=self.moderator)
park.closing_date = date.today()
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
self.assertIsNotNone(park.closing_date)
# -------------------------------------------------------------------------
# Under construction transitions
# -------------------------------------------------------------------------
def test_under_construction_to_operating_transition(self):
"""Test transition from UNDER_CONSTRUCTION to OPERATING."""
park = self._create_park(status='UNDER_CONSTRUCTION')
self.assertEqual(park.status, 'UNDER_CONSTRUCTION')
park.transition_to_operating(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'OPERATING')
# -------------------------------------------------------------------------
# Closed temp transitions
# -------------------------------------------------------------------------
def test_closed_temp_to_operating_transition(self):
"""Test transition from CLOSED_TEMP to OPERATING (reopen)."""
park = self._create_park(status='CLOSED_TEMP')
park.transition_to_operating(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'OPERATING')
def test_closed_temp_to_closed_perm_transition(self):
"""Test transition from CLOSED_TEMP to CLOSED_PERM."""
park = self._create_park(status='CLOSED_TEMP')
park.transition_to_closed_perm(user=self.moderator)
park.closing_date = date.today()
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
# -------------------------------------------------------------------------
# Closed perm transitions (to final states)
# -------------------------------------------------------------------------
def test_closed_perm_to_demolished_transition(self):
"""Test transition from CLOSED_PERM to DEMOLISHED."""
park = self._create_park(status='CLOSED_PERM')
park.transition_to_demolished(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'DEMOLISHED')
def test_closed_perm_to_relocated_transition(self):
"""Test transition from CLOSED_PERM to RELOCATED."""
park = self._create_park(status='CLOSED_PERM')
park.transition_to_relocated(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'RELOCATED')
# -------------------------------------------------------------------------
# Invalid transitions (final states)
# -------------------------------------------------------------------------
def test_demolished_cannot_transition(self):
"""Test that DEMOLISHED state cannot transition further."""
park = self._create_park(status='DEMOLISHED')
with self.assertRaises(TransitionNotAllowed):
park.transition_to_operating(user=self.moderator)
def test_relocated_cannot_transition(self):
"""Test that RELOCATED state cannot transition further."""
park = self._create_park(status='RELOCATED')
with self.assertRaises(TransitionNotAllowed):
park.transition_to_operating(user=self.moderator)
def test_operating_cannot_directly_demolish(self):
"""Test that OPERATING cannot directly transition to DEMOLISHED."""
park = self._create_park(status='OPERATING')
with self.assertRaises(TransitionNotAllowed):
park.transition_to_demolished(user=self.moderator)
def test_operating_cannot_directly_relocate(self):
"""Test that OPERATING cannot directly transition to RELOCATED."""
park = self._create_park(status='OPERATING')
with self.assertRaises(TransitionNotAllowed):
park.transition_to_relocated(user=self.moderator)
# -------------------------------------------------------------------------
# Wrapper method tests
# -------------------------------------------------------------------------
def test_reopen_wrapper_method(self):
"""Test the reopen() wrapper method."""
park = self._create_park(status='CLOSED_TEMP')
park.reopen(user=self.user)
park.refresh_from_db()
self.assertEqual(park.status, 'OPERATING')
def test_close_temporarily_wrapper_method(self):
"""Test the close_temporarily() wrapper method."""
park = self._create_park(status='OPERATING')
park.close_temporarily(user=self.user)
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_TEMP')
def test_close_permanently_wrapper_method(self):
"""Test the close_permanently() wrapper method."""
park = self._create_park(status='OPERATING')
closing = date(2025, 12, 31)
park.close_permanently(closing_date=closing, user=self.moderator)
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
self.assertEqual(park.closing_date, closing)
def test_close_permanently_without_date(self):
"""Test close_permanently() without closing_date."""
park = self._create_park(status='OPERATING')
park.close_permanently(user=self.moderator)
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
self.assertIsNone(park.closing_date)
def test_demolish_wrapper_method(self):
"""Test the demolish() wrapper method."""
park = self._create_park(status='CLOSED_PERM')
park.demolish(user=self.moderator)
park.refresh_from_db()
self.assertEqual(park.status, 'DEMOLISHED')
def test_relocate_wrapper_method(self):
"""Test the relocate() wrapper method."""
park = self._create_park(status='CLOSED_PERM')
park.relocate(user=self.moderator)
park.refresh_from_db()
self.assertEqual(park.status, 'RELOCATED')
def test_start_construction_wrapper_method(self):
"""Test the start_construction() wrapper method if applicable."""
# This depends on allowed transitions - skip if not allowed
try:
park = self._create_park(status='OPERATING')
park.start_construction(user=self.moderator)
park.refresh_from_db()
self.assertEqual(park.status, 'UNDER_CONSTRUCTION')
except TransitionNotAllowed:
# If transition from OPERATING to UNDER_CONSTRUCTION is not allowed
pass
# ============================================================================
# Park Transition History Tests
# ============================================================================
class ParkTransitionHistoryTests(TestCase):
"""Tests for Park transition history logging."""
def setUp(self):
"""Set up test fixtures."""
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
self.operator = Company.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
def _create_park(self, status='OPERATING'):
"""Helper to create a Park."""
return Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
status=status,
timezone='America/New_York'
)
def test_transition_creates_state_log(self):
"""Test that transitions create StateLog entries."""
from django_fsm_log.models import StateLog
park = self._create_park(status='OPERATING')
park.transition_to_closed_temp(user=self.moderator)
park.save()
park_ct = ContentType.objects.get_for_model(park)
log = StateLog.objects.filter(
content_type=park_ct,
object_id=park.id
).first()
self.assertIsNotNone(log)
self.assertEqual(log.state, 'CLOSED_TEMP')
self.assertEqual(log.by, self.moderator)
def test_multiple_transitions_create_multiple_logs(self):
"""Test that multiple transitions create multiple log entries."""
from django_fsm_log.models import StateLog
park = self._create_park(status='OPERATING')
park_ct = ContentType.objects.get_for_model(park)
# First transition
park.transition_to_closed_temp(user=self.moderator)
park.save()
# Second transition
park.transition_to_operating(user=self.moderator)
park.save()
logs = StateLog.objects.filter(
content_type=park_ct,
object_id=park.id
).order_by('timestamp')
self.assertEqual(logs.count(), 2)
self.assertEqual(logs[0].state, 'CLOSED_TEMP')
self.assertEqual(logs[1].state, 'OPERATING')
def test_transition_log_includes_user(self):
"""Test that transition logs include the user who made the change."""
from django_fsm_log.models import StateLog
park = self._create_park(status='OPERATING')
park.transition_to_closed_perm(user=self.moderator)
park.save()
park_ct = ContentType.objects.get_for_model(park)
log = StateLog.objects.filter(
content_type=park_ct,
object_id=park.id
).first()
self.assertEqual(log.by, self.moderator)
# ============================================================================
# Park Model Business Logic Tests
# ============================================================================
class ParkBusinessLogicTests(TestCase):
"""Tests for Park model business logic."""
def setUp(self):
"""Set up test fixtures."""
self.operator = Company.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
self.property_owner = Company.objects.create(
name='Property Owner',
description='Property owner company',
roles=['PROPERTY_OWNER']
)
def test_park_creates_with_valid_operator(self):
"""Test park can be created with valid operator."""
park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
self.assertEqual(park.operator, self.operator)
def test_park_slug_auto_generated(self):
"""Test that park slug is auto-generated from name."""
park = Park.objects.create(
name='My Amazing Theme Park',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
self.assertEqual(park.slug, 'my-amazing-theme-park')
def test_park_url_generated(self):
"""Test that frontend URL is generated on save."""
park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
self.assertIn('test-park', park.url)
def test_opening_year_computed_from_opening_date(self):
"""Test that opening_year is computed from opening_date."""
park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
opening_date=date(2020, 6, 15),
timezone='America/New_York'
)
self.assertEqual(park.opening_year, 2020)
def test_search_text_populated(self):
"""Test that search_text is populated on save."""
park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A wonderful theme park',
operator=self.operator,
timezone='America/New_York'
)
self.assertIn('test park', park.search_text)
self.assertIn('wonderful theme park', park.search_text)
self.assertIn('test operator', park.search_text)
def test_park_with_property_owner(self):
"""Test park with separate property owner."""
park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
property_owner=self.property_owner,
timezone='America/New_York'
)
self.assertEqual(park.operator, self.operator)
self.assertEqual(park.property_owner, self.property_owner)
# ============================================================================
# Park Historical Slug Tests
# ============================================================================
class ParkSlugHistoryTests(TestCase):
"""Tests for Park historical slug tracking."""
def setUp(self):
"""Set up test fixtures."""
self.operator = Company.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
def test_historical_slug_created_on_name_change(self):
"""Test that historical slug is created when name changes."""
from django.contrib.contenttypes.models import ContentType
from apps.core.history import HistoricalSlug
park = Park.objects.create(
name='Original Name',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
original_slug = park.slug
# Change name
park.name = 'New Name'
park.save()
# Check historical slug was created
park_ct = ContentType.objects.get_for_model(park)
historical = HistoricalSlug.objects.filter(
content_type=park_ct,
object_id=park.id,
slug=original_slug
).first()
self.assertIsNotNone(historical)
self.assertEqual(historical.slug, original_slug)
def test_get_by_slug_finds_current_slug(self):
"""Test get_by_slug finds park by current slug."""
park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
found_park, is_historical = Park.get_by_slug('test-park')
self.assertEqual(found_park, park)
self.assertFalse(is_historical)
def test_get_by_slug_finds_historical_slug(self):
"""Test get_by_slug finds park by historical slug."""
from django.contrib.contenttypes.models import ContentType
from apps.core.history import HistoricalSlug
park = Park.objects.create(
name='Original Name',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
original_slug = park.slug
# Change name to create historical slug
park.name = 'New Name'
park.save()
# Find by historical slug
found_park, is_historical = Park.get_by_slug(original_slug)
self.assertEqual(found_park, park)
self.assertTrue(is_historical)

View File

@@ -0,0 +1,7 @@
"""
Parks test package.
This package contains tests for the parks app including:
- Park workflow tests (test_park_workflows.py)
- Park model tests
"""

View File

@@ -0,0 +1,533 @@
"""
Integration tests for Park lifecycle workflows.
This module tests end-to-end park lifecycle workflows including:
- Park opening workflow
- Park temporary closure workflow
- Park permanent closure workflow
- Park relocation workflow
- Related ride status updates
"""
from django.test import TestCase
from django.contrib.auth import get_user_model
from django.utils import timezone
User = get_user_model()
class ParkOpeningWorkflowTests(TestCase):
"""Tests for park opening workflow."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='park_user',
email='park_user@example.com',
password='testpass123',
role='USER'
)
cls.moderator = User.objects.create_user(
username='park_mod',
email='park_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_park(self, status='OPERATING', **kwargs):
"""Helper to create a park."""
from apps.parks.models import Park, Company
operator = Company.objects.create(
name=f'Operator {status}',
roles=['OPERATOR']
)
defaults = {
'name': f'Test Park {status}',
'slug': f'test-park-{status.lower()}-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
def test_park_opens_from_under_construction(self):
"""
Test park opening from under construction state.
Flow: UNDER_CONSTRUCTION → OPERATING
"""
park = self._create_park(status='UNDER_CONSTRUCTION')
self.assertEqual(park.status, 'UNDER_CONSTRUCTION')
# Park opens
park.transition_to_operating(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'OPERATING')
class ParkTemporaryClosureWorkflowTests(TestCase):
"""Tests for park temporary closure workflow."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='temp_closure_user',
email='temp_closure@example.com',
password='testpass123',
role='USER'
)
def _create_park(self, status='OPERATING', **kwargs):
from apps.parks.models import Park, Company
operator = Company.objects.create(
name=f'Operator Temp {timezone.now().timestamp()}',
roles=['OPERATOR']
)
defaults = {
'name': f'Test Park Temp {timezone.now().timestamp()}',
'slug': f'test-park-temp-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
def test_park_temporary_closure_and_reopen(self):
"""
Test park temporary closure and reopening.
Flow: OPERATING → CLOSED_TEMP → OPERATING
"""
park = self._create_park(status='OPERATING')
self.assertEqual(park.status, 'OPERATING')
# Close temporarily (e.g., off-season)
park.transition_to_closed_temp(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_TEMP')
# Reopen
park.transition_to_operating(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'OPERATING')
class ParkPermanentClosureWorkflowTests(TestCase):
"""Tests for park permanent closure workflow."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='perm_mod',
email='perm_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_park(self, status='OPERATING', **kwargs):
from apps.parks.models import Park, Company
operator = Company.objects.create(
name=f'Operator Perm {timezone.now().timestamp()}',
roles=['OPERATOR']
)
defaults = {
'name': f'Test Park Perm {timezone.now().timestamp()}',
'slug': f'test-park-perm-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
def test_park_permanent_closure(self):
"""
Test park permanent closure from operating state.
Flow: OPERATING → CLOSED_PERM
"""
park = self._create_park(status='OPERATING')
# Close permanently
park.transition_to_closed_perm(user=self.moderator)
park.closing_date = timezone.now().date()
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
self.assertIsNotNone(park.closing_date)
def test_park_permanent_closure_from_temp(self):
"""
Test park permanent closure from temporary closure.
Flow: OPERATING → CLOSED_TEMP → CLOSED_PERM
"""
park = self._create_park(status='OPERATING')
# Temporary closure
park.transition_to_closed_temp(user=self.moderator)
park.save()
# Becomes permanent
park.transition_to_closed_perm(user=self.moderator)
park.closing_date = timezone.now().date()
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
class ParkDemolitionWorkflowTests(TestCase):
"""Tests for park demolition workflow."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='demo_mod',
email='demo_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_park(self, status='CLOSED_PERM', **kwargs):
from apps.parks.models import Park, Company
operator = Company.objects.create(
name=f'Operator Demo {timezone.now().timestamp()}',
roles=['OPERATOR']
)
defaults = {
'name': f'Test Park Demo {timezone.now().timestamp()}',
'slug': f'test-park-demo-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
def test_park_demolition_workflow(self):
"""
Test complete park demolition workflow.
Flow: OPERATING → CLOSED_PERM → DEMOLISHED
"""
park = self._create_park(status='CLOSED_PERM')
# Demolish
park.transition_to_demolished(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'DEMOLISHED')
def test_demolished_is_final_state(self):
"""Test that demolished parks cannot transition further."""
from django_fsm import TransitionNotAllowed
park = self._create_park(status='DEMOLISHED')
# Cannot transition from demolished
with self.assertRaises(TransitionNotAllowed):
park.transition_to_operating(user=self.moderator)
class ParkRelocationWorkflowTests(TestCase):
"""Tests for park relocation workflow."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='reloc_mod',
email='reloc_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_park(self, status='CLOSED_PERM', **kwargs):
from apps.parks.models import Park, Company
operator = Company.objects.create(
name=f'Operator Reloc {timezone.now().timestamp()}',
roles=['OPERATOR']
)
defaults = {
'name': f'Test Park Reloc {timezone.now().timestamp()}',
'slug': f'test-park-reloc-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
def test_park_relocation_workflow(self):
"""
Test park relocation workflow.
Flow: OPERATING → CLOSED_PERM → RELOCATED
"""
park = self._create_park(status='CLOSED_PERM')
# Relocate
park.transition_to_relocated(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'RELOCATED')
def test_relocated_is_final_state(self):
"""Test that relocated parks cannot transition further."""
from django_fsm import TransitionNotAllowed
park = self._create_park(status='RELOCATED')
# Cannot transition from relocated
with self.assertRaises(TransitionNotAllowed):
park.transition_to_operating(user=self.moderator)
class ParkWrapperMethodTests(TestCase):
"""Tests for park wrapper methods."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='wrapper_user',
email='wrapper@example.com',
password='testpass123',
role='USER'
)
cls.moderator = User.objects.create_user(
username='wrapper_mod',
email='wrapper_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_park(self, status='OPERATING', **kwargs):
from apps.parks.models import Park, Company
operator = Company.objects.create(
name=f'Operator Wrapper {timezone.now().timestamp()}',
roles=['OPERATOR']
)
defaults = {
'name': f'Test Park Wrapper {timezone.now().timestamp()}',
'slug': f'test-park-wrapper-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
def test_close_temporarily_wrapper(self):
"""Test close_temporarily wrapper method."""
park = self._create_park(status='OPERATING')
# Use wrapper method if it exists
if hasattr(park, 'close_temporarily'):
park.close_temporarily(user=self.user)
else:
park.transition_to_closed_temp(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_TEMP')
def test_reopen_wrapper(self):
"""Test reopen wrapper method."""
park = self._create_park(status='CLOSED_TEMP')
# Use wrapper method if it exists
if hasattr(park, 'reopen'):
park.reopen(user=self.user)
else:
park.transition_to_operating(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'OPERATING')
def test_close_permanently_wrapper(self):
"""Test close_permanently wrapper method."""
park = self._create_park(status='OPERATING')
closing_date = timezone.now().date()
# Use wrapper method if it exists
if hasattr(park, 'close_permanently'):
park.close_permanently(closing_date=closing_date, user=self.moderator)
else:
park.transition_to_closed_perm(user=self.moderator)
park.closing_date = closing_date
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
def test_demolish_wrapper(self):
"""Test demolish wrapper method."""
park = self._create_park(status='CLOSED_PERM')
# Use wrapper method if it exists
if hasattr(park, 'demolish'):
park.demolish(user=self.moderator)
else:
park.transition_to_demolished(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'DEMOLISHED')
def test_relocate_wrapper(self):
"""Test relocate wrapper method."""
park = self._create_park(status='CLOSED_PERM')
# Use wrapper method if it exists
if hasattr(park, 'relocate'):
park.relocate(user=self.moderator)
else:
park.transition_to_relocated(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'RELOCATED')
class ParkStateLogTests(TestCase):
"""Tests for StateLog entries on park transitions."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='log_user',
email='log_user@example.com',
password='testpass123',
role='USER'
)
cls.moderator = User.objects.create_user(
username='log_mod',
email='log_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_park(self, status='OPERATING', **kwargs):
from apps.parks.models import Park, Company
operator = Company.objects.create(
name=f'Operator Log {timezone.now().timestamp()}',
roles=['OPERATOR']
)
defaults = {
'name': f'Test Park Log {timezone.now().timestamp()}',
'slug': f'test-park-log-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
def test_transition_creates_state_log(self):
"""Test that park transitions create StateLog entries."""
from django_fsm_log.models import StateLog
from django.contrib.contenttypes.models import ContentType
park = self._create_park(status='OPERATING')
park_ct = ContentType.objects.get_for_model(park)
# Perform transition
park.transition_to_closed_temp(user=self.user)
park.save()
# Check log was created
log = StateLog.objects.filter(
content_type=park_ct,
object_id=park.id
).first()
self.assertIsNotNone(log, "StateLog entry should be created")
self.assertEqual(log.state, 'CLOSED_TEMP')
self.assertEqual(log.by, self.user)
def test_multiple_transitions_logged(self):
"""Test that multiple park transitions are all logged."""
from django_fsm_log.models import StateLog
from django.contrib.contenttypes.models import ContentType
park = self._create_park(status='OPERATING')
park_ct = ContentType.objects.get_for_model(park)
# First transition: OPERATING -> CLOSED_TEMP
park.transition_to_closed_temp(user=self.user)
park.save()
# Second transition: CLOSED_TEMP -> CLOSED_PERM
park.transition_to_closed_perm(user=self.moderator)
park.save()
# Check multiple logs created
logs = StateLog.objects.filter(
content_type=park_ct,
object_id=park.id
).order_by('timestamp')
self.assertEqual(logs.count(), 2, "Should have 2 log entries")
self.assertEqual(logs[0].state, 'CLOSED_TEMP')
self.assertEqual(logs[0].by, self.user)
self.assertEqual(logs[1].state, 'CLOSED_PERM')
self.assertEqual(logs[1].by, self.moderator)
def test_full_lifecycle_logged(self):
"""Test complete park lifecycle is logged."""
from django_fsm_log.models import StateLog
from django.contrib.contenttypes.models import ContentType
park = self._create_park(status='OPERATING')
park_ct = ContentType.objects.get_for_model(park)
# Full lifecycle: OPERATING -> CLOSED_TEMP -> OPERATING -> CLOSED_PERM -> DEMOLISHED
park.transition_to_closed_temp(user=self.user)
park.save()
park.transition_to_operating(user=self.user)
park.save()
park.transition_to_closed_perm(user=self.moderator)
park.save()
park.transition_to_demolished(user=self.moderator)
park.save()
# Check all logs created
logs = StateLog.objects.filter(
content_type=park_ct,
object_id=park.id
).order_by('timestamp')
self.assertEqual(logs.count(), 4, "Should have 4 log entries")
states = [log.state for log in logs]
self.assertEqual(states, ['CLOSED_TEMP', 'OPERATING', 'CLOSED_PERM', 'DEMOLISHED'])

View File

@@ -1 +1,926 @@
# Create your tests here.
"""
Comprehensive tests for the rides app state machine.
This module contains tests for:
- Ride status FSM transitions
- Ride transition wrapper methods
- Post-closing status automation
- Transition history logging
- Related model updates during transitions
"""
from django.test import TestCase
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.utils import timezone
from django_fsm import TransitionNotAllowed
from .models import Ride, RideModel, Company
from apps.parks.models import Park, Company as ParkCompany
from datetime import date, timedelta
User = get_user_model()
# ============================================================================
# Ride FSM Transition Tests
# ============================================================================
class RideTransitionTests(TestCase):
"""Comprehensive tests for Ride FSM transitions."""
def setUp(self):
"""Set up test fixtures."""
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123',
role='USER'
)
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
self.admin = User.objects.create_user(
username='admin',
email='admin@example.com',
password='testpass123',
role='ADMIN'
)
# Create operator and park
self.operator = ParkCompany.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
self.park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
# Create manufacturer
self.manufacturer = Company.objects.create(
name='Test Manufacturer',
description='Test manufacturer company',
roles=['MANUFACTURER']
)
def _create_ride(self, status='OPERATING', **kwargs):
"""Helper to create a Ride with specified status."""
defaults = {
'name': 'Test Ride',
'slug': 'test-ride',
'description': 'A test ride',
'park': self.park,
'manufacturer': self.manufacturer
}
defaults.update(kwargs)
return Ride.objects.create(status=status, **defaults)
# -------------------------------------------------------------------------
# Operating status transitions
# -------------------------------------------------------------------------
def test_operating_to_closed_temp_transition(self):
"""Test transition from OPERATING to CLOSED_TEMP."""
ride = self._create_ride(status='OPERATING')
self.assertEqual(ride.status, 'OPERATING')
ride.transition_to_closed_temp(user=self.user)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_TEMP')
def test_operating_to_sbno_transition(self):
"""Test transition from OPERATING to SBNO."""
ride = self._create_ride(status='OPERATING')
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
def test_operating_to_closing_transition(self):
"""Test transition from OPERATING to CLOSING."""
ride = self._create_ride(status='OPERATING')
ride.transition_to_closing(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSING')
# -------------------------------------------------------------------------
# Under construction transitions
# -------------------------------------------------------------------------
def test_under_construction_to_operating_transition(self):
"""Test transition from UNDER_CONSTRUCTION to OPERATING."""
ride = self._create_ride(status='UNDER_CONSTRUCTION')
self.assertEqual(ride.status, 'UNDER_CONSTRUCTION')
ride.transition_to_operating(user=self.user)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'OPERATING')
# -------------------------------------------------------------------------
# Closed temp transitions
# -------------------------------------------------------------------------
def test_closed_temp_to_operating_transition(self):
"""Test transition from CLOSED_TEMP to OPERATING (reopen)."""
ride = self._create_ride(status='CLOSED_TEMP')
ride.transition_to_operating(user=self.user)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'OPERATING')
def test_closed_temp_to_sbno_transition(self):
"""Test transition from CLOSED_TEMP to SBNO."""
ride = self._create_ride(status='CLOSED_TEMP')
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
def test_closed_temp_to_closed_perm_transition(self):
"""Test transition from CLOSED_TEMP to CLOSED_PERM."""
ride = self._create_ride(status='CLOSED_TEMP')
ride.transition_to_closed_perm(user=self.moderator)
ride.closing_date = date.today()
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_PERM')
# -------------------------------------------------------------------------
# SBNO transitions
# -------------------------------------------------------------------------
def test_sbno_to_operating_transition(self):
"""Test transition from SBNO to OPERATING (revival)."""
ride = self._create_ride(status='SBNO')
ride.transition_to_operating(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'OPERATING')
def test_sbno_to_closed_perm_transition(self):
"""Test transition from SBNO to CLOSED_PERM."""
ride = self._create_ride(status='SBNO')
ride.transition_to_closed_perm(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_PERM')
# -------------------------------------------------------------------------
# Closing transitions
# -------------------------------------------------------------------------
def test_closing_to_closed_perm_transition(self):
"""Test transition from CLOSING to CLOSED_PERM."""
ride = self._create_ride(status='CLOSING')
ride.transition_to_closed_perm(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_PERM')
def test_closing_to_sbno_transition(self):
"""Test transition from CLOSING to SBNO."""
ride = self._create_ride(status='CLOSING')
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
# -------------------------------------------------------------------------
# Closed perm transitions (to final states)
# -------------------------------------------------------------------------
def test_closed_perm_to_demolished_transition(self):
"""Test transition from CLOSED_PERM to DEMOLISHED."""
ride = self._create_ride(status='CLOSED_PERM')
ride.transition_to_demolished(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'DEMOLISHED')
def test_closed_perm_to_relocated_transition(self):
"""Test transition from CLOSED_PERM to RELOCATED."""
ride = self._create_ride(status='CLOSED_PERM')
ride.transition_to_relocated(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'RELOCATED')
# -------------------------------------------------------------------------
# Invalid transitions (final states)
# -------------------------------------------------------------------------
def test_demolished_cannot_transition(self):
"""Test that DEMOLISHED state cannot transition further."""
ride = self._create_ride(status='DEMOLISHED')
with self.assertRaises(TransitionNotAllowed):
ride.transition_to_operating(user=self.moderator)
def test_relocated_cannot_transition(self):
"""Test that RELOCATED state cannot transition further."""
ride = self._create_ride(status='RELOCATED')
with self.assertRaises(TransitionNotAllowed):
ride.transition_to_operating(user=self.moderator)
def test_operating_cannot_directly_demolish(self):
"""Test that OPERATING cannot directly transition to DEMOLISHED."""
ride = self._create_ride(status='OPERATING')
with self.assertRaises(TransitionNotAllowed):
ride.transition_to_demolished(user=self.moderator)
def test_operating_cannot_directly_relocate(self):
"""Test that OPERATING cannot directly transition to RELOCATED."""
ride = self._create_ride(status='OPERATING')
with self.assertRaises(TransitionNotAllowed):
ride.transition_to_relocated(user=self.moderator)
# -------------------------------------------------------------------------
# Wrapper method tests
# -------------------------------------------------------------------------
def test_open_wrapper_method(self):
"""Test the open() wrapper method."""
ride = self._create_ride(status='CLOSED_TEMP')
ride.open(user=self.user)
ride.refresh_from_db()
self.assertEqual(ride.status, 'OPERATING')
def test_close_temporarily_wrapper_method(self):
"""Test the close_temporarily() wrapper method."""
ride = self._create_ride(status='OPERATING')
ride.close_temporarily(user=self.user)
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_TEMP')
def test_mark_sbno_wrapper_method(self):
"""Test the mark_sbno() wrapper method."""
ride = self._create_ride(status='OPERATING')
ride.mark_sbno(user=self.moderator)
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
def test_mark_closing_wrapper_method(self):
"""Test the mark_closing() wrapper method."""
ride = self._create_ride(status='OPERATING')
closing = date(2025, 12, 31)
ride.mark_closing(
closing_date=closing,
post_closing_status='DEMOLISHED',
user=self.moderator
)
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSING')
self.assertEqual(ride.closing_date, closing)
self.assertEqual(ride.post_closing_status, 'DEMOLISHED')
def test_mark_closing_requires_post_closing_status(self):
"""Test that mark_closing() requires post_closing_status."""
ride = self._create_ride(status='OPERATING')
with self.assertRaises(ValidationError):
ride.mark_closing(
closing_date=date(2025, 12, 31),
post_closing_status='',
user=self.moderator
)
def test_close_permanently_wrapper_method(self):
"""Test the close_permanently() wrapper method."""
ride = self._create_ride(status='SBNO')
ride.close_permanently(user=self.moderator)
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_PERM')
def test_demolish_wrapper_method(self):
"""Test the demolish() wrapper method."""
ride = self._create_ride(status='CLOSED_PERM')
ride.demolish(user=self.moderator)
ride.refresh_from_db()
self.assertEqual(ride.status, 'DEMOLISHED')
def test_relocate_wrapper_method(self):
"""Test the relocate() wrapper method."""
ride = self._create_ride(status='CLOSED_PERM')
ride.relocate(user=self.moderator)
ride.refresh_from_db()
self.assertEqual(ride.status, 'RELOCATED')
# ============================================================================
# Ride Post-Closing Status Tests
# ============================================================================
class RidePostClosingTests(TestCase):
"""Tests for Ride post_closing_status automation."""
def setUp(self):
"""Set up test fixtures."""
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
self.operator = ParkCompany.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
self.park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
self.manufacturer = Company.objects.create(
name='Test Manufacturer',
description='Test manufacturer company',
roles=['MANUFACTURER']
)
def _create_ride(self, status='OPERATING', **kwargs):
"""Helper to create a Ride."""
defaults = {
'name': 'Test Ride',
'slug': 'test-ride',
'description': 'A test ride',
'park': self.park,
'manufacturer': self.manufacturer
}
defaults.update(kwargs)
return Ride.objects.create(status=status, **defaults)
def test_apply_post_closing_status_to_demolished(self):
"""Test apply_post_closing_status transitions to DEMOLISHED."""
yesterday = date.today() - timedelta(days=1)
ride = self._create_ride(
status='CLOSING',
closing_date=yesterday,
post_closing_status='DEMOLISHED'
)
ride.apply_post_closing_status(user=self.moderator)
ride.refresh_from_db()
self.assertEqual(ride.status, 'DEMOLISHED')
def test_apply_post_closing_status_to_relocated(self):
"""Test apply_post_closing_status transitions to RELOCATED."""
yesterday = date.today() - timedelta(days=1)
ride = self._create_ride(
status='CLOSING',
closing_date=yesterday,
post_closing_status='RELOCATED'
)
ride.apply_post_closing_status(user=self.moderator)
ride.refresh_from_db()
self.assertEqual(ride.status, 'RELOCATED')
def test_apply_post_closing_status_to_sbno(self):
"""Test apply_post_closing_status transitions to SBNO."""
yesterday = date.today() - timedelta(days=1)
ride = self._create_ride(
status='CLOSING',
closing_date=yesterday,
post_closing_status='SBNO'
)
ride.apply_post_closing_status(user=self.moderator)
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
def test_apply_post_closing_status_to_closed_perm(self):
"""Test apply_post_closing_status transitions to CLOSED_PERM."""
yesterday = date.today() - timedelta(days=1)
ride = self._create_ride(
status='CLOSING',
closing_date=yesterday,
post_closing_status='CLOSED_PERM'
)
ride.apply_post_closing_status(user=self.moderator)
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_PERM')
def test_apply_post_closing_status_not_yet_reached(self):
"""Test apply_post_closing_status does nothing if date not reached."""
tomorrow = date.today() + timedelta(days=1)
ride = self._create_ride(
status='CLOSING',
closing_date=tomorrow,
post_closing_status='DEMOLISHED'
)
ride.apply_post_closing_status(user=self.moderator)
ride.refresh_from_db()
# Status should remain CLOSING since date hasn't been reached
self.assertEqual(ride.status, 'CLOSING')
def test_apply_post_closing_status_requires_closing_status(self):
"""Test apply_post_closing_status requires CLOSING status."""
ride = self._create_ride(status='OPERATING')
with self.assertRaises(ValidationError) as ctx:
ride.apply_post_closing_status(user=self.moderator)
self.assertIn('CLOSING', str(ctx.exception))
def test_apply_post_closing_status_requires_closing_date(self):
"""Test apply_post_closing_status requires closing_date."""
ride = self._create_ride(
status='CLOSING',
post_closing_status='DEMOLISHED'
)
ride.closing_date = None
ride.save()
with self.assertRaises(ValidationError) as ctx:
ride.apply_post_closing_status(user=self.moderator)
self.assertIn('closing_date', str(ctx.exception))
def test_apply_post_closing_status_requires_post_closing_status(self):
"""Test apply_post_closing_status requires post_closing_status."""
yesterday = date.today() - timedelta(days=1)
ride = self._create_ride(
status='CLOSING',
closing_date=yesterday
)
ride.post_closing_status = None
ride.save()
with self.assertRaises(ValidationError) as ctx:
ride.apply_post_closing_status(user=self.moderator)
self.assertIn('post_closing_status', str(ctx.exception))
# ============================================================================
# Ride Transition History Tests
# ============================================================================
class RideTransitionHistoryTests(TestCase):
"""Tests for Ride transition history logging."""
def setUp(self):
"""Set up test fixtures."""
self.moderator = User.objects.create_user(
username='moderator',
email='moderator@example.com',
password='testpass123',
role='MODERATOR'
)
self.operator = ParkCompany.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
self.park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
self.manufacturer = Company.objects.create(
name='Test Manufacturer',
description='Test manufacturer company',
roles=['MANUFACTURER']
)
def _create_ride(self, status='OPERATING'):
"""Helper to create a Ride."""
return Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A test ride',
park=self.park,
manufacturer=self.manufacturer,
status=status
)
def test_transition_creates_state_log(self):
"""Test that transitions create StateLog entries."""
from django_fsm_log.models import StateLog
ride = self._create_ride(status='OPERATING')
ride.transition_to_closed_temp(user=self.moderator)
ride.save()
ride_ct = ContentType.objects.get_for_model(ride)
log = StateLog.objects.filter(
content_type=ride_ct,
object_id=ride.id
).first()
self.assertIsNotNone(log)
self.assertEqual(log.state, 'CLOSED_TEMP')
self.assertEqual(log.by, self.moderator)
def test_multiple_transitions_create_multiple_logs(self):
"""Test that multiple transitions create multiple log entries."""
from django_fsm_log.models import StateLog
ride = self._create_ride(status='OPERATING')
ride_ct = ContentType.objects.get_for_model(ride)
# First transition
ride.transition_to_closed_temp(user=self.moderator)
ride.save()
# Second transition
ride.transition_to_operating(user=self.moderator)
ride.save()
logs = StateLog.objects.filter(
content_type=ride_ct,
object_id=ride.id
).order_by('timestamp')
self.assertEqual(logs.count(), 2)
self.assertEqual(logs[0].state, 'CLOSED_TEMP')
self.assertEqual(logs[1].state, 'OPERATING')
def test_transition_log_includes_user(self):
"""Test that transition logs include the user who made the change."""
from django_fsm_log.models import StateLog
ride = self._create_ride(status='OPERATING')
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride_ct = ContentType.objects.get_for_model(ride)
log = StateLog.objects.filter(
content_type=ride_ct,
object_id=ride.id
).first()
self.assertEqual(log.by, self.moderator)
def test_post_closing_transition_logged(self):
"""Test that post_closing_status transitions are logged."""
from django_fsm_log.models import StateLog
yesterday = date.today() - timedelta(days=1)
ride = self._create_ride(status='CLOSING')
ride.closing_date = yesterday
ride.post_closing_status = 'DEMOLISHED'
ride.save()
ride.apply_post_closing_status(user=self.moderator)
ride_ct = ContentType.objects.get_for_model(ride)
log = StateLog.objects.filter(
content_type=ride_ct,
object_id=ride.id,
state='DEMOLISHED'
).first()
self.assertIsNotNone(log)
self.assertEqual(log.by, self.moderator)
# ============================================================================
# Ride Model Business Logic Tests
# ============================================================================
class RideBusinessLogicTests(TestCase):
"""Tests for Ride model business logic."""
def setUp(self):
"""Set up test fixtures."""
self.operator = ParkCompany.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
self.park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
self.manufacturer = Company.objects.create(
name='Test Manufacturer',
description='Test manufacturer company',
roles=['MANUFACTURER']
)
def test_ride_creates_with_valid_park(self):
"""Test ride can be created with valid park."""
ride = Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A test ride',
park=self.park,
manufacturer=self.manufacturer
)
self.assertEqual(ride.park, self.park)
def test_ride_slug_auto_generated(self):
"""Test that ride slug is auto-generated from name."""
ride = Ride.objects.create(
name='My Amazing Roller Coaster',
description='A test ride',
park=self.park,
manufacturer=self.manufacturer
)
self.assertEqual(ride.slug, 'my-amazing-roller-coaster')
def test_ride_url_generated(self):
"""Test that frontend URL is generated on save."""
ride = Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A test ride',
park=self.park,
manufacturer=self.manufacturer
)
self.assertIn('test-park', ride.url)
self.assertIn('test-ride', ride.url)
def test_opening_year_computed_from_opening_date(self):
"""Test that opening_year is computed from opening_date."""
ride = Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A test ride',
park=self.park,
manufacturer=self.manufacturer,
opening_date=date(2020, 6, 15)
)
self.assertEqual(ride.opening_year, 2020)
def test_search_text_populated(self):
"""Test that search_text is populated on save."""
ride = Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A thrilling roller coaster',
park=self.park,
manufacturer=self.manufacturer
)
self.assertIn('test ride', ride.search_text)
self.assertIn('thrilling roller coaster', ride.search_text)
self.assertIn('test park', ride.search_text)
self.assertIn('test manufacturer', ride.search_text)
def test_ride_slug_unique_within_park(self):
"""Test that ride slugs are unique within a park."""
Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='First ride',
park=self.park,
manufacturer=self.manufacturer
)
# Creating another ride with same name should get different slug
ride2 = Ride.objects.create(
name='Test Ride',
description='Second ride',
park=self.park,
manufacturer=self.manufacturer
)
self.assertNotEqual(ride2.slug, 'test-ride')
self.assertTrue(ride2.slug.startswith('test-ride'))
# ============================================================================
# Ride Move to Park Tests
# ============================================================================
class RideMoveTests(TestCase):
"""Tests for moving rides between parks."""
def setUp(self):
"""Set up test fixtures."""
self.operator = ParkCompany.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
self.park1 = Park.objects.create(
name='Park One',
slug='park-one',
description='First park',
operator=self.operator,
timezone='America/New_York'
)
self.park2 = Park.objects.create(
name='Park Two',
slug='park-two',
description='Second park',
operator=self.operator,
timezone='America/Los_Angeles'
)
self.manufacturer = Company.objects.create(
name='Test Manufacturer',
description='Test manufacturer company',
roles=['MANUFACTURER']
)
def test_move_ride_to_different_park(self):
"""Test moving a ride to a different park."""
ride = Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A test ride',
park=self.park1,
manufacturer=self.manufacturer
)
changes = ride.move_to_park(self.park2)
ride.refresh_from_db()
self.assertEqual(ride.park, self.park2)
self.assertEqual(changes['old_park']['id'], self.park1.id)
self.assertEqual(changes['new_park']['id'], self.park2.id)
def test_move_ride_updates_url(self):
"""Test that moving a ride updates the URL."""
ride = Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A test ride',
park=self.park1,
manufacturer=self.manufacturer
)
old_url = ride.url
changes = ride.move_to_park(self.park2)
ride.refresh_from_db()
self.assertNotEqual(ride.url, old_url)
self.assertIn('park-two', ride.url)
self.assertTrue(changes['url_changed'])
def test_move_ride_handles_slug_conflict(self):
"""Test that moving a ride handles slug conflicts in destination park."""
# Create ride in park1
ride1 = Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A test ride',
park=self.park1,
manufacturer=self.manufacturer
)
# Create ride with same slug in park2
Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='Another test ride',
park=self.park2,
manufacturer=self.manufacturer
)
# Move ride1 to park2
changes = ride1.move_to_park(self.park2)
ride1.refresh_from_db()
self.assertEqual(ride1.park, self.park2)
# Slug should have been modified to avoid conflict
self.assertNotEqual(ride1.slug, 'test-ride')
self.assertTrue(changes['slug_changed'])
# ============================================================================
# Ride Historical Slug Tests
# ============================================================================
class RideSlugHistoryTests(TestCase):
"""Tests for Ride historical slug tracking."""
def setUp(self):
"""Set up test fixtures."""
self.operator = ParkCompany.objects.create(
name='Test Operator',
description='Test operator company',
roles=['OPERATOR']
)
self.park = Park.objects.create(
name='Test Park',
slug='test-park',
description='A test park',
operator=self.operator,
timezone='America/New_York'
)
self.manufacturer = Company.objects.create(
name='Test Manufacturer',
description='Test manufacturer company',
roles=['MANUFACTURER']
)
def test_get_by_slug_finds_current_slug(self):
"""Test get_by_slug finds ride by current slug."""
ride = Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A test ride',
park=self.park,
manufacturer=self.manufacturer
)
found_ride, is_historical = Ride.get_by_slug('test-ride', park=self.park)
self.assertEqual(found_ride, ride)
self.assertFalse(is_historical)
def test_get_by_slug_with_park_filter(self):
"""Test get_by_slug filters by park."""
ride = Ride.objects.create(
name='Test Ride',
slug='test-ride',
description='A test ride',
park=self.park,
manufacturer=self.manufacturer
)
# Should find ride in correct park
found_ride, is_historical = Ride.get_by_slug('test-ride', park=self.park)
self.assertEqual(found_ride, ride)
# Should not find ride in different park
other_park = Park.objects.create(
name='Other Park',
slug='other-park',
description='Another park',
operator=self.operator,
timezone='America/New_York'
)
with self.assertRaises(Ride.DoesNotExist):
Ride.get_by_slug('test-ride', park=other_park)

View File

@@ -0,0 +1,7 @@
"""
Rides test package.
This package contains tests for the rides app including:
- Ride workflow tests (test_ride_workflows.py)
- Ride model tests
"""

View File

@@ -0,0 +1,900 @@
"""
Integration tests for Ride lifecycle workflows.
This module tests end-to-end ride lifecycle workflows including:
- Ride opening workflow
- Ride maintenance workflow
- Ride SBNO workflow
- Ride scheduled closure workflow
- Ride demolition workflow
- Ride relocation workflow
"""
from django.test import TestCase
from django.contrib.auth import get_user_model
from django.utils import timezone
from datetime import timedelta
User = get_user_model()
class RideOpeningWorkflowTests(TestCase):
"""Tests for ride opening workflow."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='ride_user',
email='ride_user@example.com',
password='testpass123',
role='USER'
)
def _create_ride(self, status='OPERATING', **kwargs):
"""Helper to create a ride with park."""
from apps.rides.models import Ride
from apps.parks.models import Park, Company
# Create manufacturer
manufacturer = Company.objects.create(
name=f'Manufacturer {timezone.now().timestamp()}',
roles=['MANUFACTURER']
)
# Create park with operator
operator = Company.objects.create(
name=f'Operator {timezone.now().timestamp()}',
roles=['OPERATOR']
)
park = Park.objects.create(
name=f'Test Park {timezone.now().timestamp()}',
slug=f'test-park-{timezone.now().timestamp()}',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
defaults = {
'name': f'Test Ride {timezone.now().timestamp()}',
'slug': f'test-ride-{timezone.now().timestamp()}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
def test_ride_opens_from_under_construction(self):
"""
Test ride opening from under construction state.
Flow: UNDER_CONSTRUCTION → OPERATING
"""
ride = self._create_ride(status='UNDER_CONSTRUCTION')
self.assertEqual(ride.status, 'UNDER_CONSTRUCTION')
# Ride opens
ride.transition_to_operating(user=self.user)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'OPERATING')
class RideMaintenanceWorkflowTests(TestCase):
"""Tests for ride maintenance (temporary closure) workflow."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='maint_user',
email='maint@example.com',
password='testpass123',
role='USER'
)
def _create_ride(self, status='OPERATING', **kwargs):
from apps.rides.models import Ride
from apps.parks.models import Park, Company
manufacturer = Company.objects.create(
name=f'Mfr Maint {timezone.now().timestamp()}',
roles=['MANUFACTURER']
)
operator = Company.objects.create(
name=f'Op Maint {timezone.now().timestamp()}',
roles=['OPERATOR']
)
park = Park.objects.create(
name=f'Park Maint {timezone.now().timestamp()}',
slug=f'park-maint-{timezone.now().timestamp()}',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
defaults = {
'name': f'Ride Maint {timezone.now().timestamp()}',
'slug': f'ride-maint-{timezone.now().timestamp()}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
def test_ride_maintenance_and_reopen(self):
"""
Test ride maintenance and reopening.
Flow: OPERATING → CLOSED_TEMP → OPERATING
"""
ride = self._create_ride(status='OPERATING')
# Close for maintenance
ride.transition_to_closed_temp(user=self.user)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_TEMP')
# Reopen after maintenance
ride.transition_to_operating(user=self.user)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'OPERATING')
class RideSBNOWorkflowTests(TestCase):
"""Tests for ride SBNO (Standing But Not Operating) workflow."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='sbno_mod',
email='sbno_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_ride(self, status='OPERATING', **kwargs):
from apps.rides.models import Ride
from apps.parks.models import Park, Company
manufacturer = Company.objects.create(
name=f'Mfr SBNO {timezone.now().timestamp()}',
roles=['MANUFACTURER']
)
operator = Company.objects.create(
name=f'Op SBNO {timezone.now().timestamp()}',
roles=['OPERATOR']
)
park = Park.objects.create(
name=f'Park SBNO {timezone.now().timestamp()}',
slug=f'park-sbno-{timezone.now().timestamp()}',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
defaults = {
'name': f'Ride SBNO {timezone.now().timestamp()}',
'slug': f'ride-sbno-{timezone.now().timestamp()}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
def test_ride_sbno_from_operating(self):
"""
Test ride becomes SBNO from operating.
Flow: OPERATING → SBNO
"""
ride = self._create_ride(status='OPERATING')
# Mark as SBNO
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
def test_ride_sbno_from_closed_temp(self):
"""
Test ride becomes SBNO from temporary closure.
Flow: OPERATING → CLOSED_TEMP → SBNO
"""
ride = self._create_ride(status='CLOSED_TEMP')
# Extended to SBNO
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
def test_ride_revival_from_sbno(self):
"""
Test ride revival from SBNO state.
Flow: SBNO → OPERATING
"""
ride = self._create_ride(status='SBNO')
# Revive the ride
ride.transition_to_operating(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'OPERATING')
def test_sbno_to_closed_perm(self):
"""
Test ride permanently closes from SBNO.
Flow: SBNO → CLOSED_PERM
"""
ride = self._create_ride(status='SBNO')
# Confirm permanent closure
ride.transition_to_closed_perm(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_PERM')
class RideScheduledClosureWorkflowTests(TestCase):
"""Tests for ride scheduled closure (CLOSING state) workflow."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='closing_mod',
email='closing_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_ride(self, status='OPERATING', **kwargs):
from apps.rides.models import Ride
from apps.parks.models import Park, Company
manufacturer = Company.objects.create(
name=f'Mfr Closing {timezone.now().timestamp()}',
roles=['MANUFACTURER']
)
operator = Company.objects.create(
name=f'Op Closing {timezone.now().timestamp()}',
roles=['OPERATOR']
)
park = Park.objects.create(
name=f'Park Closing {timezone.now().timestamp()}',
slug=f'park-closing-{timezone.now().timestamp()}',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
defaults = {
'name': f'Ride Closing {timezone.now().timestamp()}',
'slug': f'ride-closing-{timezone.now().timestamp()}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
def test_ride_mark_closing_with_date(self):
"""
Test ride marked as closing with scheduled date.
Flow: OPERATING → CLOSING (with closing_date and post_closing_status)
"""
ride = self._create_ride(status='OPERATING')
closing_date = (timezone.now() + timedelta(days=30)).date()
# Mark as closing
ride.transition_to_closing(user=self.moderator)
ride.closing_date = closing_date
ride.post_closing_status = 'DEMOLISHED'
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSING')
self.assertEqual(ride.closing_date, closing_date)
self.assertEqual(ride.post_closing_status, 'DEMOLISHED')
def test_closing_to_closed_perm(self):
"""
Test ride transitions from CLOSING to CLOSED_PERM when date reached.
Flow: CLOSING → CLOSED_PERM
"""
ride = self._create_ride(status='CLOSING')
ride.closing_date = timezone.now().date()
ride.post_closing_status = 'CLOSED_PERM'
ride.save()
# Transition when closing date reached
ride.transition_to_closed_perm(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_PERM')
def test_closing_to_sbno(self):
"""
Test ride transitions from CLOSING to SBNO.
Flow: CLOSING → SBNO
"""
ride = self._create_ride(status='CLOSING')
ride.closing_date = timezone.now().date()
ride.post_closing_status = 'SBNO'
ride.save()
# Transition to SBNO
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
class RideDemolitionWorkflowTests(TestCase):
"""Tests for ride demolition workflow."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='demo_ride_mod',
email='demo_ride_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_ride(self, status='CLOSED_PERM', **kwargs):
from apps.rides.models import Ride
from apps.parks.models import Park, Company
manufacturer = Company.objects.create(
name=f'Mfr Demo {timezone.now().timestamp()}',
roles=['MANUFACTURER']
)
operator = Company.objects.create(
name=f'Op Demo {timezone.now().timestamp()}',
roles=['OPERATOR']
)
park = Park.objects.create(
name=f'Park Demo {timezone.now().timestamp()}',
slug=f'park-demo-{timezone.now().timestamp()}',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
defaults = {
'name': f'Ride Demo {timezone.now().timestamp()}',
'slug': f'ride-demo-{timezone.now().timestamp()}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
def test_ride_demolition(self):
"""
Test ride demolition from permanently closed.
Flow: CLOSED_PERM → DEMOLISHED
"""
ride = self._create_ride(status='CLOSED_PERM')
# Demolish
ride.transition_to_demolished(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'DEMOLISHED')
def test_demolished_is_final_state(self):
"""Test that demolished rides cannot transition further."""
from django_fsm import TransitionNotAllowed
ride = self._create_ride(status='DEMOLISHED')
# Cannot transition from demolished
with self.assertRaises(TransitionNotAllowed):
ride.transition_to_operating(user=self.moderator)
class RideRelocationWorkflowTests(TestCase):
"""Tests for ride relocation workflow."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='reloc_ride_mod',
email='reloc_ride_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_ride(self, status='CLOSED_PERM', **kwargs):
from apps.rides.models import Ride
from apps.parks.models import Park, Company
manufacturer = Company.objects.create(
name=f'Mfr Reloc {timezone.now().timestamp()}',
roles=['MANUFACTURER']
)
operator = Company.objects.create(
name=f'Op Reloc {timezone.now().timestamp()}',
roles=['OPERATOR']
)
park = Park.objects.create(
name=f'Park Reloc {timezone.now().timestamp()}',
slug=f'park-reloc-{timezone.now().timestamp()}',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
defaults = {
'name': f'Ride Reloc {timezone.now().timestamp()}',
'slug': f'ride-reloc-{timezone.now().timestamp()}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
def test_ride_relocation(self):
"""
Test ride relocation from permanently closed.
Flow: CLOSED_PERM → RELOCATED
"""
ride = self._create_ride(status='CLOSED_PERM')
# Relocate
ride.transition_to_relocated(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'RELOCATED')
def test_relocated_is_final_state(self):
"""Test that relocated rides cannot transition further."""
from django_fsm import TransitionNotAllowed
ride = self._create_ride(status='RELOCATED')
# Cannot transition from relocated
with self.assertRaises(TransitionNotAllowed):
ride.transition_to_operating(user=self.moderator)
class RideWrapperMethodTests(TestCase):
"""Tests for ride wrapper methods."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='wrapper_ride_user',
email='wrapper_ride@example.com',
password='testpass123',
role='USER'
)
cls.moderator = User.objects.create_user(
username='wrapper_ride_mod',
email='wrapper_ride_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_ride(self, status='OPERATING', **kwargs):
from apps.rides.models import Ride
from apps.parks.models import Park, Company
manufacturer = Company.objects.create(
name=f'Mfr Wrapper {timezone.now().timestamp()}',
roles=['MANUFACTURER']
)
operator = Company.objects.create(
name=f'Op Wrapper {timezone.now().timestamp()}',
roles=['OPERATOR']
)
park = Park.objects.create(
name=f'Park Wrapper {timezone.now().timestamp()}',
slug=f'park-wrapper-{timezone.now().timestamp()}',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
defaults = {
'name': f'Ride Wrapper {timezone.now().timestamp()}',
'slug': f'ride-wrapper-{timezone.now().timestamp()}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
def test_close_temporarily_wrapper(self):
"""Test close_temporarily wrapper method."""
ride = self._create_ride(status='OPERATING')
if hasattr(ride, 'close_temporarily'):
ride.close_temporarily(user=self.user)
else:
ride.transition_to_closed_temp(user=self.user)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_TEMP')
def test_mark_sbno_wrapper(self):
"""Test mark_sbno wrapper method."""
ride = self._create_ride(status='OPERATING')
if hasattr(ride, 'mark_sbno'):
ride.mark_sbno(user=self.moderator)
else:
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
def test_mark_closing_wrapper(self):
"""Test mark_closing wrapper method."""
ride = self._create_ride(status='OPERATING')
closing_date = (timezone.now() + timedelta(days=30)).date()
if hasattr(ride, 'mark_closing'):
ride.mark_closing(
closing_date=closing_date,
post_closing_status='DEMOLISHED',
user=self.moderator
)
else:
ride.transition_to_closing(user=self.moderator)
ride.closing_date = closing_date
ride.post_closing_status = 'DEMOLISHED'
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSING')
def test_open_wrapper(self):
"""Test open wrapper method."""
ride = self._create_ride(status='CLOSED_TEMP')
if hasattr(ride, 'open'):
ride.open(user=self.user)
else:
ride.transition_to_operating(user=self.user)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'OPERATING')
def test_close_permanently_wrapper(self):
"""Test close_permanently wrapper method."""
ride = self._create_ride(status='SBNO')
if hasattr(ride, 'close_permanently'):
ride.close_permanently(user=self.moderator)
else:
ride.transition_to_closed_perm(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'CLOSED_PERM')
def test_demolish_wrapper(self):
"""Test demolish wrapper method."""
ride = self._create_ride(status='CLOSED_PERM')
if hasattr(ride, 'demolish'):
ride.demolish(user=self.moderator)
else:
ride.transition_to_demolished(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'DEMOLISHED')
def test_relocate_wrapper(self):
"""Test relocate wrapper method."""
ride = self._create_ride(status='CLOSED_PERM')
if hasattr(ride, 'relocate'):
ride.relocate(user=self.moderator)
else:
ride.transition_to_relocated(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'RELOCATED')
class RidePostClosingStatusAutomationTests(TestCase):
"""Tests for post_closing_status automation logic."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='auto_mod',
email='auto_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_ride(self, status='CLOSING', **kwargs):
from apps.rides.models import Ride
from apps.parks.models import Park, Company
manufacturer = Company.objects.create(
name=f'Mfr Auto {timezone.now().timestamp()}',
roles=['MANUFACTURER']
)
operator = Company.objects.create(
name=f'Op Auto {timezone.now().timestamp()}',
roles=['OPERATOR']
)
park = Park.objects.create(
name=f'Park Auto {timezone.now().timestamp()}',
slug=f'park-auto-{timezone.now().timestamp()}',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
defaults = {
'name': f'Ride Auto {timezone.now().timestamp()}',
'slug': f'ride-auto-{timezone.now().timestamp()}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
def test_apply_post_closing_status_demolished(self):
"""Test apply_post_closing_status transitions to DEMOLISHED."""
ride = self._create_ride(status='CLOSING')
ride.closing_date = timezone.now().date()
ride.post_closing_status = 'DEMOLISHED'
ride.save()
# Apply post-closing status if method exists
if hasattr(ride, 'apply_post_closing_status'):
ride.apply_post_closing_status(user=self.moderator)
else:
ride.transition_to_demolished(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'DEMOLISHED')
def test_apply_post_closing_status_relocated(self):
"""Test apply_post_closing_status transitions to RELOCATED."""
ride = self._create_ride(status='CLOSING')
ride.closing_date = timezone.now().date()
ride.post_closing_status = 'RELOCATED'
ride.save()
if hasattr(ride, 'apply_post_closing_status'):
ride.apply_post_closing_status(user=self.moderator)
else:
ride.transition_to_relocated(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'RELOCATED')
def test_apply_post_closing_status_sbno(self):
"""Test apply_post_closing_status transitions to SBNO."""
ride = self._create_ride(status='CLOSING')
ride.closing_date = timezone.now().date()
ride.post_closing_status = 'SBNO'
ride.save()
if hasattr(ride, 'apply_post_closing_status'):
ride.apply_post_closing_status(user=self.moderator)
else:
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride.refresh_from_db()
self.assertEqual(ride.status, 'SBNO')
class RideStateLogTests(TestCase):
"""Tests for StateLog entries on ride transitions."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='ride_log_user',
email='ride_log_user@example.com',
password='testpass123',
role='USER'
)
cls.moderator = User.objects.create_user(
username='ride_log_mod',
email='ride_log_mod@example.com',
password='testpass123',
role='MODERATOR'
)
def _create_ride(self, status='OPERATING', **kwargs):
from apps.rides.models import Ride
from apps.parks.models import Park, Company
manufacturer = Company.objects.create(
name=f'Mfr Log {timezone.now().timestamp()}',
roles=['MANUFACTURER']
)
operator = Company.objects.create(
name=f'Op Log {timezone.now().timestamp()}',
roles=['OPERATOR']
)
park = Park.objects.create(
name=f'Park Log {timezone.now().timestamp()}',
slug=f'park-log-{timezone.now().timestamp()}',
operator=operator,
status='OPERATING',
timezone='America/New_York'
)
defaults = {
'name': f'Ride Log {timezone.now().timestamp()}',
'slug': f'ride-log-{timezone.now().timestamp()}',
'park': park,
'manufacturer': manufacturer,
'status': status
}
defaults.update(kwargs)
return Ride.objects.create(**defaults)
def test_transition_creates_state_log(self):
"""Test that ride transitions create StateLog entries."""
from django_fsm_log.models import StateLog
from django.contrib.contenttypes.models import ContentType
ride = self._create_ride(status='OPERATING')
ride_ct = ContentType.objects.get_for_model(ride)
# Perform transition
ride.transition_to_closed_temp(user=self.user)
ride.save()
# Check log was created
log = StateLog.objects.filter(
content_type=ride_ct,
object_id=ride.id
).first()
self.assertIsNotNone(log, "StateLog entry should be created")
self.assertEqual(log.state, 'CLOSED_TEMP')
self.assertEqual(log.by, self.user)
def test_multiple_transitions_logged(self):
"""Test that multiple ride transitions are all logged."""
from django_fsm_log.models import StateLog
from django.contrib.contenttypes.models import ContentType
ride = self._create_ride(status='OPERATING')
ride_ct = ContentType.objects.get_for_model(ride)
# First transition: OPERATING -> SBNO
ride.transition_to_sbno(user=self.moderator)
ride.save()
# Second transition: SBNO -> OPERATING (revival)
ride.transition_to_operating(user=self.moderator)
ride.save()
# Check multiple logs created
logs = StateLog.objects.filter(
content_type=ride_ct,
object_id=ride.id
).order_by('timestamp')
self.assertEqual(logs.count(), 2, "Should have 2 log entries")
self.assertEqual(logs[0].state, 'SBNO')
self.assertEqual(logs[1].state, 'OPERATING')
def test_sbno_revival_workflow_logged(self):
"""Test that SBNO revival workflow is logged."""
from django_fsm_log.models import StateLog
from django.contrib.contenttypes.models import ContentType
ride = self._create_ride(status='SBNO')
ride_ct = ContentType.objects.get_for_model(ride)
# Revival: SBNO -> OPERATING
ride.transition_to_operating(user=self.moderator)
ride.save()
# Check log was created
log = StateLog.objects.filter(
content_type=ride_ct,
object_id=ride.id
).first()
self.assertIsNotNone(log, "StateLog entry should be created")
self.assertEqual(log.state, 'OPERATING')
self.assertEqual(log.by, self.moderator)
def test_full_lifecycle_logged(self):
"""Test complete ride lifecycle is logged."""
from django_fsm_log.models import StateLog
from django.contrib.contenttypes.models import ContentType
ride = self._create_ride(status='OPERATING')
ride_ct = ContentType.objects.get_for_model(ride)
# Lifecycle: OPERATING -> CLOSED_TEMP -> SBNO -> CLOSED_PERM -> DEMOLISHED
ride.transition_to_closed_temp(user=self.user)
ride.save()
ride.transition_to_sbno(user=self.moderator)
ride.save()
ride.transition_to_closed_perm(user=self.moderator)
ride.save()
ride.transition_to_demolished(user=self.moderator)
ride.save()
# Check all logs created
logs = StateLog.objects.filter(
content_type=ride_ct,
object_id=ride.id
).order_by('timestamp')
self.assertEqual(logs.count(), 4, "Should have 4 log entries")
states = [log.state for log in logs]
self.assertEqual(states, ['CLOSED_TEMP', 'SBNO', 'CLOSED_PERM', 'DEMOLISHED'])
def test_scheduled_closing_workflow_logged(self):
"""Test that scheduled closing workflow creates logs."""
from django_fsm_log.models import StateLog
from django.contrib.contenttypes.models import ContentType
ride = self._create_ride(status='OPERATING')
ride_ct = ContentType.objects.get_for_model(ride)
# Scheduled closing workflow: OPERATING -> CLOSING -> CLOSED_PERM
ride.transition_to_closing(user=self.moderator)
ride.closing_date = (timezone.now() + timedelta(days=30)).date()
ride.post_closing_status = 'DEMOLISHED'
ride.save()
ride.transition_to_closed_perm(user=self.moderator)
ride.save()
logs = StateLog.objects.filter(
content_type=ride_ct,
object_id=ride.id
).order_by('timestamp')
self.assertEqual(logs.count(), 2, "Should have 2 log entries")
self.assertEqual(logs[0].state, 'CLOSING')
self.assertEqual(logs[1].state, 'CLOSED_PERM')