feat(utils): extract RH identity context to shared utility

Move _get_rh_identity_context() and AUTH_DISABLED from rlsapi_v1.py
to src/utils/rh_identity.py for reuse by the responses endpoint.
Update rlsapi imports to use the shared module. Add unit tests.

Signed-off-by: Major Hayden <major@redhat.com>

Author: major

src/app/endpoints/rlsapi_v1.py

@@ -20,7 +20,6 @@
 import metrics
 from authentication import get_auth_dependency
 from authentication.interface import AuthTuple
-from authentication.rh_identity import RHIdentityData
 from authorization.middleware import authorize
 from client import AsyncLlamaStackClientHolder
 from configuration import configuration
@@ -54,6 +53,7 @@
     extract_token_usage,
     get_mcp_tools,
 )
+from utils.rh_identity import AUTH_DISABLED, get_rh_identity_context
 from utils.shields import run_shield_moderation
 from utils.suid import get_suid
 
@@ -65,8 +65,6 @@ class TemplateRenderError(Exception):
     """Raised when the system prompt Jinja2 template cannot be compiled."""
 
 
-# Default values when RH Identity auth is not configured
-AUTH_DISABLED = "auth_disabled"
 # Keep this tuple centralized so infer_endpoint can catch all expected backend
 # failures in one place while preserving a single telemetry/error-mapping path.
 _INFER_HANDLED_EXCEPTIONS = (
@@ -79,29 +77,8 @@ class TemplateRenderError(Exception):
 )
 
 
