Fix Job kind in tests

andreyvelich · Sep 8, 2023 · 1fd7407 · 1fd7407
1 parent dbf3d44
commit 1fd7407
Show file tree

Hide file tree

Showing 9 changed files with 87 additions and 192 deletions.
diff --git a/sdk/python/kubeflow/training/api/training_client.py b/sdk/python/kubeflow/training/api/training_client.py
@@ -200,8 +200,8 @@ def create_job(
         # Create the Training Job.
         try:
             self.custom_api.create_namespaced_custom_object(
-                constants.KUBEFLOW_GROUP,
-                constants.OPERATOR_VERSION,
+                constants.GROUP,
+                constants.VERSION,
                 namespace,
                 constants.JOB_PARAMETERS[job.kind]["plural"],
                 job,
@@ -252,8 +252,8 @@ def get_job(
 
         try:
             thread = self.custom_api.get_namespaced_custom_object(
-                constants.KUBEFLOW_GROUP,
-                constants.OPERATOR_VERSION,
+                constants.GROUP,
+                constants.VERSION,
                 namespace,
                 constants.JOB_PARAMETERS[job_kind]["plural"],
                 name,
@@ -307,8 +307,8 @@ def list_jobs(
         result = []
         try:
             thread = self.custom_api.list_namespaced_custom_object(
-                constants.KUBEFLOW_GROUP,
-                constants.OPERATOR_VERSION,
+                constants.GROUP,
+                constants.VERSION,
                 namespace,
                 constants.JOB_PARAMETERS[job_kind]["plural"],
                 async_req=True,
@@ -889,8 +889,8 @@ def update_job(
 
         try:
             self.custom_api.patch_namespaced_custom_object(
-                constants.KUBEFLOW_GROUP,
-                constants.OPERATOR_VERSION,
+                constants.GROUP,
+                constants.VERSION,
                 namespace,
                 constants.JOB_PARAMETERS[job_kind]["plural"],
                 name,
@@ -931,8 +931,8 @@ def delete_job(
 
         try:
             self.custom_api.delete_namespaced_custom_object(
-                constants.KUBEFLOW_GROUP,
-                constants.OPERATOR_VERSION,
+                constants.GROUP,
+                constants.VERSION,
                 namespace,
                 constants.JOB_PARAMETERS[job_kind]["plural"],
                 name=name,

diff --git a/sdk/python/kubeflow/training/constants/constants.py b/sdk/python/kubeflow/training/constants/constants.py
@@ -19,9 +19,9 @@
 DEFAULT_TIMEOUT = 120
 
 # Common constants.
-KUBEFLOW_GROUP = "kubeflow.org"
-OPERATOR_VERSION = "v1"
-API_VERSION = f"{KUBEFLOW_GROUP}/{OPERATOR_VERSION}"
+GROUP = "kubeflow.org"
+VERSION = "v1"
+API_VERSION = f"{GROUP}/{VERSION}"
 ISTIO_SIDECAR_INJECTION = "sidecar.istio.io/inject"
 
 # Training Job conditions.

diff --git a/sdk/python/test/e2e/test_e2e_mpijob.py b/sdk/python/test/e2e/test_e2e_mpijob.py
@@ -42,7 +42,7 @@
 logging.basicConfig(format="%(message)s")
 logging.getLogger().setLevel(logging.INFO)
 
-TRAINING_CLIENT = TrainingClient()
+TRAINING_CLIENT = TrainingClient(job_kind=constants.MPIJOB_KIND)
 JOB_NAME = "mpijob-mxnet-ci-test"
 CONTAINER_NAME = "mpi"
 GANG_SCHEDULER_NAME = os.getenv(TEST_GANG_SCHEDULER_NAME_ENV_KEY, "")
@@ -91,29 +91,18 @@ def test_sdk_e2e_with_gang_scheduling(job_namespace):
     )
 
     TRAINING_CLIENT.create_job(job=mpijob, namespace=job_namespace)
-    logging.info(f"List of created {constants.MPIJOB_KIND}s")
+    logging.info(f"List of created {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_unschedulable_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.MPIJOB_KIND,
-    )
+    verify_unschedulable_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace)
 
     TRAINING_CLIENT.update_job(patched_mpijob, JOB_NAME, job_namespace)
-    logging.info(f"List of patched {constants.MPIJOB_KIND}s")
+    logging.info(f"List of patched {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.MPIJOB_KIND,
-        timeout=900,
-    )
+    verify_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace, timeout=900)
 
-    TRAINING_CLIENT.delete_job(JOB_NAME, job_namespace)
+    TRAINING_CLIENT.delete_job(JOB_NAME)
 
 
 @pytest.mark.skipif(
@@ -148,16 +137,10 @@ def test_sdk_e2e(job_namespace):
     mpijob = generate_mpijob(job_namespace, launcher, worker)
 
     TRAINING_CLIENT.create_job(job=mpijob, namespace=job_namespace)
-    logging.info(f"List of created {constants.MPIJOB_KIND}s")
+    logging.info(f"List of created {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.MPIJOB_KIND,
-        timeout=900,
-    )
+    verify_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace, timeout=900)
 
     TRAINING_CLIENT.delete_job(JOB_NAME, job_namespace)
 
@@ -169,8 +152,8 @@ def generate_mpijob(
     scheduling_policy: Optional[KubeflowOrgV1SchedulingPolicy] = None,
 ) -> KubeflowOrgV1MPIJob:
     return KubeflowOrgV1MPIJob(
-        api_version="kubeflow.org/v1",
-        kind="MPIJob",
+        api_version=constants.API_VERSION,
+        kind=constants.MPIJOB_KIND,
         metadata=V1ObjectMeta(name=JOB_NAME, namespace=job_namespace),
         spec=KubeflowOrgV1MPIJobSpec(
             slots_per_worker=1,

diff --git a/sdk/python/test/e2e/test_e2e_mxjob.py b/sdk/python/test/e2e/test_e2e_mxjob.py
@@ -43,7 +43,7 @@
 logging.basicConfig(format="%(message)s")
 logging.getLogger().setLevel(logging.INFO)
 
-TRAINING_CLIENT = TrainingClient()
+TRAINING_CLIENT = TrainingClient(constants.MXJOB_KIND)
 JOB_NAME = "mxjob-mnist-ci-test"
 CONTAINER_NAME = "mxnet"
 GANG_SCHEDULER_NAME = os.getenv(TEST_GANG_SCHEDULER_NAME_ENV_KEY, "")
@@ -114,27 +114,16 @@ def test_sdk_e2e_with_gang_scheduling(job_namespace):
     )
 
     TRAINING_CLIENT.create_job(job=unschedulable_mxjob, namespace=job_namespace)
-    logging.info(f"List of created {constants.MXJOB_KIND}s")
+    logging.info(f"List of created {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_unschedulable_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.MXJOB_KIND,
-    )
+    verify_unschedulable_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace)
 
     TRAINING_CLIENT.update_job(schedulable_mxjob, JOB_NAME, job_namespace)
-    logging.info(f"List of patched {constants.MXJOB_KIND}s")
+    logging.info(f"List of patched {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.MXJOB_KIND,
-        timeout=900,
-    )
+    verify_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace, timeout=900)
 
     TRAINING_CLIENT.delete_job(JOB_NAME, job_namespace)
 
@@ -182,16 +171,10 @@ def test_sdk_e2e(job_namespace):
     mxjob = generate_mxjob(job_namespace, scheduler, server, worker)
 
     TRAINING_CLIENT.create_job(job=mxjob, namespace=job_namespace)
-    logging.info(f"List of created {constants.MXJOB_KIND}s")
+    logging.info(f"List of created {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.MXJOB_KIND,
-        timeout=900,
-    )
+    verify_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace, timeout=900)
 
     TRAINING_CLIENT.delete_job(JOB_NAME, job_namespace)
 
@@ -204,8 +187,8 @@ def generate_mxjob(
     scheduling_policy: Optional[KubeflowOrgV1SchedulingPolicy] = None,
 ) -> KubeflowOrgV1MXJob:
     return KubeflowOrgV1MXJob(
-        api_version="kubeflow.org/v1",
-        kind="MXJob",
+        api_version=constants.API_VERSION,
+        kind=constants.MXJOB_KIND,
         metadata=V1ObjectMeta(name=JOB_NAME, namespace=job_namespace),
         spec=KubeflowOrgV1MXJobSpec(
             job_mode="MXTrain",

diff --git a/sdk/python/test/e2e/test_e2e_paddlejob.py b/sdk/python/test/e2e/test_e2e_paddlejob.py
@@ -42,7 +42,7 @@
 logging.basicConfig(format="%(message)s")
 logging.getLogger().setLevel(logging.INFO)
 
-TRAINING_CLIENT = TrainingClient()
+TRAINING_CLIENT = TrainingClient(job_kind=constants.PADDLEJOB_KIND)
 JOB_NAME = "paddlejob-cpu-ci-test"
 CONTAINER_NAME = "paddle"
 GANG_SCHEDULER_NAME = os.getenv(TEST_GANG_SCHEDULER_NAME_ENV_KEY, "")
@@ -77,27 +77,16 @@ def test_sdk_e2e_with_gang_scheduling(job_namespace):
     )
 
     TRAINING_CLIENT.create_job(job=unschedulable_paddlejob, namespace=job_namespace)
-    logging.info(f"List of created {constants.PADDLEJOB_KIND}s")
+    logging.info(f"List of created {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_unschedulable_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.PADDLEJOB_KIND,
-    )
+    verify_unschedulable_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace)
 
     TRAINING_CLIENT.update_job(schedulable_paddlejob, JOB_NAME, job_namespace)
-    logging.info(f"List of patched {constants.PADDLEJOB_KIND}s")
+    logging.info(f"List of patched {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.PADDLEJOB_KIND,
-        timeout=900,
-    )
+    verify_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace, timeout=900)
 
     TRAINING_CLIENT.delete_job(JOB_NAME, job_namespace)
 
@@ -123,16 +112,10 @@ def test_sdk_e2e(job_namespace):
     paddlejob = generate_paddlejob(job_namespace, worker)
 
     TRAINING_CLIENT.create_job(job=paddlejob, namespace=job_namespace)
-    logging.info(f"List of created {constants.PADDLEJOB_KIND}s")
+    logging.info(f"List of created {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.PADDLEJOB_KIND,
-        timeout=900,
-    )
+    verify_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace, timeout=900)
 
     TRAINING_CLIENT.delete_job(JOB_NAME, job_namespace)
 
@@ -143,8 +126,8 @@ def generate_paddlejob(
     scheduling_policy: Optional[KubeflowOrgV1SchedulingPolicy] = None,
 ) -> KubeflowOrgV1PaddleJob:
     return KubeflowOrgV1PaddleJob(
-        api_version="kubeflow.org/v1",
-        kind="PaddleJob",
+        api_version=constants.API_VERSION,
+        kind=constants.PADDLEJOB_KIND,
         metadata=V1ObjectMeta(name=JOB_NAME, namespace=job_namespace),
         spec=KubeflowOrgV1PaddleJobSpec(
             run_policy=KubeflowOrgV1RunPolicy(

diff --git a/sdk/python/test/e2e/test_e2e_pytorchjob.py b/sdk/python/test/e2e/test_e2e_pytorchjob.py
@@ -42,7 +42,7 @@
 logging.basicConfig(format="%(message)s")
 logging.getLogger().setLevel(logging.INFO)
 
-TRAINING_CLIENT = TrainingClient()
+TRAINING_CLIENT = TrainingClient(job_kind=constants.PYTORCHJOB_KIND)
 JOB_NAME = "pytorchjob-mnist-ci-test"
 CONTAINER_NAME = "pytorch"
 GANG_SCHEDULER_NAME = os.getenv(TEST_GANG_SCHEDULER_NAME_ENV_KEY, "")
@@ -94,27 +94,16 @@ def test_sdk_e2e_with_gang_scheduling(job_namespace):
     )
 
     TRAINING_CLIENT.create_job(job=unschedulable_pytorchjob, namespace=job_namespace)
-    logging.info(f"List of created {constants.PYTORCHJOB_KIND}s")
+    logging.info(f"List of created {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_unschedulable_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.PYTORCHJOB_KIND,
-    )
+    verify_unschedulable_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace)
 
     TRAINING_CLIENT.update_job(schedulable_pytorchjob, JOB_NAME, job_namespace)
-    logging.info(f"List of patched {constants.PYTORCHJOB_KIND}s")
+    logging.info(f"List of patched {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.PYTORCHJOB_KIND,
-        timeout=900,
-    )
+    verify_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace, timeout=900)
 
     TRAINING_CLIENT.delete_job(JOB_NAME, job_namespace)
 
@@ -151,16 +140,10 @@ def test_sdk_e2e(job_namespace):
     pytorchjob = generate_pytorchjob(job_namespace, master, worker)
 
     TRAINING_CLIENT.create_job(job=pytorchjob, namespace=job_namespace)
-    logging.info(f"List of created {constants.PYTORCHJOB_KIND}s")
+    logging.info(f"List of created {TRAINING_CLIENT.job_kind}s")
     logging.info(TRAINING_CLIENT.list_jobs(job_namespace))
 
-    verify_job_e2e(
-        TRAINING_CLIENT,
-        JOB_NAME,
-        job_namespace,
-        constants.PYTORCHJOB_KIND,
-        timeout=900,
-    )
+    verify_job_e2e(TRAINING_CLIENT, JOB_NAME, job_namespace, timeout=900)
 
     TRAINING_CLIENT.delete_job(JOB_NAME, job_namespace)
 
@@ -172,8 +155,8 @@ def generate_pytorchjob(
     scheduling_policy: Optional[KubeflowOrgV1SchedulingPolicy] = None,
 ) -> KubeflowOrgV1PyTorchJob:
     return KubeflowOrgV1PyTorchJob(
-        api_version="kubeflow.org/v1",
-        kind="PyTorchJob",
+        api_version=constants.API_VERSION,
+        kind=constants.PYTORCHJOB_KIND,
         metadata=V1ObjectMeta(name=JOB_NAME, namespace=job_namespace),
         spec=KubeflowOrgV1PyTorchJobSpec(
             run_policy=KubeflowOrgV1RunPolicy(