Get D2L API ID on user service

We already have a mechanism to get this ID from roster data.
Extend it to be usable directly on user service and reuse the same method
in RosterService.

lms/models/lms_user.py

@@ -53,7 +53,7 @@ class LMSUser(CreatedUpdatedMixin, Base):
 
     @property
     def user_id(self) -> str:
-        """Alias lti_user_id to user_if for compatilbiity with models.User."""
+        """Alias lti_user_id to user_if for compatibility with models.User."""
         return self.lti_user_id
 
 

lms/services/d2l_api/client.py

@@ -27,7 +27,7 @@ class Meta:
 
 
 class _D2LTopic(Schema):
-    """Files and assigmetns are topics within a module."""
+    """Files and assignments are topics within a module."""
 
     class Meta:
         unknown = EXCLUDE

lms/services/roster.py

@@ -24,7 +24,6 @@
 from lms.models.lti_user import display_name
 from lms.services.canvas_api.client import CanvasAPIClient
 from lms.services.canvas_api.factory import canvas_api_client_factory
-from lms.services.d2l_api.client import D2LAPIClient
 from lms.services.exceptions import (
     CanvasAPIError,
     ConcurrentTokenRefreshError,
@@ -35,6 +34,7 @@
 from lms.services.lti_role_service import LTIRoleService
 from lms.services.oauth2_token import OAuth2TokenService
 from lms.services.upsert import bulk_upsert
+from lms.services.user import UserService
 
 LOG = getLogger(__name__)
 
@@ -389,7 +389,7 @@ def fetch_canvas_sections_roster(self, lms_course: LMSCourse) -> None:
         Sections are different than other rosters:
              - We fetch them via the proprietary Canvas API, not the LTI Names and Roles endpoint.
 
-             - Due to the return value of that API we don't fetch rosters for indivual sections,
+             - Due to the return value of that API we don't fetch rosters for individual sections,
                but for all sections of one course at once
 
              - The return value of the API doesn't include enough information to create unseen users
@@ -515,7 +515,7 @@ def _get_roster_users(
         for member in roster:
             lti_user_id = member.get("lti11_legacy_user_id") or member["user_id"]
             lti_v13_user_id = member["user_id"]
-            lms_api_user_id = self._get_lms_api_user_id(member, family)
+            lms_api_user_id = UserService.get_lms_api_user_id(member, family)
             name = display_name(
                 given_name=member.get("name", ""),
                 family_name=member.get("family_name", ""),
@@ -654,16 +654,6 @@ def _get_course_users(self, lms_course: LMSCourse) -> dict[str, LMSUser]:
         ).all()
         return {u.lms_api_user_id: u for u in users}
 
-    def _get_lms_api_user_id(self, member: Member, family: Family) -> str | None:
-        if family == Family.D2L:
-            return D2LAPIClient.get_api_user_id(member["user_id"])
-
-        return (
-            member.get("message", [{}])[0]
-            .get("https://purl.imsglobal.org/spec/lti/claim/custom", {})
-            .get("canvas_user_id")
-        )
-
 
 def factory(_context, request):
     return RosterService(

lms/services/user.py

@@ -6,6 +6,7 @@
 
 from lms.models import (
     AssignmentMembership,
+    Family,
     Grouping,
     LMSCourse,
     LMSCourseMembership,
@@ -22,6 +23,7 @@
 )
 from lms.models.lms_segment import LMSSegment
 from lms.services.course import CourseService
+from lms.services.lti_names_roles import Member
 from lms.services.upsert import bulk_upsert
 
 
@@ -76,8 +78,9 @@ def upsert_lms_user(self, user: User, lti_params: LTIParams) -> LMSUser:
         """Upsert LMSUser based on a User object."""
         self._db.flush()  # Make sure User has hit the DB on the current transaction
 
-        # API ID, only Canvas for now
-        lms_api_user_id = lti_params.get("custom_canvas_user_id")
+        lms_api_user_id = self.get_lms_api_user_id(
+            lti_params, user.application_instance.family
+        )
         lms_user = bulk_upsert(
             self._db,
             LMSUser,
@@ -351,6 +354,22 @@ def get_users(  # noqa: PLR0913
 
         return query.order_by(LMSUser.display_name, LMSUser.id)
 
+    @staticmethod
+    def get_lms_api_user_id(data: Member | LTIParams, family: Family):
+        if family == Family.D2L:
+            from lms.services.d2l_api.client import D2LAPIClient
+
+            return D2LAPIClient.get_api_user_id(data["user_id"])
+
+        if isinstance(data, LTIParams):
+            return data.get("custom_canvas_user_id")
+
+        return (
+            data.get("message", [{}])[0]
+            .get("https://purl.imsglobal.org/spec/lti/claim/custom", {})
+            .get("canvas_user_id")
+        )
+
 
 def factory(_context, request):
     """Service factory for the UserService."""

tests/unit/lms/services/roster_test.py

@@ -7,7 +7,6 @@
 from sqlalchemy import select
 
 from lms.models import AssignmentRoster, CourseRoster, LMSSegmentRoster
-from lms.models.family import Family
 from lms.models.lms_segment import LMSSegment
 from lms.models.lms_user import LMSUser
 from lms.models.lti_role import RoleScope, RoleType
@@ -358,69 +357,6 @@ def test_fetch_assignment_roster(
         assert roster[3].lms_user.lti_user_id == "USER_ID_INACTIVE"
         assert not roster[3].active
 
-    @pytest.mark.parametrize(
-        "family,response,lms_api_user_id",
-        [
-            (
-                None,
-                {
-                    "user_id": "USER_ID",
-                    "roles": ["ROLE1"],
-                    "status": "Active",
-                },
-                None,
-            ),
-            (
-                Family.D2L.value,
-                {"user_id": "USER_ID_API", "roles": ["ROLE1"], "status": "Active"},
-                "API",
-            ),
-            (
-                Family.CANVAS.value,
-                {
-                    "user_id": "USER_ID",
-                    "roles": ["ROLE1"],
-                    "status": "Active",
-                    "message": [
-                        {
-                            "https://purl.imsglobal.org/spec/lti/claim/custom": {
-                                "canvas_user_id": "API_ID"
-                            }
-                        }
-                    ],
-                },
-                "API_ID",
-            ),
-        ],
-    )
-    def test_fetch_assignment_roster_with_lms_api_user_id(
-        self,
-        svc,
-        lti_names_roles_service,
-        db_session,
-        lti_role_service,
-        assignment,
-        family,
-        response,
-        lms_api_user_id,
-        application_instance,
-    ):
-        application_instance.tool_consumer_info_product_family_code = family
-        lti_names_roles_service.get_context_memberships.return_value = [response]
-        lti_role_service.get_roles.return_value = [
-            factories.LTIRole(value="ROLE1"),
-        ]
-
-        svc.fetch_assignment_roster(assignment)
-
-        roster = db_session.scalars(
-            select(AssignmentRoster)
-            .order_by(AssignmentRoster.lms_user_id)
-            .where(AssignmentRoster.assignment_id == assignment.id)
-        ).all()
-
-        assert roster[0].lms_user.lms_api_user_id == lms_api_user_id
-
     @pytest.mark.parametrize(
         "known_error",
         [

tests/unit/lms/services/user_test.py

@@ -5,7 +5,8 @@
 from h_matchers import Any
 from sqlalchemy import select
 
-from lms.models import LMSUser, RoleScope, RoleType, User
+from lms.models import LMSUser, LTIParams, RoleScope, RoleType, User
+from lms.models.family import Family
 from lms.services import UserService
 from lms.services.user import UserNotFound, factory
 from tests import factories
@@ -307,6 +308,49 @@ def test_get_users_by_segment_authority_provided_id(
             student_in_assignment.h_userid
         ]
 
+    @pytest.mark.parametrize(
+        "family,data,lms_api_user_id",
+        [
+            (
+                None,
+                {
+                    "user_id": "USER_ID",
+                    "roles": ["ROLE1"],
+                    "status": "Active",
+                },
+                None,
+            ),
+            (
+                Family.D2L.value,
+                {"user_id": "USER_ID_API", "roles": ["ROLE1"], "status": "Active"},
+                "API",
+            ),
+            (
+                Family.CANVAS.value,
+                {
+                    "user_id": "USER_ID",
+                    "roles": ["ROLE1"],
+                    "status": "Active",
+                    "message": [
+                        {
+                            "https://purl.imsglobal.org/spec/lti/claim/custom": {
+                                "canvas_user_id": "API_ID"
+                            }
+                        }
+                    ],
+                },
+                "API_ID",
+            ),
+            (
+                Family.CANVAS.value,
+                LTIParams({"custom_canvas_user_id": "API_ID"}),
+                "API_ID",
+            ),
+        ],
+    )
+    def test_get_lms_api_user_id(self, service, family, data, lms_api_user_id):
+        assert service.get_lms_api_user_id(data, family) == lms_api_user_id
+
     @pytest.fixture
     def course(self, application_instance, db_session):
         course = factories.Course(application_instance=application_instance)