chore(spark): expand unit test coverage for SparkClient job methods by Neha-Nimsatkar · Pull Request #750 · kubeflow/sdk

Neha-Nimsatkar · 2026-08-10T15:22:11Z

What this PR does / why we need it

Adds proper unit test coverage for the SparkClient job management methods in kubeflow/spark/api/spark_client_test.py — submit_job, get_job, list_jobs, delete_job, wait_for_job_status, and get_job_logs.

Tests now follow the same @pytest.mark.parametrize + TestCase dataclass pattern already used in backend_test.py, so we get consistent coverage of backend delegation and input validation across the client.

Motivation

A few of the public SparkClient methods didn't have solid tests around argument forwarding, type contracts, or validation before hitting the backend. This was making it easy for delegation bugs to slip through unnoticed. This PR closes that gap and brings the test style in line with the rest of the codebase.

What changed

General: Rewrote the test suites to use @pytest.mark.parametrize with the TestCase dataclass — prints test names and checks against expected_output.
test_submit_job: Combined validation and submission scenarios into one parametrized test. Added checks that num_executors, resources_per_executor, and spark_conf get forwarded correctly, plus backend.submit_job.assert_not_called() for invalid input.
test_get_job: Covers job lookup delegation and the RuntimeError raised when a job isn't found.
test_list_jobs: Fixed status filtering to expect sets ({SparkJobStatus.RUNNING}, {SparkJobStatus.COMPLETED, SparkJobStatus.FAILED}), plus a failure case test.
test_delete_job: Covers deletion delegation and the nonexistent-job error path.
test_wait_for_job_status:
- Mock now returns a SparkJob instead of a bool.
- Added backend.wait_for_job_status.assert_not_called() check for validation failures.
- Extended matrix: default vs. custom timeouts, default vs. explicit polling intervals, single vs. multiple status sets, timeout edge cases (< 0, = 0), polling interval edge cases (<= 0, >= timeout).
test_get_job_logs: Added cases for default follow, plus explicit follow=True / follow=False.

Verification

uv run pytest kubeflow/spark/api/spark_client_test.py -v -s — all passing
uv run ruff format kubeflow/spark/api/spark_client_test.py
uv run ruff check --fix kubeflow/spark/api/spark_client_test.py
uv run pre-commit run --all-files
make verify

Fixes

Fixes #658

Checklist

Tested locally with pytest, verified with make verify
Formatted with Ruff, passes pre-commit
Docs — N/A, this is test-only

google-oss-prow · 2026-08-10T15:22:19Z

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign astefanutti for approval. For more information see the Kubernetes Code Review Process.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

github-actions · 2026-08-10T15:23:08Z

🎉 Welcome to the Kubeflow SDK! 🎉

Thanks for opening your first PR! We're happy to have you as part of our community 🚀

Here's what happens next:

If you haven't already, please check out our Contributing Guide for repo-specific guidelines and the Kubeflow Contributor Guide for general community standards
Our team will review your PR soon! cc @kubeflow/kubeflow-sdk-team

Join the community:

Slack: Join our #kubeflow-ml-experience and #kubeflow-trainer Slack channels
Meetings: Attend the Kubeflow SDK and ML Experience bi-weekly meetings

Feel free to ask questions in the comments if you need any help or clarification!
Thanks again for contributing to Kubeflow! 🙏

tariq-hasan · 2026-09-05T15:28:00Z

        )
+
+
+def test_get_job():


The new tests should use the @pytest.mark.parametrize marker with the TestCase dataclass from kubeflow.spark.test.common - similar to backend_test.py and utils_test.py.

sdk/kubeflow/spark/test/common.py

Lines 33 to 42 in 3febcd7

@dataclass

class TestCase:

"""Test case dataclass for parametrized tests."""

name: str

expected_status: str = SUCCESS

config: dict[str, Any] = field(default_factory=dict)

expected_output: Any | None = None

expected_error: type[Exception] | None = None

