Feat: model and capture local EnrollmentEvent (#2367)

benefits/core/admin.py

@@ -9,6 +9,8 @@
 from django.conf import settings
 from django.contrib import admin
 from django.contrib.auth.models import Group
+from django.http import HttpRequest
+
 from . import models
 
 logger = logging.getLogger(__name__)
@@ -120,6 +122,44 @@ def get_readonly_fields(self, request, obj=None):
             return super().get_readonly_fields(request, obj)
 
 
+@admin.register(models.EnrollmentEvent)
+class EnrollmentEventAdmin(admin.ModelAdmin):  # pragma: no cover
+    list_display = ("enrollment_datetime", "transit_agency", "enrollment_flow", "enrollment_method", "verified_by")
+
+    def get_readonly_fields(self, request: HttpRequest, obj=None):
+        return ["id"]
+
+    def has_add_permission(self, request: HttpRequest, obj=None):
+        if settings.RUNTIME_ENVIRONMENT() == settings.RUNTIME_ENVS.PROD:
+            return False
+        elif request.user and (request.user.is_superuser or is_staff_member(request.user)):
+            return True
+        else:
+            return False
+
+    def has_change_permission(self, request: HttpRequest, obj=None):
+        if settings.RUNTIME_ENVIRONMENT() == settings.RUNTIME_ENVS.PROD:
+            return False
+        elif request.user and request.user.is_superuser:
+            return True
+        else:
+            return False
+
+    def has_delete_permission(self, request: HttpRequest, obj=None):
+        if settings.RUNTIME_ENVIRONMENT() == settings.RUNTIME_ENVS.PROD:
+            return False
+        elif request.user and (request.user.is_superuser or is_staff_member(request.user)):
+            return True
+        else:
+            return False
+
+    def has_view_permission(self, request: HttpRequest, obj=None):
+        if request.user and (request.user.is_superuser or is_staff_member(request.user)):
+            return True
+        else:
+            return False
+
+
 def pre_login_user(user, request):
     logger.debug(f"Running pre-login callback for user: {user.username}")
     add_google_sso_userinfo(user, request)
@@ -160,3 +200,8 @@ def add_transit_agency_staff_user_to_group(user, request):
         agency = models.TransitAgency.objects.filter(sso_domain=user_sso_domain).first()
         if agency is not None and agency.staff_group:
             agency.staff_group.user_set.add(user)
+
+
+def is_staff_member(user):
+    staff_group = Group.objects.get(name=settings.STAFF_GROUP_NAME)
+    return staff_group.user_set.contains(user)

benefits/core/migrations/0026_enrollmentevent.py

@@ -0,0 +1,28 @@
+# Generated by Django 5.1 on 2024-09-13 04:55
+
+import uuid
+from django.db import migrations, models
+import django.db.models.deletion
+import django.utils.timezone
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0025_transitprocessor_portal_url"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="EnrollmentEvent",
+            fields=[
+                ("id", models.UUIDField(default=uuid.uuid4, primary_key=True, serialize=False)),
+                ("enrollment_method", models.TextField(choices=[("digital", "digital"), ("in_person", "in_person")])),
+                ("verified_by", models.TextField()),
+                ("enrollment_datetime", models.DateTimeField(default=django.utils.timezone.now)),
+                ("expiration_datetime", models.DateTimeField(blank=True, null=True)),
+                ("enrollment_flow", models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to="core.enrollmentflow")),
+                ("transit_agency", models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to="core.transitagency")),
+            ],
+        )
+    ]

benefits/core/migrations/local_fixtures.json

@@ -202,5 +202,53 @@
       "groups": [2],
       "user_permissions": []
     }
+  },
+  {
+    "model": "core.enrollmentevent",
+    "pk": "49328a98-f829-4009-a16a-d71ece9e51e3",
+    "fields": {
+      "transit_agency": 1,
+      "enrollment_flow": 3,
+      "enrollment_method": "digital",
+      "verified_by": "http://server:8000/verify",
+      "enrollment_datetime": "2024-09-13T16:00:00.000Z",
+      "expiration_datetime": null
+    }
+  },
+  {
+    "model": "core.enrollmentevent",
+    "pk": "b1906c39-ae91-4b17-ae71-a93c4d4e546b",
+    "fields": {
+      "transit_agency": 1,
+      "enrollment_flow": 2,
+      "enrollment_method": "in_person",
+      "verified_by": "Test User",
+      "enrollment_datetime": "2024-09-13T15:45:30.000Z",
+      "expiration_datetime": null
+    }
+  },
+  {
+    "model": "core.enrollmentevent",
+    "pk": "b2038a25-ce2a-46f0-80a0-aea49e45e1e9",
+    "fields": {
+      "transit_agency": 1,
+      "enrollment_flow": 1,
+      "enrollment_method": "digital",
+      "verified_by": "benefits-oauth-client-name",
+      "enrollment_datetime": "2024-09-12T18:25:25.000Z",
+      "expiration_datetime": null
+    }
+  },
+  {
+    "model": "core.enrollmentevent",
+    "pk": "cf28906c-5709-4055-b6fd-554563ca1286",
+    "fields": {
+      "transit_agency": 1,
+      "enrollment_flow": 4,
+      "enrollment_method": "digital",
+      "verified_by": "benefits-oauth-client-name",
+      "enrollment_datetime": "2024-09-11T20:00:00.000Z",
+      "expiration_datetime": "2025-09-12T07:00:00.000Z"
+    }
   }
 ]

