Feat: store/retrieve enrollment expiry in session

benefits/core/context_processors.py

@@ -69,6 +69,21 @@ def debug(request):
     return {"debug": session.context_dict(request)}
 
 
+def enrollment(request):
+    """Context processor adds enrollment information to request context."""
+    eligibility = session.eligibility(request)
+    expiry = session.enrollment_expiry(request)
+    reenrollment = session.enrollment_reenrollment(request)
+
+    data = {
+        "expires": expiry,
+        "reenrollment": reenrollment,
+        "supports_expiration": eligibility.supports_expiration if eligibility else False,
+    }
+
+    return {"enrollment": data}
+
+
 def origin(request):
     """Context processor adds session.origin to request context."""
     origin = session.origin(request)

benefits/core/session.py

@@ -2,6 +2,7 @@
 The core application: helpers to work with request sessions.
 """
 
+from datetime import datetime, timedelta, timezone
 import hashlib
 import logging
 import time
@@ -20,7 +21,8 @@
 _DID = "did"
 _ELIGIBILITY = "eligibility"
 _ENROLLMENT_TOKEN = "enrollment_token"
-_ENROLLMENT_TOKEN_EXP = "enrollment_token_exp"
+_ENROLLMENT_TOKEN_EXP = "enrollment_token_expiry"
+_ENROLLMENT_EXP = "enrollment_expiry"
 _LANG = "lang"
 _OAUTH_CLAIM = "oauth_claim"
 _OAUTH_TOKEN = "oauth_token"
@@ -32,29 +34,26 @@
 
 def agency(request):
     """Get the agency from the request's session, or None"""
-    logger.debug("Get session agency")
     try:
         return models.TransitAgency.by_id(request.session[_AGENCY])
     except (KeyError, models.TransitAgency.DoesNotExist):
-        logger.debug("Can't get agency from session")
         return None
 
 
 def active_agency(request):
     """True if the request's session is configured with an active agency. False otherwise."""
-    logger.debug("Get session active agency flag")
     a = agency(request)
     return a and a.active
 
 
 def context_dict(request):
     """The request's session context as a dict."""
