NVIDIA-NeMo · marcusds · May 28, 2026 · May 28, 2026 · May 28, 2026 · May 28, 2026
@@ -175,8 +175,9 @@ def test_run_config_rejects_aggregate_fields(self) -> None:
     def test_rejects_legacy_backend_argument(self):
         backend = _FakeDirectBackend(single_result=_empty_evaluation_result(), multi_result=_empty_benchmark_result())
 
+        legacy_kwargs: dict = {"backend": backend}
         with pytest.raises(TypeError, match="backend"):
-            Evaluator(backend=backend)  # type: ignore[call-arg]
+            Evaluator(**legacy_kwargs)
 
     @pytest.mark.asyncio
     async def test_run_uses_offline_params_without_request_fail_fast(self):

@@ -127,7 +127,8 @@ def test_remove_missing_field(self):
 
     def test_remove_from_none(self):
         pf = ParsedFilter(operation=None, _field_map=SampleFilter._get_entity_field_map())
-        assert pf.remove("status") is None
+        result = pf.remove("status")
+        assert result is None
 
     def test_remove_non_eq_not_removed(self):
         op = ComparisonOperation(operator=FilterOperator.LIKE, field="name", value="llama")

@@ -46,7 +46,9 @@
 from .client import TEST_ADMIN_EMAIL, TEST_USER_EMAIL, ClientContext, create_test_client
 from .docker import (
     DEFAULT_RETRY_CONFIG,
+    MOCK_NIM_IMAGE_TAG,
     MOCK_NIM_NGINX_CONF,
+    MOCK_SIDECAR_IMAGE_TAG,
     MODELS_CONTROLLER_MANAGED_LABEL,
     DockerRetryConfig,
     DockerTestContext,

@@ -135,29 +135,29 @@ def create_docker_client(fail_message: str | None = None) -> docker.DockerClient
         A validated Docker client.
 
     Raises:
-        pytest.fail: If Docker client cannot be created or daemon is not responding.
+        pytest.fail.Exception: If Docker client cannot be created or daemon is not responding.
     """
     try:
         client = docker.from_env()
     except DockerException as e:
         msg = fail_message or "Docker client initialization failed"
-        pytest.fail(
+        raise pytest.fail.Exception(
             f"{msg}: {e}\n\n"
             "Please ensure Docker is installed and the Docker daemon is running:\n"
             "  - macOS/Windows: Start Docker Desktop\n"
             "  - Linux: Run 'sudo systemctl start docker' or 'sudo service docker start'\n"
             "  - Verify with: 'docker info'"
-        )
+        ) from e
 
     # Verify the daemon is actually responding
     try:
         client.ping()
     except DockerException as e:
-        pytest.fail(
+        raise pytest.fail.Exception(
             f"Docker daemon is not responding: {e}\n\n"
             "The Docker client was created but cannot communicate with the daemon.\n"
             "Please ensure the Docker daemon is running."
-        )
+        ) from e
 
     return client
 

@@ -8,7 +8,7 @@
 import string
 import tempfile
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 import pandas as pd
 from typing_extensions import Self
@@ -17,6 +17,7 @@
 
 if TYPE_CHECKING:
     from nemo_platform import NeMoPlatform
+    from nemo_platform.types.safe_synthesizer import SafeSynthesizerJobConfigParam
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -227,13 +228,12 @@
     def _resolve_datasource(self, **kwargs) -> None:
         if self._data_source_path is not None:
             return  # already uploaded; reuse the cached result
-        match self._data_source:
-            case pd.DataFrame() as df:
-                pass
-            case str(url):
-                df = pd.read_csv(url, **kwargs)
-            case _:
-                raise ValueError("Data source must be a pandas DataFrame or a URL")
+        if isinstance(self._data_source, pd.DataFrame):
+            df = self._data_source
+        elif isinstance(self._data_source, str):
+            df = pd.read_csv(self._data_source, **kwargs)
+        else:
+            raise ValueError("Data source must be a pandas DataFrame or a URL")
 
         tmp_path: Path | None = None
         try:
@@ -310,8 +310,8 @@
         spec = self._build_job_spec()
         response = self._client.safe_synthesizer.jobs.create(
             workspace=self._workspace,
-            spec=spec,
-            **kwargs,  # type: ignore  # spec accepts dict at runtime
+            spec=cast("SafeSynthesizerJobConfigParam", spec),
+            **kwargs,
         )
         return SafeSynthesizerJob(response.name, self._client, workspace=self._workspace)
 

@@ -52,14 +52,6 @@
 CONTEXT_WINDOW = 80
 
 
-@dataclass(frozen=True)
-class PatternSpec:
-    name: str
-    regex: re.Pattern[str]
-    guard: Callable[[str, re.Match[str]], bool] | None = None
-    mask: Callable[[str], str] = lambda s: _mask_middle(s)
-
-
 # --------------------------------------------------------------------------- #
 # Masking helpers
 # --------------------------------------------------------------------------- #
@@ -72,6 +64,14 @@ def _mask_middle(value: str, keep: int = 2) -> str:
     return f"{value[:keep]}{'*' * (len(value) - keep * 2)}{value[-keep:]}"
 
 
+@dataclass(frozen=True)
+class PatternSpec:
+    name: str
+    regex: re.Pattern[str]
+    guard: Callable[[str, re.Match[str]], bool] | None = None
+    mask: Callable[[str], str] = _mask_middle
+
+
 def _mask_email(value: str) -> str:
     local, _, domain = value.partition("@")
     if not domain:

@@ -418,7 +418,7 @@ def schema_tree(spec_file: str = typer.Argument(..., help="Path to OpenAPI speci
         print_verbose("\n[bold magenta]Schema Dependency Tree[/bold magenta]")
         print_verbose("Top-level schemas (used directly in endpoints) are shown at the root level")
         print_verbose("Dependent schemas are shown as children\n")
-        print_verbose("Unused schemas: ", ", ".join(sorted(unused_schemas)), style="bold yellow")
+        print_verbose(f"Unused schemas: {', '.join(sorted(unused_schemas))}", style="bold yellow")
 
         print_schema_tree(tree)
 

@@ -350,7 +350,12 @@ async def list_entities(
             filter_op=combined_filter,
             relationship_child_workspaces=accessible_workspaces,
         )
-    elif accessible_workspaces is None or workspace in accessible_workspaces:
+    else:
+        raise_if_workspace_inaccessible(
+            accessible_workspaces,
+            workspace,
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+        )
         # Check if workspace is being deleted (404 for user requests)
         await validate_workspace_not_deleting(workspace_repository, auth_client, workspace)
 
@@ -364,12 +369,6 @@ async def list_entities(
             filter_op=filter,
             relationship_child_workspaces=accessible_workspaces,
         )
-    else:
-        raise_if_workspace_inaccessible(
-            accessible_workspaces,
-            workspace,
-            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
-        )
 
     return EntitiesPage(
         data=entities,

@@ -208,6 +208,7 @@ class ModelRouterMiddleware(NemoInferenceMiddleware):
     REQUEST_MUTATION_KEY = "x_original_model"
 
     def __init__(self, target_model_entity_id: str) -> None:
+        super().__init__()
         self._target = target_model_entity_id
 
     async def on_startup(self) -> None:

@@ -4,6 +4,7 @@
 import asyncio
 import json
 import logging
+from typing import cast
 
 import nmp.evaluator.app.values as app
 import nmp.evaluator.entities as entities
@@ -129,9 +130,17 @@ async def register_result_entity(
     log.info("Registering result entity", extra={"aggregate_scores_path": aggregate_scores_path})
 
     if getattr(job, "metric", None):
-        result_entity = load_metric_result_entity(aggregate_scores_path, job, config)
+        result_entity = load_metric_result_entity(
+            aggregate_scores_path,
+            cast("app.MetricJob", job),
+            config,
+        )
     elif getattr(job, "benchmark", None):
-        result_entity = load_benchmark_result_entity(aggregate_scores_path, job, config)
+        result_entity = load_benchmark_result_entity(
+            aggregate_scores_path,
+            cast("app.BenchmarkJob", job),
+            config,
+        )
     else:
         raise ValueError(f"unsupported job {type(job)}")
 
@@ -167,9 +176,11 @@ def load_benchmark_result_entity(
     if isinstance(job.benchmark, app.Benchmark):
         metric_refs = [metric.metric_ref for metric in job.benchmark.metrics]
         dataset_ref = job.benchmark.dataset
-        benchmark_ref = job.benchmark.name
+        benchmark_ref = app.BenchmarkRef(root=job.benchmark.name)
     elif isinstance(job.benchmark, app.SystemBenchmark):
-        benchmark_ref = f"{SYSTEM_WORKSPACE}/{job.benchmark.name}"
+        benchmark_ref = app.BenchmarkRef(root=f"{SYSTEM_WORKSPACE}/{job.benchmark.name}")
+    else:
+        raise ValueError(f"Unsupported benchmark type: {type(job.benchmark).__name__}")
 
     return entities.BenchmarkJobResult(
         name=config.NEMO_JOB_ID,

@@ -302,12 +302,10 @@ def run_task():
     enable_synthesis: bool = job_config.get("enable_synthesis", True)
     logger.info(f"enable_synthesis={enable_synthesis}")
 
-    nss_job_config: SafeSynthesizerJobConfig
-    match job_config:
-        case dict():
-            nss_job_config = SafeSynthesizerJobConfig.model_validate(job_config)
-        case _:
-            raise ValueError(f"Config must be a dictionary or a string: {job_config}")
+    if isinstance(job_config, dict):
+        nss_job_config = SafeSynthesizerJobConfig.model_validate(job_config)
+    else:
+        raise ValueError(f"Config must be a dictionary or a string: {job_config}")
     logger.info(f"Nemo Safe Synthesizer runtime job config: {nss_job_config.model_dump_json(indent=2)}")
 
     save_path = Path(os.environ.get(EPHEMERAL_TASK_STORAGE_PATH_ENVVAR, DEFAULT_TASK_STORAGE_PATH))

@@ -118,6 +118,7 @@ def test_agent_polled_status() -> None:
         commands = session.get_bash_commands()
     except Exception:
         pytest.skip("trace_reader not available")
+        return
 
     status_checks = [
         cmd for cmd in commands if "jobs" in cmd and ("get-status" in cmd or "get_status" in cmd or "status" in cmd)
@@ -134,6 +135,7 @@ def test_agent_investigated_failure() -> None:
         commands = session.get_bash_commands()
     except Exception:
         pytest.skip("trace_reader not available")
+        return
 
     fail_investigation = [cmd for cmd in commands if "gpu-fail-job" in cmd or "fail-job" in cmd or "fail_job" in cmd]
     assert len(fail_investigation) >= 2, (