diff --git a/e2e/auditor/test_audit_job.py b/e2e/auditor/test_audit_job.py index b375842c71..dc6b9be7a1 100644 --- a/e2e/auditor/test_audit_job.py +++ b/e2e/auditor/test_audit_job.py @@ -20,6 +20,8 @@ import pytest from nemo_platform import NeMoPlatform +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient from nmp.testing import add_mock_provider, short_unique_name from e2e.auditor.utils import minimal_audit_config, unique_name @@ -47,7 +49,7 @@ def _chat_completion(content: str = "I'm happy to help!") -> dict: def _wait_for_audit_job(sdk: NeMoPlatform, job_name: str, workspace: str) -> str: deadline = time.monotonic() + AUDIT_JOB_TIMEOUT_SECONDS while time.monotonic() < deadline: - status_resp = sdk.jobs.get_status(name=job_name, workspace=workspace) + status_resp = client_from_platform(sdk, JobsClient).get_job_status(name=job_name, workspace=workspace) status = str(status_resp.status) if status in TERMINAL_STATUSES: return status @@ -57,9 +59,10 @@ def _wait_for_audit_job(sdk: NeMoPlatform, job_name: str, workspace: str) -> str def _cleanup_audit_job(sdk: NeMoPlatform, job_name: str, workspace: str) -> None: with suppress(Exception): - sdk.jobs.cancel(name=job_name, workspace=workspace) + jobs = client_from_platform(sdk, JobsClient) + jobs.cancel_job(name=job_name, workspace=workspace) with suppress(Exception): - sdk.jobs.delete(name=job_name, workspace=workspace) + jobs.delete_job(name=job_name, workspace=workspace) def _add_mock_provider_or_skip(sdk: NeMoPlatform, workspace: str, name: str) -> str: diff --git a/e2e/test_anonymizer_plugin.py b/e2e/test_anonymizer_plugin.py index 2a58030710..d38816ffdc 100644 --- a/e2e/test_anonymizer_plugin.py +++ b/e2e/test_anonymizer_plugin.py @@ -36,8 +36,10 @@ ) from nemo_anonymizer_plugin.sdk.resources import AnonymizerPreviewResult from nemo_platform import NeMoPlatform +from nemo_platform_plugin.client.adapter import client_from_platform from nemo_platform_plugin.files.client import FilesClient from nemo_platform_plugin.files.types import CreateFilesetRequest +from nemo_platform_plugin.jobs.client import JobsClient from nmp.testing import MockProviderResponse, add_mock_provider, short_unique_name pytestmark = [ @@ -324,9 +326,10 @@ def _wait_for_anonymizer_job(job: AnonymizerJobResource, *, timeout_seconds: flo def _cleanup_anonymizer_job(sdk: NeMoPlatform, job_name: str) -> None: with suppress(Exception): - sdk.jobs.cancel(name=job_name, workspace=sdk.workspace) + jobs = client_from_platform(sdk, JobsClient) + jobs.cancel_job(name=job_name, workspace=sdk.workspace) with suppress(Exception): - sdk.jobs.delete(name=job_name, workspace=sdk.workspace) + jobs.delete_job(name=job_name, workspace=sdk.workspace) @pytest.fixture(scope="module") diff --git a/e2e/test_evaluator_plugin.py b/e2e/test_evaluator_plugin.py index 3b2a9f2330..4b8bb9c464 100644 --- a/e2e/test_evaluator_plugin.py +++ b/e2e/test_evaluator_plugin.py @@ -48,6 +48,8 @@ from nemo_evaluator_sdk.values.scores import JSONScoreParser, RangeScore from nemo_platform import APIConnectionError, APIStatusError, NeMoPlatform from nemo_platform.types.inference import ModelProvider +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient from nmp.testing import add_mock_provider, short_unique_name, wait_for_model_entity from nmp.testing.e2e import wait_for_platform_job from nmp.testing.utils import ensure_passthrough_virtual_model @@ -287,9 +289,10 @@ def _create_ready_mock_model( def _cleanup_evaluator_job(sdk: NeMoPlatform, job_name: str) -> None: with suppress(Exception): - sdk.jobs.cancel(name=job_name, workspace=sdk.workspace) + jobs = client_from_platform(sdk, JobsClient) + jobs.cancel_job(name=job_name, workspace=sdk.workspace) with suppress(Exception): - sdk.jobs.delete(name=job_name, workspace=sdk.workspace) + jobs.delete_job(name=job_name, workspace=sdk.workspace) def _wait_for_evaluator_job(job: EvaluatorJobResource) -> None: @@ -869,7 +872,9 @@ def test_gym_agent_evaluate_job_invalid_config_fails( job = wait_for_platform_job(evaluator_sdk, job_name, evaluator_workspace, timeout=240) assert job.status.lower() == "error", f"job {job_name!r} ended {job.status!r}" - job_status = evaluator_sdk.jobs.get_status(workspace=evaluator_workspace, name=job_name) - assert job_status.steps[0].status == "error" + job_status = client_from_platform(evaluator_sdk, JobsClient).get_job_status( + workspace=evaluator_workspace, name=job_name + ) + assert job_status.data().steps[0].status == "error" finally: _cleanup_evaluator_job(evaluator_sdk, job_name) diff --git a/e2e/test_jobs.py b/e2e/test_jobs.py index 60ccdd1ddb..f15ecc443a 100644 --- a/e2e/test_jobs.py +++ b/e2e/test_jobs.py @@ -16,7 +16,10 @@ import pytest from nemo_platform import NeMoPlatform, NotFoundError +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient from nemo_platform_plugin.jobs.constants import DEFAULT_JOB_STORAGE_PATH +from nemo_platform_plugin.jobs.types import CreatePlatformJobRequest from nmp.testing.e2e import wait_for_job_logs, wait_for_platform_job from e2e.services_pool import RunningServices @@ -38,10 +41,10 @@ def _job_diagnostic_message(sdk: NeMoPlatform, job, workspace: str, prefix: str) if job.error_details: parts.append(f"Error details: {job.error_details}") try: - logs = sdk.jobs.get_logs(workspace=workspace, name=job.name) - if logs.data: - parts.append(f"Job logs ({len(logs.data)} entries):") - for entry in logs.data: + logs = list(client_from_platform(sdk, JobsClient).list_job_logs(workspace=workspace, name=job.name).items()) + if logs: + parts.append(f"Job logs ({len(logs)} entries):") + for entry in logs: parts.append(f" - {entry.message}") except Exception as log_err: parts.append(f"Could not fetch job logs: {log_err}") @@ -57,23 +60,29 @@ def test_basic_platform_job_lifecycle(sdk: NeMoPlatform, workspace: str): 3. Verify job reaches completed status 4. Retrieve and check step logs """ - job = sdk.jobs.create( - workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "echo-step", - "executor": { - "provider": "cpu", - "container": { - "command": ["echo", "Hello from e2e test!"], + job = ( + client_from_platform(sdk, JobsClient) + .create_job( + workspace=workspace, + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ + { + "name": "echo-step", + "executor": { + "provider": "cpu", + "container": { + "command": ["echo", "Hello from e2e test!"], + }, + }, }, - }, + ], }, - ], - }, + ), + ) + .data() ) completed_job = wait_for_platform_job(sdk, job.name, workspace) @@ -100,23 +109,29 @@ def test_job_logs_across_multiple_batches(sdk: NeMoPlatform, workspace: str): [f'echo "Log message {i} of {num_logs}"; sleep {delay_seconds}' for i in range(1, num_logs + 1)] ) - job = sdk.jobs.create( - workspace=workspace, - source=JOB_SOURCE, - spec={"test": "multi-batch-logs"}, - platform_spec={ - "steps": [ - { - "name": "multi-log-step", - "executor": { - "provider": "cpu", - "container": { - "command": ["sh", "-c", log_command], + job = ( + client_from_platform(sdk, JobsClient) + .create_job( + workspace=workspace, + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "multi-batch-logs"}, + platform_spec={ + "steps": [ + { + "name": "multi-log-step", + "executor": { + "provider": "cpu", + "container": { + "command": ["sh", "-c", log_command], + }, + }, }, - }, + ], }, - ], - }, + ), + ) + .data() ) completed_job = wait_for_platform_job(sdk, job.name, workspace, timeout=120) @@ -139,26 +154,36 @@ def test_job_logs_across_multiple_batches(sdk: NeMoPlatform, workspace: str): def test_job_config_is_readable(sdk: NeMoPlatform, workspace: str): """Test that a job can read its configuration via $NEMO_JOB_STEP_CONFIG_FILE_PATH.""" - job = sdk.jobs.create( - workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "config-step", - "executor": { - "provider": "cpu", - "container": { - "command": ["sh", "-c", "echo 'Step config:'; cat $NEMO_JOB_STEP_CONFIG_FILE_PATH;"], + job = ( + client_from_platform(sdk, JobsClient) + .create_job( + workspace=workspace, + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ + { + "name": "config-step", + "executor": { + "provider": "cpu", + "container": { + "command": [ + "sh", + "-c", + "echo 'Step config:'; cat $NEMO_JOB_STEP_CONFIG_FILE_PATH;", + ], + }, + }, + "config": { + "message": "Hello from job config!", + }, }, - }, - "config": { - "message": "Hello from job config!", - }, + ], }, - ], - }, + ), + ) + .data() ) completed_job = wait_for_platform_job(sdk, job.name, workspace) @@ -177,51 +202,54 @@ def test_job_passing_data_between_steps(sdk: NeMoPlatform, workspace: str): "name": "NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH", "value": DEFAULT_JOB_STORAGE_PATH, } - job = sdk.jobs.create( + jobs = client_from_platform(sdk, JobsClient) + job = jobs.create_job( workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "generate-data-step", - "executor": { - "provider": "cpu", - "container": { - "command": [ - "sh", - "-c", - "echo 'Data from first step' > $NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH/data.txt", - ], + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ + { + "name": "generate-data-step", + "executor": { + "provider": "cpu", + "container": { + "command": [ + "sh", + "-c", + "echo 'Data from first step' > $NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH/data.txt", + ], + }, }, + "environment": [persistent_storage_env], }, - "environment": [persistent_storage_env], - }, - { - "name": "consume-data-step", - "executor": { - "provider": "cpu", - "container": { - "command": [ - "sh", - "-c", - "echo 'Consuming data:'; cat $NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH/data.txt", - ], + { + "name": "consume-data-step", + "executor": { + "provider": "cpu", + "container": { + "command": [ + "sh", + "-c", + "echo 'Consuming data:'; cat $NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH/data.txt", + ], + }, }, + "environment": [persistent_storage_env], }, - "environment": [persistent_storage_env], - }, - ], - }, - ) + ], + }, + ), + ).data() completed_job = wait_for_platform_job(sdk, job.name, workspace) assert completed_job.status == "completed", _job_diagnostic_message( sdk, completed_job, workspace, f"Job failed with status: {completed_job.status}" ) - step_logs = sdk.jobs.get_logs(workspace=workspace, name=job.name) - all_messages = " ".join(log.message for log in step_logs.data) + step_logs = list(jobs.list_job_logs(workspace=workspace, name=job.name).items()) + all_messages = " ".join(log.message for log in step_logs) assert "Data from first step" in all_messages, "Second step did not receive data from first step" @@ -235,29 +263,35 @@ def test_job_using_secret_environment_variable(sdk: NeMoPlatform, workspace: str secret_deleted = False try: - job = sdk.jobs.create( - workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "secret-envvar-step", - "executor": { - "provider": "cpu", - "container": { - "command": ["sh", "-c", 'echo "Secret value is: $SECRET_ENV_VAR"'], - }, - }, - "environment": [ + job = ( + client_from_platform(sdk, JobsClient) + .create_job( + workspace=workspace, + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ { - "name": "SECRET_ENV_VAR", - "from_secret": {"name": secret.name}, + "name": "secret-envvar-step", + "executor": { + "provider": "cpu", + "container": { + "command": ["sh", "-c", 'echo "Secret value is: $SECRET_ENV_VAR"'], + }, + }, + "environment": [ + { + "name": "SECRET_ENV_VAR", + "from_secret": {"name": secret.name}, + }, + ], }, ], }, - ], - }, + ), + ) + .data() ) completed_job = wait_for_platform_job(sdk, job.name, workspace) @@ -283,23 +317,29 @@ def test_job_using_secret_environment_variable(sdk: NeMoPlatform, workspace: str def test_job_with_expected_failure(sdk: NeMoPlatform, workspace: str): """Test that a job correctly reports failure when a step exits non-zero.""" - job = sdk.jobs.create( - workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "failing-step", - "executor": { - "provider": "cpu", - "container": { - "command": ["sh", "-c", "echo 'This step will fail'; exit 1;"], + job = ( + client_from_platform(sdk, JobsClient) + .create_job( + workspace=workspace, + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ + { + "name": "failing-step", + "executor": { + "provider": "cpu", + "container": { + "command": ["sh", "-c", "echo 'This step will fail'; exit 1;"], + }, + }, }, - }, + ], }, - ], - }, + ), + ) + .data() ) completed_job = wait_for_platform_job(sdk, job.name, workspace) @@ -312,26 +352,29 @@ def test_job_with_expected_failure(sdk: NeMoPlatform, workspace: str): def test_job_cancel_immediately(sdk: NeMoPlatform, workspace: str): """Test that a job can be created and then cancelled immediately.""" - job = sdk.jobs.create( + jobs = client_from_platform(sdk, JobsClient) + job = jobs.create_job( workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "long-running-step", - "executor": { - "provider": "cpu", - "container": { - "command": ["sh", "-c", "sleep 60"], + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ + { + "name": "long-running-step", + "executor": { + "provider": "cpu", + "container": { + "command": ["sh", "-c", "sleep 60"], + }, }, }, - }, - ], - }, - ) + ], + }, + ), + ).data() - sdk.jobs.cancel(workspace=workspace, name=job.name) + jobs.cancel_job(workspace=workspace, name=job.name) cancelled_job = wait_for_platform_job(sdk, job.name, workspace) assert cancelled_job.status == "cancelled", _job_diagnostic_message( @@ -341,31 +384,34 @@ def test_job_cancel_immediately(sdk: NeMoPlatform, workspace: str): def test_job_cancel_once_active(sdk: NeMoPlatform, workspace: str): """Test that an active job can be cancelled.""" - job = sdk.jobs.create( + jobs = client_from_platform(sdk, JobsClient) + job = jobs.create_job( workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "long-running-step", - "executor": { - "provider": "cpu", - "container": { - "command": ["sh", "-c", "sleep 300"], + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ + { + "name": "long-running-step", + "executor": { + "provider": "cpu", + "container": { + "command": ["sh", "-c", "sleep 300"], + }, }, }, - }, - ], - }, - ) + ], + }, + ), + ).data() active_job = wait_for_platform_job(sdk, job.name, workspace, status_to_check="active") assert active_job.status == "active", _job_diagnostic_message( sdk, active_job, workspace, f"Job did not become active, status: {active_job.status}" ) - sdk.jobs.cancel(workspace=workspace, name=job.name) + jobs.cancel_job(workspace=workspace, name=job.name) cancelled_job = wait_for_platform_job(sdk, job.name, workspace) assert cancelled_job.status == "cancelled", _job_diagnostic_message( @@ -380,36 +426,39 @@ def test_job_cancel_once_active(sdk: NeMoPlatform, workspace: str): def test_job_pause_resume(sdk: NeMoPlatform, workspace: str): """Test that a job can be paused and then resumed after being paused.""" - job = sdk.jobs.create( + jobs = client_from_platform(sdk, JobsClient) + job = jobs.create_job( workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "long-running-step-pause-resume", - "executor": { - "provider": "cpu", - "container": { - # Short sleep so the job completes quickly after resume. - # The pause/resume cycle is what we're testing, not the workload. - "command": ["sh", "-c", "sleep 30"], + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ + { + "name": "long-running-step-pause-resume", + "executor": { + "provider": "cpu", + "container": { + # Short sleep so the job completes quickly after resume. + # The pause/resume cycle is what we're testing, not the workload. + "command": ["sh", "-c", "sleep 30"], + }, }, }, - }, - ], - }, - ) + ], + }, + ), + ).data() active_job = wait_for_platform_job(sdk, job.name, workspace, status_to_check="active") assert active_job.status == "active", f"Job did not become active, status: {active_job.status}" - sdk.jobs.pause(workspace=workspace, name=job.name) + jobs.pause_job(workspace=workspace, name=job.name) paused_job = wait_for_platform_job(sdk, job.name, workspace, status_to_check="paused") assert paused_job.status == "paused", f"Job should have been paused but has status: {paused_job.status}" - sdk.jobs.resume(workspace=workspace, name=job.name) + jobs.resume_job(workspace=workspace, name=job.name) resumed_job = wait_for_platform_job(sdk, job.name, workspace, status_to_check="active") assert resumed_job.status in ("active", "completed"), ( @@ -422,34 +471,37 @@ def test_job_pause_resume(sdk: NeMoPlatform, workspace: str): def test_job_pause_and_cancel(sdk: NeMoPlatform, workspace: str): """Test that a job can be paused and then cancelled after being paused.""" - job = sdk.jobs.create( + jobs = client_from_platform(sdk, JobsClient) + job = jobs.create_job( workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "long-running-step-pause-cancel", - "executor": { - "provider": "cpu", - "container": { - "command": ["sh", "-c", "sleep 30"], + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ + { + "name": "long-running-step-pause-cancel", + "executor": { + "provider": "cpu", + "container": { + "command": ["sh", "-c", "sleep 30"], + }, }, }, - }, - ], - }, - ) + ], + }, + ), + ).data() active_job = wait_for_platform_job(sdk, job.name, workspace, status_to_check="active") assert active_job.status == "active", f"Job did not become active, status: {active_job.status}" - sdk.jobs.pause(workspace=workspace, name=job.name) + jobs.pause_job(workspace=workspace, name=job.name) paused_job = wait_for_platform_job(sdk, job.name, workspace, status_to_check="paused") assert paused_job.status == "paused", f"Job should have been paused but has status: {paused_job.status}" - sdk.jobs.cancel(workspace=workspace, name=job.name) + jobs.cancel_job(workspace=workspace, name=job.name) cancelled_job = wait_for_platform_job(sdk, job.name, workspace) assert cancelled_job.status == "cancelled", f"Job should have been cancelled but has status: {cancelled_job.status}" @@ -464,82 +516,88 @@ def test_job_using_additional_volume(sdk: NeMoPlatform, workspace: str, _service # not affect unrelated jobs that also request persistent job storage. profile = ADDITIONAL_VOLUME_PROFILE if _services_instance.config_path is None else "default" - job = sdk.jobs.create( + jobs = client_from_platform(sdk, JobsClient) + job = jobs.create_job( workspace=workspace, - source=JOB_SOURCE, - spec={"test": "data-between-steps"}, - platform_spec={ - "steps": [ - { - "name": "write-data", - "executor": { - "provider": "cpu", - "profile": profile, - "container": { - "command": [ - "sh", - "-c", - "echo 'Hello, World!' > /mnt/additional_storage/shared_data.txt; " - "echo 'Successfully wrote data to persistent storage';", - ], + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "data-between-steps"}, + platform_spec={ + "steps": [ + { + "name": "write-data", + "executor": { + "provider": "cpu", + "profile": profile, + "container": { + "command": [ + "sh", + "-c", + "echo 'Hello, World!' > /mnt/additional_storage/shared_data.txt; " + "echo 'Successfully wrote data to persistent storage';", + ], + }, }, }, - }, - { - "name": "read-data", - "executor": { - "provider": "cpu", - "profile": profile, - "container": { - "command": [ - "sh", - "-c", - "cat /mnt/additional_storage/shared_data.txt; " - "echo 'Successfully read data from persistent storage';", - ], + { + "name": "read-data", + "executor": { + "provider": "cpu", + "profile": profile, + "container": { + "command": [ + "sh", + "-c", + "cat /mnt/additional_storage/shared_data.txt; " + "echo 'Successfully read data from persistent storage';", + ], + }, }, }, - }, - ], - }, - ) + ], + }, + ), + ).data() completed_job = wait_for_platform_job(sdk, job.name, workspace) assert completed_job.status == "completed", f"Job failed with status: {completed_job.status}" - step_logs = sdk.jobs.get_logs(workspace=workspace, name=job.name) - assert len(step_logs.data) == 3, "Expected three step logs" - assert "Successfully wrote data to persistent storage" in step_logs.data[0].message - assert "Hello, World!" in step_logs.data[1].message - assert "Successfully read data from persistent storage" in step_logs.data[2].message + step_logs = list(jobs.list_job_logs(workspace=workspace, name=job.name).items()) + assert len(step_logs) == 3, "Expected three step logs" + assert "Successfully wrote data to persistent storage" in step_logs[0].message + assert "Hello, World!" in step_logs[1].message + assert "Successfully read data from persistent storage" in step_logs[2].message @pytest.mark.container_only @pytest.mark.parametrize("bad_image", ["__invalid_ubuntu:image", "ubuntu:does-not-exist-1234"]) def test_job_invalid_image_format(sdk: NeMoPlatform, workspace: str, bad_image: str): """Test that a job with a bad image fails appropriately.""" - job = sdk.jobs.create( + jobs = client_from_platform(sdk, JobsClient) + job = jobs.create_job( workspace=workspace, - source=JOB_SOURCE, - spec={"test": "value"}, - platform_spec={ - "steps": [ - { - "name": "bad-image-step", - "executor": { - "provider": "cpu", - "container": { - "image": bad_image, - "command": ["echo", "This should not run"], + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "value"}, + platform_spec={ + "steps": [ + { + "name": "bad-image-step", + "executor": { + "provider": "cpu", + "container": { + "image": bad_image, + "command": ["echo", "This should not run"], + }, }, }, - }, - ], - }, - ) + ], + }, + ), + ).data() completed_job = wait_for_platform_job(sdk, job.name, workspace) assert completed_job.status == "error", f"Job should have failed but has status: {completed_job.status}" - job_status = sdk.jobs.get_status(workspace=workspace, name=job.name) - assert job_status.steps[0].status == "error", "Step should have failed" + job_status = jobs.get_job_status(workspace=workspace, name=job.name) + assert job_status.data().steps[0].status == "error", "Step should have failed" diff --git a/e2e/test_nemo_agents_execute_job.py b/e2e/test_nemo_agents_execute_job.py index 026e9ad778..4dc8c71cf2 100644 --- a/e2e/test_nemo_agents_execute_job.py +++ b/e2e/test_nemo_agents_execute_job.py @@ -13,6 +13,8 @@ import pytest from nemo_agents_plugin.entities import NEMO_AGENTS_SPEC_CONFIG_FORMAT from nemo_platform import NeMoPlatform +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient from nmp.testing import MockProviderResponse, add_mock_provider from nmp.testing.e2e import wait_for_platform_job @@ -37,7 +39,7 @@ def _job_diagnostic_message(sdk: NeMoPlatform, job: Any, workspace: str, prefix: if job.error_details: parts.append(f"Error details: {job.error_details}") try: - logs = sdk.jobs.get_logs(workspace=workspace, name=job.name) + logs = client_from_platform(sdk, JobsClient).list_job_logs(workspace=workspace, name=job.name) if logs.data: parts.append(f"Job logs ({len(logs.data)} entries):") for entry in logs.data: diff --git a/e2e/test_safe_synthesizer.py b/e2e/test_safe_synthesizer.py index b773eafd1a..c4147ef4d7 100644 --- a/e2e/test_safe_synthesizer.py +++ b/e2e/test_safe_synthesizer.py @@ -38,6 +38,7 @@ from nemo_platform_plugin.client.adapter import client_from_platform from nemo_platform_plugin.files.client import FilesClient from nemo_platform_plugin.files.types import CreateFilesetRequest +from nemo_platform_plugin.jobs.client import JobsClient pytestmark = [ pytest.mark.timeout(600), @@ -320,15 +321,16 @@ def _result_names(results: dict[str, Any]) -> set[str]: def _status_details(sdk: NeMoPlatform, workspace: str, job_name: str) -> str: details = [f"Safe Synthesizer job {job_name} did not complete successfully."] + jobs = client_from_platform(sdk, JobsClient) with suppress(Exception): - job = sdk.jobs.retrieve(job_name, workspace=workspace) + job = jobs.get_job(name=job_name, workspace=workspace).data() details.append(f"Job: {job.model_dump_json(indent=2)}") with suppress(Exception): - status = sdk.jobs.get_status(job_name, workspace=workspace) + status = jobs.get_job_status(name=job_name, workspace=workspace).data() details.append(f"Status: {status.model_dump_json(indent=2)}") with suppress(Exception): - logs = sdk.jobs.get_logs(job_name, workspace=workspace) - tail = logs.data[-30:] if logs.data else [] + log_entries = list(jobs.list_job_logs(name=job_name, workspace=workspace).items()) + tail = log_entries[-30:] if log_entries else [] details.append("Recent logs:") details.extend(f"[{entry.job_step}] {entry.message}" for entry in tail) return "\n".join(details) @@ -350,7 +352,7 @@ def _wait_for_status( while time.monotonic() < deadline: try: - status_info = sdk.jobs.get_status(job_name, workspace=workspace) + status_info = client_from_platform(sdk, JobsClient).get_job_status(name=job_name, workspace=workspace) status = str(status_info.status) if not history or history[-1] != status: history.append(status) diff --git a/packages/nmp_common/tests/sdk_factory/test_sdk.py b/packages/nmp_common/tests/sdk_factory/test_sdk.py index e981cb5f53..a2f0bda99c 100644 --- a/packages/nmp_common/tests/sdk_factory/test_sdk.py +++ b/packages/nmp_common/tests/sdk_factory/test_sdk.py @@ -8,7 +8,9 @@ import httpx import pytest from nemo_platform_ext.auth.helpers import NMPOIDCConfig +from nemo_platform_plugin.client.adapter import client_from_platform from nemo_platform_plugin.client.constants import WORKLOAD_IDENTITY_TOKEN_FILE_ENVVAR +from nemo_platform_plugin.jobs.client import JobsClient from nmp.common.config import Configuration, PlatformConfig from nmp.common.http_clients import shared_async_http_client, shared_sync_http_client from nmp.common.sdk_factory import ( @@ -111,7 +113,7 @@ def capture_request(request: httpx.Request) -> httpx.Response: sdk = get_platform_sdk(http_client=http_client) assert str(sdk.base_url).rstrip("/") == "http://nemo-platform-api:8080" - sdk.jobs.list(workspace="default") + client_from_platform(sdk, JobsClient).list_jobs(workspace="default") assert len(captured_requests) == 1 assert str(captured_requests[0].url) == "http://nemo-platform-api:8080/apis/jobs/v2/workspaces/default/jobs" diff --git a/packages/nmp_testing/src/nmp/testing/e2e/customizer.py b/packages/nmp_testing/src/nmp/testing/e2e/customizer.py index ff2f837d59..3a5ef35b0c 100644 --- a/packages/nmp_testing/src/nmp/testing/e2e/customizer.py +++ b/packages/nmp_testing/src/nmp/testing/e2e/customizer.py @@ -17,6 +17,8 @@ import pytest from nemo_platform import NeMoPlatform from nemo_platform.types.inference import ContainerExecutorConfigParam, ModelDeploymentConfigModelSpecParam +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient logger = logging.getLogger(__name__) @@ -154,7 +156,7 @@ def log_status_details(status_details: dict | None, prefix: str = "status_detail logger.info(json.dumps(status_details, indent=2)) -def save_job_logs_to_file(sdk: NeMoPlatform, job_name: str, workspace: str) -> tuple[Path | None, list | None]: +def save_job_logs_to_file(jobs: JobsClient, job_name: str, workspace: str) -> tuple[Path | None, list | None]: """Save complete job logs to a file for CI artifact collection. Returns: @@ -164,13 +166,13 @@ def save_job_logs_to_file(sdk: NeMoPlatform, job_name: str, workspace: str) -> t LOGS_DIR.mkdir(exist_ok=True) log_file = LOGS_DIR / f"{job_name}.log" try: - logs = sdk.customization.jobs.get_logs(job_name, workspace=workspace) - if logs.data: + logs = list(jobs.list_job_logs(name=job_name, workspace=workspace).items()) + if logs: with log_file.open("w") as f: - for log_entry in logs.data: + for log_entry in logs: f.write(f"[{log_entry.job_step}] {log_entry.message}\n") - logger.info(f"Saved {len(logs.data)} log entries to {log_file}") - return log_file, logs.data + logger.info(f"Saved {len(logs)} log entries to {log_file}") + return log_file, logs else: logger.warning(f"No log data returned by SDK for job {job_name} — logs may not be available yet") except Exception as e: @@ -178,7 +180,7 @@ def save_job_logs_to_file(sdk: NeMoPlatform, job_name: str, workspace: str) -> t status_file = LOGS_DIR / f"{job_name}-status.json" try: - job_status = sdk.customization.jobs.get_status(job_name, workspace=workspace) + job_status = jobs.get_job_status(name=job_name, workspace=workspace).data() status_file.write_text(job_status.model_dump_json(indent=2)) logger.info(f"Saved job status to {status_file}") except Exception as e: @@ -194,14 +196,15 @@ def get_job_failure_details(sdk: NeMoPlatform, job_name: str, workspace: str) -> Also saves complete logs to a file for CI artifact collection. """ details = [f"Job {job_name} failed. Details:"] + jobs = client_from_platform(sdk, JobsClient) try: - job_status = sdk.customization.jobs.get_status(job_name, workspace=workspace) + job_status = jobs.get_job_status(name=job_name, workspace=workspace).data() details.append(f"\nJob Status: {job_status.model_dump_json(indent=2)}") except Exception as e: details.append(f"\nFailed to get job status: {e}") - log_file, log_entries = save_job_logs_to_file(sdk, job_name, workspace) + log_file, log_entries = save_job_logs_to_file(jobs, job_name, workspace) if log_file: details.append(f"\nFull logs saved to: {log_file}") @@ -243,7 +246,7 @@ def wait_for_customization_job( ): """Wait for a customization job to reach a terminal state. - Uses ``sdk.customization.jobs.get_status()`` to poll, logs training + Uses ``client_from_platform(sdk, JobsClient).get_job_status()`` to poll, logs training progress from the steps structure, and returns the full job object via ``sdk.customization.jobs.retrieve()`` once terminal. @@ -265,6 +268,7 @@ def wait_for_customization_job( start_time = time.time() last_status = None consecutive_errors = 0 + jobs = client_from_platform(sdk, JobsClient) while True: elapsed = time.time() - start_time @@ -275,7 +279,7 @@ def wait_for_customization_job( ) try: - status = sdk.customization.jobs.get_status(name=job_name, workspace=workspace) + status = jobs.get_job_status(name=job_name, workspace=workspace).data() consecutive_errors = 0 except (httpx.TimeoutException, httpx.ConnectError, ConnectionError, OSError) as exc: consecutive_errors += 1 @@ -303,7 +307,7 @@ def wait_for_customization_job( time.sleep(poll_interval) - return sdk.customization.jobs.retrieve(job_name, workspace=workspace) + return jobs.get_job(name=job_name, workspace=workspace) def wait_for_model_spec( diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/controller.py b/plugins/nemo-insights/src/nemo_insights_plugin/controller.py index e784041839..801f70ab9a 100644 --- a/plugins/nemo-insights/src/nemo_insights_plugin/controller.py +++ b/plugins/nemo-insights/src/nemo_insights_plugin/controller.py @@ -26,7 +26,8 @@ NemoEntityConflictError, NemoEntityNotFoundError, ) -from nemo_platform_plugin.jobs.api_factory import PlatformJobSpec +from nemo_platform_plugin.jobs.client import AsyncJobsClient +from nemo_platform_plugin.jobs.types import CreatePlatformJobRequest, ListJobsQueryParams, PlatformJobSpec from nemo_platform_plugin.sdk_provider import get_async_platform_sdk logger = logging.getLogger(__name__) @@ -66,6 +67,7 @@ class InsightsAnalysisController(NemoController): def __init__(self) -> None: self._sdk: AsyncNeMoPlatform | None = None self._entities: NemoEntitiesClient | None = None + self._jobs: AsyncJobsClient | None = None self._config: InsightsConfig | None = None @property @@ -76,6 +78,10 @@ def sdk(self) -> AsyncNeMoPlatform: def entities(self) -> NemoEntitiesClient: return _require(self._entities, "entities") + @property + def jobs(self) -> AsyncJobsClient: + return _require(self._jobs, "jobs") + @property def insights_config(self) -> InsightsConfig: return _require(self._config, "insights_config") @@ -91,6 +97,7 @@ async def on_startup(self) -> None: self._config = get_nemo_config(InsightsConfig) self._sdk = get_async_platform_sdk(as_service="insights", internal=True) self._entities = NemoEntitiesClient(client_from_platform(self._sdk, AsyncEntitiesClient)) + self._jobs = client_from_platform(self._sdk, AsyncJobsClient) logger.info("InsightsAnalysisController started.") async def on_shutdown(self) -> None: @@ -193,11 +200,13 @@ async def _has_enough_new_traces(self, config: AnalysisConfig, status: AnalysisR async def _has_active_job(self, config: AnalysisConfig) -> bool: try: - jobs = self.sdk.jobs.list( + jobs = await self.jobs.list_jobs( workspace=config.workspace, - filter=cast(Any, {"source": "insights", "status": _ACTIVE_JOB_STATUSES}), - page_size=100, - sort="-created_at", + query_params=ListJobsQueryParams( + filter=cast(Any, {"source": "insights", "status": _ACTIVE_JOB_STATUSES}), + page_size=100, + sort="-created_at", + ), ) except Exception: logger.debug( @@ -208,8 +217,8 @@ async def _has_active_job(self, config: AnalysisConfig) -> bool: return True try: - async for job in jobs: - if _job_targets_agent(job, config.agent): + async for item in jobs.items(): + if _job_targets_agent(item, config.agent): return True except Exception: logger.debug( @@ -251,14 +260,18 @@ async def _submit_analysis_job( spec=spec, job_name=job_name, ) - await self.sdk.jobs.create( - workspace=config.workspace, - source="insights", - name=job_name, - spec=spec.model_dump(mode="json"), - platform_spec=platform_spec, - custom_fields={"insights_analysis_agent": config.agent}, - ) + ( + await self.jobs.create_job( + workspace=config.workspace, + body=CreatePlatformJobRequest( + source="insights", + name=job_name, + spec=spec.model_dump(mode="json"), + platform_spec=platform_spec, + custom_fields={"insights_analysis_agent": config.agent}, + ), + ) + ).data() logger.info( "Submitted insights analysis job '%s' for agent '%s' in workspace '%s'", job_name, @@ -267,13 +280,16 @@ async def _submit_analysis_job( ) async def _compile_job_spec(self, *, workspace: str, spec: AnalyzeSpec, job_name: str) -> PlatformJobSpec: - return await AnalyzeJob.compile( - workspace=workspace, - spec=spec, - entity_client=self.entities, - job_name=job_name, - async_sdk=self.sdk, - profile=self.insights_config.analyst.job_profile, + return cast( + PlatformJobSpec, + await AnalyzeJob.compile( + workspace=workspace, + spec=spec, + entity_client=self.entities, + job_name=job_name, + async_sdk=self.sdk, + profile=self.insights_config.analyst.job_profile, + ), ) diff --git a/plugins/nemo-insights/tests/test_periodic_analysis.py b/plugins/nemo-insights/tests/test_periodic_analysis.py index 33fd353c97..df89717d32 100644 --- a/plugins/nemo-insights/tests/test_periodic_analysis.py +++ b/plugins/nemo-insights/tests/test_periodic_analysis.py @@ -6,7 +6,7 @@ from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace -from typing import cast +from typing import Any, cast from zoneinfo import ZoneInfo import httpx @@ -42,6 +42,7 @@ from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityNotFoundError from nemo_platform_plugin.job_context import JobContext, StoragePaths from nemo_platform_plugin.job_results import JobResults +from nemo_platform_plugin.jobs.client import AsyncJobsClient from nemo_platform_plugin.jobs.constants import ( DEFAULT_JOB_STORAGE_PATH, PERSISTENT_JOB_STORAGE_PATH_ENVVAR, @@ -753,26 +754,27 @@ async def test_analyze_job_compile_requests_storage_without_provider_credentials class _AsyncJobList: def __init__(self, jobs: list[SimpleNamespace]) -> None: - self.jobs = jobs + self._jobs = jobs - def __aiter__(self): - return self._iter() - - async def _iter(self): - for job in self.jobs: + async def items(self): + for job in self._jobs: yield job class _AsyncJobs: + """Duck-typed AsyncJobsClient: records created jobs, serves list_jobs items.""" + def __init__(self, jobs: list[SimpleNamespace] | None = None) -> None: - self.created: list[dict[str, object]] = [] + self.created: list[Any] = [] self.jobs = list(jobs or []) - async def create(self, **kwargs: object) -> SimpleNamespace: - self.created.append(kwargs) - return SimpleNamespace(name=kwargs.get("name")) + async def create_job(self, *, workspace: str, body: Any) -> SimpleNamespace: + del workspace + self.created.append(body) + return SimpleNamespace(data=lambda: SimpleNamespace(name=body.name)) - def list(self, **_: object) -> _AsyncJobList: + async def list_jobs(self, *, workspace: str, query_params: Any = None) -> _AsyncJobList: + del workspace, query_params return _AsyncJobList(self.jobs) @@ -813,6 +815,7 @@ def _controller( sdk = _AsyncSdk(jobs=jobs) entities = _Entities(run_status=run_status) controller._sdk = cast(AsyncNeMoPlatform, sdk) + controller._jobs = cast(AsyncJobsClient, sdk.jobs) controller._entities = cast(NemoEntitiesClient, entities) return controller, sdk, entities @@ -843,15 +846,15 @@ async def test_controller_submits_due_job(monkeypatch: pytest.MonkeyPatch) -> No ) async def fake_compile_job_spec(**_: object) -> dict[str, list[object]]: - return {"steps": []} + return {"steps": [{"name": "step-one", "executor": {"provider": "cpu", "container": {"image": "x"}}}]} monkeypatch.setattr(controller, "_compile_job_spec", fake_compile_job_spec) await controller._reconcile_config(config) assert len(sdk.jobs.created) == 1 created = sdk.jobs.created[0] - created_spec = cast(dict[str, object], created["spec"]) - assert created["source"] == "insights" + created_spec = cast(dict[str, object], created.spec) + assert created.source == "insights" assert created_spec["agent"] == "research-agent" assert created_spec["since"] is None assert created_spec["default_model"] == "default/gpt-5" @@ -891,7 +894,7 @@ async def test_controller_skips_active_job(monkeypatch: pytest.MonkeyPatch) -> N ) async def fake_compile_job_spec(**_: object) -> dict[str, list[object]]: - return {"steps": []} + return {"steps": [{"name": "step-one", "executor": {"provider": "cpu", "container": {"image": "x"}}}]} monkeypatch.setattr(controller, "_compile_job_spec", fake_compile_job_spec) await controller._reconcile_config(config) diff --git a/plugins/nemo-iron-swarm/src/nemo_iron_swarm_plugin/jobs/hitl.py b/plugins/nemo-iron-swarm/src/nemo_iron_swarm_plugin/jobs/hitl.py index 5a0d8c76e5..c7ab4a1412 100644 --- a/plugins/nemo-iron-swarm/src/nemo_iron_swarm_plugin/jobs/hitl.py +++ b/plugins/nemo-iron-swarm/src/nemo_iron_swarm_plugin/jobs/hitl.py @@ -23,6 +23,9 @@ IronSwarmRunError, ) from nemo_iron_swarm_plugin.jobs.synth_client import SynthClient +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient +from nemo_platform_plugin.jobs.types import JobStatusDetailsUpdate logger = logging.getLogger(__name__) @@ -85,7 +88,9 @@ def publish(self, kind: str, payload: dict[str, Any]) -> None: # the interview until the poll deadline with no explanation). for attempt in range(1, _PUBLISH_MAX_ATTEMPTS + 1): try: - self._sdk.jobs.update_status_details(self._name, workspace=self._workspace, body=body) + client_from_platform(self._sdk, JobsClient).update_job_status_details( + name=self._name, workspace=self._workspace, body=JobStatusDetailsUpdate(root=body) + ) return except Exception: if attempt == _PUBLISH_MAX_ATTEMPTS: @@ -101,7 +106,11 @@ def await_response(self, kind: str) -> list[dict[str, Any]]: deadline = time.monotonic() + self._timeout while time.monotonic() < deadline: try: - job = self._sdk.jobs.retrieve(self._name, workspace=self._workspace) + job = ( + client_from_platform(self._sdk, JobsClient) + .get_job(name=self._name, workspace=self._workspace) + .data() + ) except Exception: # a transient poll failure must not abort a minutes-long human wait logger.warning("status_details poll failed for job %s; retrying", self._name, exc_info=True) time.sleep(self._poll_interval) diff --git a/plugins/nemo-iron-swarm/tests/unit/test_synth_hitl.py b/plugins/nemo-iron-swarm/tests/unit/test_synth_hitl.py index fe4acedab2..09070dc6f1 100644 --- a/plugins/nemo-iron-swarm/tests/unit/test_synth_hitl.py +++ b/plugins/nemo-iron-swarm/tests/unit/test_synth_hitl.py @@ -5,6 +5,7 @@ from __future__ import annotations +import json from types import SimpleNamespace from typing import Any, cast @@ -71,18 +72,44 @@ def test_drive_synth_hitl_relays_interview_then_review() -> None: def test_status_details_channel_publishes_and_matches_round() -> None: published: dict[str, Any] = {} - class _Jobs: - def update_status_details(self, _name: str, *, workspace: str, body: dict[str, Any]) -> None: - published.update(body) + job = { + "id": "job1", + "attempt_id": "att-1", + "name": "job1", + "workspace": "default", + "source": "test", + "spec": {}, + "platform_spec": { + "steps": [{"name": "step-one", "executor": {"provider": "cpu", "container": {"image": "x"}}}] + }, + "fileset": "fs-1", + "status": "active", + "status_details": {"interview_response": {"round": 1, "answers": [{"gap": "g", "answer": "a"}]}}, + } - def retrieve(self, _name: str, *, workspace: str) -> Any: - return SimpleNamespace( - status_details={"interview_response": {"round": 1, "answers": [{"gap": "g", "answer": "a"}]}} - ) - - channel = hitl.StatusDetailsChannel( - SimpleNamespace(jobs=_Jobs()), name="job1", workspace="default", poll_interval=0.0 + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "PATCH" and request.url.path.endswith("/status-details"): + if request.content: + published.update(json.loads(request.content)) + return httpx.Response(200, json={}) + if request.method == "GET" and request.url.path.endswith("/jobs/job1"): + return httpx.Response(200, json=job) + return httpx.Response(404) + + # The channel wraps the SDK via ``client_from_platform`` into a typed JobsClient; + # model that with a real JobsClient over a mocked transport. + http_client = httpx.Client(transport=httpx.MockTransport(handler), base_url="http://platform:8080") + platform = SimpleNamespace( + base_url="http://platform:8080", + workspace="default", + _custom_headers={"Authorization": "Bearer x"}, + _client=http_client, + timeout=None, + max_retries=2, + _prepare_url=lambda url: url, ) + + channel = hitl.StatusDetailsChannel(platform, name="job1", workspace="default", poll_interval=0.0) channel.publish("interview", {"questions": [{"gap": "g"}]}) assert published["interview"]["round"] == 1 assert channel.await_response("interview") == [{"gap": "g", "answer": "a"}] diff --git a/services/core/jobs/tests/integration/test_jobs_auth_propagation.py b/services/core/jobs/tests/integration/test_jobs_auth_propagation.py index 1efbc6cab7..1cd5b885f9 100644 --- a/services/core/jobs/tests/integration/test_jobs_auth_propagation.py +++ b/services/core/jobs/tests/integration/test_jobs_auth_propagation.py @@ -18,6 +18,9 @@ import pytest from nemo_platform import NeMoPlatform +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient +from nemo_platform_plugin.jobs.types import CreatePlatformJobRequest from nmp.core.files.service import FilesService from nmp.core.jobs.service import JobsService from nmp.testing import as_user, create_test_client, short_unique_name, unique_email @@ -51,30 +54,33 @@ def test_auth_context_stripped_for_regular_user(self, sdk: NeMoPlatform): creator_sdk = as_user(sdk, creator_email, groups=["team-alpha"]) - creator_sdk.jobs.create( + jobs = client_from_platform(creator_sdk, JobsClient) + jobs.create_job( workspace=workspace, - name=job_name, - source="auth-propagation-test", - spec={}, - platform_spec={ - "steps": [ - { - "name": "test-step", - "executor": { - "provider": "cpu", - "profile": "default", - "container": { - "image": "busybox:latest", - "entrypoint": ["entrypoint"], - "command": ["command"], + body=CreatePlatformJobRequest( + name=job_name, + source="auth-propagation-test", + spec={}, + platform_spec={ + "steps": [ + { + "name": "test-step", + "executor": { + "provider": "cpu", + "profile": "default", + "container": { + "image": "busybox:latest", + "entrypoint": ["entrypoint"], + "command": ["command"], + }, }, }, - }, - ] - }, + ] + }, + ), ) - steps = list(creator_sdk.jobs.steps.list(job_name, workspace=workspace)) + steps = list(jobs.list_steps(name=job_name, workspace=workspace).items()) assert len(steps) == 1 assert steps[0].auth_context is None, "Regular user should not see auth_context" @@ -87,31 +93,36 @@ def test_auth_context_visible_to_service_principal(self, sdk: NeMoPlatform): creator_sdk = as_user(sdk, creator_email, groups=creator_groups) - creator_sdk.jobs.create( + jobs = client_from_platform(creator_sdk, JobsClient) + jobs.create_job( workspace=workspace, - name=job_name, - source="auth-propagation-test", - spec={}, - platform_spec={ - "steps": [ - { - "name": "test-step", - "executor": { - "provider": "cpu", - "profile": "default", - "container": { - "image": "busybox:latest", - "entrypoint": ["entrypoint"], - "command": ["command"], + body=CreatePlatformJobRequest( + name=job_name, + source="auth-propagation-test", + spec={}, + platform_spec={ + "steps": [ + { + "name": "test-step", + "executor": { + "provider": "cpu", + "profile": "default", + "container": { + "image": "busybox:latest", + "entrypoint": ["entrypoint"], + "command": ["command"], + }, }, }, - }, - ] - }, + ] + }, + ), ) service_sdk = _as_service_principal(sdk) - steps = list(service_sdk.jobs.steps.list(job_name, workspace=workspace)) + steps = list( + client_from_platform(service_sdk, JobsClient).list_steps(name=job_name, workspace=workspace).items() + ) assert len(steps) == 1 step = steps[0] diff --git a/services/core/jobs/tests/integration/test_jobs_secrets_access.py b/services/core/jobs/tests/integration/test_jobs_secrets_access.py index a151106a8e..e1252280d4 100644 --- a/services/core/jobs/tests/integration/test_jobs_secrets_access.py +++ b/services/core/jobs/tests/integration/test_jobs_secrets_access.py @@ -18,6 +18,8 @@ import pytest from nemo_platform import NeMoPlatform from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient +from nemo_platform_plugin.jobs.types import CreatePlatformJobRequest from nemo_platform_plugin.secrets.client import SecretsClient from nemo_platform_plugin.secrets.types import PlatformSecretCreateRequest from nmp.core.files.service import FilesService @@ -93,12 +95,18 @@ def test_create_job_with_secret_user_has_access_succeeds(self, sdk: NeMoPlatform ) user_sdk = as_user(sdk, user_email) - job = user_sdk.jobs.create( - workspace=workspace, - name=job_name, - source="integration-test", - spec={}, - platform_spec=_platform_spec_with_secret(secret_name), + job = ( + client_from_platform(user_sdk, JobsClient) + .create_job( + workspace=workspace, + body=CreatePlatformJobRequest( + name=job_name, + source="integration-test", + spec={}, + platform_spec=_platform_spec_with_secret(secret_name), + ), + ) + .data() ) assert job.id is not None @@ -133,12 +141,14 @@ def test_create_job_with_secret_user_lacks_access_fails(self, sdk: NeMoPlatform) secret_ref = f"{workspace_other}/{secret_name}" with pytest.raises(Exception) as exc_info: - user_sdk.jobs.create( + client_from_platform(user_sdk, JobsClient).create_job( workspace=workspace_own, - name=job_name, - source="integration-test", - spec={}, - platform_spec=_platform_spec_with_secret(secret_ref), + body=CreatePlatformJobRequest( + name=job_name, + source="integration-test", + spec={}, + platform_spec=_platform_spec_with_secret(secret_ref), + ), ) msg = str(exc_info.value).lower() @@ -160,12 +170,14 @@ def test_create_job_with_nonexistent_secret_fails(self, sdk: NeMoPlatform): user_sdk = as_user(sdk, user_email) with pytest.raises(Exception) as exc_info: - user_sdk.jobs.create( + client_from_platform(user_sdk, JobsClient).create_job( workspace=workspace, - name=job_name, - source="integration-test", - spec={}, - platform_spec=_platform_spec_with_secret("nonexistent-secret-name"), + body=CreatePlatformJobRequest( + name=job_name, + source="integration-test", + spec={}, + platform_spec=_platform_spec_with_secret("nonexistent-secret-name"), + ), ) msg = str(exc_info.value).lower() diff --git a/tests/agentic-use/jobs-execute-gpu-cli/tests/test_outputs.py b/tests/agentic-use/jobs-execute-gpu-cli/tests/test_outputs.py index d3a2465cca..e3b0ea495a 100644 --- a/tests/agentic-use/jobs-execute-gpu-cli/tests/test_outputs.py +++ b/tests/agentic-use/jobs-execute-gpu-cli/tests/test_outputs.py @@ -12,7 +12,7 @@ import time import pytest -from nemo_platform import NeMoPlatform +from nemo_platform_plugin.jobs.client import JobsClient WORKSPACE = "gpu-job-workspace" @@ -31,20 +31,20 @@ def _make_unsigned_jwt() -> str: @pytest.fixture -def client() -> NeMoPlatform: +def client() -> JobsClient: nmp_base_url = os.environ.get("NMP_BASE_URL", "http://localhost:8080") - return NeMoPlatform( + return JobsClient( base_url=nmp_base_url, workspace=WORKSPACE, - access_token=_make_unsigned_jwt(), + auth=_make_unsigned_jwt(), ) -def _wait_for_terminal(client: NeMoPlatform, job_name: str, max_wait: int = 60) -> str: +def _wait_for_terminal(client: JobsClient, job_name: str, max_wait: int = 60) -> str: """Wait for a job to reach a terminal status.""" for _ in range(max_wait // 5): try: - resp = client.jobs.get_status(name=job_name, workspace=WORKSPACE) + resp = client.get_job_status(name=job_name, workspace=WORKSPACE) status = resp.status if hasattr(resp, "status") else str(resp) if status in ("completed", "error", "cancelled"): return status @@ -52,16 +52,16 @@ def _wait_for_terminal(client: NeMoPlatform, job_name: str, max_wait: int = 60) pass time.sleep(5) try: - resp = client.jobs.get_status(name=job_name, workspace=WORKSPACE) + resp = client.get_job_status(name=job_name, workspace=WORKSPACE) return resp.status if hasattr(resp, "status") else str(resp) except Exception: return "unknown" -def _find_job_by_name(client: NeMoPlatform, name: str): +def _find_job_by_name(client: JobsClient, name: str): """Find a specific job by name.""" - jobs = client.jobs.list(workspace=WORKSPACE) - for job in jobs.data: + jobs = client.list_jobs(workspace=WORKSPACE).items() + for job in jobs: if job.name == name: return job return None @@ -70,37 +70,37 @@ def _find_job_by_name(client: NeMoPlatform, name: str): # --- Job existence and completion checks --- -def test_multiple_jobs_created(client: NeMoPlatform) -> None: +def test_multiple_jobs_created(client: JobsClient) -> None: """Verify that at least 3 jobs were created.""" - jobs = client.jobs.list(workspace=WORKSPACE) - assert len(jobs.data) >= 3, f"Expected at least 3 jobs, found {len(jobs.data)}: {[j.name for j in jobs.data]}" + jobs = list(client.list_jobs(workspace=WORKSPACE).items()) + assert len(jobs) >= 3, f"Expected at least 3 jobs, found {len(jobs)}: {[j.name for j in jobs]}" -def test_gpu_verify_job_completed(client: NeMoPlatform) -> None: +def test_gpu_verify_job_completed(client: JobsClient) -> None: """Verify gpu-verify-job reached completed status (nvidia-smi ran on GPU).""" job = _find_job_by_name(client, "gpu-verify-job") assert job is not None, ( - f"Job 'gpu-verify-job' not found. Jobs: {[j.name for j in client.jobs.list(workspace=WORKSPACE).data]}" + f"Job 'gpu-verify-job' not found. Jobs: {[j.name for j in client.list_jobs(workspace=WORKSPACE).items()]}" ) status = _wait_for_terminal(client, "gpu-verify-job") assert status == "completed", f"Job 'gpu-verify-job' has status '{status}', expected 'completed'." -def test_gpu_compute_job_completed(client: NeMoPlatform) -> None: +def test_gpu_compute_job_completed(client: JobsClient) -> None: """Verify gpu-compute-job reached completed status.""" job = _find_job_by_name(client, "gpu-compute-job") assert job is not None, ( - f"Job 'gpu-compute-job' not found. Jobs: {[j.name for j in client.jobs.list(workspace=WORKSPACE).data]}" + f"Job 'gpu-compute-job' not found. Jobs: {[j.name for j in client.list_jobs(workspace=WORKSPACE).items()]}" ) status = _wait_for_terminal(client, "gpu-compute-job") assert status == "completed", f"Job 'gpu-compute-job' has status '{status}', expected 'completed'." -def test_gpu_fail_job_errored(client: NeMoPlatform) -> None: +def test_gpu_fail_job_errored(client: JobsClient) -> None: """Verify gpu-fail-job reached error status (exit code 1).""" job = _find_job_by_name(client, "gpu-fail-job") assert job is not None, ( - f"Job 'gpu-fail-job' not found. Jobs: {[j.name for j in client.jobs.list(workspace=WORKSPACE).data]}" + f"Job 'gpu-fail-job' not found. Jobs: {[j.name for j in client.list_jobs(workspace=WORKSPACE).items()]}" ) status = _wait_for_terminal(client, "gpu-fail-job") assert status == "error", f"Job 'gpu-fail-job' has status '{status}', expected 'error'." diff --git a/tests/agentic-use/jobs-multistep-cpu-cli/tests/test_outputs.py b/tests/agentic-use/jobs-multistep-cpu-cli/tests/test_outputs.py index 5935cb6881..b0e6f05497 100644 --- a/tests/agentic-use/jobs-multistep-cpu-cli/tests/test_outputs.py +++ b/tests/agentic-use/jobs-multistep-cpu-cli/tests/test_outputs.py @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 """Verify that the agent created multiple jobs, handled failure, and recovered. @@ -20,7 +20,7 @@ import time import pytest -from nemo_platform import NeMoPlatform +from nemo_platform_plugin.jobs.client import JobsClient from trace_reader import get_session WORKSPACE = "job-test-workspace" @@ -40,20 +40,20 @@ def _make_unsigned_jwt() -> str: @pytest.fixture -def client() -> NeMoPlatform: +def client() -> JobsClient: nmp_base_url = os.environ.get("NMP_BASE_URL", "http://localhost:8080") - return NeMoPlatform( + return JobsClient( base_url=nmp_base_url, workspace=WORKSPACE, - access_token=_make_unsigned_jwt(), + auth=_make_unsigned_jwt(), ) -def _wait_for_terminal(client: NeMoPlatform, job_name: str, max_wait: int = 30) -> str: +def _wait_for_terminal(client: JobsClient, job_name: str, max_wait: int = 30) -> str: """Wait for a job to reach a terminal status, with a safety timeout.""" for _ in range(max_wait // 5): try: - resp = client.jobs.get_status(name=job_name, workspace=WORKSPACE) + resp = client.get_job_status(name=job_name, workspace=WORKSPACE) status = resp.status if hasattr(resp, "status") else str(resp) if status in ("completed", "error", "cancelled"): return status @@ -62,16 +62,16 @@ def _wait_for_terminal(client: NeMoPlatform, job_name: str, max_wait: int = 30) time.sleep(5) # Return whatever we got try: - resp = client.jobs.get_status(name=job_name, workspace=WORKSPACE) + resp = client.get_job_status(name=job_name, workspace=WORKSPACE) return resp.status if hasattr(resp, "status") else str(resp) except Exception: return "unknown" -def _find_job_by_name(client: NeMoPlatform, name: str): +def _find_job_by_name(client: JobsClient, name: str): """Find a specific job by name.""" - jobs = client.jobs.list(workspace=WORKSPACE) - for job in jobs.data: + jobs = client.list_jobs(workspace=WORKSPACE).items() + for job in jobs: if job.name == name: return job return None @@ -80,30 +80,30 @@ def _find_job_by_name(client: NeMoPlatform, name: str): # --- Job existence checks --- -def test_multiple_jobs_created(client: NeMoPlatform) -> None: +def test_multiple_jobs_created(client: JobsClient) -> None: """Verify that at least 3 jobs were created.""" - jobs = client.jobs.list(workspace=WORKSPACE) - assert len(jobs.data) >= 3, ( - f"Expected at least 3 jobs in workspace '{WORKSPACE}', found {len(jobs.data)}: {[j.name for j in jobs.data]}" + jobs = list(client.list_jobs(workspace=WORKSPACE).items()) + assert len(jobs) >= 3, ( + f"Expected at least 3 jobs in workspace '{WORKSPACE}', found {len(jobs)}: {[j.name for j in jobs]}" ) -def test_success_job_completed(client: NeMoPlatform) -> None: +def test_success_job_completed(client: JobsClient) -> None: """Verify success-job reached completed status.""" job = _find_job_by_name(client, "success-job") assert job is not None, ( "Job 'success-job' not found. " - f"Jobs in workspace: {[j.name for j in client.jobs.list(workspace=WORKSPACE).data]}" + f"Jobs in workspace: {[j.name for j in client.list_jobs(workspace=WORKSPACE).items()]}" ) status = _wait_for_terminal(client, "success-job") assert status == "completed", f"Job 'success-job' has status '{status}', expected 'completed'." -def test_fail_job_errored(client: NeMoPlatform) -> None: +def test_fail_job_errored(client: JobsClient) -> None: """Verify fail-job reached error status (exit code 1).""" job = _find_job_by_name(client, "fail-job") assert job is not None, ( - f"Job 'fail-job' not found. Jobs in workspace: {[j.name for j in client.jobs.list(workspace=WORKSPACE).data]}" + f"Job 'fail-job' not found. Jobs in workspace: {[j.name for j in client.list_jobs(workspace=WORKSPACE).items()]}" ) status = _wait_for_terminal(client, "fail-job") assert status == "error", ( @@ -111,12 +111,12 @@ def test_fail_job_errored(client: NeMoPlatform) -> None: ) -def test_recovery_job_completed(client: NeMoPlatform) -> None: +def test_recovery_job_completed(client: JobsClient) -> None: """Verify recovery-job reached completed status after the failure.""" job = _find_job_by_name(client, "recovery-job") assert job is not None, ( "Job 'recovery-job' not found. " - f"Jobs in workspace: {[j.name for j in client.jobs.list(workspace=WORKSPACE).data]}" + f"Jobs in workspace: {[j.name for j in client.list_jobs(workspace=WORKSPACE).items()]}" ) status = _wait_for_terminal(client, "recovery-job") assert status == "completed", f"Job 'recovery-job' has status '{status}', expected 'completed'." diff --git a/tests/auth/integration/jobs_auth_helpers.py b/tests/auth/integration/jobs_auth_helpers.py index 09188f4de4..40fecf760f 100644 --- a/tests/auth/integration/jobs_auth_helpers.py +++ b/tests/auth/integration/jobs_auth_helpers.py @@ -3,6 +3,7 @@ from collections.abc import Iterator from contextlib import contextmanager +from typing import Any from nemo_platform import NeMoPlatform @@ -16,5 +17,5 @@ def managed_admin_workspace(admin_sdk: NeMoPlatform, workspace_name: str) -> Ite admin_sdk.workspaces.delete(workspace_name) -def job_exists_in_pages(jobs_page: object, job_name: str) -> bool: - return any(item.name == job_name for page in jobs_page.iter_pages() for item in page.data) +def job_exists_in_pages(items: Iterator[Any], job_name: str) -> bool: + return any(item.name == job_name for item in items) diff --git a/tests/auth/integration/test_jobs_auth.py b/tests/auth/integration/test_jobs_auth.py index cae83ca9c7..62e2a466a9 100644 --- a/tests/auth/integration/test_jobs_auth.py +++ b/tests/auth/integration/test_jobs_auth.py @@ -25,6 +25,8 @@ PlatformJobSpec, PlatformJobStep, ) +from nemo_platform_plugin.jobs.client import JobsClient +from nemo_platform_plugin.jobs.types import CreatePlatformJobRequest from nmp.common.entities import ALL_WORKSPACES from nmp.core.jobs.controllers.diagnostics import collect_job_diagnostics from nmp.testing import TEST_ADMIN_EMAIL, grant_workspace_role, short_unique_name, unique_email @@ -96,26 +98,32 @@ def test_job_principal_propagation(services_pool_sdk: NeMoPlatform): grant_workspace_role(admin_sdk, workspace=workspace_name, principal=user_email, roles=["Editor"]) user_sdk = _as_bearer_user(services_pool_sdk, user_email, principal_id=_oidc_subject()) - job = user_sdk.jobs.create( - workspace=workspace_name, - source=JOB_SOURCE, - spec={"test": "auth-propagation"}, - platform_spec=PlatformJobSpec( - steps=[ - PlatformJobStep( - name="auth-test-step", - executor=CPUExecutionProviderSpec( - provider="cpu", - container=ContainerSpec( - entrypoint=["nemo-platform"], - command=["run", "task", "--task", "nmp.hello_world.tasks.hello_world"], - ), - ), - environment=[EnvironmentVariable(name="BUSY_LOOP_DURATION_SECONDS", value="0")], - config={"message": "auth propagation test"}, - ) - ] - ), + job = ( + client_from_platform(user_sdk, JobsClient) + .create_job( + workspace=workspace_name, + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "auth-propagation"}, + platform_spec=PlatformJobSpec( + steps=[ + PlatformJobStep( + name="auth-test-step", + executor=CPUExecutionProviderSpec( + provider="cpu", + container=ContainerSpec( + entrypoint=["nemo-platform"], + command=["run", "task", "--task", "nmp.hello_world.tasks.hello_world"], + ), + ), + environment=[EnvironmentVariable(name="BUSY_LOOP_DURATION_SECONDS", value="0")], + config={"message": "auth propagation test"}, + ) + ] + ), + ), + ) + .data() ) completed_job = wait_for_platform_job(user_sdk, job.name, workspace_name) @@ -155,28 +163,34 @@ def test_job_cannot_access_unauthorized_workspace(services_pool_sdk: NeMoPlatfor files = client_from_platform(owner_sdk, FilesClient) files.create_fileset(workspace=restricted_workspace, body=CreateFilesetRequest(name=fileset_name)) - job = other_sdk.jobs.create( - workspace=runner_workspace, - source=JOB_SOURCE, - spec={"test": "auth-denial"}, - platform_spec=PlatformJobSpec( - steps=[ - PlatformJobStep( - name="access-test-step", - executor=CPUExecutionProviderSpec( - provider="cpu", - container=ContainerSpec( - entrypoint=["nemo-platform"], - command=["run", "task", "--task", "nmp.hello_world.tasks.access_fileset"], - ), - ), - config={ - "workspace": restricted_workspace, - "fileset": fileset_name, - }, - ) - ] - ), + job = ( + client_from_platform(other_sdk, JobsClient) + .create_job( + workspace=runner_workspace, + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "auth-denial"}, + platform_spec=PlatformJobSpec( + steps=[ + PlatformJobStep( + name="access-test-step", + executor=CPUExecutionProviderSpec( + provider="cpu", + container=ContainerSpec( + entrypoint=["nemo-platform"], + command=["run", "task", "--task", "nmp.hello_world.tasks.access_fileset"], + ), + ), + config={ + "workspace": restricted_workspace, + "fileset": fileset_name, + }, + ) + ] + ), + ), + ) + .data() ) completed_job = wait_for_platform_job(other_sdk, job.name, runner_workspace) @@ -190,7 +204,11 @@ def test_job_cannot_access_unauthorized_workspace(services_pool_sdk: NeMoPlatfor ) assert completed_job.status == "error" - tasks_response = other_sdk.jobs.tasks.list("access-test-step", job=job.name, workspace=runner_workspace) + tasks_response = ( + client_from_platform(other_sdk, JobsClient) + .list_job_step_tasks(name="access-test-step", job=job.name, workspace=runner_workspace) + .data() + ) if not tasks_response.data: _log_auth_job_diagnostics( other_sdk, @@ -222,28 +240,33 @@ def test_job_admin_can_list_jobs_in_all_workspaces(services_pool_sdk: NeMoPlatfo grant_workspace_role(admin_sdk, workspace=workspace_name, principal=user_email, roles=["Editor"]) user_sdk = _as_bearer_user(services_pool_sdk, user_email, principal_id=_oidc_subject()) - job = user_sdk.jobs.create( - workspace=workspace_name, - source=JOB_SOURCE, - spec={"test": "admin-list"}, - platform_spec=PlatformJobSpec( - steps=[ - PlatformJobStep( - name="admin-list-step", - executor=CPUExecutionProviderSpec( - provider="cpu", - container=ContainerSpec( - command=["echo", "admin list jobs"], - ), - ), - ) - ] - ), + job = ( + client_from_platform(user_sdk, JobsClient) + .create_job( + workspace=workspace_name, + body=CreatePlatformJobRequest( + source=JOB_SOURCE, + spec={"test": "admin-list"}, + platform_spec=PlatformJobSpec( + steps=[ + PlatformJobStep( + name="admin-list-step", + executor=CPUExecutionProviderSpec( + provider="cpu", + container=ContainerSpec( + command=["echo", "admin list jobs"], + ), + ), + ) + ] + ), + ), + ) + .data() ) completed_job = wait_for_platform_job(user_sdk, job.name, workspace_name) assert completed_job.status == "completed" - jobs = admin_sdk.jobs.list(workspace=ALL_WORKSPACES) - assert jobs.pagination is not None - assert job_exists_in_pages(jobs, job.name) + jobs = client_from_platform(admin_sdk, JobsClient).list_jobs(workspace=ALL_WORKSPACES) + assert job_exists_in_pages(jobs.items(), job.name) diff --git a/tests/auth/test_jobs_auth_helpers.py b/tests/auth/test_jobs_auth_helpers.py index 18ef7d109e..1204feec7d 100644 --- a/tests/auth/test_jobs_auth_helpers.py +++ b/tests/auth/test_jobs_auth_helpers.py @@ -1,6 +1,8 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +from collections.abc import Iterator + import pytest from tests.auth.integration.jobs_auth_helpers import job_exists_in_pages, managed_admin_workspace @@ -28,14 +30,13 @@ def __init__(self, name: str) -> None: self.name = name -class _StubPage: - def __init__(self, pages: list["_StubPage"], job_names: list[str]) -> None: - self._pages = pages - self.data = [_StubJob(name) for name in job_names] - self.pagination = object() +class _StubJobsResponse: + def __init__(self, job_names: list[str]) -> None: + self._job_names = job_names - def iter_pages(self): - yield from self._pages + def items(self) -> Iterator[_StubJob]: + for name in self._job_names: + yield _StubJob(name) def test_managed_admin_workspace_deletes_workspace_after_success() -> None: @@ -60,8 +61,6 @@ def test_managed_admin_workspace_deletes_workspace_after_failure() -> None: def test_job_exists_in_pages_checks_later_pages() -> None: - page_two = _StubPage([], ["target-job"]) - page_one = _StubPage([], ["other-job"]) - page_one._pages = [page_one, page_two] + jobs_response = _StubJobsResponse(["other-job", "target-job"]) - assert job_exists_in_pages(page_one, "target-job") is True + assert job_exists_in_pages(jobs_response.items(), "target-job") is True diff --git a/tests/auth_idp/contracts/test_jobs.py b/tests/auth_idp/contracts/test_jobs.py index 169579f1bd..2752d64473 100644 --- a/tests/auth_idp/contracts/test_jobs.py +++ b/tests/auth_idp/contracts/test_jobs.py @@ -2,6 +2,9 @@ # SPDX-License-Identifier: Apache-2.0 import pytest +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient +from nemo_platform_plugin.jobs.types import CreatePlatformJobRequest from nmp.testing import grant_workspace_role from nmp.testing.e2e import wait_for_job_logs, wait_for_platform_job @@ -31,34 +34,40 @@ def test_provider_workload_job_runs_via_workload_profile( roles=["Viewer", "JobRunner"], ) - job = e2e_setup_sdk.jobs.create( - workspace=auth_idp_workspace, - source=f"{auth_idp_case.id}-workload-job", - spec={"test": "workload-job"}, - platform_spec={ - "steps": [ - { - "name": "workload-workspace-get", - "executor": { - "provider": "cpu", - "profile": "workload", - "container": { - "image": nmp_api_image(), - "entrypoint": ["nemo-platform"], - "command": [ - "run", - "task", - "--task", - "nmp.hello_world.tasks.workload_workspace_get", - ], - }, - }, - "config": { - "workspace": auth_idp_workspace, - }, - } - ] - }, + job = ( + client_from_platform(e2e_setup_sdk, JobsClient) + .create_job( + workspace=auth_idp_workspace, + body=CreatePlatformJobRequest( + source=f"{auth_idp_case.id}-workload-job", + spec={"test": "workload-job"}, + platform_spec={ + "steps": [ + { + "name": "workload-workspace-get", + "executor": { + "provider": "cpu", + "profile": "workload", + "container": { + "image": nmp_api_image(), + "entrypoint": ["nemo-platform"], + "command": [ + "run", + "task", + "--task", + "nmp.hello_world.tasks.workload_workspace_get", + ], + }, + }, + "config": { + "workspace": auth_idp_workspace, + }, + } + ] + }, + ), + ) + .data() ) completed_job = wait_for_platform_job(e2e_setup_sdk, job.name, auth_idp_workspace, timeout=240)