Add test utilities and state machine diagrams for FSM models

- Introduced reusable test utilities in `backend/tests/utils` for FSM transitions, HTMX interactions, and common scenarios.
- Added factory functions for creating test submissions, parks, rides, and photo submissions.
- Implemented assertion helpers for verifying state changes, toast notifications, and transition logs.
- Created comprehensive state machine diagrams for all FSM-enabled models in `docs/STATE_DIAGRAMS.md`, detailing states, transitions, and guard conditions.
This commit is contained in:
pacnpal
2025-12-22 08:55:39 -05:00
parent b508434574
commit 45d97b6e68
71 changed files with 8608 additions and 633 deletions

View File

@@ -0,0 +1,229 @@
# FSM HTMX Browser Testing Checklist
This checklist provides manual testing steps to verify the FSM+HTMX integration
works correctly across different browsers and scenarios.
## Quick Start
Before testing, ensure:
1. Development server is running: `uv run manage.py runserver`
2. Test users exist: `uv run manage.py create_test_users`
3. Test data exists: `uv run manage.py loaddata test_data`
## Moderation Dashboard
### Edit Submissions
- [ ] **Approve submission** shows green toast and updates status badge to "Approved"
- [ ] **Reject submission** shows red toast and updates status badge to "Rejected"
- [ ] **Escalate submission** shows yellow toast and updates status badge to "Escalated"
- [ ] Confirmation dialog appears for approve/reject/escalate actions
- [ ] Loading spinner appears during transition (visible briefly)
- [ ] Row updates without full page reload
- [ ] Multiple transitions in sequence work correctly
- [ ] Transition buttons disappear after final state reached (Approved/Rejected)
- [ ] Status badge color matches state (green=approved, red=rejected, yellow=escalated)
- [ ] Toast notification auto-dismisses after ~3-5 seconds
### Photo Submissions
- [ ] **Approve photo** shows success toast and creates approved photo
- [ ] **Reject photo** shows warning toast and updates status
- [ ] Photo preview remains visible during transition
- [ ] Loading state shown while processing image approval
### Moderation Queue
- [ ] **Start** button transitions item to "In Progress"
- [ ] **Complete** button transitions item to "Completed"
- [ ] **Cancel** button transitions item to "Cancelled"
- [ ] Queue items update in real-time without refresh
- [ ] Assignment updates correctly when claiming item
### Bulk Operations
- [ ] **Cancel** pending operation shows confirmation and cancels
- [ ] Progress indicator updates during operation
- [ ] Cannot cancel completed operations (button hidden)
## Park Detail Page
### Status Transitions (as Moderator)
- [ ] **Close Temporarily** button visible for operating parks
- [ ] Status badge updates to "Temporarily Closed" (yellow)
- [ ] **Reopen** button appears after temporary closure
- [ ] Status badge updates back to "Operating" (green)
- [ ] **Close Permanently** transitions to "Permanently Closed" (red)
- [ ] **Mark as Demolished** available only from permanently closed
- [ ] **Mark as Relocated** available only from permanently closed
- [ ] Status badge updates to "Demolished" (gray) or "Relocated" (gray)
### Status Actions Section
- [ ] Available transitions update based on current state
- [ ] Buttons hidden for invalid transitions (can't reopen operating park)
- [ ] Confirmation dialogs appear for destructive transitions
- [ ] Toast notifications appear for all transitions
- [ ] Section refreshes via HTMX swap without full page reload
### Unauthorized Users
- [ ] Regular users cannot see transition buttons
- [ ] Status badge visible but not clickable for regular users
- [ ] Moderator buttons visible only to moderators
## Ride Detail Page
### Status Transitions (as Moderator)
- [ ] **Close Temporarily** button works correctly
- [ ] **Mark SBNO** transitions to "Standing But Not Operating"
- [ ] **Set Closing** transitions to "Closing" status
- [ ] Closing status with future date shows countdown (if implemented)
- [ ] **Reopen** button appears for closed/SBNO rides
- [ ] **Close Permanently** available for operating/closed_temp rides
- [ ] **Mark as Demolished** available only from permanently closed
- [ ] **Mark as Relocated** available only from permanently closed
### SBNO State
- [ ] SBNO status badge shows distinct styling (orange/amber)
- [ ] Reopen available from SBNO state
- [ ] Close Permanently available from SBNO state
## Error Handling
### Invalid Transitions
- [ ] Attempting invalid transition shows error toast (red)
- [ ] Error message is user-friendly (not technical exception)
- [ ] Database state unchanged after failed transition
- [ ] UI remains in consistent state after error
### Permission Denied
- [ ] Unauthorized transition attempt shows 403 error toast
- [ ] Error message says "Permission denied" or similar
- [ ] No partial state change occurs
### Network Errors
- [ ] Offline/network failure shows error toast
- [ ] "Connection failed" or similar message displayed
- [ ] Retry option available (if applicable)
- [ ] UI recovers gracefully from network errors
### Server Errors
- [ ] 500 errors show user-friendly message
- [ ] No raw exception/traceback shown to user
- [ ] Error toast has red/danger styling
## Loading States
- [ ] Loading spinner/indicator appears immediately on button click
- [ ] Button becomes disabled during transition
- [ ] Spinner disappears when response received
- [ ] Multiple rapid clicks don't cause duplicate requests
## Confirmation Dialogs
- [ ] Dialog appears for dangerous transitions (reject, cancel, demolish)
- [ ] Dialog shows appropriate warning message
- [ ] Clicking "Cancel" prevents transition
- [ ] Clicking "OK/Confirm" executes transition
- [ ] ESC key dismisses dialog without action
- [ ] Dialog backdrop click dismisses without action
## Cross-Browser Testing
Test all scenarios above in each browser:
### Desktop
- [ ] Chrome/Chromium (latest)
- [ ] Firefox (latest)
- [ ] Safari/WebKit (latest)
- [ ] Edge (latest)
### Mobile
- [ ] Mobile Safari (iOS 15+)
- [ ] Chrome Mobile (Android)
- [ ] Samsung Internet (Android)
### Notes
- Test with different viewport sizes (mobile, tablet, desktop)
- Verify touch interactions work on mobile
- Check that modals/dialogs are properly positioned on mobile
## Accessibility
### Keyboard Navigation
- [ ] All transition buttons focusable via Tab key
- [ ] Enter key activates focused button
- [ ] Escape key closes confirmation dialogs
- [ ] Focus indicator visible on all interactive elements
- [ ] Focus returns to appropriate element after transition
### Screen Reader
- [ ] Status changes announced to screen readers
- [ ] Toast notifications announced (aria-live region)
- [ ] Button labels are descriptive
- [ ] Loading states communicated to screen readers
### Visual
- [ ] Color contrast meets WCAG AA standards
- [ ] Status colors distinguishable without relying solely on color
- [ ] Focus indicators have sufficient contrast
- [ ] Text remains readable at 200% zoom
## Performance
- [ ] Transitions complete in under 500ms (typical)
- [ ] No visible flickering during HTMX swap
- [ ] Memory usage stable after multiple transitions
- [ ] No console errors during normal operation
## Test Data Reset
After testing, reset test data:
```bash
uv run manage.py flush_test_data
uv run manage.py loaddata test_data
```
## Reporting Issues
When reporting issues:
1. Note browser and version
2. Include console errors (if any)
3. Describe expected vs actual behavior
4. Include network request/response details
5. Screenshot or video if possible
## Automated Testing
Run automated E2E tests to complement manual testing:
```bash
# Run all FSM E2E tests
pytest backend/tests/e2e/test_moderation_fsm.py
pytest backend/tests/e2e/test_park_ride_fsm.py
pytest backend/tests/e2e/test_fsm_permissions.py
pytest backend/tests/e2e/test_fsm_error_handling.py
# Run with specific browser
pytest --browser firefox backend/tests/e2e/
# Run with headed mode (visible browser)
pytest --headed backend/tests/e2e/
# Run integration tests (faster, no browser)
pytest backend/tests/integration/test_fsm_transition_view.py
```

View File

@@ -34,10 +34,10 @@ def setup_page(page: Page):
@pytest.fixture
def auth_page(page: Page):
def auth_page(page: Page, live_server):
"""Fixture for authenticated page"""
# Login
page.goto("http://localhost:8000/accounts/login/")
# Login using live_server URL
page.goto(f"{live_server.url}/accounts/login/")
page.get_by_label("Username").fill("testuser")
page.get_by_label("Password").fill("testpass123")
page.get_by_role("button", name="Sign In").click()
@@ -46,10 +46,10 @@ def auth_page(page: Page):
@pytest.fixture
def mod_page(page: Page):
def mod_page(page: Page, live_server):
"""Fixture for moderator page"""
# Login as moderator
page.goto("http://localhost:8000/accounts/login/")
# Login as moderator using live_server URL
page.goto(f"{live_server.url}/accounts/login/")
page.get_by_label("Username").fill("moderator")
page.get_by_label("Password").fill("modpass123")
page.get_by_role("button", name="Sign In").click()
@@ -58,10 +58,10 @@ def mod_page(page: Page):
@pytest.fixture
def test_park(auth_page: Page):
def test_park(auth_page: Page, live_server):
"""Fixture for test park"""
# Create test park
auth_page.goto("http://localhost:8000/parks/create/")
# Create test park using live_server URL
auth_page.goto(f"{live_server.url}/parks/create/")
auth_page.get_by_label("Name").fill("Test Park")
auth_page.get_by_label("Location").fill("Orlando, FL")
auth_page.get_by_label("Description").fill("A test theme park")
@@ -72,10 +72,10 @@ def test_park(auth_page: Page):
@pytest.fixture
def test_ride(test_park: Page):
def test_ride(test_park: Page, live_server):
"""Fixture for test ride"""
# Create test ride
test_park.goto("http://localhost:8000/rides/create/")
# Create test ride using live_server URL
test_park.goto(f"{live_server.url}/rides/create/")
test_park.get_by_label("Name").fill("Test Ride")
test_park.get_by_label("Park").select_option("Test Park")
test_park.get_by_label("Type").select_option("Roller Coaster")
@@ -87,10 +87,10 @@ def test_ride(test_park: Page):
@pytest.fixture
def test_review(test_park: Page):
def test_review(test_park: Page, live_server):
"""Fixture for test review"""
# Create test review
test_park.goto("http://localhost:8000/parks/test-park/")
# Create test review using live_server URL
test_park.goto(f"{live_server.url}/parks/test-park/")
test_park.get_by_role("tab", name="Reviews").click()
test_park.get_by_role("button", name="Write Review").click()
test_park.get_by_label("Rating").select_option("5")
@@ -99,3 +99,310 @@ def test_review(test_park: Page):
test_park.get_by_role("button", name="Submit Review").click()
yield test_park
# =============================================================================
# FSM Testing Fixtures
# =============================================================================
@pytest.fixture
def admin_page(page: Page, live_server):
"""Fixture for admin/superuser page"""
# Login as admin using live_server URL
page.goto(f"{live_server.url}/accounts/login/")
page.get_by_label("Username").fill("admin")
page.get_by_label("Password").fill("adminpass123")
page.get_by_role("button", name="Sign In").click()
yield page
@pytest.fixture
def submission_pending(db):
"""Create a pending EditSubmission for FSM testing."""
from django.contrib.auth import get_user_model
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
# Get or create test user
user, _ = User.objects.get_or_create(
username="fsm_test_submitter",
defaults={"email": "fsm_test@example.com"}
)
user.set_password("testpass123")
user.save()
# Get a park
park = Park.objects.first()
if not park:
pytest.skip("No parks available for testing")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "FSM test submission"},
reason="FSM e2e test",
status="PENDING"
)
yield submission
# Cleanup
try:
submission.delete()
except Exception:
pass
@pytest.fixture
def submission_approved(db):
"""Create an approved EditSubmission for FSM testing."""
from django.contrib.auth import get_user_model
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
user, _ = User.objects.get_or_create(
username="fsm_test_submitter_approved",
defaults={"email": "fsm_approved@example.com"}
)
park = Park.objects.first()
if not park:
pytest.skip("No parks available for testing")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Already approved"},
reason="FSM approved test",
status="APPROVED"
)
yield submission
try:
submission.delete()
except Exception:
pass
@pytest.fixture
def park_operating(db):
"""Create an operating Park for FSM testing."""
from tests.factories import ParkFactory
park = ParkFactory(
name="FSM Test Park Operating",
slug="fsm-test-park-operating",
status="OPERATING"
)
yield park
@pytest.fixture
def park_closed_temp(db):
"""Create a temporarily closed Park for FSM testing."""
from tests.factories import ParkFactory
park = ParkFactory(
name="FSM Test Park Closed Temp",
slug="fsm-test-park-closed-temp",
status="CLOSED_TEMP"
)
yield park
@pytest.fixture
def park_closed_perm(db):
"""Create a permanently closed Park for FSM testing."""
from tests.factories import ParkFactory
from datetime import date, timedelta
park = ParkFactory(
name="FSM Test Park Closed Perm",
slug="fsm-test-park-closed-perm",
status="CLOSED_PERM",
closing_date=date.today() - timedelta(days=365)
)
yield park
@pytest.fixture
def ride_operating(db, park_operating):
"""Create an operating Ride for FSM testing."""
from tests.factories import RideFactory
ride = RideFactory(
name="FSM Test Ride Operating",
slug="fsm-test-ride-operating",
park=park_operating,
status="OPERATING"
)
yield ride
@pytest.fixture
def ride_sbno(db, park_operating):
"""Create an SBNO Ride for FSM testing."""
from tests.factories import RideFactory
ride = RideFactory(
name="FSM Test Ride SBNO",
slug="fsm-test-ride-sbno",
park=park_operating,
status="SBNO"
)
yield ride
@pytest.fixture
def ride_closed_perm(db, park_operating):
"""Create a permanently closed Ride for FSM testing."""
from tests.factories import RideFactory
from datetime import date, timedelta
ride = RideFactory(
name="FSM Test Ride Closed Perm",
slug="fsm-test-ride-closed-perm",
park=park_operating,
status="CLOSED_PERM",
closing_date=date.today() - timedelta(days=365)
)
yield ride
@pytest.fixture
def queue_item_pending(db):
"""Create a pending ModerationQueue item for FSM testing."""
from django.contrib.auth import get_user_model
from apps.moderation.models import ModerationQueue
User = get_user_model()
user, _ = User.objects.get_or_create(
username="fsm_queue_flagger",
defaults={"email": "fsm_queue@example.com"}
)
queue_item = ModerationQueue.objects.create(
item_type="CONTENT_REVIEW",
status="PENDING",
priority="MEDIUM",
title="FSM Test Queue Item",
description="Queue item for FSM e2e testing",
flagged_by=user
)
yield queue_item
try:
queue_item.delete()
except Exception:
pass
@pytest.fixture
def bulk_operation_pending(db):
"""Create a pending BulkOperation for FSM testing."""
from django.contrib.auth import get_user_model
from apps.moderation.models import BulkOperation
User = get_user_model()
user, _ = User.objects.get_or_create(
username="fsm_bulk_creator",
defaults={"email": "fsm_bulk@example.com", "is_staff": True}
)
operation = BulkOperation.objects.create(
operation_type="IMPORT",
status="PENDING",
priority="MEDIUM",
description="FSM Test Bulk Operation",
parameters={"test": True},
created_by=user,
total_items=10
)
yield operation
try:
operation.delete()
except Exception:
pass
# =============================================================================
# Helper Fixtures
# =============================================================================
@pytest.fixture
def live_server(live_server_url):
"""Provide the live server URL for tests.
Note: This fixture is provided by pytest-django. The live_server_url
fixture provides the URL as a string.
"""
class LiveServer:
url = live_server_url
return LiveServer()
@pytest.fixture
def moderator_user(db):
"""Get or create a moderator user for testing."""
from django.contrib.auth import get_user_model
User = get_user_model()
user, _ = User.objects.get_or_create(
username="moderator",
defaults={
"email": "moderator@example.com",
"is_staff": True
}
)
user.set_password("modpass123")
user.save()
return user
@pytest.fixture
def regular_user(db):
"""Get or create a regular user for testing."""
from django.contrib.auth import get_user_model
User = get_user_model()
user, _ = User.objects.get_or_create(
username="testuser",
defaults={"email": "testuser@example.com"}
)
user.set_password("testpass123")
user.save()
return user

