Merge pull request #1396 from Open-Source-Legal/claude/resolve-issue-1333-QwoCf

Type annotate users/tasks.py and graduate from mypy baseline (#1333)

Author: JSv4

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **Mypy: graduated `opencontractserver/users/tasks.py` out of the baseline** (Issue #1333 follow-up): `tasks.py` was the last `opencontractserver.users` module still suppressed in `mypy.ini`. PR #1370 left it untyped because the file is only loaded when `settings.USE_AUTH0=True`, so it never failed at runtime under the test settings; the typing gap kept the package short of the issue's "all four packages at ≥80% return-annotation coverage" Done-When criterion. Added return + parameter annotations to all five Auth0 sync tasks (`get_new_auth0_token`, `apply_data_to_user`, `sync_remote_user`, `ensure_valid_auth0_token`, `get_user_details_async`), introduced a module-level docstring documenting the `USE_AUTH0` gating, and removed the `[mypy-opencontractserver.users.tasks] ignore_errors = True` section. Local `data` rebound from request body (`dict[str, str]`) to response payload (`dict[str, Any]`) was split into two distinctly-named variables (`request_data` / `payload`) so the types are unambiguous; behavior is unchanged. No callers needed updating — `config/graphql_auth0_auth/utils.py` still consumes `sync_remote_user.delay(...)` exactly as before.
+
 ### Fixed
 
 - **`test_superuser_sees_all_queryset` miscounts personal corpuses by 1** (Issue #1394, `opencontractserver/tests/test_visibility_managers.py`, `opencontractserver/tests/test_resolvers.py`): Two `VisibleToUserTests.test_superuser_sees_all_queryset` cases asserted that `Corpus.objects.visible_to_user(superuser).count() == 4` (public + private + 2 personal), but the actual count is 5 because the test DB starts with a pre-existing personal corpus owned by django-guardian's `AnonymousUser` (created during fixture setup before/around the username-based skip in `opencontractserver/users/signals.py::user_created_signal`). The assertion is now scoped to corpuses created by the test's two users (`creator__in=[self.user, self.superuser]`), making it resilient to any fixture-level corpuses that exist at test DB init time. Production code is unchanged.

mypy.ini

@@ -1095,10 +1095,7 @@ ignore_errors = True
 [mypy-opencontractserver.tests.websocket.test_unified_agent_consumer]
 ignore_errors = True
 
-# --- opencontractserver.users (1 file) ---
-
-[mypy-opencontractserver.users.tasks]
-ignore_errors = True
+# --- opencontractserver.users ---
 
 # ``opencontractserver.users.models.User`` defines reverse relations to
 # ``opencontractserver.corpuses.models.Corpus``, ``.CorpusFolder``, and

opencontractserver/users/tasks.py

@@ -1,9 +1,18 @@
+"""Celery tasks for synchronising user data with Auth0.
+
+These tasks are defined only when ``settings.USE_AUTH0`` is enabled — they
+fetch a Machine-to-Machine token, look up the remote profile via the Auth0
+Management API, and copy the result onto the local :class:`User` row.
+"""
+
 import datetime
 import json
 import logging
+from typing import Any, Optional
 
 import requests
 from celery import chain
+from celery.result import AsyncResult
 from django.conf import settings
 from django.contrib.auth import get_user_model
 
@@ -21,34 +30,34 @@
 if settings.USE_AUTH0:
 
     @celery_app.task()
-    def get_new_auth0_token():
+    def get_new_auth0_token() -> Optional[str]:
 
         # print("get_new_auth0_token")
         url = f"https://{auth0_settings.AUTH0_DOMAIN}/oauth/token"
         # print(url)
 
-        headers = {"content-type": "application/json"}
+        headers: dict[str, str] = {"content-type": "application/json"}
         # print(headers)
 
-        data = {
+        request_data: dict[str, str] = {
             "grant_type": auth0_settings.AUTH0_M2M_MANAGEMENT_GRANT_TYPE,
             "client_id": auth0_settings.AUTH0_M2M_MANAGEMENT_API_ID,
             "client_secret": auth0_settings.AUTH0_M2M_MANAGEMENT_API_SECRET,
             "audience": f"https://{auth0_settings.AUTH0_DOMAIN}/api/v2/",
         }
-        # print(f"Machine-2-Machine Request data: {data}")
+        # print(f"Machine-2-Machine Request data: {request_data}")
 
-        response = requests.post(url, headers=headers, json=data)
+        response = requests.post(url, headers=headers, json=request_data)
 
         # print("Auth0 Response:")
         # print(response.status_code)
         # print(response.text)
 
         if response.status_code == 200:
-            data = json.loads(response.text)
-            # print(data)
-            access_token = data["access_token"]
-            expires_in = data["expires_in"]
+            payload: dict[str, Any] = json.loads(response.text)
+            # print(payload)
+            access_token: str = payload["access_token"]
+            expires_in: int = payload["expires_in"]
 
             newToken = Auth0APIToken()
             newToken.token = access_token
@@ -60,11 +69,11 @@ def get_new_auth0_token():
 
             return newToken.token
 
-        else:
-            logger.error("Error retrieving access token to Auth0.")
+        logger.error("Error retrieving access token to Auth0.")
+        return None
 
     @celery_app.task()
-    def apply_data_to_user(data, userPk):
+    def apply_data_to_user(data: dict[str, Any], userPk: str) -> None:
 
         # print(f"apply_data_to_user() - userPk is: {userPk}\nData: {data}")
 
@@ -103,7 +112,7 @@ def apply_data_to_user(data, userPk):
                 )
 
     @celery_app.task()
-    def sync_remote_user(user_pk):
+    def sync_remote_user(user_pk: str) -> AsyncResult:
 
         refresh = False
         tokens = Auth0APIToken.objects.all()
@@ -133,7 +142,7 @@ def sync_remote_user(user_pk):
         return data.apply_async()
 
     @celery_app.task()
-    def ensure_valid_auth0_token():
+    def ensure_valid_auth0_token() -> Optional[str]:
 
         tokens = Auth0APIToken.objects.all()
 
@@ -147,6 +156,7 @@ def ensure_valid_auth0_token():
             for tok in tokens:
                 tok.delete()
                 return get_new_auth0_token.delay().get()
+            return None
         else:
             if tokens[0].expiration_Date < datetime.datetime.now(datetime.timezone.utc):
                 # print("Token has expired. Refetching from Auth0")
@@ -157,8 +167,8 @@ def ensure_valid_auth0_token():
                 return tokens[0].token
 
     @celery_app.task
-    def get_user_details_async(token, auth0_Id):
-        headers = {
+    def get_user_details_async(token: str, auth0_Id: str) -> dict[str, Any]:
+        headers: dict[str, str] = {
             "Authorization": f"Bearer {token}",
         }
         url = f"https://{auth0_settings.AUTH0_DOMAIN}/api/v2/users/{auth0_Id}"