benefits/core/models.py

@@ -5,12 +5,14 @@
 from functools import cached_property
 import importlib
 import logging
+import uuid
 
 from django.conf import settings
 from django.core.exceptions import ValidationError
 from django.contrib.auth.models import Group, User
 from django.db import models
 from django.urls import reverse
+from django.utils import timezone
 
 import requests
 
@@ -422,3 +424,30 @@ def for_user(user: User):
 
         # the loop above returns the first match found. Return None if no match was found.
         return None
+
+
+class EnrollmentMethods:
+    DIGITAL = "digital"
+    IN_PERSON = "in_person"
+
+
+class EnrollmentEvent(models.Model):
+    """A record of a successful enrollment."""
+
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4)
+    transit_agency = models.ForeignKey(TransitAgency, on_delete=models.PROTECT)
+    enrollment_flow = models.ForeignKey(EnrollmentFlow, on_delete=models.PROTECT)
+    enrollment_method = models.TextField(
+        choices={
+            EnrollmentMethods.DIGITAL: EnrollmentMethods.DIGITAL,
+            EnrollmentMethods.IN_PERSON: EnrollmentMethods.IN_PERSON,
+        }
+    )
+    verified_by = models.TextField()
+    enrollment_datetime = models.DateTimeField(default=timezone.now)
+    expiration_datetime = models.DateTimeField(blank=True, null=True)
+
+    def __str__(self):
+        dt = timezone.localtime(self.enrollment_datetime)
+        ts = dt.strftime("%b %d, %Y, %I:%M %p")
+        return f"{ts}, {self.transit_agency}, {self.enrollment_flow}"

benefits/enrollment/views.py

@@ -15,6 +15,7 @@
 from benefits.core import session
 from benefits.core.middleware import EligibleSessionRequired, FlowSessionRequired, pageview_decorator
 
+from benefits.core import models
 from . import analytics, forms
 from .enrollment import Status, request_card_tokenization_access, enroll
 
@@ -67,6 +68,19 @@ def index(request):
 
         match (status):
             case Status.SUCCESS:
+                agency = session.agency(request)
+                flow = session.flow(request)
+                expiry = session.enrollment_expiry(request)
+                verified_by = flow.claims_provider.client_name if flow.uses_claims_verification else flow.eligibility_api_url
+                event = models.EnrollmentEvent.objects.create(
+                    transit_agency=agency,
+                    enrollment_flow=flow,
+                    enrollment_method=models.EnrollmentMethods.DIGITAL,
+                    verified_by=verified_by,
+                    expiration_datetime=expiry,
+                )
+                event.save()
+                analytics.returned_success(request, flow.group_id)
                 return success(request)
 
             case Status.SYSTEM_ERROR:
@@ -168,6 +182,5 @@ def success(request):
         # if they click the logout button, they are taken to the new route
         session.update(request, origin=reverse(routes.LOGGED_OUT))
 
-    analytics.returned_success(request, flow.group_id)
     context = {"redirect_to": request.path}
     return TemplateResponse(request, flow.enrollment_success_template, context)

benefits/in_person/views.py

@@ -9,8 +9,7 @@
 
 
 from benefits.routes import routes
-from benefits.core import session
-from benefits.core.models import EnrollmentFlow
+from benefits.core import models, session
 from benefits.enrollment.enrollment import Status, request_card_tokenization_access, enroll
 
 from benefits.in_person import forms
@@ -29,7 +28,7 @@ def eligibility(request):
 
         if form.is_valid():
             flow_id = form.cleaned_data.get("flow")
-            flow = EnrollmentFlow.objects.get(id=flow_id)
+            flow = models.EnrollmentFlow.objects.get(id=flow_id)
             session.update(request, flow=flow)
 
             in_person_enrollment = reverse(routes.IN_PERSON_ENROLLMENT)
@@ -79,6 +78,18 @@ def enrollment(request):
 
         match (status):
             case Status.SUCCESS:
+                agency = session.agency(request)
+                flow = session.flow(request)
+                expiry = session.enrollment_expiry(request)
+                verified_by = f"{request.user.first_name} {request.user.last_name}"
+                event = models.EnrollmentEvent.objects.create(
+                    transit_agency=agency,
+                    enrollment_flow=flow,
+                    enrollment_method=models.EnrollmentMethods.IN_PERSON,
+                    verified_by=verified_by,
+                    expiration_datetime=expiry,
+                )
+                event.save()
                 return redirect(routes.IN_PERSON_ENROLLMENT_SUCCESS)
 
             case Status.SYSTEM_ERROR:

