feat(traffic): Celery lifecycle task — per-minute activation + expiry

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: hc-sousa

src/traffic/migrations/0003_periodic_task_run_lifecycle.py

@@ -0,0 +1,48 @@
+"""Register the per-minute traffic lifecycle task (activation + expiry)."""
+
+from django.db import migrations
+
+
+def _every_minute_schedule(apps):
+    CrontabSchedule = apps.get_model('django_celery_beat', 'CrontabSchedule')
+    schedule, _ = CrontabSchedule.objects.get_or_create(
+        minute='*',
+        hour='*',
+        day_of_week='*',
+        day_of_month='*',
+        month_of_year='*',
+        defaults={'timezone': 'Atlantic/Azores'},
+    )
+    if schedule.timezone != 'Atlantic/Azores':
+        schedule.timezone = 'Atlantic/Azores'
+        schedule.save(update_fields=['timezone'])
+    return schedule
+
+
+def register_lifecycle_task(apps, schema_editor):
+    PeriodicTask = apps.get_model('django_celery_beat', 'PeriodicTask')
+    every_minute = _every_minute_schedule(apps)
+    PeriodicTask.objects.update_or_create(
+        name='traffic.run_lifecycle',
+        defaults={
+            'task': 'traffic.run_lifecycle',
+            'crontab': every_minute,
+            'enabled': True,
+        },
+    )
+
+
+def unregister_lifecycle_task(apps, schema_editor):
+    PeriodicTask = apps.get_model('django_celery_beat', 'PeriodicTask')
+    PeriodicTask.objects.filter(name='traffic.run_lifecycle').delete()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('traffic', '0002_seed_default_categories'),
+        ('django_celery_beat', '__latest__'),
+    ]
+
+    operations = [
+        migrations.RunPython(register_lifecycle_task, unregister_lifecycle_task),
+    ]

src/traffic/tasks.py

@@ -0,0 +1,18 @@
+"""Traffic Celery tasks."""
+
+from __future__ import annotations
+
+import logging
+
+from celery import shared_task
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(name='traffic.run_lifecycle')
+def run_lifecycle_task() -> dict:
+    from traffic.services import run_lifecycle
+
+    counts = run_lifecycle()
+    logger.info('traffic.run_lifecycle counts=%s', counts)
+    return {'status': 'ok', **counts}

src/traffic/tests/test_tasks.py

@@ -0,0 +1,38 @@
+"""U4 task test: the Celery task drives the lifecycle transitions."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+import pytest
+from django.utils import timezone
+
+from tenancy.services import for_island, get_or_create_default_island
+from traffic.models import TrafficCategory, TrafficReport
+from traffic.tasks import run_lifecycle_task
+
+pytestmark = pytest.mark.django_db
+
+
+def test_run_lifecycle_task_transitions():
+    island = get_or_create_default_island()
+    now = timezone.now()
+    with for_island(island):
+        cat = TrafficCategory.objects.create(
+            island=island, name='Radar', slug='radar', is_schedulable=True,
+        )
+        due = TrafficReport.objects.create(
+            island=island, category=cat, latitude=37.78, longitude=-25.50,
+            status=TrafficReport.SCHEDULED, active_from=now - timedelta(minutes=1),
+            expires_at=now + timedelta(hours=1),
+        )
+        stale = TrafficReport.objects.create(
+            island=island, category=cat, latitude=37.78, longitude=-25.50,
+            status=TrafficReport.ACTIVE, expires_at=now - timedelta(minutes=1),
+        )
+
+    result = run_lifecycle_task()
+
+    assert result['status'] == 'ok'
+    assert TrafficReport.objects.unscoped().get(id=due.id).status == TrafficReport.ACTIVE
+    assert TrafficReport.objects.unscoped().get(id=stale.id).status == TrafficReport.EXPIRED