feat: Implement initial schema and add various API, service, and management command enhancements across the application.

This commit is contained in:
pacnpal
2026-01-01 15:13:01 -05:00
parent c95f99ca10
commit b243b17af7
413 changed files with 11164 additions and 17433 deletions

View File

@@ -22,33 +22,24 @@ class ParkOpeningWorkflowTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='park_user',
email='park_user@example.com',
password='testpass123',
role='USER'
username="park_user", email="park_user@example.com", password="testpass123", role="USER"
)
cls.moderator = User.objects.create_user(
username='park_mod',
email='park_mod@example.com',
password='testpass123',
role='MODERATOR'
username="park_mod", email="park_mod@example.com", password="testpass123", role="MODERATOR"
)
def _create_park(self, status='OPERATING', **kwargs):
def _create_park(self, status="OPERATING", **kwargs):
"""Helper to create a park."""
from apps.parks.models import Company, Park
operator = Company.objects.create(
name=f'Operator {status}',
roles=['OPERATOR']
)
operator = Company.objects.create(name=f"Operator {status}", roles=["OPERATOR"])
defaults = {
'name': f'Test Park {status}',
'slug': f'test-park-{status.lower()}-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
"name": f"Test Park {status}",
"slug": f"test-park-{status.lower()}-{timezone.now().timestamp()}",
"operator": operator,
"status": status,
"timezone": "America/New_York",
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
@@ -59,16 +50,16 @@ class ParkOpeningWorkflowTests(TestCase):
Flow: UNDER_CONSTRUCTION → OPERATING
"""
park = self._create_park(status='UNDER_CONSTRUCTION')
park = self._create_park(status="UNDER_CONSTRUCTION")
self.assertEqual(park.status, 'UNDER_CONSTRUCTION')
self.assertEqual(park.status, "UNDER_CONSTRUCTION")
# Park opens
park.transition_to_operating(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'OPERATING')
self.assertEqual(park.status, "OPERATING")
class ParkTemporaryClosureWorkflowTests(TestCase):
@@ -77,26 +68,20 @@ class ParkTemporaryClosureWorkflowTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='temp_closure_user',
email='temp_closure@example.com',
password='testpass123',
role='USER'
username="temp_closure_user", email="temp_closure@example.com", password="testpass123", role="USER"
)
def _create_park(self, status='OPERATING', **kwargs):
def _create_park(self, status="OPERATING", **kwargs):
from apps.parks.models import Company, Park
operator = Company.objects.create(
name=f'Operator Temp {timezone.now().timestamp()}',
roles=['OPERATOR']
)
operator = Company.objects.create(name=f"Operator Temp {timezone.now().timestamp()}", roles=["OPERATOR"])
defaults = {
'name': f'Test Park Temp {timezone.now().timestamp()}',
'slug': f'test-park-temp-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
"name": f"Test Park Temp {timezone.now().timestamp()}",
"slug": f"test-park-temp-{timezone.now().timestamp()}",
"operator": operator,
"status": status,
"timezone": "America/New_York",
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
@@ -107,23 +92,23 @@ class ParkTemporaryClosureWorkflowTests(TestCase):
Flow: OPERATING → CLOSED_TEMP → OPERATING
"""
park = self._create_park(status='OPERATING')
park = self._create_park(status="OPERATING")
self.assertEqual(park.status, 'OPERATING')
self.assertEqual(park.status, "OPERATING")
# Close temporarily (e.g., off-season)
park.transition_to_closed_temp(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_TEMP')
self.assertEqual(park.status, "CLOSED_TEMP")
# Reopen
park.transition_to_operating(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'OPERATING')
self.assertEqual(park.status, "OPERATING")
class ParkPermanentClosureWorkflowTests(TestCase):
@@ -132,26 +117,20 @@ class ParkPermanentClosureWorkflowTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='perm_mod',
email='perm_mod@example.com',
password='testpass123',
role='MODERATOR'
username="perm_mod", email="perm_mod@example.com", password="testpass123", role="MODERATOR"
)
def _create_park(self, status='OPERATING', **kwargs):
def _create_park(self, status="OPERATING", **kwargs):
from apps.parks.models import Company, Park
operator = Company.objects.create(
name=f'Operator Perm {timezone.now().timestamp()}',
roles=['OPERATOR']
)
operator = Company.objects.create(name=f"Operator Perm {timezone.now().timestamp()}", roles=["OPERATOR"])
defaults = {
'name': f'Test Park Perm {timezone.now().timestamp()}',
'slug': f'test-park-perm-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
"name": f"Test Park Perm {timezone.now().timestamp()}",
"slug": f"test-park-perm-{timezone.now().timestamp()}",
"operator": operator,
"status": status,
"timezone": "America/New_York",
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
@@ -162,7 +141,7 @@ class ParkPermanentClosureWorkflowTests(TestCase):
Flow: OPERATING → CLOSED_PERM
"""
park = self._create_park(status='OPERATING')
park = self._create_park(status="OPERATING")
# Close permanently
park.transition_to_closed_perm(user=self.moderator)
@@ -170,7 +149,7 @@ class ParkPermanentClosureWorkflowTests(TestCase):
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
self.assertEqual(park.status, "CLOSED_PERM")
self.assertIsNotNone(park.closing_date)
def test_park_permanent_closure_from_temp(self):
@@ -179,7 +158,7 @@ class ParkPermanentClosureWorkflowTests(TestCase):
Flow: OPERATING → CLOSED_TEMP → CLOSED_PERM
"""
park = self._create_park(status='OPERATING')
park = self._create_park(status="OPERATING")
# Temporary closure
park.transition_to_closed_temp(user=self.moderator)
@@ -191,7 +170,7 @@ class ParkPermanentClosureWorkflowTests(TestCase):
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
self.assertEqual(park.status, "CLOSED_PERM")
class ParkDemolitionWorkflowTests(TestCase):
@@ -200,26 +179,20 @@ class ParkDemolitionWorkflowTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='demo_mod',
email='demo_mod@example.com',
password='testpass123',
role='MODERATOR'
username="demo_mod", email="demo_mod@example.com", password="testpass123", role="MODERATOR"
)
def _create_park(self, status='CLOSED_PERM', **kwargs):
def _create_park(self, status="CLOSED_PERM", **kwargs):
from apps.parks.models import Company, Park
operator = Company.objects.create(
name=f'Operator Demo {timezone.now().timestamp()}',
roles=['OPERATOR']
)
operator = Company.objects.create(name=f"Operator Demo {timezone.now().timestamp()}", roles=["OPERATOR"])
defaults = {
'name': f'Test Park Demo {timezone.now().timestamp()}',
'slug': f'test-park-demo-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
"name": f"Test Park Demo {timezone.now().timestamp()}",
"slug": f"test-park-demo-{timezone.now().timestamp()}",
"operator": operator,
"status": status,
"timezone": "America/New_York",
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
@@ -230,20 +203,20 @@ class ParkDemolitionWorkflowTests(TestCase):
Flow: OPERATING → CLOSED_PERM → DEMOLISHED
"""
park = self._create_park(status='CLOSED_PERM')
park = self._create_park(status="CLOSED_PERM")
# Demolish
park.transition_to_demolished(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'DEMOLISHED')
self.assertEqual(park.status, "DEMOLISHED")
def test_demolished_is_final_state(self):
"""Test that demolished parks cannot transition further."""
from django_fsm import TransitionNotAllowed
park = self._create_park(status='DEMOLISHED')
park = self._create_park(status="DEMOLISHED")
# Cannot transition from demolished
with self.assertRaises(TransitionNotAllowed):
@@ -256,26 +229,20 @@ class ParkRelocationWorkflowTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.moderator = User.objects.create_user(
username='reloc_mod',
email='reloc_mod@example.com',
password='testpass123',
role='MODERATOR'
username="reloc_mod", email="reloc_mod@example.com", password="testpass123", role="MODERATOR"
)
def _create_park(self, status='CLOSED_PERM', **kwargs):
def _create_park(self, status="CLOSED_PERM", **kwargs):
from apps.parks.models import Company, Park
operator = Company.objects.create(
name=f'Operator Reloc {timezone.now().timestamp()}',
roles=['OPERATOR']
)
operator = Company.objects.create(name=f"Operator Reloc {timezone.now().timestamp()}", roles=["OPERATOR"])
defaults = {
'name': f'Test Park Reloc {timezone.now().timestamp()}',
'slug': f'test-park-reloc-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
"name": f"Test Park Reloc {timezone.now().timestamp()}",
"slug": f"test-park-reloc-{timezone.now().timestamp()}",
"operator": operator,
"status": status,
"timezone": "America/New_York",
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
@@ -286,20 +253,20 @@ class ParkRelocationWorkflowTests(TestCase):
Flow: OPERATING → CLOSED_PERM → RELOCATED
"""
park = self._create_park(status='CLOSED_PERM')
park = self._create_park(status="CLOSED_PERM")
# Relocate
park.transition_to_relocated(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'RELOCATED')
self.assertEqual(park.status, "RELOCATED")
def test_relocated_is_final_state(self):
"""Test that relocated parks cannot transition further."""
from django_fsm import TransitionNotAllowed
park = self._create_park(status='RELOCATED')
park = self._create_park(status="RELOCATED")
# Cannot transition from relocated
with self.assertRaises(TransitionNotAllowed):
@@ -312,71 +279,62 @@ class ParkWrapperMethodTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='wrapper_user',
email='wrapper@example.com',
password='testpass123',
role='USER'
username="wrapper_user", email="wrapper@example.com", password="testpass123", role="USER"
)
cls.moderator = User.objects.create_user(
username='wrapper_mod',
email='wrapper_mod@example.com',
password='testpass123',
role='MODERATOR'
username="wrapper_mod", email="wrapper_mod@example.com", password="testpass123", role="MODERATOR"
)
def _create_park(self, status='OPERATING', **kwargs):
def _create_park(self, status="OPERATING", **kwargs):
from apps.parks.models import Company, Park
operator = Company.objects.create(
name=f'Operator Wrapper {timezone.now().timestamp()}',
roles=['OPERATOR']
)
operator = Company.objects.create(name=f"Operator Wrapper {timezone.now().timestamp()}", roles=["OPERATOR"])
defaults = {
'name': f'Test Park Wrapper {timezone.now().timestamp()}',
'slug': f'test-park-wrapper-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
"name": f"Test Park Wrapper {timezone.now().timestamp()}",
"slug": f"test-park-wrapper-{timezone.now().timestamp()}",
"operator": operator,
"status": status,
"timezone": "America/New_York",
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
def test_close_temporarily_wrapper(self):
"""Test close_temporarily wrapper method."""
park = self._create_park(status='OPERATING')
park = self._create_park(status="OPERATING")
# Use wrapper method if it exists
if hasattr(park, 'close_temporarily'):
if hasattr(park, "close_temporarily"):
park.close_temporarily(user=self.user)
else:
park.transition_to_closed_temp(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_TEMP')
self.assertEqual(park.status, "CLOSED_TEMP")
def test_reopen_wrapper(self):
"""Test reopen wrapper method."""
park = self._create_park(status='CLOSED_TEMP')
park = self._create_park(status="CLOSED_TEMP")
# Use wrapper method if it exists
if hasattr(park, 'reopen'):
if hasattr(park, "reopen"):
park.reopen(user=self.user)
else:
park.transition_to_operating(user=self.user)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'OPERATING')
self.assertEqual(park.status, "OPERATING")
def test_close_permanently_wrapper(self):
"""Test close_permanently wrapper method."""
park = self._create_park(status='OPERATING')
park = self._create_park(status="OPERATING")
closing_date = timezone.now().date()
# Use wrapper method if it exists
if hasattr(park, 'close_permanently'):
if hasattr(park, "close_permanently"):
park.close_permanently(closing_date=closing_date, user=self.moderator)
else:
park.transition_to_closed_perm(user=self.moderator)
@@ -384,35 +342,35 @@ class ParkWrapperMethodTests(TestCase):
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'CLOSED_PERM')
self.assertEqual(park.status, "CLOSED_PERM")
def test_demolish_wrapper(self):
"""Test demolish wrapper method."""
park = self._create_park(status='CLOSED_PERM')
park = self._create_park(status="CLOSED_PERM")
# Use wrapper method if it exists
if hasattr(park, 'demolish'):
if hasattr(park, "demolish"):
park.demolish(user=self.moderator)
else:
park.transition_to_demolished(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'DEMOLISHED')
self.assertEqual(park.status, "DEMOLISHED")
def test_relocate_wrapper(self):
"""Test relocate wrapper method."""
park = self._create_park(status='CLOSED_PERM')
park = self._create_park(status="CLOSED_PERM")
# Use wrapper method if it exists
if hasattr(park, 'relocate'):
if hasattr(park, "relocate"):
park.relocate(user=self.moderator)
else:
park.transition_to_relocated(user=self.moderator)
park.save()
park.refresh_from_db()
self.assertEqual(park.status, 'RELOCATED')
self.assertEqual(park.status, "RELOCATED")
class ParkStateLogTests(TestCase):
@@ -421,32 +379,23 @@ class ParkStateLogTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create_user(
username='log_user',
email='log_user@example.com',
password='testpass123',
role='USER'
username="log_user", email="log_user@example.com", password="testpass123", role="USER"
)
cls.moderator = User.objects.create_user(
username='log_mod',
email='log_mod@example.com',
password='testpass123',
role='MODERATOR'
username="log_mod", email="log_mod@example.com", password="testpass123", role="MODERATOR"
)
def _create_park(self, status='OPERATING', **kwargs):
def _create_park(self, status="OPERATING", **kwargs):
from apps.parks.models import Company, Park
operator = Company.objects.create(
name=f'Operator Log {timezone.now().timestamp()}',
roles=['OPERATOR']
)
operator = Company.objects.create(name=f"Operator Log {timezone.now().timestamp()}", roles=["OPERATOR"])
defaults = {
'name': f'Test Park Log {timezone.now().timestamp()}',
'slug': f'test-park-log-{timezone.now().timestamp()}',
'operator': operator,
'status': status,
'timezone': 'America/New_York'
"name": f"Test Park Log {timezone.now().timestamp()}",
"slug": f"test-park-log-{timezone.now().timestamp()}",
"operator": operator,
"status": status,
"timezone": "America/New_York",
}
defaults.update(kwargs)
return Park.objects.create(**defaults)
@@ -456,7 +405,7 @@ class ParkStateLogTests(TestCase):
from django.contrib.contenttypes.models import ContentType
from django_fsm_log.models import StateLog
park = self._create_park(status='OPERATING')
park = self._create_park(status="OPERATING")
park_ct = ContentType.objects.get_for_model(park)
# Perform transition
@@ -464,13 +413,10 @@ class ParkStateLogTests(TestCase):
park.save()
# Check log was created
log = StateLog.objects.filter(
content_type=park_ct,
object_id=park.id
).first()
log = StateLog.objects.filter(content_type=park_ct, object_id=park.id).first()
self.assertIsNotNone(log, "StateLog entry should be created")
self.assertEqual(log.state, 'CLOSED_TEMP')
self.assertEqual(log.state, "CLOSED_TEMP")
self.assertEqual(log.by, self.user)
def test_multiple_transitions_logged(self):
@@ -478,7 +424,7 @@ class ParkStateLogTests(TestCase):
from django.contrib.contenttypes.models import ContentType
from django_fsm_log.models import StateLog
park = self._create_park(status='OPERATING')
park = self._create_park(status="OPERATING")
park_ct = ContentType.objects.get_for_model(park)
# First transition: OPERATING -> CLOSED_TEMP
@@ -490,15 +436,12 @@ class ParkStateLogTests(TestCase):
park.save()
# Check multiple logs created
logs = StateLog.objects.filter(
content_type=park_ct,
object_id=park.id
).order_by('timestamp')
logs = StateLog.objects.filter(content_type=park_ct, object_id=park.id).order_by("timestamp")
self.assertEqual(logs.count(), 2, "Should have 2 log entries")
self.assertEqual(logs[0].state, 'CLOSED_TEMP')
self.assertEqual(logs[0].state, "CLOSED_TEMP")
self.assertEqual(logs[0].by, self.user)
self.assertEqual(logs[1].state, 'CLOSED_PERM')
self.assertEqual(logs[1].state, "CLOSED_PERM")
self.assertEqual(logs[1].by, self.moderator)
def test_full_lifecycle_logged(self):
@@ -506,7 +449,7 @@ class ParkStateLogTests(TestCase):
from django.contrib.contenttypes.models import ContentType
from django_fsm_log.models import StateLog
park = self._create_park(status='OPERATING')
park = self._create_park(status="OPERATING")
park_ct = ContentType.objects.get_for_model(park)
# Full lifecycle: OPERATING -> CLOSED_TEMP -> OPERATING -> CLOSED_PERM -> DEMOLISHED
@@ -523,11 +466,8 @@ class ParkStateLogTests(TestCase):
park.save()
# Check all logs created
logs = StateLog.objects.filter(
content_type=park_ct,
object_id=park.id
).order_by('timestamp')
logs = StateLog.objects.filter(content_type=park_ct, object_id=park.id).order_by("timestamp")
self.assertEqual(logs.count(), 4, "Should have 4 log entries")
states = [log.state for log in logs]
self.assertEqual(states, ['CLOSED_TEMP', 'OPERATING', 'CLOSED_PERM', 'DEMOLISHED'])
self.assertEqual(states, ["CLOSED_TEMP", "OPERATING", "CLOSED_PERM", "DEMOLISHED"])

View File

@@ -55,9 +55,7 @@ class ParkQueryOptimizationTests(TestCase):
# Should be a small number of queries (main query + prefetch)
# The exact count depends on prefetch_related configuration
self.assertLessEqual(
len(context.captured_queries),
5,
f"Expected <= 5 queries, got {len(context.captured_queries)}"
len(context.captured_queries), 5, f"Expected <= 5 queries, got {len(context.captured_queries)}"
)
def test_optimized_for_detail_query_count(self):
@@ -72,9 +70,7 @@ class ParkQueryOptimizationTests(TestCase):
# Should be a reasonable number of queries
self.assertLessEqual(
len(context.captured_queries),
10,
f"Expected <= 10 queries, got {len(context.captured_queries)}"
len(context.captured_queries), 10, f"Expected <= 10 queries, got {len(context.captured_queries)}"
)
def test_with_location_includes_location(self):
@@ -94,10 +90,10 @@ class ParkQueryOptimizationTests(TestCase):
if result.exists():
first = result.first()
# Should include these fields
self.assertIn('id', first)
self.assertIn('name', first)
self.assertIn('slug', first)
self.assertIn('status', first)
self.assertIn("id", first)
self.assertIn("name", first)
self.assertIn("slug", first)
self.assertIn("status", first)
def test_search_autocomplete_limits_results(self):
"""Verify search_autocomplete respects limit parameter."""
@@ -148,7 +144,7 @@ class CompanyQueryOptimizationTests(TestCase):
if result.exists():
first = result.first()
# Should have ride_count attribute
self.assertTrue(hasattr(first, 'ride_count'))
self.assertTrue(hasattr(first, "ride_count"))
def test_operators_with_park_count_includes_annotation(self):
"""Verify operators_with_park_count adds park count annotations."""
@@ -156,7 +152,7 @@ class CompanyQueryOptimizationTests(TestCase):
if result.exists():
first = result.first()
# Should have operated_parks_count attribute
self.assertTrue(hasattr(first, 'operated_parks_count'))
self.assertTrue(hasattr(first, "operated_parks_count"))
class ComputedFieldMaintenanceTests(TestCase):