__test__ = False

Thanks for the pointer @tariq-hasan! I've refactored all new job management test methods (get_job, list_jobs, delete_job, wait_for_job_status, and get_job_logs) to use @pytest.mark.parametrize with the TestCase dataclass from kubeflow.spark.test.common.

All unit tests, ruff linting, and pre-commit checks are passing cleanly. Please have a look!

tariq-hasan · 2026-09-05T15:29:13Z

/ok-to-test
/retest

Neha-Nimsatkar · 2026-09-06T07:27:45Z

Hi @tariq-hasan, I've updated the tests to use @pytest.mark.parametrize with the TestCase dataclass pattern from kubeflow.spark.test.common, and all unit tests and pre-commit hooks are passing locally.

The E2E Test matrix hit an infrastructure disconnect across the self-hosted runners ("self-hosted runner lost communication with the server"). Could you please click "Re-run failed jobs" when you get a chance? Thanks!

tariq-hasan · 2026-09-06T20:06:53Z

/retest

tariq-hasan · 2026-09-06T20:12:20Z

/ok-to-test
/retest

tariq-hasan · 2026-09-06T21:39:20Z

/retitle chore(spark): expand unit test coverage for SparkClient job methods

tariq-hasan · 2026-09-06T21:47:05Z

@@ -173,3 +173,158 @@ def test_submit_job_success(job, options):
            spark_conf=None,
            options=options,
        )


Please combine test_submit_job_validation and test_submit_job_success into a single test_submit_job with the pytest.mark.parametrize marker and the TestCase dataclass usage.

Let's address this as well.

In addition please add a case with all of num_executors, resources_per_executor and spark_conf set, asserting each reaches the backend.

tariq-hasan · 2026-09-06T21:49:42Z

+        ),
+    ],
+)
+def test_wait_for_job_status(test_case: TestCase):


For test_wait_for_job_status please test this behavior as well.

sdk/kubeflow/spark/api/spark_client.py

Lines 285 to 288 in 3febcd7

common_utils.validate_wait_for_job_status(

polling_interval,

timeout,

)

tariq-hasan · 2026-09-06T21:52:22Z

