feat: E2E testing infrastructure with Playwright

commcare_connect/audit/data_access.py

@@ -335,6 +335,7 @@ def fetch_visits_for_ids(self, visit_ids: list[int], opportunity_id: int | None
         return self.pipeline.fetch_raw_visits(
             opportunity_id=opp_id,
             filter_visit_ids=set(visit_ids),
+            include_images=True,
         )
 
     # =========================================================================

commcare_connect/labs/analysis/backends/sql/backend.py

@@ -103,6 +103,7 @@ def fetch_raw_visits(
         skip_form_json: bool = False,
         filter_visit_ids: set[int] | None = None,
         tolerance_pct: int = 100,
+        include_images: bool = False,
     ) -> list[dict]:
         """
         Fetch raw visit data from SQL cache or API.
@@ -120,7 +121,7 @@ def fetch_raw_visits(
 
         # Cache miss or force refresh - fetch from API
         logger.info(f"[SQL] Raw cache MISS for opp {opportunity_id}, fetching from API")
-        csv_bytes = self._fetch_from_api(opportunity_id, access_token)
+        csv_bytes = self._fetch_from_api(opportunity_id, access_token, include_images=include_images)
 
         # Parse full data (always with form_json for storage)
         visit_dicts = parse_csv_bytes(csv_bytes, opportunity_id, skip_form_json=False)
@@ -131,8 +132,11 @@ def fetch_raw_visits(
         logger.info(f"[SQL] Stored {visit_count} visits to RawVisitCache")
 
         # Apply filters for return value
+        # Normalize to strings for comparison — visit_id is CharField in cache
+        # but parse_csv_bytes returns int IDs, and callers may pass either type.
         if filter_visit_ids:
-            visit_dicts = [v for v in visit_dicts if v.get("id") in filter_visit_ids]
+            str_filter = {str(vid) for vid in filter_visit_ids}
+            visit_dicts = [v for v in visit_dicts if str(v.get("id")) in str_filter]
 
         if skip_form_json:
             for v in visit_dicts:
@@ -309,9 +313,11 @@ def _load_from_cache(
         logger.info(f"[SQL] Loaded {len(visits)} visits from RawVisitCache")
         return visits
 
-    def _fetch_from_api(self, opportunity_id: int, access_token: str) -> bytes:
+    def _fetch_from_api(self, opportunity_id: int, access_token: str, include_images: bool = False) -> bytes:
         """Fetch raw CSV bytes from Connect API."""
         url = f"{settings.CONNECT_PRODUCTION_URL}/export/opportunity/{opportunity_id}/user_visits/"
+        if include_images:
+            url += "?images=true"
         headers = {
             "Authorization": f"Bearer {access_token}",
             "Accept-Encoding": "gzip, deflate",

commcare_connect/labs/analysis/pipeline.py

@@ -124,6 +124,7 @@ def fetch_raw_visits(
         skip_form_json: bool = False,
         filter_visit_ids: set[int] | None = None,
         force_refresh: bool = False,
+        include_images: bool = False,
     ) -> list[dict]:
         """
         Fetch raw visit data. Backend handles caching internally.
@@ -162,6 +163,7 @@ def fetch_raw_visits(
             force_refresh=force_refresh,
             skip_form_json=skip_form_json,
             filter_visit_ids=filter_visit_ids,
+            include_images=include_images,
         )
 
     def has_valid_raw_cache(self, opportunity_id: int | None = None) -> bool:

commcare_connect/labs/analysis/sse_streaming.py

@@ -74,7 +74,7 @@ def stream_data(self, request) -> Generator[str, None, None]:
     heartbeat_enabled = True
     heartbeat_interval = 20  # seconds between heartbeat comments
 
-    def get(self, request):
+    def get(self, request, **kwargs):
         """
         Handle GET request and return streaming response.
 

commcare_connect/labs/middleware.py

@@ -135,7 +135,7 @@ def __call__(self, request: HttpRequest) -> HttpResponse:
             return HttpResponseRedirect(prod_url)
 
         # Whitelisted path - require authentication (except login/oauth/logout and admin)
-        public_paths = ["/labs/login/", "/labs/initiate/", "/labs/callback/", "/labs/logout/"]
+        public_paths = ["/labs/login/", "/labs/initiate/", "/labs/callback/", "/labs/logout/", "/labs/test-auth/"]
 
         # Admin URLs don't require OAuth authentication (they use Django's standard auth)
         if not path.startswith("/admin/") and path not in public_paths:

commcare_connect/labs/urls.py

@@ -1,6 +1,6 @@
 from django.urls import include, path
 
-from commcare_connect.labs import views
+from commcare_connect.labs import views, views_test_auth
 from commcare_connect.labs.analysis import views as analysis_views
 from commcare_connect.labs.integrations.commcare import oauth_views as commcare_oauth_views
 from commcare_connect.labs.integrations.connect import oauth_views as connect_oauth_views
@@ -18,6 +18,8 @@
     path("callback/", connect_oauth_views.labs_oauth_callback, name="oauth_callback"),
     path("logout/", connect_oauth_views.labs_logout, name="logout"),
     path("dashboard/", connect_oauth_views.labs_dashboard, name="dashboard"),
+    # E2E test auth (DEBUG only)
+    path("test-auth/", views_test_auth.test_auth_view, name="test_auth"),
     # Labs Overview
     path("overview/", views.LabsOverviewView.as_view(), name="overview"),
     # Scout data agent

commcare_connect/labs/views_test_auth.py

@@ -0,0 +1,83 @@
+"""
+DEBUG-only view to inject a real OAuth session for Playwright E2E tests.
+
+Reads the CLI token from TokenManager, introspects it against production,
+fetches org data, and writes labs_oauth into the Django session — exactly
+like BaseLabsURLTest does for the Django test client.
+"""
+
+import logging
+from datetime import datetime
+
+from django.conf import settings
+from django.http import JsonResponse
+from django.utils import timezone
+from django.views.decorators.http import require_GET
+
+from commcare_connect.labs.integrations.connect.cli import TokenManager
+from commcare_connect.labs.integrations.connect.oauth import (
+    fetch_user_organization_data,
+    introspect_token,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@require_GET
+def test_auth_view(request):
+    """Inject a real OAuth session for E2E testing. DEBUG only."""
+    if not settings.DEBUG:
+        return JsonResponse({"error": "Only available in DEBUG mode"}, status=403)
+
+    token_manager = TokenManager()
+    token_data = token_manager.load_token()
+
+    if not token_data:
+        return JsonResponse({"error": "No CLI token found. Run: python manage.py get_cli_token"}, status=401)
+
+    if token_manager.is_expired():
+        return JsonResponse({"error": "CLI token expired. Run: python manage.py get_cli_token"}, status=401)
+
+    access_token = token_data["access_token"]
+
+    # Introspect token to get user profile
+    profile_data = introspect_token(
+        access_token=access_token,
+        client_id=settings.CONNECT_OAUTH_CLIENT_ID,
+        client_secret=settings.CONNECT_OAUTH_CLIENT_SECRET,
+        production_url=settings.CONNECT_PRODUCTION_URL,
+    )
+    if not profile_data:
+        return JsonResponse({"error": "Token introspection failed"}, status=401)
+
+    # Fetch org data
+    org_data = fetch_user_organization_data(access_token)
+    if not org_data:
+        return JsonResponse({"error": "Failed to fetch organization data"}, status=500)
+
+    # Convert expires_at from ISO string to timestamp
+    if "expires_at" in token_data and isinstance(token_data["expires_at"], str):
+        expires_at = datetime.fromisoformat(token_data["expires_at"]).timestamp()
+    else:
+        expires_in = token_data.get("expires_in", 1209600)
+        expires_at = (timezone.now() + timezone.timedelta(seconds=expires_in)).timestamp()
+
+    # Write session — same structure as the OAuth callback
+    request.session["labs_oauth"] = {
+        "access_token": access_token,
+        "refresh_token": token_data.get("refresh_token", ""),
+        "expires_at": expires_at,
+        "user_profile": {
+            "id": profile_data.get("id"),
+            "username": profile_data.get("username"),
+            "email": profile_data.get("email"),
+            "first_name": profile_data.get("first_name", ""),
+            "last_name": profile_data.get("last_name", ""),
+        },
+        "organization_data": org_data,
+    }
+
+    return JsonResponse({
+        "success": True,
+        "username": profile_data.get("username"),
+    })