-    logger.debug("Get session context dict")
     return {
         _AGENCY: agency(request).slug if active_agency(request) else None,
         _DEBUG: debug(request),
         _DID: did(request),
         _ELIGIBILITY: eligibility(request),
+        _ENROLLMENT_EXP: enrollment_expiry(request),
         _ENROLLMENT_TOKEN: enrollment_token(request),
         _ENROLLMENT_TOKEN_EXP: enrollment_token_expiry(request),
         _LANG: language(request),
@@ -69,7 +68,6 @@ def context_dict(request):
 
 def debug(request):
     """Get the DEBUG flag from the request's session."""
-    logger.debug("Get session debug flag")
     return bool(request.session.get(_DEBUG, False))
 
 
@@ -84,7 +82,6 @@ def did(request):
 
     See more: https://help.amplitude.com/hc/en-us/articles/115003135607-Track-unique-users-in-Amplitude
     """
-    logger.debug("Get session did")
     d = request.session.get(_DID)
     if not d:
         reset(request)
@@ -94,7 +91,6 @@ def did(request):
 
 def eligibility(request):
     """Get the confirmed models.EligibilityType from the request's session, or None"""
-    logger.debug("Get session confirmed eligibility")
     eligibility = request.session.get(_ELIGIBILITY)
     if eligibility:
         return models.EligibilityType.get(eligibility)
@@ -104,41 +100,52 @@ def eligibility(request):
 
 def eligible(request):
     """True if the request's session is configured with an active agency and has confirmed eligibility. False otherwise."""
-    logger.debug("Get session eligible flag")
     return active_agency(request) and agency(request).supports_type(eligibility(request))
 
 
+def enrollment_expiry(request):
+    """Get the expiry date for a user's enrollment from session, or None."""
+    expiry = request.session.get(_ENROLLMENT_EXP)
+    if expiry:
+        return datetime.fromtimestamp(expiry, tz=timezone.utc)
+    else:
+        return None
+
+
+def enrollment_reenrollment(request):
+    """Get the reenrollment date for a user's enrollment from session, or None."""
+    expiry = enrollment_expiry(request)
+    elig = eligibility(request)
+
+    if elig and elig.supports_expiration and expiry:
+        return expiry - timedelta(days=elig.expiration_reenrollment_days)
+    else:
+        return None
+
+
 def enrollment_token(request):
     """Get the enrollment token from the request's session, or None."""
-    logger.debug("Get session enrollment token")
     return request.session.get(_ENROLLMENT_TOKEN)
 
 
 def enrollment_token_expiry(request):
     """Get the enrollment token's expiry time from the request's session, or None."""
-    logger.debug("Get session enrollment token expiry")
     return request.session.get(_ENROLLMENT_TOKEN_EXP)
 
 
 def enrollment_token_valid(request):
     """True if the request's session is configured with a valid token. False otherwise."""
     if bool(enrollment_token(request)):
-        logger.debug("Session contains an enrollment token")
         exp = enrollment_token_expiry(request)
-
         # ensure token does not expire in the next 5 seconds
         valid = exp is None or exp > (time.time() + 5)
-
-        logger.debug(f"Session enrollment token is {'valid' if valid else 'expired'}")
         return valid
     else:
-        logger.debug("Session does not contain a valid enrollment token")
         return False
 
 
 def language(request):
     """Get the language configured for the request."""
-    logger.debug("Get session language")
     return request.LANGUAGE_CODE
 
 
@@ -154,19 +161,16 @@ def logout(request):
 
 def oauth_token(request):
     """Get the oauth token from the request's session, or None"""
-    logger.debug("Get session oauth token")
     return request.session.get(_OAUTH_TOKEN)
 
 
 def oauth_claim(request):
     """Get the oauth claim from the request's session, or None"""
-    logger.debug("Get session oauth claim")
     return request.session.get(_OAUTH_CLAIM)
 
 
 def origin(request):
     """Get the origin for the request's session, or the default core:index."""
-    logger.debug("Get session origin")
     return request.session.get(_ORIGIN, reverse("core:index"))
 
 
@@ -201,7 +205,6 @@ def start(request):
 
     See more: https://help.amplitude.com/hc/en-us/articles/115002323627-Tracking-Sessions
     """
-    logger.debug("Get session time")
     s = request.session.get(_START)
     if not s:
         reset(request)
@@ -223,7 +226,6 @@ def uid(request):
     here a value is set on anonymous users anyway, as the users never sign-in
     and become de-anonymized to this app / Amplitude.
     """
-    logger.debug("Get session uid")
     u = request.session.get(_UID)
     if not u:
         reset(request)
@@ -236,6 +238,7 @@ def update(
     agency=None,
     debug=None,
     eligibility_types=None,
+    enrollment_expiry=None,
     enrollment_token=None,
     enrollment_token_exp=None,
     oauth_token=None,
@@ -245,13 +248,10 @@ def update(
 ):
     """Update the request's session with non-null values."""
     if agency is not None and isinstance(agency, models.TransitAgency):
-        logger.debug(f"Update session {_AGENCY}")
         request.session[_AGENCY] = agency.id
     if debug is not None:
-        logger.debug(f"Update session {_DEBUG}")
         request.session[_DEBUG] = debug
     if eligibility_types is not None and isinstance(eligibility_types, list):
-        logger.debug(f"Update session {_ELIGIBILITY}")
         if len(eligibility_types) > 1:
             raise NotImplementedError("Multiple eligibilities are not supported at this time.")
         elif len(eligibility_types) == 1:
@@ -262,29 +262,31 @@ def update(
         else:
             # empty list, clear session eligibility
             request.session[_ELIGIBILITY] = None
+    if isinstance(enrollment_expiry, datetime):
+        if enrollment_expiry.tzinfo is None or enrollment_expiry.tzinfo.utcoffset(enrollment_expiry) is None:
+            # this is a naive datetime instance, update tzinfo for UTC
+            # see notes under https://docs.python.org/3/library/datetime.html#datetime.datetime.timestamp
+            # > There is no method to obtain the POSIX timestamp directly from a naive datetime instance representing UTC time.
+            # > If your application uses this convention and your system timezone is not set to UTC, you can obtain the POSIX
+            # > timestamp by supplying tzinfo=timezone.utc
+            enrollment_expiry = enrollment_expiry.replace(tzinfo=timezone.utc)
+        request.session[_ENROLLMENT_EXP] = enrollment_expiry.timestamp()
     if enrollment_token is not None:
-        logger.debug(f"Update session {_ENROLLMENT_TOKEN}")
         request.session[_ENROLLMENT_TOKEN] = enrollment_token
         request.session[_ENROLLMENT_TOKEN_EXP] = enrollment_token_exp
     if oauth_token is not None:
-        logger.debug(f"Update session {_OAUTH_TOKEN}")
         request.session[_OAUTH_TOKEN] = oauth_token
     if oauth_claim is not None:
-        logger.debug(f"Update session {_OAUTH_CLAIM}")
         request.session[_OAUTH_CLAIM] = oauth_claim
     if origin is not None:
-        logger.debug(f"Update session {_ORIGIN}")
         request.session[_ORIGIN] = origin
     if verifier is not None and isinstance(verifier, models.EligibilityVerifier):
-        logger.debug(f"Update session {_VERIFIER}")
         request.session[_VERIFIER] = verifier.id
 
 
 def verifier(request):
     """Get the verifier from the request's session, or None"""
-    logger.debug("Get session verifier")
     try:
         return models.EligibilityVerifier.by_id(request.session[_VERIFIER])
     except (KeyError, models.EligibilityVerifier.DoesNotExist):
-        logger.debug("Can't get verifier from session")
         return None

tests/pytest/core/test_context_processors.py

@@ -1,4 +1,8 @@
-from benefits.core.context_processors import unique_values
+from datetime import datetime, timedelta, timezone
+import pytest
+
+from benefits.core import session
+from benefits.core.context_processors import unique_values, enrollment
 
 
 def test_unique_values():
@@ -7,3 +11,31 @@ def test_unique_values():
     new_list = unique_values(original_list)
 
     assert new_list == ["a", "b", "c", "zzz", "d"]
+
+
+@pytest.mark.django_db
+def test_enrollment_default(app_request):
+    context = enrollment(app_request)
+
+    assert "enrollment" in context
+    assert context["enrollment"] == {"expires": None, "reenrollment": None, "supports_expiration": False}
+
+
+@pytest.mark.django_db
+def test_enrollment_expiration(app_request, model_EligibilityType_supports_expiration, model_TransitAgency):
+    model_TransitAgency.eligibility_types.add(model_EligibilityType_supports_expiration)
+    model_TransitAgency.save()
+
+    expiry = datetime.now(tz=timezone.utc)
+    reenrollment = expiry - timedelta(days=model_EligibilityType_supports_expiration.expiration_reenrollment_days)
+
+    session.update(
+        app_request,
+        agency=model_TransitAgency,
+        eligibility_types=[model_EligibilityType_supports_expiration.name],
+        enrollment_expiry=expiry,
+    )
+
+    context = enrollment(app_request)
+
+    assert context["enrollment"] == {"expires": expiry, "reenrollment": reenrollment, "supports_expiration": True}

tests/pytest/core/test_session.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timedelta, timezone
 import time
 
 from django.contrib.sessions.middleware import SessionMiddleware
@@ -74,6 +75,66 @@ def test_eligibile_True(model_TransitAgency, app_request):
     assert session.eligible(app_request)
 
 
+@pytest.mark.django_db
+def test_enrollment_expiry_default(app_request):
+    assert session.enrollment_expiry(app_request) is None
+
+
+@pytest.mark.django_db
+def test_enrollment_expiry_not_datetime(app_request):
+    session.update(app_request, enrollment_expiry="2024-03-25T00:00:00Z")
+
+    assert session.enrollment_expiry(app_request) is None
+
+
+@pytest.mark.django_db
+def test_enrollment_expiry_datetime_timezone_utc(app_request):
+    expiry = datetime.now(tz=timezone.utc)
+
+    session.update(app_request, enrollment_expiry=expiry)
+
+    assert session.enrollment_expiry(app_request) == expiry
+
+
+@pytest.mark.django_db
+def test_enrollment_expiry_datetime_timezone_naive(app_request):
+    expiry = datetime.now()
+    assert expiry.tzinfo is None
+
+    session.update(app_request, enrollment_expiry=expiry)
+    session_expiry = session.enrollment_expiry(app_request)
+
+    assert all(
+        [
+            session_expiry.year == expiry.year,
+            session_expiry.month == expiry.month,
+            session_expiry.day == expiry.day,
+            session_expiry.hour == expiry.hour,
+            session_expiry.minute == expiry.minute,
+            session_expiry.second == expiry.second,
+            session_expiry.tzinfo == timezone.utc,
+        ]
+    )
+
+
+@pytest.mark.django_db
+def test_enrollment_reenrollment(app_request, model_EligibilityType_supports_expiration, model_TransitAgency):
+    model_TransitAgency.eligibility_types.add(model_EligibilityType_supports_expiration)
+    model_TransitAgency.save()
+
+    expiry = datetime.now(tz=timezone.utc)
+    expected_reenrollment = expiry - timedelta(days=model_EligibilityType_supports_expiration.expiration_reenrollment_days)
+
+    session.update(
+        app_request,
+        agency=model_TransitAgency,
+        eligibility_types=[model_EligibilityType_supports_expiration.name],
+        enrollment_expiry=expiry,
+    )
+
+    assert session.enrollment_reenrollment(app_request) == expected_reenrollment
+
+
 @pytest.mark.django_db
 def test_enrollment_token_default(app_request):
     assert session.enrollment_token(app_request) is None