+        TestCase(
+            name="get job by name",
+            expected_status=SUCCESS,
+            config={"job_name": "spark-job-123"},


Please ensure name and expected_status get used as well - otherwise they would be redundant. The same applies for other tests as well.

Also unhappy paths where errors are raised should be tested in addition to happy paths.

tariq-hasan · 2026-09-06T21:54:50Z

+        TestCase(
+            name="list jobs filtered by status",
+            expected_status=SUCCESS,
+            config={"status": "Running"},


We should use SparkJobStatus.RUNNING.

tariq-hasan · 2026-09-06T21:56:27Z

+        result = client.wait_for_job_status(
+            name=test_case.config["job_name"],
+            status=test_case.config["status"],
+            timeout=test_case.config["timeout"],


We're not really using polling interval here but the test case has polling interval so the test passes by accident. We need to test both default and overrides.

tariq-hasan · 2026-09-06T22:07:52Z

+            expected_status=SUCCESS,
+            config={
+                "job_name": "spark-job-123",
+                "status": "Completed",


We should use enum here - SparkJobStatus.COMPLETED.

Neha-Nimsatkar · 2026-09-07T19:29:21Z

Thanks for the detailed review, @tariq-hasan!

I have addressed all the feedback and updated the test suite:

Used SparkJobStatus enums (RUNNING and COMPLETED) instead of hardcoded strings.
Added assertions for expected_status and error handling across test cases so name and status definitions are actively validated.
Added unhappy path coverage for get_job, list_jobs, and delete_job.
Updated wait_for_job_status to explicitly test both default (2s) and overridden polling intervals, as well as parameter validation for invalid timeout values.
Verified all 21 unit tests, Ruff linting/formatting, and pre-commit checks pass locally.

tariq-hasan

Hi @Neha-Nimsatkar! I have added a few comments.

tariq-hasan · 2026-09-12T17:24:16Z

@@ -173,3 +173,158 @@ def test_submit_job_success(job, options):
            spark_conf=None,
            options=options,
        )


Let's address this as well.

In addition please add a case with all of num_executors, resources_per_executor and spark_conf set, asserting each reaches the backend.

tariq-hasan · 2026-09-12T17:41:00Z

+        TestCase(
+            name="list jobs filtered by status",
+            expected_status=SUCCESS,
+            config={"status": SparkJobStatus.RUNNING},


Let's fix the type contracts everywhere in the tests. For example this one should be

Suggested change

config={"status": SparkJobStatus.RUNNING},

config={"status": {SparkJobStatus.RUNNING}},

tariq-hasan · 2026-09-12T17:42:41Z

+    """Test wait_for_job_status delegation and validation."""
+    with patch("kubeflow.spark.api.spark_client.KubernetesBackend") as mock_backend:
+        backend = mock_backend.return_value
+        backend.wait_for_job_status.return_value = True


backend.wait_for_job_status.return_value should return a mock SparkJob not a boolean.

tariq-hasan · 2026-09-12T17:45:11Z

+            expected_error=ValueError,
+        ),
+    ],
+)


Let's also add cases for polling_interval <= 0, polling_interval >= timeout and timeout=0.

And cases for default and custom timeout as well.

The ideal test case matrix would cover

default status invocation

default timeout invocation

default polling interval

explicit polling interval

single status {RUNNING}

multiple statuses {COMPLETED, FAILED}

invalid timeout < 0

timeout = 0

polling_interval <= 0

polling_interval >= timeout

successful wait

backend timeout

tariq-hasan · 2026-09-12T17:57:43Z

+    ],
+)
+def test_wait_for_job_status(test_case: TestCase):
+    """Test wait_for_job_status delegation and validation."""


For invalid inputs let's add something along the lines of

with pytest.raises(ValueError): client.wait_for_job_status(...) backend.wait_for_job_status.assert_not_called()

The idea is that validation must occur before delegation to the polling function and this invariant needs to be asserted in unit tests.

tariq-hasan · 2026-09-12T18:03:29Z

+        backend.get_job_logs.assert_called_once_with(
+            name=test_case.config["job_name"],
+            follow=test_case.config["follow"],
+        )


Let's print test name throughout all the tests. You can check kubeflow/spark/backends/kubernetes/backend_test.py for the general pattern.

In addition let's use expected_output from the TestCase dataclass more comprehensively in the tests.

tariq-hasan · 2026-09-12T18:23:46Z

+            expected_output=["log stream 1", "log stream 2"],
+        ),
+    ],
+)


Let's have a default follow invocation that does not use the follow parameter in its input.

Neha-Nimsatkar · 2026-09-17T17:13:10Z

Hi @tariq-hasan ,

All suggested changes are in commit 38faf0d:
Full test matrix added (timeouts, polling intervals, invalid inputs)
Added validation assertions (assert_not_called)
Fixed type contracts and mock returns
All CI checks are green and comments are resolved. Could you please take a look when you have time?
Thankyou ...

RobuRishabh

@Neha-Nimsatkar Thanks for working on this, the validation-before-delegation assertions and the full matrix look great. One remaining gap: the "default status/timeout/polling interval" cases skip the assertion exactly when the default is in play (if "status" in test_case.config), so they'd pass even if the wrong default were forwarded. Could you assert the complete expected call (e.g., assert_called_once_with(name=..., status={SparkJobStatus.COMPLETED}, timeout=600, polling_interval=2) with .get() fallbacks)? Also, the test_submit_job failure cases currently inject the error via side_effect, so they test propagation rather than validation its worth renaming to reflect that.

Signed-off-by: Neha-Nimsatkar <1492neha@gmail.com>

Neha-Nimsatkar · 2026-10-16T04:48:37Z