-def _get_rh_identity_context(request: Request) -> tuple[str, str]:
-    """Extract org_id and system_id from RH Identity request state.
-
-    When RH Identity authentication is configured, the auth dependency stores
-    the RHIdentityData object in request.state.rh_identity_data. This function
-    extracts the org_id and system_id for telemetry purposes.
-
-    Args:
-        request: The FastAPI request object.
-
-    Returns:
-        Tuple of (org_id, system_id). Returns ("auth_disabled", "auth_disabled")
-        when RH Identity auth is not configured or data is unavailable.
-    """
-    rh_identity: Optional[RHIdentityData] = getattr(
-        request.state, "rh_identity_data", None
-    )
-    if rh_identity is None:
-        return AUTH_DISABLED, AUTH_DISABLED
-
-    org_id = rh_identity.get_org_id() or AUTH_DISABLED
-    system_id = rh_identity.get_user_id() or AUTH_DISABLED
-    return org_id, system_id
+# Backward-compatible alias so existing test imports continue to work.
+_get_rh_identity_context = get_rh_identity_context
 
 
 infer_responses: dict[int | str, dict[str, Any]] = {
@@ -369,7 +346,7 @@ def _queue_splunk_event(  # pylint: disable=too-many-arguments,too-many-position
         input_tokens: Number of prompt tokens consumed by the LLM call.
         output_tokens: Number of completion tokens produced by the LLM call.
     """
-    org_id, system_id = _get_rh_identity_context(request)
+    org_id, system_id = get_rh_identity_context(request)
     systeminfo = infer_request.context.systeminfo
 
     event_data = InferenceEventData(
@@ -530,7 +507,7 @@ def _resolve_quota_subject(request: Request, auth: AuthTuple) -> str | None:
     if quota_subject == "user_id":
         return user_id
 
-    org_id, system_id = _get_rh_identity_context(request)
+    org_id, system_id = get_rh_identity_context(request)
 
     if quota_subject == "org_id":
         if org_id == AUTH_DISABLED:

src/utils/rh_identity.py

@@ -0,0 +1,39 @@
+"""Utility functions for extracting RH Identity context for telemetry.
+
+This module provides functions to extract organization and system identifiers
+from Red Hat Identity request state for telemetry and logging purposes.
+"""
+
+from typing import Final, Optional
+
+from starlette.requests import Request
+
+from authentication.rh_identity import RHIdentityData
+
+# Default value when RH Identity auth is not configured
+AUTH_DISABLED: Final[str] = "auth_disabled"
+
+
+def get_rh_identity_context(request: Request) -> tuple[str, str]:
+    """Extract org_id and system_id from RH Identity request state.
+
+    When RH Identity authentication is configured, the auth dependency stores
+    the RHIdentityData object in request.state.rh_identity_data. This function
+    extracts the org_id and system_id for telemetry purposes.
+
+    Args:
+        request: The FastAPI request object.
+
+    Returns:
+        Tuple of (org_id, system_id). Returns ("auth_disabled", "auth_disabled")
+        when RH Identity auth is not configured or data is unavailable.
+    """
+    rh_identity: Optional[RHIdentityData] = getattr(
+        request.state, "rh_identity_data", None
+    )
+    if rh_identity is None:
+        return AUTH_DISABLED, AUTH_DISABLED
+
+    org_id = rh_identity.get_org_id() or AUTH_DISABLED
+    system_id = rh_identity.get_user_id() or AUTH_DISABLED
+    return org_id, system_id

tests/unit/utils/README.md

@@ -39,6 +39,9 @@ Unit tests for utils/query.py functions.
 ## [test_responses.py](test_responses.py)
 Unit tests for utils/responses.py functions.
 
+## [test_rh_identity.py](test_rh_identity.py)
+Unit tests for utils/rh_identity module.
+
 ## [test_shields.py](test_shields.py)
 Unit tests for utils/shields.py functions.
 

tests/unit/utils/test_rh_identity.py

@@ -0,0 +1,69 @@
+"""Unit tests for utils/rh_identity module."""
+
+import pytest
+from pytest_mock import MockerFixture
+
+from authentication.rh_identity import RHIdentityData
+from utils.rh_identity import AUTH_DISABLED, get_rh_identity_context
+
+
+def test_auth_disabled_constant() -> None:
+    """Verify AUTH_DISABLED constant value."""
+    assert AUTH_DISABLED == "auth_disabled"
+
+
+@pytest.mark.parametrize(
+    ("rh_identity_setup", "expected_org_id", "expected_system_id"),
+    [
+        pytest.param(
+            {"org_id": "org123", "user_id": "sys456"},
+            "org123",
+            "sys456",
+            id="identity_present",
+        ),
+        pytest.param(
+            None,
+            AUTH_DISABLED,
+            AUTH_DISABLED,
+            id="identity_absent",
+        ),
+        pytest.param(
+            {"org_id": "", "user_id": "sys456"},
+            AUTH_DISABLED,
+            "sys456",
+            id="empty_org_id",
+        ),
+        pytest.param(
+            {"org_id": "org123", "user_id": ""},
+            "org123",
+            AUTH_DISABLED,
+            id="empty_user_id",
+        ),
+    ],
+)
+def test_get_rh_identity_context(
+    mocker: MockerFixture,
+    rh_identity_setup: dict[str, str] | None,
+    expected_org_id: str,
+    expected_system_id: str,
+) -> None:
+    """Test get_rh_identity_context extracts or defaults org/system IDs."""
+    mock_request = mocker.Mock()
+
+    if rh_identity_setup is not None:
+        mock_rh_identity = mocker.Mock(spec=RHIdentityData)
+        mock_rh_identity.get_org_id.return_value = rh_identity_setup["org_id"]
+        mock_rh_identity.get_user_id.return_value = rh_identity_setup["user_id"]
+        mock_request.state = mocker.Mock()
+        mock_request.state.rh_identity_data = mock_rh_identity
+    else:
+        mock_request.state = mocker.Mock(spec=[])
+
+    org_id, system_id = get_rh_identity_context(mock_request)
+
+    assert org_id == expected_org_id
+    assert system_id == expected_system_id
+
+    if rh_identity_setup is not None:
+        mock_rh_identity.get_org_id.assert_called_once()
+        mock_rh_identity.get_user_id.assert_called_once()