diff --git a/kubeflow/spark/api/spark_client.py b/kubeflow/spark/api/spark_client.py index 332eaa286..f5035c2e7 100644 --- a/kubeflow/spark/api/spark_client.py +++ b/kubeflow/spark/api/spark_client.py @@ -64,7 +64,7 @@ def connect( base_url: str | None = None, token: str | None = None, num_executors: int | None = None, - resources_per_executor: dict[str, str] | None = None, + resources_per_executor: dict | None = None, spark_conf: dict[str, str] | None = None, driver: Driver | None = None, executor: Executor | None = None, @@ -175,7 +175,7 @@ def submit_job( self, job: FileJob | FuncJob, num_executors: int | None = None, - resources_per_executor: dict[str, str] | None = None, + resources_per_executor: dict | None = None, spark_conf: dict[str, str] | None = None, options: list | None = None, ) -> str: diff --git a/kubeflow/spark/backends/kubernetes/utils.py b/kubeflow/spark/backends/kubernetes/utils.py index 1885bce04..81d445eb9 100644 --- a/kubeflow/spark/backends/kubernetes/utils.py +++ b/kubeflow/spark/backends/kubernetes/utils.py @@ -114,8 +114,8 @@ def _resolve_driver_resources( Tuple of (cores, memory). Raises: - ValueError: - If the configured CPU or memory values are invalid. + ValueError: If the configured CPU or memory values are invalid. + TypeError: If an unsupported CPU type (such as bool) is passed. """ cores = constants.DEFAULT_DRIVER_CPU @@ -136,7 +136,7 @@ def _resolve_driver_resources( def _resolve_executor_resources( executor: Executor | None = None, num_executors: int | None = None, - resources_per_executor: dict[str, str] | None = None, + resources_per_executor: dict | None = None, ) -> tuple[int, int, str]: """Resolve executor configuration. @@ -154,8 +154,8 @@ def _resolve_executor_resources( Tuple containing ``(instances, cores, memory)``. Raises: - ValueError: - If the configured CPU or memory values are invalid. + ValueError: If the configured CPU or memory values are invalid. + TypeError: If an unsupported CPU type (such as bool) is passed. """ if executor and executor.num_instances is not None: @@ -257,11 +257,15 @@ def _validate_cpu_value(cpu: str | int | None) -> int: Integer CPU core value. Raises: - ValueError: If CPU value is invalid. + ValueError: If CPU value is invalid or non-positive. + TypeError: If an unsupported type (such as bool) is passed. """ if cpu is None: raise ValueError("CPU value cannot be None") + if isinstance(cpu, bool): + raise TypeError("Invalid CPU type 'bool'. Expected str or int.") + if isinstance(cpu, int): cores = float(cpu) @@ -279,13 +283,19 @@ def _validate_cpu_value(cpu: str | int | None) -> int: f"Invalid CPU value '{cpu}'. Decimal milli-CPU values are not supported." ) - cores = int(milli_cpu) / 1000 + try: + cores = int(milli_cpu) / 1000 + except ValueError as e: + raise ValueError(f"Invalid CPU value '{cpu}'.") from e else: - cores = float(cpu) + try: + cores = float(cpu) + except ValueError as e: + raise ValueError(f"Invalid CPU value '{cpu}'.") from e else: - raise ValueError(f"Invalid CPU type '{type(cpu)}'. Expected str or int.") + raise TypeError(f"Invalid CPU type '{type(cpu).__name__}'. Expected str or int.") if not math.isfinite(cores) or cores <= 0: raise ValueError(f"Invalid CPU value: {cpu!r}") @@ -473,7 +483,7 @@ def get_spark_connect_driver_spec( def get_spark_connect_executor_spec( executor: Executor | None = None, num_executors: int | None = None, - resources_per_executor: dict[str, str] | None = None, + resources_per_executor: dict | None = None, ) -> models.SparkV1alpha1ExecutorSpec: """Convert SDK Executor to API ExecutorSpec. @@ -490,8 +500,8 @@ def get_spark_connect_executor_spec( API ExecutorSpec model. Raises: - ValueError: - If the configured executor resources are invalid. + ValueError: If the configured executor resources are invalid. + TypeError: If an unsupported CPU type (such as bool) is passed. """ instances, cores, memory = _resolve_executor_resources( executor, @@ -511,7 +521,7 @@ def build_spark_connect_cr( namespace: str, spark_version: str | None = None, num_executors: int | None = None, - resources_per_executor: dict[str, str] | None = None, + resources_per_executor: dict | None = None, spark_conf: dict[str, str] | None = None, driver: Driver | None = None, executor: Executor | None = None, @@ -542,8 +552,8 @@ def build_spark_connect_cr( SparkConnect CR as typed Pydantic model. Raises: - ValueError: - If the provided driver or executor resource configuration is invalid. + ValueError: If the provided driver or executor resource configuration is invalid. + TypeError: If an unsupported CPU type (such as bool) is passed. """ _validate_spark_conf(spark_conf) @@ -688,7 +698,7 @@ def get_spark_job_driver_spec( def get_spark_job_executor_spec( num_executors: int | None = None, - resources_per_executor: dict[str, str] | None = None, + resources_per_executor: dict | None = None, ) -> models.SparkV1beta2ExecutorSpec: """Build ExecutorSpec for SparkApplication. @@ -802,7 +812,7 @@ def get_spark_application_cr_from_file_job( main_file: str, arguments: list[str] | None = None, num_executors: int | None = None, - resources_per_executor: dict[str, str] | None = None, + resources_per_executor: dict | None = None, options: list | None = None, backend: Any | None = None, spark_conf: dict[str, str] | None = None, @@ -865,7 +875,7 @@ def get_spark_application_cr_from_func_job( func: Callable, func_args: dict[str, Any] | None = None, num_executors: int | None = None, - resources_per_executor: dict[str, str] | None = None, + resources_per_executor: dict | None = None, options: list | None = None, backend: Any | None = None, spark_conf: dict[str, str] | None = None, diff --git a/kubeflow/spark/backends/kubernetes/utils_test.py b/kubeflow/spark/backends/kubernetes/utils_test.py index 35170d3b7..9f8b9284f 100644 --- a/kubeflow/spark/backends/kubernetes/utils_test.py +++ b/kubeflow/spark/backends/kubernetes/utils_test.py @@ -39,6 +39,7 @@ get_spark_application_cr_from_file_job, get_spark_application_cr_from_func_job, get_spark_application_info_from_cr, + get_spark_connect_executor_spec, get_spark_connect_info_from_cr, get_spark_job_driver_spec, get_spark_job_executor_spec, @@ -98,11 +99,6 @@ def mock_k8s_backend(): backend.__class__ = KubernetesBackend return backend - def test_missing_hostname(self): - """U16: Missing hostname raises ValueError.""" - with pytest.raises(ValueError, match="Host is required"): - validate_spark_connect_url("sc://:15002") - @pytest.fixture def spark_connect_resource(minimal_spec): @@ -966,6 +962,19 @@ def test_generate_job_name(test_case: TestCase) -> None: }, expected_error=ValueError, ), + TestCase( + name="invalid cpu types", + expected_status=FAILED, + config={ + "cases": [ + True, + False, + [1], + {"cpu": 1}, + ], + }, + expected_error=TypeError, + ), ], ) def test_validate_cpu_value(test_case: TestCase) -> None: @@ -1892,4 +1901,10 @@ def test_get_spark_application_info_from_cr( assert job.creation_timestamp == creation_timestamp assert job.num_executors == 5 + +def test_get_spark_connect_executor_spec_bool_cpu_raises_type_error(): + """Verify that boolean CPU values raise TypeError via public construction path.""" + with pytest.raises(TypeError): + get_spark_connect_executor_spec(resources_per_executor={"cpu": True}) + print("test execution complete") diff --git a/kubeflow/spark/types/types.py b/kubeflow/spark/types/types.py index e1779139d..fbc3716dd 100644 --- a/kubeflow/spark/types/types.py +++ b/kubeflow/spark/types/types.py @@ -87,7 +87,7 @@ class Driver: """ image: str | None = None - resources: dict[str, str] | None = None + resources: dict | None = None java_options: str | None = None service_account: str | None = None @@ -121,7 +121,7 @@ class Executor: """ num_instances: int | None = None - resources_per_executor: dict[str, str] | None = None + resources_per_executor: dict | None = None java_options: str | None = None