@Neha-Nimsatkar Thanks for working on this, the validation-before-delegation assertions and the full matrix look great. One remaining gap: the "default status/timeout/polling interval" cases skip the assertion exactly when the default is in play (if "status" in test_case.config), so they'd pass even if the wrong default were forwarded. Could you assert the complete expected call (e.g., assert_called_once_with(name=..., status={SparkJobStatus.COMPLETED}, timeout=600, polling_interval=2) with .get() fallbacks)? Also, the test_submit_job failure cases currently inject the error via side_effect, so they test propagation rather than validation its worth renaming to reflect that.

Thanks @RobuRishabh, I have addressed both points:

Removed the conditional if checks in (test_wait_for_job_status) and now assert the full arguments, using .get() fallbacks for the defaults.
Renamed the (test_submit_job) side-effect cases so they show backend error propagation instead of validation.

Please take another look when you get time.

Co-authored-by: Tariq Hasan <mmtariquehsn@gmail.com> Signed-off-by: Neha Nimsatkar <1492neha@gmail.com>

RobuRishabh · 2026-10-20T15:52:21Z

+            name="single status {RUNNING}",
+            expected_status=SUCCESS,
+            config={
+                "job_name": "spark-job-123",
+                "status": {SparkJobStatus.RUNNING},
+                "timeout": 60,
+            },
+            expected_output=SparkJob(name="spark-job-123", namespace="default"),
+        ),
+        TestCase(
+            name="multiple statuses {COMPLETED, FAILED}",
+            expected_status=SUCCESS,
+            config={
+                "job_name": "spark-job-123",
+                "status": {SparkJobStatus.COMPLETED, SparkJobStatus.FAILED},
+                "timeout": 60,
+            },
+            expected_output=SparkJob(name="spark-job-123", namespace="default"),
+        ),
+        TestCase(


single status {RUNNING} and multiple statuses {COMPLETED, FAILED} duplicate the path of default polling interval (346), same kwargs shape, assertion just echoes config back; set contents don't change forwarding behavior. So either drop both or merge into one non-default-status case

Done . Merged both cases into a single non-default status test case to eliminate redundant coverage.

Signed-off-by: Neha Nimsatkar <1492neha@gmail.com>

RobuRishabh · 2026-10-20T19:05:43Z

-        # If we got an exception but expected success, fail
        assert test_case.expected_status == FAILED, f"Unexpected exception in {test_case.name}: {e}"
-        # Validate the exception type if specified
        if test_case.expected_error:


Removing these explanatory comments from test_create_and_connect is unrelated to the job-methods coverage this PR describes. Mind reverting to keep the diff surgical?

Restored both explanatory comments ...

Signed-off-by: Neha Nimsatkar <1492neha@gmail.com>

google-oss-prow Bot requested review from astefanutti, kramaranya and szaher August 10, 2026 15:22

google-oss-prow Bot added the size/M label Aug 10, 2026

Neha-Nimsatkar changed the title ~~test(spark): expand unit test coverage for SparkClient job methods~~ test(spark): Expand unit test coverage for SparkClient job methods Aug 10, 2026

Neha-Nimsatkar changed the title ~~test(spark): Expand unit test coverage for SparkClient job methods~~ fix(spark): expand unit test coverage for SparkClient job methods Aug 10, 2026

tariq-hasan suggested changes Sep 5, 2026

View reviewed changes

google-oss-prow Bot assigned tariq-hasan Sep 5, 2026

google-oss-prow Bot added ok-to-test size/L and removed size/M labels Sep 5, 2026

google-oss-prow Bot changed the title ~~fix(spark): expand unit test coverage for SparkClient job methods~~ chore(spark): expand unit test coverage for SparkClient job methods Sep 6, 2026

tariq-hasan suggested changes Sep 6, 2026

View reviewed changes

tariq-hasan suggested changes Sep 12, 2026

View reviewed changes

tariq-hasan mentioned this pull request Sep 12, 2026

chore(spark): add SparkClient API delegation coverage #658

Closed

Neha-Nimsatkar force-pushed the test/spark-client-coverage branch from ebc20a8 to 38faf0d Compare September 13, 2026 18:08

RobuRishabh added this to Kubeflow SDK Sep 21, 2026

RobuRishabh moved this to Needs Triage in Kubeflow SDK Sep 21, 2026

tariq-hasan moved this from Needs Triage to In Progress in Kubeflow SDK Sep 21, 2026

Neha-Nimsatkar mentioned this pull request Sep 29, 2026

Add Neha-Nimsatkar to Kubeflow org kubeflow/internal-acls#978

Open

2 tasks done

RobuRishabh reviewed Oct 14, 2026

View reviewed changes

chore(spark): expand unit test coverage for SparkClient job methods

8d4eee1

Signed-off-by: Neha-Nimsatkar <1492neha@gmail.com>

Neha-Nimsatkar force-pushed the test/spark-client-coverage branch from 4743f92 to 8d4eee1 Compare October 16, 2026 04:43

tariq-hasan suggested changes Oct 18, 2026

View reviewed changes

Comment thread kubeflow/spark/api/spark_client_test.py

github-project-automation Bot moved this from In Progress to Changes Requested in Kubeflow SDK Oct 18, 2026

Update kubeflow/spark/api/spark_client_test.py

03676eb

Co-authored-by: Tariq Hasan <mmtariquehsn@gmail.com> Signed-off-by: Neha Nimsatkar <1492neha@gmail.com>

google-oss-prow Bot added size/XL and removed size/L labels Oct 20, 2026

RobuRishabh reviewed Oct 20, 2026

View reviewed changes

test(spark): merge redundant status test cases

35b66b8

Signed-off-by: Neha Nimsatkar <1492neha@gmail.com>

google-oss-prow Bot added size/L and removed size/XL labels Oct 20, 2026

RobuRishabh reviewed Oct 20, 2026

View reviewed changes

Neha-Nimsatkar added 2 commits October 22, 2026 08:44

test(spark): restore explanatory comments in test_create_and_connect

1bf511c

Signed-off-by: Neha Nimsatkar <1492neha@gmail.com>

ci: re-trigger build after transient mirror sync error

6bd8e96

Signed-off-by: Neha Nimsatkar <1492neha@gmail.com>

Neha-Nimsatkar force-pushed the test/spark-client-coverage branch from b22255f to 6bd8e96 Compare October 22, 2026 10:57

	@dataclass
	class TestCase:
	"""Test case dataclass for parametrized tests."""

	name: str
	expected_status: str = SUCCESS
	config: dict[str, Any] = field(default_factory=dict)
	expected_output: Any \| None = None
	expected_error: type[Exception] \| None = None
	__test__ = False

	common_utils.validate_wait_for_job_status(
	polling_interval,
	timeout,
	)

	config={"status": SparkJobStatus.RUNNING},
	config={"status": {SparkJobStatus.RUNNING}},

Conversation

Neha-Nimsatkar commented Aug 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What this PR does / why we need it

Motivation

What changed

Verification

Fixes

Checklist

Uh oh!

google-oss-prow Bot commented Aug 10, 2026

Uh oh!

github-actions Bot commented Aug 10, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tariq-hasan commented Sep 5, 2026

Uh oh!

Neha-Nimsatkar commented Sep 6, 2026

Uh oh!

tariq-hasan commented Sep 6, 2026

Uh oh!

tariq-hasan commented Sep 6, 2026

Uh oh!

tariq-hasan commented Sep 6, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Neha-Nimsatkar commented Sep 7, 2026

Uh oh!

tariq-hasan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Neha-Nimsatkar commented Sep 17, 2026

Uh oh!

RobuRishabh left a comment

Choose a reason for hiding this comment

Uh oh!

Neha-Nimsatkar commented Oct 16, 2026

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Neha-Nimsatkar commented Aug 10, 2026 •

edited

Loading