diff --git a/plugins/nemo-auditor/openapi/openapi.yaml b/plugins/nemo-auditor/openapi/openapi.yaml index 8c51dbe286..27e2dc0212 100644 --- a/plugins/nemo-auditor/openapi/openapi.yaml +++ b/plugins/nemo-auditor/openapi/openapi.yaml @@ -312,6 +312,43 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/auditor/v2/workspaces/{workspace}/jobs/audit/{job}/results/artifacts/download: + get: + tags: + - Auditor Jobs + summary: Download Audit Artifacts + description: Stream an aggregate tar.gz of all garak report artifacts for an + audit job. + operationId: download_audit_artifacts_apis_auditor_v2_workspaces__workspace__jobs_audit__job__results_artifacts_download_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: job + in: path + required: true + schema: + type: string + title: Job + responses: + '200': + description: Aggregate gzip archive of all garak report artifacts. + content: + application/gzip: + schema: + type: string + format: binary + '404': + description: No report artifacts found for the job. + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/auditor/v2/workspaces/{workspace}/jobs/audit/{job}/results/{name}: get: tags: diff --git a/plugins/nemo-auditor/src/nemo_auditor/api/v2/artifacts.py b/plugins/nemo-auditor/src/nemo_auditor/api/v2/artifacts.py new file mode 100644 index 0000000000..e6023cd368 --- /dev/null +++ b/plugins/nemo-auditor/src/nemo_auditor/api/v2/artifacts.py @@ -0,0 +1,107 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Aggregate artifact download for audit jobs. + +Registered before the generic ``/{name}/download`` catch-all so FastAPI +matches this specific path first. Fetches the individual garak report +results already stored by the job and streams them back as a single +``artifacts.tar.gz`` — no additional storage. +""" + +from __future__ import annotations + +import logging +import tarfile +import tempfile +from pathlib import Path + +import anyio.to_thread +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException +from fastapi.responses import FileResponse +from nemo_auditor.authz import scope +from nemo_platform import AsyncNeMoPlatform +from nemo_platform_plugin.authz import CallerKind, path_rule +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.client.errors import NotFoundError +from nemo_platform_plugin.dependencies import get_sdk_client +from nemo_platform_plugin.jobs.client import AsyncJobsClient +from nemo_platform_plugin.jobs.result_manager import result_manager_factory + +logger = logging.getLogger(__name__) + +router = APIRouter() + +_AUDIT_SCOPE = scope.child("audit") +_AUDIT_READ_PERMISSION = _AUDIT_SCOPE.permission( + "read", + description="Read auditor.audit jobs, including status, logs, and results", +) + +# The individual result names published by AuditJob, in preferred archive order. +_REPORT_RESULT_NAMES = ("report-jsonl", "report-html", "report-hitlog-jsonl") + + +@router.get( + "/jobs/audit/{job}/results/artifacts/download", + response_class=FileResponse, + responses={ + 200: { + "description": "Aggregate gzip archive of all garak report artifacts.", + "content": {"application/gzip": {"schema": {"type": "string", "format": "binary"}}}, + }, + 404: {"description": "No report artifacts found for the job."}, + }, +) +@_AUDIT_SCOPE.read +@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[_AUDIT_READ_PERMISSION]) +async def download_audit_artifacts( + workspace: str, + job: str, + background_tasks: BackgroundTasks, + sdk: AsyncNeMoPlatform = Depends(get_sdk_client), +) -> FileResponse: + """Stream an aggregate tar.gz of all garak report artifacts for an audit job.""" + jobs_client = client_from_platform(sdk, AsyncJobsClient) + result_manager = result_manager_factory(job_name=job, workspace=workspace, files_sdk=sdk) + + tmp_dir = tempfile.TemporaryDirectory() + artifact_tmps = [] + try: + tmp = Path(tmp_dir.name) + + for result_name in _REPORT_RESULT_NAMES: + try: + result_info = (await jobs_client.get_job_result(name=result_name, job=job, workspace=workspace)).data() + except NotFoundError: + continue + + artifact_tmps.append(await result_manager.download_artifact(artifact_url=result_info.artifact_url)) + + if not artifact_tmps: + raise HTTPException( + status_code=404, + detail=f"No report artifacts found for audit job '{job}' in workspace '{workspace}'.", + ) + + tar_path = tmp / "artifacts.tar.gz" + + def _create_archive() -> None: + with tarfile.open(tar_path, "w:gz") as tar: + for artifact_tmp in artifact_tmps: + tar.add(artifact_tmp.path, arcname=artifact_tmp.path.name) + + await anyio.to_thread.run_sync(_create_archive) + + def _cleanup(): + for t in artifact_tmps: + t.cleanup_tmp_dir() + tmp_dir.cleanup() + + background_tasks.add_task(_cleanup) + return FileResponse(path=str(tar_path), filename="artifacts.tar.gz", media_type="application/gzip") + except Exception: + for t in artifact_tmps: + t.cleanup_tmp_dir() + tmp_dir.cleanup() + raise diff --git a/plugins/nemo-auditor/src/nemo_auditor/jobs/audit.py b/plugins/nemo-auditor/src/nemo_auditor/jobs/audit.py index fe244da61e..c559737921 100644 --- a/plugins/nemo-auditor/src/nemo_auditor/jobs/audit.py +++ b/plugins/nemo-auditor/src/nemo_auditor/jobs/audit.py @@ -627,7 +627,6 @@ def _on_sigterm(signum, frame): if has_reports else {} ) - n_failed = len(list(failed_dir.iterdir())) status = "completed" if n_failed == 0 else "partial" self.report_progress( diff --git a/plugins/nemo-auditor/src/nemo_auditor/service.py b/plugins/nemo-auditor/src/nemo_auditor/service.py index 96b6576c63..acceb3743f 100644 --- a/plugins/nemo-auditor/src/nemo_auditor/service.py +++ b/plugins/nemo-auditor/src/nemo_auditor/service.py @@ -22,7 +22,7 @@ class AuditorPluginService(NemoService): dependencies: ClassVar[list[str]] = ["entities"] def get_routers(self) -> list[RouterSpec]: - from nemo_auditor.api.v2 import configs, targets + from nemo_auditor.api.v2 import artifacts, configs, targets healthz_router = APIRouter() @@ -57,6 +57,12 @@ async def healthz() -> dict[str, object]: description="Audit target CRUD.", prefix=crud_prefix, ), + RouterSpec( + router=artifacts.router, + tag="Auditor Jobs", + description="Audit job aggregate artifact download.", + prefix=crud_prefix, + ), RouterSpec( add_job_routes(AuditJob, authz=scope.child("audit")), tag="Auditor Jobs", diff --git a/plugins/nemo-auditor/tests/test_api_artifacts.py b/plugins/nemo-auditor/tests/test_api_artifacts.py new file mode 100644 index 0000000000..e6ebd24fdc --- /dev/null +++ b/plugins/nemo-auditor/tests/test_api_artifacts.py @@ -0,0 +1,236 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the aggregate artifact download route (GET /results/artifacts/download). + +The endpoint fetches individual garak report results from the Jobs service and +streams them back as a single tar.gz — no stored 'artifacts' result. Tests use +FastAPI's TestClient with dependency_overrides and targeted patches so no real +platform or fileset is needed. +""" + +from __future__ import annotations + +import io +import tarfile +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import FastAPI +from fastapi.routing import APIRoute +from fastapi.testclient import TestClient +from nemo_auditor.api.v2 import artifacts as artifacts_module +from nemo_auditor.service import AuditorPluginService +from nemo_platform_plugin.client.errors import NotFoundError +from nemo_platform_plugin.dependencies import get_sdk_client +from nemo_platform_plugin.jobs.file_manager import TmpDirPath + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_result_response(artifact_url: str = "fileset://default/fs/r") -> MagicMock: + """A jobs-client response whose .data() returns a result info with artifact_url.""" + result_info = MagicMock() + result_info.artifact_url = artifact_url + resp = MagicMock() + resp.data.return_value = result_info + return resp + + +def _make_tmp(tmp_path: Path, filename: str, content: bytes = b"data") -> TmpDirPath: + """Real TmpDirPath backed by a file under pytest's tmp_path.""" + d = tmp_path / filename + d.mkdir() + f = d / filename + f.write_bytes(content) + return TmpDirPath(path=f, tmp_dir=d) + + +def _tar_names(content: bytes) -> list[str]: + with tarfile.open(fileobj=io.BytesIO(content), mode="r:gz") as tar: + return tar.getnames() + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def mock_sdk() -> MagicMock: + return MagicMock() + + +@pytest.fixture +def test_app(mock_sdk: MagicMock) -> FastAPI: + app = FastAPI() + app.include_router( + artifacts_module.router, + prefix="/apis/auditor/v2/workspaces/{workspace}", + ) + app.dependency_overrides[get_sdk_client] = lambda: mock_sdk + return app + + +@pytest.fixture +def client(test_app: FastAPI) -> TestClient: + return TestClient(test_app, raise_server_exceptions=False) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestDownloadAuditArtifacts: + def test_all_results_present_returns_200_with_valid_targz( + self, client: TestClient, mock_sdk: MagicMock, tmp_path: Path + ) -> None: + jsonl_tmp = _make_tmp(tmp_path, "run.report.jsonl", b"jsonl") + html_tmp = _make_tmp(tmp_path, "run.report.html", b"") + hitlog_tmp = _make_tmp(tmp_path, "run.hitlog.jsonl", b"hitlog") + + mock_jobs = AsyncMock() + mock_jobs.get_job_result = AsyncMock( + side_effect=[ + _make_result_response("fileset://default/fs/report-jsonl"), + _make_result_response("fileset://default/fs/report-html"), + _make_result_response("fileset://default/fs/report-hitlog-jsonl"), + ] + ) + + mock_rm = AsyncMock() + mock_rm.download_artifact = AsyncMock(side_effect=[jsonl_tmp, html_tmp, hitlog_tmp]) + + with ( + patch.object(artifacts_module, "client_from_platform", return_value=mock_jobs), + patch.object(artifacts_module, "result_manager_factory", return_value=mock_rm), + ): + resp = client.get("/apis/auditor/v2/workspaces/default/jobs/audit/job-1/results/artifacts/download") + + assert resp.status_code == 200 + assert "gzip" in resp.headers["content-type"] + names = _tar_names(resp.content) + assert "run.report.jsonl" in names + assert "run.report.html" in names + assert "run.hitlog.jsonl" in names + + def test_missing_optional_results_are_skipped( + self, client: TestClient, mock_sdk: MagicMock, tmp_path: Path + ) -> None: + jsonl_tmp = _make_tmp(tmp_path, "run.report.jsonl", b"jsonl") + + mock_jobs = AsyncMock() + mock_jobs.get_job_result = AsyncMock( + side_effect=[ + _make_result_response("fileset://default/fs/report-jsonl"), + NotFoundError(MagicMock()), + NotFoundError(MagicMock()), + ] + ) + + mock_rm = AsyncMock() + mock_rm.download_artifact = AsyncMock(return_value=jsonl_tmp) + + with ( + patch.object(artifacts_module, "client_from_platform", return_value=mock_jobs), + patch.object(artifacts_module, "result_manager_factory", return_value=mock_rm), + ): + resp = client.get("/apis/auditor/v2/workspaces/default/jobs/audit/job-1/results/artifacts/download") + + assert resp.status_code == 200 + names = _tar_names(resp.content) + assert "run.report.jsonl" in names + assert len(names) == 1 + + def test_no_results_at_all_returns_404(self, client: TestClient, mock_sdk: MagicMock) -> None: + mock_jobs = AsyncMock() + mock_jobs.get_job_result = AsyncMock(side_effect=NotFoundError(MagicMock())) + + mock_rm = AsyncMock() + + with ( + patch.object(artifacts_module, "client_from_platform", return_value=mock_jobs), + patch.object(artifacts_module, "result_manager_factory", return_value=mock_rm), + ): + resp = client.get("/apis/auditor/v2/workspaces/default/jobs/audit/job-1/results/artifacts/download") + + assert resp.status_code == 404 + assert "job-1" in resp.json()["detail"] + + def test_result_names_queried_in_order(self, client: TestClient, mock_sdk: MagicMock, tmp_path: Path) -> None: + jsonl_tmp = _make_tmp(tmp_path, "run.report.jsonl") + html_tmp = _make_tmp(tmp_path, "run.report.html") + hitlog_tmp = _make_tmp(tmp_path, "run.hitlog.jsonl") + + mock_jobs = AsyncMock() + mock_jobs.get_job_result = AsyncMock( + side_effect=[ + _make_result_response(), + _make_result_response(), + _make_result_response(), + ] + ) + + mock_rm = AsyncMock() + mock_rm.download_artifact = AsyncMock(side_effect=[jsonl_tmp, html_tmp, hitlog_tmp]) + + with ( + patch.object(artifacts_module, "client_from_platform", return_value=mock_jobs), + patch.object(artifacts_module, "result_manager_factory", return_value=mock_rm), + ): + client.get("/apis/auditor/v2/workspaces/default/jobs/audit/job-1/results/artifacts/download") + + queried = [call.kwargs["name"] for call in mock_jobs.get_job_result.await_args_list] + assert queried == ["report-jsonl", "report-html", "report-hitlog-jsonl"] + + def test_workspace_and_job_forwarded_to_jobs_client( + self, client: TestClient, mock_sdk: MagicMock, tmp_path: Path + ) -> None: + tmp = _make_tmp(tmp_path, "run.report.jsonl") + + mock_jobs = AsyncMock() + mock_jobs.get_job_result = AsyncMock( + side_effect=[_make_result_response(), NotFoundError(MagicMock()), NotFoundError(MagicMock())] + ) + + mock_rm = AsyncMock() + mock_rm.download_artifact = AsyncMock(return_value=tmp) + + with ( + patch.object(artifacts_module, "client_from_platform", return_value=mock_jobs), + patch.object(artifacts_module, "result_manager_factory", return_value=mock_rm), + ): + client.get("/apis/auditor/v2/workspaces/prod/jobs/audit/my-job/results/artifacts/download") + + first_call = mock_jobs.get_job_result.await_args_list[0] + assert first_call.kwargs["workspace"] == "prod" + assert first_call.kwargs["job"] == "my-job" + + +class TestArtifactsRouteWiring: + def test_artifacts_download_route_is_mounted(self) -> None: + service = AuditorPluginService() + paths: set[str] = set() + for spec in service.get_routers(): + for route in spec.router.routes: + if isinstance(route, APIRoute) and "GET" in route.methods: + paths.add(f"/apis/auditor{spec.prefix}{route.path}") + assert "/apis/auditor/v2/workspaces/{workspace}/jobs/audit/{job}/results/artifacts/download" in paths + + def test_artifacts_route_precedes_generic_name_route(self) -> None: + # FastAPI matches in registration order; the specific /artifacts/download path + # must appear before the wildcard /{name}/download catch-all within each router. + service = AuditorPluginService() + all_get_paths: list[str] = [] + for spec in service.get_routers(): + for route in spec.router.routes: + if isinstance(route, APIRoute) and "GET" in route.methods: + all_get_paths.append(f"/apis/auditor{spec.prefix}{route.path}") + + artifacts_idx = next(i for i, p in enumerate(all_get_paths) if p.endswith("/artifacts/download")) + generic_idx = next(i for i, p in enumerate(all_get_paths) if p.endswith("/{name}/download")) + assert artifacts_idx < generic_idx