benefits/settings.py

@@ -25,18 +25,25 @@ def _filter_empty(ls):
 ALLOWED_HOSTS = _filter_empty(os.environ.get("DJANGO_ALLOWED_HOSTS", "localhost").split(","))
 
 
+class RUNTIME_ENVS:
+    LOCAL = "local"
+    DEV = "dev"
+    TEST = "test"
+    PROD = "prod"
+
+
 def RUNTIME_ENVIRONMENT():
     """Helper calculates the current runtime environment from ALLOWED_HOSTS."""
 
     # usage of django.conf.settings.ALLOWED_HOSTS here (rather than the module variable directly)
     # is to ensure dynamic calculation, e.g. for unit tests and elsewhere this setting is needed
-    env = "local"
+    env = RUNTIME_ENVS.LOCAL
     if "dev-benefits.calitp.org" in settings.ALLOWED_HOSTS:
-        env = "dev"
+        env = RUNTIME_ENVS.DEV
     elif "test-benefits.calitp.org" in settings.ALLOWED_HOSTS:
-        env = "test"
+        env = RUNTIME_ENVS.TEST
     elif "benefits.calitp.org" in settings.ALLOWED_HOSTS:
-        env = "prod"
+        env = RUNTIME_ENVS.PROD
     return env
 
 

tests/pytest/conftest.py

@@ -1,4 +1,5 @@
 from unittest.mock import create_autospec
+from django.contrib.auth.models import User
 from django.contrib.sessions.middleware import SessionMiddleware
 from django.middleware.locale import LocaleMiddleware
 from django.utils import timezone
@@ -33,6 +34,11 @@ def app_request(rf):
     return app_request
 
 
+@pytest.fixture
+def model_User():
+    return User.objects.create(is_active=True, is_staff=True, first_name="Test", last_name="User")
+
+
 # autouse this fixture so we never call out to the real secret store
 @pytest.fixture(autouse=True)
 def mock_models_get_secret_by_name(mocker):

tests/pytest/core/test_models.py

@@ -1,10 +1,12 @@
+from datetime import timedelta
 from django.conf import settings
 from django.contrib.auth.models import Group, User
 from django.core.exceptions import ValidationError
+from django.utils import timezone
 
 import pytest
 
-from benefits.core.models import SecretNameField, EnrollmentFlow, TransitAgency
+from benefits.core.models import SecretNameField, EnrollmentFlow, TransitAgency, EnrollmentEvent, EnrollmentMethods
 import benefits.secrets
 
 
@@ -375,3 +377,53 @@ def test_TransitAgency_for_user_in_group_not_linked_to_any_agency():
     user.groups.add(group)
 
     assert TransitAgency.for_user(user) is None
+
+
+@pytest.mark.django_db
+def test_EnrollmentEvent_create(model_TransitAgency, model_EnrollmentFlow):
+    ts = timezone.now()
+    event = EnrollmentEvent.objects.create(
+        transit_agency=model_TransitAgency,
+        enrollment_flow=model_EnrollmentFlow,
+        enrollment_method=EnrollmentMethods.DIGITAL,
+        verified_by="Test",
+    )
+
+    assert event.transit_agency == model_TransitAgency
+    assert event.enrollment_flow == model_EnrollmentFlow
+    assert event.enrollment_method == EnrollmentMethods.DIGITAL
+    assert event.verified_by == "Test"
+    # default enrollment_datetime should be nearly the same as the timestamp
+    assert event.enrollment_datetime - ts <= timedelta(milliseconds=1)
+
+    dt = timezone.datetime(2024, 9, 13, 13, 30, 0, tzinfo=timezone.get_default_timezone())
+    expiry = dt + timedelta(weeks=52)
+    event = EnrollmentEvent(
+        transit_agency=model_TransitAgency,
+        enrollment_flow=model_EnrollmentFlow,
+        enrollment_method=EnrollmentMethods.DIGITAL,
+        verified_by="Test",
+        enrollment_datetime=dt,
+        expiration_datetime=expiry,
+    )
+    # enrollment_datetime should equal the given value exactly
+    assert event.enrollment_datetime == dt
+    assert event.expiration_datetime == expiry
+
+
+@pytest.mark.django_db
+def test_EnrollmentEvent_str(model_TransitAgency, model_EnrollmentFlow):
+    ts = timezone.datetime(2024, 9, 13, 13, 30, 0, tzinfo=timezone.get_default_timezone())
+
+    event = EnrollmentEvent.objects.create(
+        transit_agency=model_TransitAgency,
+        enrollment_flow=model_EnrollmentFlow,
+        enrollment_method=EnrollmentMethods.DIGITAL,
+        verified_by="Test",
+        enrollment_datetime=ts,
+    )
+    event_str = str(event)
+
+    assert "Sep 13, 2024, 01:30 PM" in event_str
+    assert str(event.transit_agency) in event_str
+    assert str(event.enrollment_flow) in event_str