feat(jobs): auth e2e testing and http_client DI

packages/nmp_common/src/nmp/common/sdk_factory.py

@@ -16,12 +16,12 @@
 
 logger = logging.getLogger(__name__)
 
-# Test-only: HTTP client to use for SDK requests in test context.
-# Set by test fixtures to route requests through the test transport.
+# Test-only: HTTP clients to use for SDK requests in test context.
+# Set by test fixtures to route requests through the in-process test transport.
 #
-# TODO: Remove this module-level variable once all direct get_async_platform_sdk()
-# callers are migrated to use DependencyProvider. See architecture/docs/http-client-injection.md
-# for migration path and best practices.
+# TODO: Remove these module-level variables once all direct get_platform_sdk() /
+# get_async_platform_sdk() callers are migrated to use DependencyProvider. See
+# architecture/docs/http-client-injection.md for migration path and best practices.
 _test_http_client: Optional[httpx.AsyncClient] = None
 
 
@@ -134,6 +134,7 @@ def _get_default_headers(
 def get_platform_sdk(
     as_service: str | None = None,
     internal: bool = False,
+    http_client: httpx.Client | None = None,
     on_behalf_of: str | Principal | None = None,
 ) -> NeMoPlatform:
     """
@@ -146,6 +147,7 @@ def get_platform_sdk(
                    If None and auth is enabled, propagates the current user's auth context.
         internal: If True, mark all requests from this SDK as internal requests.
                  Use this for controllers and background tasks that make internal API calls.
+        http_client: Optional sync HTTP client to use for requests.
         on_behalf_of: Optional principal ID to use for on-behalf-of authorization.
 
     Returns:
@@ -154,14 +156,14 @@ def get_platform_sdk(
     headers = _get_default_headers(as_service, internal, on_behalf_of)
     sdk = NeMoPlatform(
         base_url=_base_url_from_config(),
-        http_client=shared_sync_http_client(),
+        http_client=http_client or shared_sync_http_client(),
         default_headers=headers if headers else None,
     )
     sdk._prepare_url = _create_url_router(sdk._prepare_url)
     return sdk
 
 
-def get_task_sdk(as_service: str) -> NeMoPlatform:
+def get_task_sdk(as_service: str, http_client: httpx.Client | None = None) -> NeMoPlatform:
     """Create an SDK for use inside a task container with on-behalf-of auth.
 
     Reads the job creator's principal from the NMP_PRINCIPAL environment variable
@@ -170,6 +172,7 @@ def get_task_sdk(as_service: str) -> NeMoPlatform:
 
     Args:
         as_service: Service name for the service principal (e.g., "customizer").
+        http_client: Optional sync HTTP client to use for requests.
 
     Returns:
         Configured NeMoPlatform SDK with internal + on-behalf-of headers.
@@ -183,6 +186,7 @@ def get_task_sdk(as_service: str) -> NeMoPlatform:
     return get_platform_sdk(
         as_service=as_service,
         internal=True,
+        http_client=http_client,
         on_behalf_of=principal.effective_principal if principal else None,
     )
 
@@ -337,17 +341,23 @@ class PlatformSDKProvider:
     discovered automatically when ``nmp-common`` is installed.
     """
 
-    def get_task_sdk(self, service_name: str) -> NeMoPlatform:
-        return get_task_sdk(service_name)
+    def get_task_sdk(self, service_name: str, http_client: httpx.Client | None = None) -> NeMoPlatform:
+        return get_task_sdk(service_name, http_client=http_client)
 
     def get_platform_sdk(
         self,
         *,
         as_service: str | None = None,
         internal: bool = False,
+        http_client: httpx.Client | None = None,
         on_behalf_of: str | Principal | None = None,
     ) -> NeMoPlatform:
-        return get_platform_sdk(as_service=as_service, internal=internal, on_behalf_of=on_behalf_of)
+        return get_platform_sdk(
+            as_service=as_service,
+            internal=internal,
+            http_client=http_client,
+            on_behalf_of=on_behalf_of,
+        )
 
     def get_async_platform_sdk(
         self,

packages/nmp_common/tests/sdk_factory/test_sdk.py

@@ -5,6 +5,7 @@
 from unittest.mock import patch
 
 import pytest
+from fastapi.testclient import TestClient
 from nmp.common.config import Configuration, PlatformConfig
 from nmp.common.sdk_factory import (
     get_async_platform_sdk,
@@ -156,6 +157,16 @@ def test_get_task_sdk_without_principal(monkeypatch: pytest.MonkeyPatch):
     assert "X-NMP-Principal-On-Behalf-Of" not in sdk.default_headers
 
 
+def test_get_task_sdk_uses_explicit_sync_http_client(monkeypatch: pytest.MonkeyPatch):
+    """get_task_sdk should use an explicitly provided sync HTTP client."""
+    monkeypatch.delenv("NMP_PRINCIPAL", raising=False)
+    client = TestClient(lambda scope, receive, send: None)
+
+    sdk = get_task_sdk(as_service="customizer", http_client=client)
+
+    assert sdk._client is client
+
+
 def test_get_request_scoped_sdk_merges_otel_and_auth_headers():
     """Test that get_request_scoped_sdk merges OTEL and auth headers."""
     base_sdk = get_async_platform_sdk()

packages/nmp_testing/src/nmp/testing/client.py

@@ -384,9 +384,9 @@ def _add_service(
             app.state.access_log = access_log_instance
 
         # Configure module-level http client as FALLBACK for direct callers of
-        # get_async_platform_sdk() that don't use DependencyProvider. The primary injection
-        # path is through DependencyProvider (see below). This module-level variable will
-        # be removed once all direct callers are migrated.
+        # get_async_platform_sdk()/get_platform_sdk() that don't use DependencyProvider.
+        # The primary injection path is through DependencyProvider (see below). These
+        # module-level variables will be removed once all direct callers are migrated.
         # See architecture/docs/http-client-injection.md for details.
         sdk_factory_module._test_http_client = async_http_client
         stack.callback(lambda: setattr(sdk_factory_module, "_test_http_client", None))

services/core/jobs/pyproject.toml

@@ -60,3 +60,12 @@ dev = [
 [tool.pytest.ini_options]
 # testpaths = ["tests"]  # Commented to prevent namespace collision with root pytest
 asyncio_mode = "auto"
+markers = [
+    "integration: Service integration tests - test individual service interfaces and interactions (uses ASGI, mocks external services via SDK)",
+    "e2e: End-to-end tests - test complete customer workflows on deployed infrastructure (Helm/Docker Compose)",
+    "regression: Regression tests - test individual functional microservices for baseline functionality",
+    "canary: Canary tests - test deployed integration environments like top of tree",
+    "slow: Tests that take a long time to run",
+    "skip_in_ci: Tests that should be skipped in CI environment",
+    "unit: Unit tests - test single classes/functions with no infrastructure dependencies",
+]

services/core/jobs/tests/integration/test_task_auth_runtime.py

@@ -0,0 +1,147 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Integration tests for task-side auth propagation via ``NMP_PRINCIPAL``.
+
+These tests cover the runtime half of the jobs auth propagation story:
+
+- the task receives ``NMP_PRINCIPAL``
+- ``get_task_sdk(as_service=...)`` converts that into service + on-behalf-of headers
+- downstream services authorize based on the delegated user's permissions
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from contextlib import redirect_stdout
+from io import StringIO
+from types import ModuleType
+
+import pytest
+from nemo_platform import PermissionDeniedError
+from nmp.core.secrets.service import SecretsService
+from nmp.testing import (
+    TEST_ADMIN_EMAIL,
+    ClientContext,
+    as_user,
+    create_test_client,
+    grant_workspace_role,
+    short_unique_name,
+    unique_email,
+)
+
+
+def _secret_access_task_module() -> ModuleType:
+    module = ModuleType("task_auth_runtime_test_module")
+
+    def run(*, http_client) -> int:
+        from nmp.common.sdk_factory import get_task_sdk
+
+        workspace = os.environ["NEMO_JOB_WORKSPACE"]
+        secret_name = os.environ["NEMO_TEST_SECRET_NAME"]
+
+        result = get_task_sdk(as_service="jobs", http_client=http_client).secrets.access(
+            workspace=workspace,
+            name=secret_name,
+        )
+        print(result.value)
+        return 0
+
+    module.run = run
+    return module
+
+
+class TestTaskRuntimeAuthPropagation:
+    def test_task_sdk_accesses_secret_on_behalf_of_creator(self):
+        workspace = short_unique_name("task-obo")
+        secret_name = short_unique_name("secret")
+        secret_value = "task-visible-secret"
+        creator_email = unique_email("creator")
+
+        with create_test_client(
+            SecretsService,
+            auth_enabled=True,
+            access_log=True,
+            client_type=ClientContext,
+            workspaces=[workspace],
+        ) as ctx:
+            admin_sdk = as_user(ctx.sdk, TEST_ADMIN_EMAIL)
+            admin_sdk.secrets.create(workspace=workspace, name=secret_name, value=secret_value)
+            grant_workspace_role(
+                admin_sdk,
+                workspace=workspace,
+                principal=creator_email,
+                roles=["Viewer"],
+            )
+
+            ctx.access_log.clear()
+            stdout = StringIO()
+            with (
+                redirect_stdout(stdout),
+                pytest.MonkeyPatch.context() as monkeypatch,
+            ):
+                monkeypatch.setenv("NEMO_JOB_WORKSPACE", workspace)
+                monkeypatch.setenv("NEMO_TEST_SECRET_NAME", secret_name)
+                monkeypatch.setenv(
+                    "NMP_PRINCIPAL",
+                    json.dumps(
+                        {
+                            "id": creator_email,
+                            "email": creator_email,
+                            "groups": [],
+                        }
+                    ),
+                )
+                exit_code = _secret_access_task_module().run(http_client=ctx.test_client)
+
+            assert exit_code == 0
+            assert secret_value in stdout.getvalue()
+
+            request = ctx.access_log.assert_has_request(
+                method="GET",
+                path_contains=f"/apis/secrets/v2/workspaces/{workspace}/secrets/{secret_name}/access",
+                principal_id="service:jobs",
+            )
+            assert request.on_behalf_of == creator_email
+
+    def test_task_sdk_denies_secret_access_when_creator_lacks_permission(self):
+        workspace = short_unique_name("task-deny")
+        secret_name = short_unique_name("secret")
+        creator_email = unique_email("creator")
+
+        with create_test_client(
+            SecretsService,
+            auth_enabled=True,
+            access_log=True,
+            client_type=ClientContext,
+            workspaces=[workspace],
+        ) as ctx:
+            admin_sdk = as_user(ctx.sdk, TEST_ADMIN_EMAIL)
+            admin_sdk.secrets.create(workspace=workspace, name=secret_name, value="secret-value")
+
+            ctx.access_log.clear()
+            with (
+                pytest.MonkeyPatch.context() as monkeypatch,
+                pytest.raises(PermissionDeniedError),
+            ):
+                monkeypatch.setenv("NEMO_JOB_WORKSPACE", workspace)
+                monkeypatch.setenv("NEMO_TEST_SECRET_NAME", secret_name)
+                monkeypatch.setenv(
+                    "NMP_PRINCIPAL",
+                    json.dumps(
+                        {
+                            "id": creator_email,
+                            "email": creator_email,
+                            "groups": [],
+                        }
+                    ),
+                )
+                _secret_access_task_module().run(http_client=ctx.test_client)
+
+            request = ctx.access_log.assert_has_request(
+                method="GET",
+                path_contains=f"/apis/secrets/v2/workspaces/{workspace}/secrets/{secret_name}/access",
+                principal_id="service:jobs",
+            )
+            assert request.on_behalf_of == creator_email