View File

@@ -0,0 +1,668 @@
"""
E2E Tests for FSM Error Handling and Loading States
Tests error scenarios and loading indicators during FSM transitions:
- Invalid transitions show error toast
- Loading indicator appears during transition
- Network error shows error toast
- Validation error shows user-friendly message
- Confirm dialog appears for dangerous transitions
- Cancel confirm dialog prevents transition
These tests verify:
- Error toast appears with appropriate message
- Loading spinner appears during requests
- Network errors are handled gracefully
- Confirmation dialogs work correctly
- User-friendly error messages are displayed
"""
import pytest
from playwright.sync_api import Page, expect
class TestInvalidTransitionErrors:
"""Tests for error handling when attempting invalid transitions."""
def test_invalid_transition_shows_error_toast(
self, mod_page: Page, live_server, db
):
"""Test that attempting an invalid transition shows an error toast."""
from apps.parks.models import Park
# Get an operating park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Attempt an invalid transition via direct API call
# For example, trying to reopen an already operating park
response = mod_page.evaluate(f"""
async () => {{
const response = await fetch('/core/fsm/parks/park/{park.pk}/transition/transition_to_operating/', {{
method: 'POST',
headers: {{
'HX-Request': 'true',
'Content-Type': 'application/x-www-form-urlencoded',
'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]')?.value || ''
}},
credentials: 'include'
}});
return {{
status: response.status,
hxTrigger: response.headers.get('HX-Trigger')
}};
}}
""")
# Should return error status (400)
if response:
assert response.get('status') in [400, 403]
# Check for error toast in HX-Trigger header
hx_trigger = response.get('hxTrigger')
if hx_trigger:
assert 'showToast' in hx_trigger
assert 'error' in hx_trigger.lower()
def test_already_transitioned_shows_error(
self, mod_page: Page, live_server, db
):
"""Test that trying to approve an already-approved submission shows error."""
from django.contrib.auth import get_user_model
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
# Create an already-approved submission
user, _ = User.objects.get_or_create(
username="testsubmitter2",
defaults={"email": "testsubmitter2@example.com"}
)
park = Park.objects.first()
if not park:
pytest.skip("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Already approved"},
reason="Already approved test",
status="APPROVED" # Already approved
)
try:
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
# Try to approve again via direct API call
response = mod_page.evaluate(f"""
async () => {{
const response = await fetch('/core/fsm/moderation/editsubmission/{submission.pk}/transition/transition_to_approved/', {{
method: 'POST',
headers: {{
'HX-Request': 'true',
'Content-Type': 'application/x-www-form-urlencoded',
'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]')?.value || ''
}},
credentials: 'include'
}});
return {{
status: response.status,
hxTrigger: response.headers.get('HX-Trigger')
}};
}}
""")
# Should return error status
if response:
assert response.get('status') in [400, 403]
finally:
submission.delete()
def test_nonexistent_transition_shows_error(
self, mod_page: Page, live_server, db
):
"""Test that requesting a non-existent transition shows error."""
from apps.parks.models import Park
park = Park.objects.first()
if not park:
pytest.skip("No park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Try to call a non-existent transition
response = mod_page.evaluate(f"""
async () => {{
const response = await fetch('/core/fsm/parks/park/{park.pk}/transition/nonexistent_transition/', {{
method: 'POST',
headers: {{
'HX-Request': 'true',
'Content-Type': 'application/x-www-form-urlencoded',
'X-CSRFToken': document.querySelector('[name=csrfmiddlewaretoken]')?.value || ''
}},
credentials: 'include'
}});
return {{
status: response.status,
hxTrigger: response.headers.get('HX-Trigger')
}};
}}
""")
# Should return error status (400 or 404)
if response:
assert response.get('status') in [400, 404]
class TestLoadingIndicators:
"""Tests for loading indicator visibility during transitions."""
def test_loading_indicator_appears_during_transition(
self, mod_page: Page, live_server, db
):
"""Verify loading spinner appears during HTMX transition."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_actions = mod_page.locator('[data-park-status-actions]')
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
if not close_temp_btn.is_visible():
pytest.skip("Close Temporarily button not visible")
# Add a route to slow down the request so we can see loading state
mod_page.route("**/core/fsm/**", lambda route: (
mod_page.wait_for_timeout(500),
route.continue_()
))
# Handle confirmation dialog
mod_page.on("dialog", lambda dialog: dialog.accept())
# Click and immediately check for loading indicator
close_temp_btn.click()
# Check for htmx-indicator class or spinner
loading_indicator = mod_page.locator('.htmx-indicator, .htmx-request .spinner, [class*="loading"]')
# The loading indicator should appear (may be brief)
# We wait a short time for it to appear
try:
expect(loading_indicator.first).to_be_visible(timeout=1000)
except Exception:
# Loading indicator may have already disappeared if response was fast
pass
# Wait for transition to complete
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
def test_button_disabled_during_transition(
self, mod_page: Page, live_server, db
):
"""Test that transition button is disabled during request."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_actions = mod_page.locator('[data-park-status-actions]')
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
if not close_temp_btn.is_visible():
pytest.skip("Close Temporarily button not visible")
# Add a route to slow down the request
mod_page.route("**/core/fsm/**", lambda route: (
mod_page.wait_for_timeout(1000),
route.continue_()
))
mod_page.on("dialog", lambda dialog: dialog.accept())
# Click button
close_temp_btn.click()
# Check if button becomes disabled or has htmx-request class
expect(close_temp_btn).to_have_attribute("disabled", "", timeout=500)
class TestNetworkErrorHandling:
"""Tests for handling network errors during transitions."""
def test_network_error_shows_error_toast(
self, mod_page: Page, live_server, db
):
"""Test that network errors show appropriate error toast."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Abort network requests to simulate network error
mod_page.route("**/core/fsm/**", lambda route: route.abort("failed"))
status_actions = mod_page.locator('[data-park-status-actions]')
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
if not close_temp_btn.is_visible():
pytest.skip("Close Temporarily button not visible")
mod_page.on("dialog", lambda dialog: dialog.accept())
# Click button - should trigger network error
close_temp_btn.click()
# HTMX should show error indication
# Check for error toast or htmx error event
error_indicator = mod_page.locator('[data-toast].error, .htmx-error, [class*="error"]')
# May show as toast or inline error
try:
expect(error_indicator.first).to_be_visible(timeout=5000)
except Exception:
# Error may be handled differently
pass
# Verify database state was NOT changed
park.refresh_from_db()
assert park.status == "OPERATING"
def test_server_error_shows_user_friendly_message(
self, mod_page: Page, live_server, db
):
"""Test that server errors show user-friendly messages."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Return 500 error to simulate server error
mod_page.route("**/core/fsm/**", lambda route: route.fulfill(
status=500,
headers={"HX-Trigger": '{"showToast": {"message": "An unexpected error occurred", "type": "error"}}'},
body=""
))
status_actions = mod_page.locator('[data-park-status-actions]')
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
if not close_temp_btn.is_visible():
pytest.skip("Close Temporarily button not visible")
mod_page.on("dialog", lambda dialog: dialog.accept())
close_temp_btn.click()
# Should show user-friendly error message
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Should not show technical error details to user
expect(toast).not_to_contain_text("Traceback")
expect(toast).not_to_contain_text("Exception")
class TestConfirmationDialogs:
"""Tests for confirmation dialogs on dangerous transitions."""
def test_confirm_dialog_appears_for_reject_transition(
self, mod_page: Page, live_server, db
):
"""Test that confirmation dialog appears for reject transition."""
from django.contrib.auth import get_user_model
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
user, _ = User.objects.get_or_create(
username="testsubmitter3",
defaults={"email": "testsubmitter3@example.com"}
)
park = Park.objects.first()
if not park:
pytest.skip("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Confirm dialog test"},
reason="Confirm dialog test",
status="PENDING"
)
dialog_shown = {"shown": False}
def handle_dialog(dialog):
dialog_shown["shown"] = True
assert "reject" in dialog.message.lower() or "sure" in dialog.message.lower()
dialog.dismiss() # Cancel the action
try:
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
mod_page.on("dialog", handle_dialog)
submission_row = mod_page.locator(
f'[data-submission-id="{submission.pk}"]'
)
if submission_row.is_visible():
reject_btn = submission_row.get_by_role("button", name="Reject")
if reject_btn.is_visible():
reject_btn.click()
# Give time for dialog to appear
mod_page.wait_for_timeout(500)
# Verify dialog was shown
assert dialog_shown["shown"], "Confirmation dialog should have been shown"
finally:
submission.delete()
def test_cancel_confirm_dialog_prevents_transition(
self, mod_page: Page, live_server, db
):
"""Test that canceling the confirmation dialog prevents the transition."""
from django.contrib.auth import get_user_model
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
user, _ = User.objects.get_or_create(
username="testsubmitter4",
defaults={"email": "testsubmitter4@example.com"}
)
park = Park.objects.first()
if not park:
pytest.skip("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Cancel confirm test"},
reason="Cancel confirm test",
status="PENDING"
)
try:
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
# Dismiss (cancel) the dialog
mod_page.on("dialog", lambda dialog: dialog.dismiss())
submission_row = mod_page.locator(
f'[data-submission-id="{submission.pk}"]'
)
if submission_row.is_visible():
reject_btn = submission_row.get_by_role("button", name="Reject")
if reject_btn.is_visible():
reject_btn.click()
# Wait a moment
mod_page.wait_for_timeout(500)
# Verify submission status was NOT changed
submission.refresh_from_db()
assert submission.status == "PENDING"
finally:
submission.delete()
def test_accept_confirm_dialog_executes_transition(
self, mod_page: Page, live_server, db
):
"""Test that accepting the confirmation dialog executes the transition."""
from django.contrib.auth import get_user_model
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
user, _ = User.objects.get_or_create(
username="testsubmitter5",
defaults={"email": "testsubmitter5@example.com"}
)
park = Park.objects.first()
if not park:
pytest.skip("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Accept confirm test"},
reason="Accept confirm test",
status="PENDING"
)
try:
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
# Accept the dialog
mod_page.on("dialog", lambda dialog: dialog.accept())
submission_row = mod_page.locator(
f'[data-submission-id="{submission.pk}"]'
)
if submission_row.is_visible():
reject_btn = submission_row.get_by_role("button", name="Reject")
if reject_btn.is_visible():
reject_btn.click()
# Wait for transition to complete
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify submission status WAS changed
submission.refresh_from_db()
assert submission.status == "REJECTED"
finally:
submission.delete()
class TestValidationErrors:
"""Tests for validation error handling."""
def test_validation_error_shows_specific_message(
self, mod_page: Page, live_server, db
):
"""Test that validation errors show specific error messages."""
# This test depends on having transitions that require additional data
# For example, a transition that requires a reason field
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Some transitions might require additional data (like closing_date for close_permanently)
# If validation fails, it should show a specific message
# This is a placeholder test - actual behavior depends on FSM configuration
# The test verifies that validation errors are handled gracefully
pass
class TestToastNotificationBehavior:
"""Tests for toast notification appearance and behavior."""
def test_success_toast_auto_dismisses(
self, mod_page: Page, live_server, db
):
"""Test that success toast auto-dismisses after timeout."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_actions = mod_page.locator('[data-park-status-actions]')
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
if not close_temp_btn.is_visible():
pytest.skip("Close Temporarily button not visible")
mod_page.on("dialog", lambda dialog: dialog.accept())
close_temp_btn.click()
# Toast should appear
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Toast should auto-dismiss after timeout (typically 3-5 seconds)
# Wait for auto-dismiss
expect(toast).not_to_be_visible(timeout=10000)
def test_error_toast_has_correct_styling(
self, mod_page: Page, live_server, db
):
"""Test that error toast has correct red/danger styling."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Simulate an error response
mod_page.route("**/core/fsm/**", lambda route: route.fulfill(
status=400,
headers={
"HX-Trigger": '{"showToast": {"message": "Test error message", "type": "error"}}'
},
body=""
))
status_actions = mod_page.locator('[data-park-status-actions]')
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
if not close_temp_btn.is_visible():
pytest.skip("Close Temporarily button not visible")
mod_page.on("dialog", lambda dialog: dialog.accept())
close_temp_btn.click()
# Error toast should appear with error styling
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Should have error/danger styling (red)
expect(toast).to_have_class(/error|danger|bg-red|text-red/)
def test_success_toast_has_correct_styling(
self, mod_page: Page, live_server, db
):
"""Test that success toast has correct green/success styling."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
# Reset any previous state
park.status = "OPERATING"
park.save()
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_actions = mod_page.locator('[data-park-status-actions]')
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
if not close_temp_btn.is_visible():
pytest.skip("Close Temporarily button not visible")
mod_page.on("dialog", lambda dialog: dialog.accept())
close_temp_btn.click()
# Success toast should appear with success styling
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Should have success styling (green)
expect(toast).to_have_class(/success|bg-green|text-green/)

View File

@@ -0,0 +1,505 @@
"""
E2E Tests for FSM Permission Guards
Tests that unauthorized users cannot execute FSM transitions:
- Unauthenticated users cannot see transition buttons
- Regular users cannot approve submissions
- Regular users cannot change park/ride status
- Moderators can approve but not admin-only transitions
- Transition buttons hidden when not allowed
These tests verify:
- Transition buttons are NOT visible for unauthorized users
- Direct POST requests return 403 Forbidden
- Database state does NOT change after failed transition attempt
- Error toast displays "Permission denied" message
"""
import pytest
from playwright.sync_api import Page, expect
class TestUnauthenticatedUserPermissions:
"""Tests for unauthenticated user permission guards."""
def test_unauthenticated_user_cannot_see_moderation_dashboard(
self, page: Page, live_server
):
"""Test that unauthenticated users are redirected from moderation dashboard."""
# Navigate to moderation dashboard without logging in
response = page.goto(f"{live_server.url}/moderation/dashboard/")
# Should be redirected to login page or see access denied
# Check URL contains login or access denied
current_url = page.url
assert "login" in current_url or "denied" in current_url or response.status == 403
def test_unauthenticated_user_cannot_see_transition_buttons(
self, page: Page, live_server, db
):
"""Test that unauthenticated users cannot see transition buttons on park detail."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
page.goto(f"{live_server.url}/parks/{park.slug}/")
page.wait_for_load_state("networkidle")
# Status action buttons should NOT be visible
status_actions = page.locator('[data-park-status-actions]')
# Either the section doesn't exist or the buttons are not there
if status_actions.is_visible():
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
expect(close_temp_btn).not_to_be_visible()
def test_unauthenticated_direct_post_returns_403(
self, page: Page, live_server, db
):
"""Test that direct POST to FSM endpoint returns 403 for unauthenticated user."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
# Attempt to POST directly to FSM transition endpoint
response = page.request.post(
f"{live_server.url}/core/fsm/parks/park/{park.pk}/transition/transition_to_closed_temp/",
headers={"HX-Request": "true"}
)
# Should get 403 Forbidden
assert response.status == 403 or response.status == 302 # 302 redirect to login
# Verify database state did NOT change
park.refresh_from_db()
assert park.status == "OPERATING"
class TestRegularUserPermissions:
"""Tests for regular (non-moderator) user permission guards."""
def test_regular_user_cannot_approve_submission(
self, auth_page: Page, live_server, db
):
"""Test that regular users cannot approve submissions."""
from django.contrib.auth import get_user_model
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
# Create a pending submission
user = User.objects.filter(username="testuser").first()
if not user:
pytest.skip("Test user not found")
park = Park.objects.first()
if not park:
pytest.skip("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Test change"},
reason="Permission test",
status="PENDING"
)
try:
# Navigate to moderation dashboard as regular user
auth_page.goto(f"{live_server.url}/moderation/dashboard/")
# Regular user should be redirected or denied
current_url = auth_page.url
# If somehow on dashboard, verify no approve button
if "dashboard" in current_url:
submission_row = auth_page.locator(
f'[data-submission-id="{submission.pk}"]'
)
if submission_row.is_visible():
approve_btn = submission_row.get_by_role("button", name="Approve")
expect(approve_btn).not_to_be_visible()
# Try direct POST - should be denied
response = auth_page.request.post(
f"{live_server.url}/core/fsm/moderation/editsubmission/{submission.pk}/transition/transition_to_approved/",
headers={"HX-Request": "true"}
)
# Should be denied (403 or 302 redirect)
assert response.status in [302, 403]
# Verify database state did NOT change
submission.refresh_from_db()
assert submission.status == "PENDING"
finally:
submission.delete()
def test_regular_user_cannot_change_park_status(
self, auth_page: Page, live_server, db
):
"""Test that regular users cannot change park status."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
auth_page.goto(f"{live_server.url}/parks/{park.slug}/")
auth_page.wait_for_load_state("networkidle")
# Status action buttons should NOT be visible to regular user
status_actions = auth_page.locator('[data-park-status-actions]')
if status_actions.is_visible():
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
expect(close_temp_btn).not_to_be_visible()
# Try direct POST - should be denied
response = auth_page.request.post(
f"{live_server.url}/core/fsm/parks/park/{park.pk}/transition/transition_to_closed_temp/",
headers={"HX-Request": "true"}
)
# Should be denied
assert response.status in [302, 400, 403]
# Verify database state did NOT change
park.refresh_from_db()
assert park.status == "OPERATING"
def test_regular_user_cannot_change_ride_status(
self, auth_page: Page, live_server, db
):
"""Test that regular users cannot change ride status."""
from apps.rides.models import Ride
ride = Ride.objects.filter(status="OPERATING").first()
if not ride:
pytest.skip("No operating ride available")
auth_page.goto(
f"{live_server.url}/parks/{ride.park.slug}/rides/{ride.slug}/"
)
auth_page.wait_for_load_state("networkidle")
# Status action buttons should NOT be visible to regular user
status_actions = auth_page.locator('[data-ride-status-actions]')
if status_actions.is_visible():
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
expect(close_temp_btn).not_to_be_visible()
# Try direct POST - should be denied
response = auth_page.request.post(
f"{live_server.url}/core/fsm/rides/ride/{ride.pk}/transition/transition_to_closed_temp/",
headers={"HX-Request": "true"}
)
# Should be denied
assert response.status in [302, 400, 403]
# Verify database state did NOT change
ride.refresh_from_db()
assert ride.status == "OPERATING"
class TestModeratorPermissions:
"""Tests for moderator-specific permission guards."""
def test_moderator_can_approve_submission(
self, mod_page: Page, live_server, db
):
"""Test that moderators CAN see and use approve button."""
from django.contrib.auth import get_user_model
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
# Create a pending submission
user = User.objects.filter(username="testuser").first()
if not user:
user = User.objects.create_user(
username="testuser",
email="testuser@example.com",
password="testpass123"
)
park = Park.objects.first()
if not park:
pytest.skip("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Test change for moderator"},
reason="Moderator permission test",
status="PENDING"
)
try:
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
# Moderator should be able to see the submission
submission_row = mod_page.locator(
f'[data-submission-id="{submission.pk}"]'
)
if submission_row.is_visible():
# Should see approve button
approve_btn = submission_row.get_by_role("button", name="Approve")
expect(approve_btn).to_be_visible()
finally:
submission.delete()
def test_moderator_can_change_park_status(
self, mod_page: Page, live_server, db
):
"""Test that moderators CAN see and use park status change buttons."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Status action buttons SHOULD be visible to moderator
status_actions = mod_page.locator('[data-park-status-actions]')
if status_actions.is_visible():
# Should see close temporarily button
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
expect(close_temp_btn).to_be_visible()
def test_moderator_cannot_access_admin_only_transitions(
self, mod_page: Page, live_server, db
):
"""Test that moderators CANNOT access admin-only transitions."""
# This test verifies that certain transitions require admin privileges
# Specific transitions depend on the FSM configuration
from apps.parks.models import Park
# Get a permanently closed park for testing admin-only demolish
park = Park.objects.filter(status="CLOSED_PERM").first()
if not park:
# Create one
park = Park.objects.filter(status="OPERATING").first()
if park:
park.status = "CLOSED_PERM"
park.save()
else:
pytest.skip("No park available for testing")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Check for admin-only buttons (if any are configured)
# The specific buttons that should be hidden depend on the FSM configuration
status_actions = mod_page.locator('[data-park-status-actions]')
# If there are admin-only transitions, verify they're hidden
# This is a placeholder - actual admin-only transitions depend on configuration
admin_only_btn = status_actions.get_by_role(
"button", name="Force Delete" # Example admin-only action
)
expect(admin_only_btn).not_to_be_visible()
class TestPermissionDeniedErrorHandling:
"""Tests for error handling when permission is denied."""
def test_permission_denied_shows_error_toast(
self, auth_page: Page, live_server, db
):
"""Test that permission denied errors show appropriate toast."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
# Navigate to the page first
auth_page.goto(f"{live_server.url}/parks/{park.slug}/")
auth_page.wait_for_load_state("networkidle")
# Make the request programmatically with HTMX header
response = auth_page.evaluate("""
async () => {
const response = await fetch('/core/fsm/parks/park/""" + str(park.pk) + """/transition/transition_to_closed_temp/', {
method: 'POST',
headers: {
'HX-Request': 'true',
'Content-Type': 'application/x-www-form-urlencoded'
},
credentials: 'include'
});
return {
status: response.status,
hxTrigger: response.headers.get('HX-Trigger')
};
}
""")
# Check if error toast was triggered
if response and response.get('status') in [400, 403]:
hx_trigger = response.get('hxTrigger')
if hx_trigger:
assert 'showToast' in hx_trigger
assert 'error' in hx_trigger.lower() or 'denied' in hx_trigger.lower()
def test_database_state_unchanged_on_permission_denied(
self, auth_page: Page, live_server, db
):
"""Test that database state is unchanged when permission is denied."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
original_status = park.status
# Attempt unauthorized transition via direct fetch
auth_page.goto(f"{live_server.url}/parks/{park.slug}/")
auth_page.wait_for_load_state("networkidle")
auth_page.evaluate("""
async () => {
await fetch('/core/fsm/parks/park/""" + str(park.pk) + """/transition/transition_to_closed_temp/', {
method: 'POST',
headers: {
'HX-Request': 'true',
'Content-Type': 'application/x-www-form-urlencoded'
},
credentials: 'include'
});
}
""")
# Verify database state did NOT change
park.refresh_from_db()
assert park.status == original_status
class TestTransitionButtonVisibility:
"""Tests for correct transition button visibility based on permissions and state."""
def test_transition_button_hidden_when_state_invalid(
self, mod_page: Page, live_server, db
):
"""Test that transition buttons are hidden when the current state is invalid."""
from apps.parks.models import Park
# Get an operating park
park = Park.objects.filter(status="OPERATING").first()
if not park:
pytest.skip("No operating park available")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_actions = mod_page.locator('[data-park-status-actions]')
# Reopen button should NOT be visible for operating park
# (can't reopen something that's already operating)
reopen_btn = status_actions.get_by_role("button", name="Reopen")
expect(reopen_btn).not_to_be_visible()
# Demolish should NOT be visible for operating park
# (can only demolish from CLOSED_PERM)
demolish_btn = status_actions.get_by_role("button", name="Mark as Demolished")
expect(demolish_btn).not_to_be_visible()
def test_correct_buttons_shown_for_closed_temp_state(
self, mod_page: Page, live_server, db
):
"""Test that correct buttons are shown for temporarily closed state."""
from apps.parks.models import Park
park = Park.objects.filter(status="CLOSED_TEMP").first()
if not park:
# Create one
park = Park.objects.filter(status="OPERATING").first()
if park:
park.status = "CLOSED_TEMP"
park.save()
else:
pytest.skip("No park available for testing")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_actions = mod_page.locator('[data-park-status-actions]')
# Reopen button SHOULD be visible
reopen_btn = status_actions.get_by_role("button", name="Reopen")
expect(reopen_btn).to_be_visible()
# Close Temporarily should NOT be visible (already closed)
close_temp_btn = status_actions.get_by_role(
"button", name="Close Temporarily"
)
expect(close_temp_btn).not_to_be_visible()
def test_correct_buttons_shown_for_closed_perm_state(
self, mod_page: Page, live_server, db
):
"""Test that correct buttons are shown for permanently closed state."""
from apps.parks.models import Park
park = Park.objects.filter(status="CLOSED_PERM").first()
if not park:
# Create one
park = Park.objects.filter(status="OPERATING").first()
if park:
park.status = "CLOSED_PERM"
park.save()
else:
pytest.skip("No park available for testing")
mod_page.goto(f"{live_server.url}/parks/{park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_actions = mod_page.locator('[data-park-status-actions]')
# Demolish/Relocate buttons SHOULD be visible
demolish_btn = status_actions.get_by_role("button", name="Mark as Demolished")
relocate_btn = status_actions.get_by_role("button", name="Mark as Relocated")
# At least one of these should be visible for CLOSED_PERM
visible = demolish_btn.is_visible() or relocate_btn.is_visible()
assert visible, "Expected demolish or relocate button for CLOSED_PERM state"
# Reopen should still be visible to restore to operating
reopen_btn = status_actions.get_by_role("button", name="Reopen")
# May or may not be visible depending on FSM configuration

View File

@@ -0,0 +1,477 @@
"""
E2E Tests for Moderation FSM Transitions
Tests the complete HTMX FSM transition flow for moderation-related models:
- EditSubmission: approve, reject, escalate
- PhotoSubmission: approve, reject
- ModerationQueue: complete, cancel
- BulkOperation: cancel
These tests verify:
- Status badge updates correctly after transition
- Toast notifications appear with correct message/type
- HTMX swap occurs without full page reload
- StateLog entry created in database
"""
import pytest
from playwright.sync_api import Page, expect
from django.contrib.contenttypes.models import ContentType
@pytest.fixture
def pending_submission(db):
"""Create a pending EditSubmission for testing."""
from django.contrib.auth import get_user_model
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
User = get_user_model()
# Get or create test user
user, _ = User.objects.get_or_create(
username="testsubmitter",
defaults={"email": "testsubmitter@example.com"}
)
user.set_password("testpass123")
user.save()
# Get or create a park for the submission
park = Park.objects.first()
if not park:
pytest.skip("No parks available for testing")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Updated park description for testing"},
reason="E2E test submission",
status="PENDING"
)
yield submission
# Cleanup
submission.delete()
@pytest.fixture
def pending_photo_submission(db):
"""Create a pending PhotoSubmission for testing."""
from django.contrib.auth import get_user_model
from apps.moderation.models import PhotoSubmission
from apps.parks.models import Park
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
# Get or create test user
user, _ = User.objects.get_or_create(
username="testphotosubmitter",
defaults={"email": "testphotosubmitter@example.com"}
)
user.set_password("testpass123")
user.save()
# Get or create a park
park = Park.objects.first()
if not park:
pytest.skip("No parks available for testing")
content_type = ContentType.objects.get_for_model(Park)
# Check if CloudflareImage model exists and has entries
try:
from django_cloudflareimages_toolkit.models import CloudflareImage
photo = CloudflareImage.objects.first()
if not photo:
pytest.skip("No CloudflareImage available for testing")
except ImportError:
pytest.skip("CloudflareImage not available")
submission = PhotoSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
photo=photo,
caption="E2E test photo",
status="PENDING"
)
yield submission
# Cleanup
submission.delete()
class TestEditSubmissionTransitions:
"""Tests for EditSubmission FSM transitions via HTMX."""
def test_submission_approve_transition_as_moderator(
self, mod_page: Page, pending_submission, live_server
):
"""Test approving an EditSubmission as a moderator."""
# Navigate to moderation dashboard
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
# Wait for the page to load
mod_page.wait_for_load_state("networkidle")
# Find the submission row
submission_row = mod_page.locator(f'[data-submission-id="{pending_submission.pk}"]')
# Verify initial status is pending
status_badge = submission_row.locator('[data-status-badge]')
expect(status_badge).to_contain_text("Pending")
# Click the approve button
approve_btn = submission_row.get_by_role("button", name="Approve")
# Handle confirmation dialog if present
mod_page.on("dialog", lambda dialog: dialog.accept())
# Click and wait for HTMX swap
approve_btn.click()
# Wait for toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
expect(toast).to_contain_text("approved")
# Verify status badge updated
expect(status_badge).to_contain_text("Approved")
# Verify database state
from apps.moderation.models import EditSubmission
pending_submission.refresh_from_db()
assert pending_submission.status == "APPROVED"
def test_submission_reject_transition_as_moderator(
self, mod_page: Page, pending_submission, live_server
):
"""Test rejecting an EditSubmission as a moderator."""
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
submission_row = mod_page.locator(f'[data-submission-id="{pending_submission.pk}"]')
# Verify initial status
status_badge = submission_row.locator('[data-status-badge]')
expect(status_badge).to_contain_text("Pending")
# Click reject button
reject_btn = submission_row.get_by_role("button", name="Reject")
# Handle confirmation dialog
mod_page.on("dialog", lambda dialog: dialog.accept())
reject_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
expect(toast).to_contain_text("rejected")
# Verify status badge updated (should show red/danger styling)
expect(status_badge).to_contain_text("Rejected")
# Verify database state
from apps.moderation.models import EditSubmission
pending_submission.refresh_from_db()
assert pending_submission.status == "REJECTED"
def test_submission_escalate_transition_as_moderator(
self, mod_page: Page, pending_submission, live_server
):
"""Test escalating an EditSubmission as a moderator."""
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
submission_row = mod_page.locator(f'[data-submission-id="{pending_submission.pk}"]')
# Verify initial status
status_badge = submission_row.locator('[data-status-badge]')
expect(status_badge).to_contain_text("Pending")
# Click escalate button
escalate_btn = submission_row.get_by_role("button", name="Escalate")
# Handle confirmation dialog
mod_page.on("dialog", lambda dialog: dialog.accept())
escalate_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
expect(toast).to_contain_text("escalated")
# Verify status badge updated
expect(status_badge).to_contain_text("Escalated")
# Verify database state
from apps.moderation.models import EditSubmission
pending_submission.refresh_from_db()
assert pending_submission.status == "ESCALATED"
class TestPhotoSubmissionTransitions:
"""Tests for PhotoSubmission FSM transitions via HTMX."""
def test_photo_submission_approve_transition(
self, mod_page: Page, pending_photo_submission, live_server
):
"""Test approving a PhotoSubmission as a moderator."""
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
# Click on Photos tab if it exists
photos_tab = mod_page.get_by_role("tab", name="Photos")
if photos_tab.is_visible():
photos_tab.click()
# Find the photo submission row
submission_row = mod_page.locator(
f'[data-photo-submission-id="{pending_photo_submission.pk}"]'
)
if not submission_row.is_visible():
pytest.skip("Photo submission not visible in dashboard")
# Click approve button
approve_btn = submission_row.get_by_role("button", name="Approve")
mod_page.on("dialog", lambda dialog: dialog.accept())
approve_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
expect(toast).to_contain_text("approved")
# Verify database state
from apps.moderation.models import PhotoSubmission
pending_photo_submission.refresh_from_db()
assert pending_photo_submission.status == "APPROVED"
def test_photo_submission_reject_transition(
self, mod_page: Page, pending_photo_submission, live_server
):
"""Test rejecting a PhotoSubmission as a moderator."""
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
# Click on Photos tab if it exists
photos_tab = mod_page.get_by_role("tab", name="Photos")
if photos_tab.is_visible():
photos_tab.click()
# Find the photo submission row
submission_row = mod_page.locator(
f'[data-photo-submission-id="{pending_photo_submission.pk}"]'
)
if not submission_row.is_visible():
pytest.skip("Photo submission not visible in dashboard")
# Click reject button
reject_btn = submission_row.get_by_role("button", name="Reject")
mod_page.on("dialog", lambda dialog: dialog.accept())
reject_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
expect(toast).to_contain_text("rejected")
# Verify database state
from apps.moderation.models import PhotoSubmission
pending_photo_submission.refresh_from_db()
assert pending_photo_submission.status == "REJECTED"
class TestModerationQueueTransitions:
"""Tests for ModerationQueue FSM transitions via HTMX."""
@pytest.fixture
def pending_queue_item(self, db):
"""Create a pending ModerationQueue item for testing."""
from django.contrib.auth import get_user_model
from apps.moderation.models import ModerationQueue
User = get_user_model()
user, _ = User.objects.get_or_create(
username="testflagger",
defaults={"email": "testflagger@example.com"}
)
queue_item = ModerationQueue.objects.create(
item_type="CONTENT_REVIEW",
status="PENDING",
priority="MEDIUM",
title="E2E Test Queue Item",
description="Queue item for E2E testing",
flagged_by=user
)
yield queue_item
queue_item.delete()
def test_moderation_queue_start_transition(
self, mod_page: Page, pending_queue_item, live_server
):
"""Test starting work on a ModerationQueue item."""
mod_page.goto(f"{live_server.url}/moderation/queue/")
mod_page.wait_for_load_state("networkidle")
# Find the queue item row
queue_row = mod_page.locator(f'[data-queue-id="{pending_queue_item.pk}"]')
if not queue_row.is_visible():
pytest.skip("Queue item not visible")
# Click start button
start_btn = queue_row.get_by_role("button", name="Start")
start_btn.click()
# Verify status updated to IN_PROGRESS
status_badge = queue_row.locator('[data-status-badge]')
expect(status_badge).to_contain_text("In Progress", timeout=5000)
# Verify database state
from apps.moderation.models import ModerationQueue
pending_queue_item.refresh_from_db()
assert pending_queue_item.status == "IN_PROGRESS"
def test_moderation_queue_complete_transition(
self, mod_page: Page, pending_queue_item, live_server
):
"""Test completing a ModerationQueue item."""
# First set status to IN_PROGRESS
pending_queue_item.status = "IN_PROGRESS"
pending_queue_item.save()
mod_page.goto(f"{live_server.url}/moderation/queue/")
mod_page.wait_for_load_state("networkidle")
queue_row = mod_page.locator(f'[data-queue-id="{pending_queue_item.pk}"]')
if not queue_row.is_visible():
pytest.skip("Queue item not visible")
# Click complete button
complete_btn = queue_row.get_by_role("button", name="Complete")
mod_page.on("dialog", lambda dialog: dialog.accept())
complete_btn.click()
# Verify toast and status
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
from apps.moderation.models import ModerationQueue
pending_queue_item.refresh_from_db()
assert pending_queue_item.status == "COMPLETED"
class TestBulkOperationTransitions:
"""Tests for BulkOperation FSM transitions via HTMX."""
@pytest.fixture
def pending_bulk_operation(self, db):
"""Create a pending BulkOperation for testing."""
from django.contrib.auth import get_user_model
from apps.moderation.models import BulkOperation
User = get_user_model()
user, _ = User.objects.get_or_create(
username="testadmin",
defaults={"email": "testadmin@example.com", "is_staff": True}
)
operation = BulkOperation.objects.create(
operation_type="IMPORT",
status="PENDING",
priority="MEDIUM",
description="E2E Test Bulk Operation",
parameters={"test": True},
created_by=user,
total_items=10
)
yield operation
operation.delete()
def test_bulk_operation_cancel_transition(
self, mod_page: Page, pending_bulk_operation, live_server
):
"""Test canceling a BulkOperation."""
mod_page.goto(f"{live_server.url}/moderation/bulk-operations/")
mod_page.wait_for_load_state("networkidle")
# Find the operation row
operation_row = mod_page.locator(
f'[data-bulk-operation-id="{pending_bulk_operation.pk}"]'
)
if not operation_row.is_visible():
pytest.skip("Bulk operation not visible")
# Click cancel button
cancel_btn = operation_row.get_by_role("button", name="Cancel")
mod_page.on("dialog", lambda dialog: dialog.accept())
cancel_btn.click()
# Verify toast
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
expect(toast).to_contain_text("cancel")
# Verify database state
from apps.moderation.models import BulkOperation
pending_bulk_operation.refresh_from_db()
assert pending_bulk_operation.status == "CANCELLED"
class TestTransitionLoadingStates:
"""Tests for loading indicators during FSM transitions."""
def test_loading_indicator_appears_during_transition(
self, mod_page: Page, pending_submission, live_server
):
"""Verify loading spinner appears during HTMX transition."""
mod_page.goto(f"{live_server.url}/moderation/dashboard/")
mod_page.wait_for_load_state("networkidle")
submission_row = mod_page.locator(
f'[data-submission-id="{pending_submission.pk}"]'
)
# Get approve button and associated loading indicator
approve_btn = submission_row.get_by_role("button", name="Approve")
# Slow down network to see loading state
mod_page.route("**/*", lambda route: route.continue_())
mod_page.on("dialog", lambda dialog: dialog.accept())
# Start transition
approve_btn.click()
# Check for htmx-indicator visibility (may be brief)
# The indicator should become visible during the request
loading_indicator = submission_row.locator('.htmx-indicator')
# Wait for transition to complete
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)

View File

@@ -0,0 +1,591 @@
"""
E2E Tests for Park and Ride FSM Status Transitions
Tests the complete HTMX FSM transition flow for park and ride status changes:
- Park: close_temporarily, reopen, close_permanently, demolish, relocate
- Ride: close_temporarily, mark_sbno, close_permanently, demolish, relocate
These tests verify:
- Status badge updates in detail page header
- Available transitions update based on current state
- Toast notifications appear with correct message/type
- HTMX swap occurs without full page reload
- StateLog entry created in database
"""
import pytest
from playwright.sync_api import Page, expect
from datetime import date, timedelta
@pytest.fixture
def operating_park(db):
"""Create an operating Park for testing status transitions."""
from apps.parks.models import Park
from tests.factories import ParkFactory
# Use factory to create a complete park
park = ParkFactory(
name="E2E Test Park",
slug="e2e-test-park",
status="OPERATING"
)
yield park
# Cleanup handled by factory
@pytest.fixture
def operating_ride(db, operating_park):
"""Create an operating Ride for testing status transitions."""
from tests.factories import RideFactory
ride = RideFactory(
name="E2E Test Ride",
slug="e2e-test-ride",
park=operating_park,
status="OPERATING"
)
yield ride
class TestParkStatusTransitions:
"""Tests for Park FSM status transitions via HTMX."""
def test_park_close_temporarily_as_moderator(
self, mod_page: Page, operating_park, live_server
):
"""Test closing a park temporarily as a moderator."""
mod_page.goto(f"{live_server.url}/parks/{operating_park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Verify initial status badge shows Operating
status_section = mod_page.locator('[data-park-status-actions]')
status_badge = mod_page.locator('[data-status-badge]')
expect(status_badge).to_contain_text("Operating")
# Find and click "Close Temporarily" button
close_temp_btn = status_section.get_by_role(
"button", name="Close Temporarily"
)
if not close_temp_btn.is_visible():
# May be in a dropdown menu
actions_dropdown = status_section.locator('[data-actions-dropdown]')
if actions_dropdown.is_visible():
actions_dropdown.click()
close_temp_btn = mod_page.get_by_role(
"button", name="Close Temporarily"
)
# Handle confirmation dialog
mod_page.on("dialog", lambda dialog: dialog.accept())
close_temp_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("Temporarily Closed", timeout=5000)
# Verify database state
from apps.parks.models import Park
operating_park.refresh_from_db()
assert operating_park.status == "CLOSED_TEMP"
def test_park_reopen_from_closed_temp(
self, mod_page: Page, operating_park, live_server
):
"""Test reopening a temporarily closed park."""
# First close the park temporarily
operating_park.status = "CLOSED_TEMP"
operating_park.save()
mod_page.goto(f"{live_server.url}/parks/{operating_park.slug}/")
mod_page.wait_for_load_state("networkidle")
# Verify initial status badge shows Temporarily Closed
status_badge = mod_page.locator('[data-status-badge]')
expect(status_badge).to_contain_text("Temporarily Closed")
# Find and click "Reopen" button
status_section = mod_page.locator('[data-park-status-actions]')
reopen_btn = status_section.get_by_role("button", name="Reopen")
mod_page.on("dialog", lambda dialog: dialog.accept())
reopen_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("Operating", timeout=5000)
# Verify database state
from apps.parks.models import Park
operating_park.refresh_from_db()
assert operating_park.status == "OPERATING"
def test_park_close_permanently_as_moderator(
self, mod_page: Page, operating_park, live_server
):
"""Test closing a park permanently as a moderator."""
mod_page.goto(f"{live_server.url}/parks/{operating_park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-park-status-actions]')
status_badge = mod_page.locator('[data-status-badge]')
# Find and click "Close Permanently" button
close_perm_btn = status_section.get_by_role(
"button", name="Close Permanently"
)
if not close_perm_btn.is_visible():
actions_dropdown = status_section.locator('[data-actions-dropdown]')
if actions_dropdown.is_visible():
actions_dropdown.click()
close_perm_btn = mod_page.get_by_role(
"button", name="Close Permanently"
)
mod_page.on("dialog", lambda dialog: dialog.accept())
close_perm_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("Permanently Closed", timeout=5000)
# Verify database state
from apps.parks.models import Park
operating_park.refresh_from_db()
assert operating_park.status == "CLOSED_PERM"
def test_park_demolish_from_closed_perm(
self, mod_page: Page, operating_park, live_server
):
"""Test transitioning a permanently closed park to demolished."""
# Set park to permanently closed
operating_park.status = "CLOSED_PERM"
operating_park.closing_date = date.today() - timedelta(days=365)
operating_park.save()
mod_page.goto(f"{live_server.url}/parks/{operating_park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-park-status-actions]')
status_badge = mod_page.locator('[data-status-badge]')
# Find and click "Mark as Demolished" button
demolish_btn = status_section.get_by_role("button", name="Mark as Demolished")
if not demolish_btn.is_visible():
actions_dropdown = status_section.locator('[data-actions-dropdown]')
if actions_dropdown.is_visible():
actions_dropdown.click()
demolish_btn = mod_page.get_by_role(
"button", name="Mark as Demolished"
)
mod_page.on("dialog", lambda dialog: dialog.accept())
demolish_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("Demolished", timeout=5000)
# Verify database state
from apps.parks.models import Park
operating_park.refresh_from_db()
assert operating_park.status == "DEMOLISHED"
def test_park_available_transitions_update(
self, mod_page: Page, operating_park, live_server
):
"""Test that available transitions update based on current state."""
mod_page.goto(f"{live_server.url}/parks/{operating_park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-park-status-actions]')
# Operating park should have Close Temporarily and Close Permanently
expect(
status_section.get_by_role("button", name="Close Temporarily")
).to_be_visible()
# Should NOT have Reopen (not applicable for Operating state)
expect(
status_section.get_by_role("button", name="Reopen")
).not_to_be_visible()
# Now close temporarily and verify buttons change
operating_park.status = "CLOSED_TEMP"
operating_park.save()
mod_page.reload()
mod_page.wait_for_load_state("networkidle")
# Now should have Reopen button
expect(
status_section.get_by_role("button", name="Reopen")
).to_be_visible()
class TestRideStatusTransitions:
"""Tests for Ride FSM status transitions via HTMX."""
def test_ride_close_temporarily_as_moderator(
self, mod_page: Page, operating_ride, live_server
):
"""Test closing a ride temporarily as a moderator."""
mod_page.goto(
f"{live_server.url}/parks/{operating_ride.park.slug}/rides/{operating_ride.slug}/"
)
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-ride-status-actions]')
status_badge = mod_page.locator('[data-status-badge]')
expect(status_badge).to_contain_text("Operating")
# Find and click "Close Temporarily" button
close_temp_btn = status_section.get_by_role(
"button", name="Close Temporarily"
)
mod_page.on("dialog", lambda dialog: dialog.accept())
close_temp_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("Temporarily Closed", timeout=5000)
# Verify database state
from apps.rides.models import Ride
operating_ride.refresh_from_db()
assert operating_ride.status == "CLOSED_TEMP"
def test_ride_mark_sbno_as_moderator(
self, mod_page: Page, operating_ride, live_server
):
"""Test marking a ride as Standing But Not Operating (SBNO)."""
mod_page.goto(
f"{live_server.url}/parks/{operating_ride.park.slug}/rides/{operating_ride.slug}/"
)
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-ride-status-actions]')
status_badge = mod_page.locator('[data-status-badge]')
# Find and click "Mark SBNO" button
sbno_btn = status_section.get_by_role("button", name="Mark SBNO")
if not sbno_btn.is_visible():
actions_dropdown = status_section.locator('[data-actions-dropdown]')
if actions_dropdown.is_visible():
actions_dropdown.click()
sbno_btn = mod_page.get_by_role("button", name="Mark SBNO")
mod_page.on("dialog", lambda dialog: dialog.accept())
sbno_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("SBNO", timeout=5000)
# Verify database state
from apps.rides.models import Ride
operating_ride.refresh_from_db()
assert operating_ride.status == "SBNO"
def test_ride_reopen_from_closed_temp(
self, mod_page: Page, operating_ride, live_server
):
"""Test reopening a temporarily closed ride."""
# First close the ride temporarily
operating_ride.status = "CLOSED_TEMP"
operating_ride.save()
mod_page.goto(
f"{live_server.url}/parks/{operating_ride.park.slug}/rides/{operating_ride.slug}/"
)
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-ride-status-actions]')
status_badge = mod_page.locator('[data-status-badge]')
# Find and click "Reopen" button
reopen_btn = status_section.get_by_role("button", name="Reopen")
mod_page.on("dialog", lambda dialog: dialog.accept())
reopen_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("Operating", timeout=5000)
# Verify database state
from apps.rides.models import Ride
operating_ride.refresh_from_db()
assert operating_ride.status == "OPERATING"
def test_ride_close_permanently_as_moderator(
self, mod_page: Page, operating_ride, live_server
):
"""Test closing a ride permanently as a moderator."""
mod_page.goto(
f"{live_server.url}/parks/{operating_ride.park.slug}/rides/{operating_ride.slug}/"
)
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-ride-status-actions]')
status_badge = mod_page.locator('[data-status-badge]')
# Find and click "Close Permanently" button
close_perm_btn = status_section.get_by_role(
"button", name="Close Permanently"
)
if not close_perm_btn.is_visible():
actions_dropdown = status_section.locator('[data-actions-dropdown]')
if actions_dropdown.is_visible():
actions_dropdown.click()
close_perm_btn = mod_page.get_by_role(
"button", name="Close Permanently"
)
mod_page.on("dialog", lambda dialog: dialog.accept())
close_perm_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("Permanently Closed", timeout=5000)
# Verify database state
from apps.rides.models import Ride
operating_ride.refresh_from_db()
assert operating_ride.status == "CLOSED_PERM"
def test_ride_demolish_from_closed_perm(
self, mod_page: Page, operating_ride, live_server
):
"""Test transitioning a permanently closed ride to demolished."""
# Set ride to permanently closed
operating_ride.status = "CLOSED_PERM"
operating_ride.closing_date = date.today() - timedelta(days=365)
operating_ride.save()
mod_page.goto(
f"{live_server.url}/parks/{operating_ride.park.slug}/rides/{operating_ride.slug}/"
)
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-ride-status-actions]')
status_badge = mod_page.locator('[data-status-badge]')
# Find and click "Mark as Demolished" button
demolish_btn = status_section.get_by_role(
"button", name="Mark as Demolished"
)
if not demolish_btn.is_visible():
actions_dropdown = status_section.locator('[data-actions-dropdown]')
if actions_dropdown.is_visible():
actions_dropdown.click()
demolish_btn = mod_page.get_by_role(
"button", name="Mark as Demolished"
)
mod_page.on("dialog", lambda dialog: dialog.accept())
demolish_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("Demolished", timeout=5000)
# Verify database state
from apps.rides.models import Ride
operating_ride.refresh_from_db()
assert operating_ride.status == "DEMOLISHED"
def test_ride_relocate_from_closed_perm(
self, mod_page: Page, operating_ride, live_server
):
"""Test transitioning a permanently closed ride to relocated."""
# Set ride to permanently closed
operating_ride.status = "CLOSED_PERM"
operating_ride.closing_date = date.today() - timedelta(days=365)
operating_ride.save()
mod_page.goto(
f"{live_server.url}/parks/{operating_ride.park.slug}/rides/{operating_ride.slug}/"
)
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-ride-status-actions]')
status_badge = mod_page.locator('[data-status-badge]')
# Find and click "Mark as Relocated" button
relocate_btn = status_section.get_by_role(
"button", name="Mark as Relocated"
)
if not relocate_btn.is_visible():
actions_dropdown = status_section.locator('[data-actions-dropdown]')
if actions_dropdown.is_visible():
actions_dropdown.click()
relocate_btn = mod_page.get_by_role(
"button", name="Mark as Relocated"
)
mod_page.on("dialog", lambda dialog: dialog.accept())
relocate_btn.click()
# Verify toast notification
toast = mod_page.locator('[data-toast]')
expect(toast).to_be_visible(timeout=5000)
# Verify status badge updated
expect(status_badge).to_contain_text("Relocated", timeout=5000)
# Verify database state
from apps.rides.models import Ride
operating_ride.refresh_from_db()
assert operating_ride.status == "RELOCATED"
class TestRideClosingWorkflow:
"""Tests for the special CLOSING status workflow with automatic transitions."""
def test_ride_set_closing_with_future_date(
self, mod_page: Page, operating_ride, live_server
):
"""Test setting a ride to CLOSING status with a future closing date."""
mod_page.goto(
f"{live_server.url}/parks/{operating_ride.park.slug}/rides/{operating_ride.slug}/"
)
mod_page.wait_for_load_state("networkidle")
status_section = mod_page.locator('[data-ride-status-actions]')
# Find and click "Set Closing" button
set_closing_btn = status_section.get_by_role(
"button", name="Set Closing"
)
if set_closing_btn.is_visible():
mod_page.on("dialog", lambda dialog: dialog.accept())
set_closing_btn.click()
# Verify status badge updated
status_badge = mod_page.locator('[data-status-badge]')
expect(status_badge).to_contain_text("Closing", timeout=5000)
# Verify database state
from apps.rides.models import Ride
operating_ride.refresh_from_db()
assert operating_ride.status == "CLOSING"
else:
pytest.skip("Set Closing button not available")
def test_ride_closing_shows_countdown(
self, mod_page: Page, operating_ride, live_server
):
"""Test that a ride in CLOSING status shows a countdown to closing date."""
# Set ride to CLOSING with future date
future_date = date.today() + timedelta(days=30)
operating_ride.status = "CLOSING"
operating_ride.closing_date = future_date
operating_ride.save()
mod_page.goto(
f"{live_server.url}/parks/{operating_ride.park.slug}/rides/{operating_ride.slug}/"
)
mod_page.wait_for_load_state("networkidle")
# Verify closing countdown is displayed
closing_info = mod_page.locator('[data-closing-countdown]')
if closing_info.is_visible():
expect(closing_info).to_contain_text("30")
else:
# May just show the status badge
status_badge = mod_page.locator('[data-status-badge]')
expect(status_badge).to_contain_text("Closing")
class TestStatusBadgeStyling:
"""Tests for correct status badge styling based on state."""
def test_operating_status_badge_style(
self, mod_page: Page, operating_park, live_server
):
"""Test that Operating status has correct green styling."""
mod_page.goto(f"{live_server.url}/parks/{operating_park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_badge = mod_page.locator('[data-status-badge]')
expect(status_badge).to_have_class(/bg-green|text-green|success/)
def test_closed_temp_status_badge_style(
self, mod_page: Page, operating_park, live_server
):
"""Test that Temporarily Closed status has correct yellow/warning styling."""
operating_park.status = "CLOSED_TEMP"
operating_park.save()
mod_page.goto(f"{live_server.url}/parks/{operating_park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_badge = mod_page.locator('[data-status-badge]')
expect(status_badge).to_have_class(/bg-yellow|text-yellow|warning/)
def test_closed_perm_status_badge_style(
self, mod_page: Page, operating_park, live_server
):
"""Test that Permanently Closed status has correct red/danger styling."""
operating_park.status = "CLOSED_PERM"
operating_park.save()
mod_page.goto(f"{live_server.url}/parks/{operating_park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_badge = mod_page.locator('[data-status-badge]')
expect(status_badge).to_have_class(/bg-red|text-red|danger/)
def test_demolished_status_badge_style(
self, mod_page: Page, operating_park, live_server
):
"""Test that Demolished status has correct gray styling."""
operating_park.status = "DEMOLISHED"
operating_park.save()
mod_page.goto(f"{live_server.url}/parks/{operating_park.slug}/")
mod_page.wait_for_load_state("networkidle")
status_badge = mod_page.locator('[data-status-badge]')
expect(status_badge).to_have_class(/bg-gray|text-gray|muted/)

View File

@@ -0,0 +1,6 @@
"""
Integration tests for ThrillWiki.
These tests verify that different components of the application work together correctly
without requiring a browser (unlike E2E tests).
"""

View File

@@ -0,0 +1,792 @@
"""
Integration Tests for FSMTransitionView
Tests the FSMTransitionView without a browser, using Django's test client.
These tests verify:
- FSMTransitionView handles HTMX requests correctly
- HX-Trigger headers contain proper toast data
- Correct partial templates rendered for each model
- Permission validation before transition execution
These are faster than E2E tests and don't require Playwright.
"""
import json
import pytest
from django.test import TestCase, Client
from django.urls import reverse
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
class TestFSMTransitionViewHTMX(TestCase):
"""Tests for FSMTransitionView with HTMX requests."""
@classmethod
def setUpTestData(cls):
"""Set up test data for all tests in this class."""
# Create regular user
cls.user = User.objects.create_user(
username="testuser",
email="testuser@example.com",
password="testpass123"
)
# Create moderator user
cls.moderator = User.objects.create_user(
username="moderator",
email="moderator@example.com",
password="modpass123",
is_staff=True
)
# Create admin user
cls.admin = User.objects.create_user(
username="admin",
email="admin@example.com",
password="adminpass123",
is_staff=True,
is_superuser=True
)
def setUp(self):
"""Set up test client for each test."""
self.client = Client()
def test_fsm_transition_view_with_htmx_header(self):
"""Test that FSMTransitionView handles HTMX requests correctly."""
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
self.client.login(username="moderator", password="modpass123")
# Create a pending submission
park = Park.objects.first()
if not park:
self.skipTest("No park available for testing")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=self.user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Test change"},
reason="Integration test",
status="PENDING"
)
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "moderation",
"model_name": "editsubmission",
"pk": submission.pk,
"transition_name": "transition_to_approved"
}
)
# Make request with HTMX header
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
# Should return 200 OK
self.assertEqual(response.status_code, 200)
# Should have HX-Trigger header
self.assertIn("HX-Trigger", response)
# Parse HX-Trigger header
trigger_data = json.loads(response["HX-Trigger"])
self.assertIn("showToast", trigger_data)
self.assertEqual(trigger_data["showToast"]["type"], "success")
# Cleanup
submission.delete()
def test_fsm_transition_view_without_htmx_header(self):
"""Test that FSMTransitionView handles non-HTMX requests correctly."""
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
self.client.login(username="moderator", password="modpass123")
park = Park.objects.first()
if not park:
self.skipTest("No park available for testing")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=self.user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Test change non-htmx"},
reason="Integration test non-htmx",
status="PENDING"
)
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "moderation",
"model_name": "editsubmission",
"pk": submission.pk,
"transition_name": "transition_to_approved"
}
)
# Make request WITHOUT HTMX header
response = self.client.post(url)
# Should return 200 OK with JSON response
self.assertEqual(response.status_code, 200)
self.assertEqual(response["Content-Type"], "application/json")
# Parse JSON response
data = response.json()
self.assertTrue(data["success"])
self.assertIn("message", data)
# Cleanup
submission.delete()
def test_fsm_transition_view_returns_correct_partial(self):
"""Test that FSMTransitionView returns correct partial template."""
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
self.client.login(username="moderator", password="modpass123")
park = Park.objects.first()
if not park:
self.skipTest("No park available for testing")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=self.user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Test partial"},
reason="Partial test",
status="PENDING"
)
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "moderation",
"model_name": "editsubmission",
"pk": submission.pk,
"transition_name": "transition_to_approved"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
# Response should contain HTML (partial template)
self.assertIn("text/html", response["Content-Type"])
# Cleanup
submission.delete()
def test_fsm_transition_view_adds_toast_trigger(self):
"""Test that FSMTransitionView adds correct HX-Trigger for toast."""
from apps.parks.models import Park
self.client.login(username="moderator", password="modpass123")
park = Park.objects.filter(status="OPERATING").first()
if not park:
self.skipTest("No operating park available for testing")
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "parks",
"model_name": "park",
"pk": park.pk,
"transition_name": "transition_to_closed_temp"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
# Parse HX-Trigger header
trigger_data = json.loads(response["HX-Trigger"])
# Verify toast structure
self.assertIn("showToast", trigger_data)
self.assertIn("message", trigger_data["showToast"])
self.assertIn("type", trigger_data["showToast"])
self.assertEqual(trigger_data["showToast"]["type"], "success")
# Reset park status for other tests
park.status = "OPERATING"
park.save()
def test_fsm_transition_view_handles_invalid_model(self):
"""Test that FSMTransitionView handles invalid model gracefully."""
self.client.login(username="moderator", password="modpass123")
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "nonexistent",
"model_name": "fakemodel",
"pk": 1,
"transition_name": "fake_transition"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
# Should return 404
self.assertEqual(response.status_code, 404)
# Should have error toast trigger
self.assertIn("HX-Trigger", response)
trigger_data = json.loads(response["HX-Trigger"])
self.assertEqual(trigger_data["showToast"]["type"], "error")
def test_fsm_transition_view_handles_invalid_transition(self):
"""Test that FSMTransitionView handles invalid transition name."""
from apps.parks.models import Park
self.client.login(username="moderator", password="modpass123")
park = Park.objects.first()
if not park:
self.skipTest("No park available for testing")
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "parks",
"model_name": "park",
"pk": park.pk,
"transition_name": "nonexistent_transition"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
# Should return 400 Bad Request
self.assertEqual(response.status_code, 400)
# Should have error toast trigger
self.assertIn("HX-Trigger", response)
trigger_data = json.loads(response["HX-Trigger"])
self.assertEqual(trigger_data["showToast"]["type"], "error")
def test_fsm_transition_view_validates_permissions(self):
"""Test that FSMTransitionView validates user permissions."""
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
# Login as regular user (not moderator)
self.client.login(username="testuser", password="testpass123")
park = Park.objects.first()
if not park:
self.skipTest("No park available for testing")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=self.user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Permission test"},
reason="Permission test",
status="PENDING"
)
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "moderation",
"model_name": "editsubmission",
"pk": submission.pk,
"transition_name": "transition_to_approved"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
# Should return 400 or 403 (permission denied)
self.assertIn(response.status_code, [400, 403])
# Verify submission was NOT changed
submission.refresh_from_db()
self.assertEqual(submission.status, "PENDING")
# Cleanup
submission.delete()
class TestFSMTransitionViewParkModel(TestCase):
"""Tests for FSMTransitionView with Park model."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username="mod_park",
email="mod_park@example.com",
password="modpass123",
is_staff=True
)
def setUp(self):
self.client = Client()
self.client.login(username="mod_park", password="modpass123")
def test_park_close_temporarily_transition(self):
"""Test park close temporarily transition via view."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
self.skipTest("No operating park available")
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "parks",
"model_name": "park",
"pk": park.pk,
"transition_name": "transition_to_closed_temp"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
self.assertEqual(response.status_code, 200)
# Verify park status changed
park.refresh_from_db()
self.assertEqual(park.status, "CLOSED_TEMP")
# Reset for other tests
park.status = "OPERATING"
park.save()
def test_park_reopen_transition(self):
"""Test park reopen transition via view."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
self.skipTest("No park available")
# First close the park
park.status = "CLOSED_TEMP"
park.save()
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "parks",
"model_name": "park",
"pk": park.pk,
"transition_name": "transition_to_operating"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
self.assertEqual(response.status_code, 200)
# Verify park status changed
park.refresh_from_db()
self.assertEqual(park.status, "OPERATING")
def test_park_slug_based_transition(self):
"""Test FSM transition using slug instead of pk."""
from apps.parks.models import Park
park = Park.objects.filter(status="OPERATING").first()
if not park:
self.skipTest("No operating park available")
url = reverse(
"core:fsm_transition_by_slug",
kwargs={
"app_label": "parks",
"model_name": "park",
"slug": park.slug,
"transition_name": "transition_to_closed_temp"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
self.assertEqual(response.status_code, 200)
# Verify park status changed
park.refresh_from_db()
self.assertEqual(park.status, "CLOSED_TEMP")
# Reset for other tests
park.status = "OPERATING"
park.save()
class TestFSMTransitionViewRideModel(TestCase):
"""Tests for FSMTransitionView with Ride model."""
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username="mod_ride",
email="mod_ride@example.com",
password="modpass123",
is_staff=True
)
def setUp(self):
self.client = Client()
self.client.login(username="mod_ride", password="modpass123")
def test_ride_close_temporarily_transition(self):
"""Test ride close temporarily transition via view."""
from apps.rides.models import Ride
ride = Ride.objects.filter(status="OPERATING").first()
if not ride:
self.skipTest("No operating ride available")
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "rides",
"model_name": "ride",
"pk": ride.pk,
"transition_name": "transition_to_closed_temp"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
self.assertEqual(response.status_code, 200)
# Verify ride status changed
ride.refresh_from_db()
self.assertEqual(ride.status, "CLOSED_TEMP")
# Reset for other tests
ride.status = "OPERATING"
ride.save()
def test_ride_mark_sbno_transition(self):
"""Test ride mark SBNO transition via view."""
from apps.rides.models import Ride
ride = Ride.objects.filter(status="OPERATING").first()
if not ride:
self.skipTest("No operating ride available")
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "rides",
"model_name": "ride",
"pk": ride.pk,
"transition_name": "transition_to_sbno"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
self.assertEqual(response.status_code, 200)
# Verify ride status changed
ride.refresh_from_db()
self.assertEqual(ride.status, "SBNO")
# Reset for other tests
ride.status = "OPERATING"
ride.save()
class TestFSMTransitionViewModerationModels(TestCase):
"""Tests for FSMTransitionView with moderation models."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username="submitter",
email="submitter@example.com",
password="testpass123"
)
cls.moderator = User.objects.create_user(
username="mod_moderation",
email="mod_moderation@example.com",
password="modpass123",
is_staff=True
)
def setUp(self):
self.client = Client()
self.client.login(username="mod_moderation", password="modpass123")
def test_edit_submission_approve_transition(self):
"""Test EditSubmission approve transition."""
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
park = Park.objects.first()
if not park:
self.skipTest("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=self.user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Approve test"},
reason="Approve test",
status="PENDING"
)
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "moderation",
"model_name": "editsubmission",
"pk": submission.pk,
"transition_name": "transition_to_approved"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
self.assertEqual(response.status_code, 200)
# Verify submission status changed
submission.refresh_from_db()
self.assertEqual(submission.status, "APPROVED")
# Cleanup
submission.delete()
def test_edit_submission_reject_transition(self):
"""Test EditSubmission reject transition."""
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
park = Park.objects.first()
if not park:
self.skipTest("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=self.user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Reject test"},
reason="Reject test",
status="PENDING"
)
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "moderation",
"model_name": "editsubmission",
"pk": submission.pk,
"transition_name": "transition_to_rejected"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
self.assertEqual(response.status_code, 200)
# Verify submission status changed
submission.refresh_from_db()
self.assertEqual(submission.status, "REJECTED")
# Cleanup
submission.delete()
def test_edit_submission_escalate_transition(self):
"""Test EditSubmission escalate transition."""
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
park = Park.objects.first()
if not park:
self.skipTest("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=self.user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "Escalate test"},
reason="Escalate test",
status="PENDING"
)
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "moderation",
"model_name": "editsubmission",
"pk": submission.pk,
"transition_name": "transition_to_escalated"
}
)
response = self.client.post(
url,
HTTP_HX_REQUEST="true"
)
self.assertEqual(response.status_code, 200)
# Verify submission status changed
submission.refresh_from_db()
self.assertEqual(submission.status, "ESCALATED")
# Cleanup
submission.delete()
class TestFSMTransitionViewStateLog(TestCase):
"""Tests that FSM transitions create proper StateLog entries."""
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username="submitter_log",
email="submitter_log@example.com",
password="testpass123"
)
cls.moderator = User.objects.create_user(
username="mod_log",
email="mod_log@example.com",
password="modpass123",
is_staff=True
)
def setUp(self):
self.client = Client()
self.client.login(username="mod_log", password="modpass123")
def test_transition_creates_state_log(self):
"""Test that FSM transition creates a StateLog entry."""
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
from django_fsm_log.models import StateLog
park = Park.objects.first()
if not park:
self.skipTest("No park available")
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=self.user,
content_type=content_type,
object_id=park.pk,
submission_type="EDIT",
changes={"description": "StateLog test"},
reason="StateLog test",
status="PENDING"
)
# Count existing StateLog entries
initial_log_count = StateLog.objects.filter(
content_type=ContentType.objects.get_for_model(EditSubmission),
object_id=submission.pk
).count()
url = reverse(
"core:fsm_transition",
kwargs={
"app_label": "moderation",
"model_name": "editsubmission",
"pk": submission.pk,
"transition_name": "transition_to_approved"
}
)
self.client.post(
url,
HTTP_HX_REQUEST="true"
)
# Check that a new StateLog entry was created
new_log_count = StateLog.objects.filter(
content_type=ContentType.objects.get_for_model(EditSubmission),
object_id=submission.pk
).count()
self.assertEqual(new_log_count, initial_log_count + 1)
# Verify the StateLog entry details
latest_log = StateLog.objects.filter(
content_type=ContentType.objects.get_for_model(EditSubmission),
object_id=submission.pk
).latest('timestamp')
self.assertEqual(latest_log.state, "APPROVED")
self.assertEqual(latest_log.by, self.moderator)
# Cleanup
submission.delete()

View File

@@ -0,0 +1,28 @@
"""
Test utilities for ThrillWiki.
This package provides reusable test utilities, helpers, and fixtures
for testing FSM transitions, HTMX interactions, and other common scenarios.
"""
from .fsm_test_helpers import (
create_test_submission,
create_test_park,
create_test_ride,
assert_status_changed,
assert_state_log_created,
assert_toast_triggered,
wait_for_htmx_swap,
verify_transition_buttons_visible,
)
__all__ = [
"create_test_submission",
"create_test_park",
"create_test_ride",
"assert_status_changed",
"assert_state_log_created",
"assert_toast_triggered",
"wait_for_htmx_swap",
"verify_transition_buttons_visible",
]

View File

@@ -0,0 +1,571 @@
"""
FSM Test Helpers
Reusable utility functions for testing FSM transitions:
- Factory functions for creating test objects in specific states
- Assertion helpers for common FSM test scenarios
- Playwright helpers for HTMX swap verification
- Toast notification verification utilities
"""
import json
from typing import Any, Dict, List, Optional, Type
from django.db.models import Model
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from django.http import HttpResponse
User = get_user_model()
# =============================================================================
# Factory Functions
# =============================================================================
def create_test_submission(
status: str = "PENDING",
user: Optional[User] = None,
park: Optional[Model] = None,
submission_type: str = "EDIT",
changes: Optional[Dict[str, Any]] = None,
reason: str = "Test submission",
**kwargs
) -> "EditSubmission":
"""
Create a test EditSubmission with the given status.
Args:
status: The initial status (PENDING, APPROVED, REJECTED, ESCALATED)
user: The submitting user (created if not provided)
park: The park to edit (first park used if not provided)
submission_type: EDIT or CREATE
changes: The changes JSON (default: {"description": "Test change"})
reason: The submission reason
**kwargs: Additional fields to set on the submission
Returns:
EditSubmission instance
"""
from apps.moderation.models import EditSubmission
from apps.parks.models import Park
# Get or create user
if user is None:
user, _ = User.objects.get_or_create(
username="test_submitter",
defaults={"email": "test_submitter@example.com"}
)
user.set_password("testpass123")
user.save()
# Get or create park
if park is None:
park = Park.objects.first()
if not park:
raise ValueError("No park available for testing")
# Default changes
if changes is None:
changes = {"description": "Test change"}
content_type = ContentType.objects.get_for_model(Park)
submission = EditSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
submission_type=submission_type,
changes=changes,
reason=reason,
status=status,
**kwargs
)
return submission
def create_test_park(
status: str = "OPERATING",
name: Optional[str] = None,
slug: Optional[str] = None,
**kwargs
) -> "Park":
"""
Create a test Park with the given status.
Args:
status: The initial status (OPERATING, CLOSED_TEMP, CLOSED_PERM, DEMOLISHED, RELOCATED)
name: Park name (auto-generated if not provided)
slug: Park slug (auto-generated if not provided)
**kwargs: Additional fields to set on the park
Returns:
Park instance
"""
from tests.factories import ParkFactory
if name is None:
import random
name = f"Test Park {random.randint(1000, 9999)}"
if slug is None:
from django.utils.text import slugify
slug = slugify(name)
park = ParkFactory(
name=name,
slug=slug,
status=status,
**kwargs
)
return park
def create_test_ride(
status: str = "OPERATING",
name: Optional[str] = None,
slug: Optional[str] = None,
park: Optional[Model] = None,
**kwargs
) -> "Ride":
"""
Create a test Ride with the given status.
Args:
status: The initial status (OPERATING, CLOSED_TEMP, SBNO, CLOSING, CLOSED_PERM, DEMOLISHED, RELOCATED)
name: Ride name (auto-generated if not provided)
slug: Ride slug (auto-generated if not provided)
park: The park this ride belongs to (created if not provided)
**kwargs: Additional fields to set on the ride
Returns:
Ride instance
"""
from tests.factories import RideFactory
if name is None:
import random
name = f"Test Ride {random.randint(1000, 9999)}"
if slug is None:
from django.utils.text import slugify
slug = slugify(name)
ride_kwargs = {
"name": name,
"slug": slug,
"status": status,
**kwargs
}
if park is not None:
ride_kwargs["park"] = park
ride = RideFactory(**ride_kwargs)
return ride
def create_test_photo_submission(
status: str = "PENDING",
user: Optional[User] = None,
park: Optional[Model] = None,
**kwargs
) -> "PhotoSubmission":
"""
Create a test PhotoSubmission with the given status.
Args:
status: The initial status (PENDING, APPROVED, REJECTED, ESCALATED)
user: The submitting user (created if not provided)
park: The park for the photo (first park used if not provided)
**kwargs: Additional fields to set on the submission
Returns:
PhotoSubmission instance
"""
from apps.moderation.models import PhotoSubmission
from apps.parks.models import Park
# Get or create user
if user is None:
user, _ = User.objects.get_or_create(
username="test_photo_submitter",
defaults={"email": "test_photo@example.com"}
)
user.set_password("testpass123")
user.save()
# Get or create park
if park is None:
park = Park.objects.first()
if not park:
raise ValueError("No park available for testing")
content_type = ContentType.objects.get_for_model(Park)
# Get a photo if available
try:
from django_cloudflareimages_toolkit.models import CloudflareImage
photo = CloudflareImage.objects.first()
if not photo:
raise ValueError("No CloudflareImage available for testing")
except ImportError:
raise ValueError("CloudflareImage not available")
submission = PhotoSubmission.objects.create(
user=user,
content_type=content_type,
object_id=park.pk,
photo=photo,
caption="Test photo submission",
status=status,
**kwargs
)
return submission
# =============================================================================
# Assertion Helpers
# =============================================================================
def assert_status_changed(obj: Model, expected_status: str) -> None:
"""
Assert that an object's status has changed to the expected value.
Args:
obj: The model instance to check
expected_status: The expected status value
Raises:
AssertionError: If status doesn't match expected
"""
obj.refresh_from_db()
actual_status = getattr(obj, "status", None)
assert actual_status == expected_status, (
f"Expected status '{expected_status}', got '{actual_status}'"
)
def assert_state_log_created(
obj: Model,
transition_name: str,
user: Optional[User] = None,
expected_state: Optional[str] = None
) -> None:
"""
Assert that a StateLog entry was created for a transition.
Args:
obj: The model instance that was transitioned
transition_name: The name of the transition (optional, for verification)
user: The user who performed the transition (optional)
expected_state: The expected final state in the log
Raises:
AssertionError: If no matching StateLog entry found
"""
from django_fsm_log.models import StateLog
content_type = ContentType.objects.get_for_model(obj)
logs = StateLog.objects.filter(
content_type=content_type,
object_id=obj.pk
).order_by('-timestamp')
assert logs.exists(), "No StateLog entries found for object"
latest_log = logs.first()
if expected_state is not None:
assert latest_log.state == expected_state, (
f"Expected state '{expected_state}' in log, got '{latest_log.state}'"
)
if user is not None:
assert latest_log.by == user, (
f"Expected log by user '{user}', got '{latest_log.by}'"
)
def assert_toast_triggered(
response: HttpResponse,
message: Optional[str] = None,
toast_type: str = "success"
) -> None:
"""
Assert that the response contains an HX-Trigger header with toast data.
Args:
response: The HTTP response to check
message: Expected message substring (optional)
toast_type: Expected toast type ('success', 'error', 'warning', 'info')
Raises:
AssertionError: If toast trigger not found or doesn't match
"""
assert "HX-Trigger" in response, "Response missing HX-Trigger header"
trigger_data = json.loads(response["HX-Trigger"])
assert "showToast" in trigger_data, "HX-Trigger missing showToast event"
toast_data = trigger_data["showToast"]
assert toast_data.get("type") == toast_type, (
f"Expected toast type '{toast_type}', got '{toast_data.get('type')}'"
)
if message is not None:
assert message in toast_data.get("message", ""), (
f"Expected '{message}' in toast message, got '{toast_data.get('message')}'"
)
def assert_no_status_change(obj: Model, original_status: str) -> None:
"""
Assert that an object's status has NOT changed from the original.
Args:
obj: The model instance to check
original_status: The original status value
Raises:
AssertionError: If status has changed
"""
obj.refresh_from_db()
actual_status = getattr(obj, "status", None)
assert actual_status == original_status, (
f"Status should not have changed from '{original_status}', but is now '{actual_status}'"
)
# =============================================================================
# Playwright Helpers
# =============================================================================
def wait_for_htmx_swap(
page,
target_selector: str,
timeout: int = 5000
) -> None:
"""
Wait for an HTMX swap to complete on a target element.
Args:
page: Playwright page object
target_selector: CSS selector for the target element
timeout: Maximum time to wait in milliseconds
"""
# Wait for the htmx:afterSwap event
page.wait_for_function(
f"""
() => {{
const el = document.querySelector('{target_selector}');
return el && !el.classList.contains('htmx-request');
}}
""",
timeout=timeout
)
def verify_transition_buttons_visible(
page,
transitions: List[str],
container_selector: str = "[data-status-actions]"
) -> Dict[str, bool]:
"""
Verify which transition buttons are visible on the page.
Args:
page: Playwright page object
transitions: List of transition names to check (e.g., ["Approve", "Reject"])
container_selector: CSS selector for the container holding the buttons
Returns:
Dict mapping transition name to visibility boolean
"""
container = page.locator(container_selector)
results = {}
for transition in transitions:
button = container.get_by_role("button", name=transition)
results[transition] = button.is_visible()
return results
def get_status_badge_text(page, badge_selector: str = "[data-status-badge]") -> str:
"""
Get the text content of the status badge.
Args:
page: Playwright page object
badge_selector: CSS selector for the status badge
Returns:
The text content of the status badge
"""
badge = page.locator(badge_selector)
return badge.text_content().strip() if badge.is_visible() else ""
def get_status_badge_class(page, badge_selector: str = "[data-status-badge]") -> str:
"""
Get the class attribute of the status badge.
Args:
page: Playwright page object
badge_selector: CSS selector for the status badge
Returns:
The class attribute of the status badge
"""
badge = page.locator(badge_selector)
return badge.get_attribute("class") or ""
def wait_for_toast(page, toast_selector: str = "[data-toast]", timeout: int = 5000):
"""
Wait for a toast notification to appear.
Args:
page: Playwright page object
toast_selector: CSS selector for the toast element
timeout: Maximum time to wait in milliseconds
Returns:
The toast element locator
"""
toast = page.locator(toast_selector)
toast.wait_for(state="visible", timeout=timeout)
return toast
def wait_for_toast_dismiss(
page,
toast_selector: str = "[data-toast]",
timeout: int = 10000
) -> None:
"""
Wait for a toast notification to be dismissed.
Args:
page: Playwright page object
toast_selector: CSS selector for the toast element
timeout: Maximum time to wait in milliseconds
"""
toast = page.locator(toast_selector)
toast.wait_for(state="hidden", timeout=timeout)
def click_and_confirm(page, button_locator, accept: bool = True) -> None:
"""
Click a button and handle the confirmation dialog.
Args:
page: Playwright page object
button_locator: The button locator to click
accept: Whether to accept (True) or dismiss (False) the dialog
"""
def handle_dialog(dialog):
if accept:
dialog.accept()
else:
dialog.dismiss()
page.on("dialog", handle_dialog)
button_locator.click()
# =============================================================================
# Test Client Helpers
# =============================================================================
def make_htmx_post(client, url: str, data: Optional[Dict] = None) -> HttpResponse:
"""
Make a POST request with HTMX headers.
Args:
client: Django test client
url: The URL to POST to
data: Optional POST data
Returns:
HttpResponse from the request
"""
return client.post(
url,
data=data or {},
HTTP_HX_REQUEST="true"
)
def make_htmx_get(client, url: str) -> HttpResponse:
"""
Make a GET request with HTMX headers.
Args:
client: Django test client
url: The URL to GET
Returns:
HttpResponse from the request
"""
return client.get(
url,
HTTP_HX_REQUEST="true"
)
def get_fsm_transition_url(
app_label: str,
model_name: str,
pk: int,
transition_name: str,
use_slug: bool = False,
slug: Optional[str] = None
) -> str:
"""
Generate the URL for an FSM transition.
Args:
app_label: The Django app label (e.g., 'moderation')
model_name: The model name in lowercase (e.g., 'editsubmission')
pk: The object's primary key
transition_name: The name of the transition method
use_slug: Whether to use slug-based URL
slug: The object's slug (required if use_slug is True)
Returns:
The transition URL string
"""
from django.urls import reverse
if use_slug:
if slug is None:
raise ValueError("slug is required when use_slug is True")
return reverse(
"core:fsm_transition_by_slug",
kwargs={
"app_label": app_label,
"model_name": model_name,
"slug": slug,
"transition_name": transition_name
}
)
else:
return reverse(
"core:fsm_transition",
kwargs={
"app_label": app_label,
"model_name": model_name,
"pk": pk,
"transition_name": transition_name
}
)