remove: JobDefinition and update tests

Author: VanillaSpoon

src/codeflare_sdk/__init__.py

@@ -14,6 +14,6 @@
     get_cluster,
 )
 
-from .job import JobDefinition, Job, RayJobClient
+from .job import RayJobClient
 
 from .utils import generate_cert

src/codeflare_sdk/job/__init__.py

@@ -1,3 +1 @@
-from .jobs import JobDefinition, Job
-
 from .ray_jobs import RayJobClient

src/codeflare_sdk/job/jobs.py

@@ -5,7 +5,7 @@
 from torchx.specs.api import AppState, is_terminal
 
 from codeflare_sdk import Cluster, ClusterConfiguration, TokenAuthentication
-from codeflare_sdk.job.jobs import JobDefinition
+from codeflare_sdk.job import RayJobClient
 
 import pytest
 
@@ -97,18 +97,25 @@ def assert_jobsubmit_withoutLogin(self, cluster):
 
     def assert_jobsubmit_withlogin(self, cluster):
         self.assert_appwrapper_exists()
-        jobdef = JobDefinition(
-            name="mnist",
-            script="./tests/e2e/mnist.py",
-            scheduler_args={"requirements": "./tests/e2e/mnist_pip_requirements.txt"},
+        auth_token = run_oc_command(["whoami", "--show-token=true"])
+        ray_dashboard = cluster.cluster_dashboard_uri()
+        header = {"Authorization": f"Bearer {auth_token}"}
+        client = RayJobClient(address=ray_dashboard, headers=header, verify=True)
+
+        # Submit the job
+        submission_id = client.submit_job(
+            entrypoint="python mnist.py",
+            runtime_env={
+                "working_dir": "./tests/e2e/",
+                "pip": "mnist_pip_requirements.txt",
+            },
         )
-        job = jobdef.submit(cluster)
-
+        print(f"Submitted job with ID: {submission_id}")
         done = False
         time = 0
         timeout = 900
         while not done:
-            status = job.status()
+            status = client.get_job_status(submission_id)
             if is_terminal(status.state):
                 break
             if not done:
@@ -118,11 +125,12 @@ def assert_jobsubmit_withlogin(self, cluster):
                 sleep(5)
                 time += 5
 
-        print(job.status())
-        self.assert_job_completion(status)
+        logs = client.get_job_logs(submission_id)
+        print(logs)
 
-        print(job.logs())
+        self.assert_job_completion(status)
 
+        client.delete_job(submission_id)
         cluster.down()
 
     def assert_appwrapper_exists(self):

tests/e2e/mnist_raycluster_sdk_oauth_test.py

@@ -10,7 +10,7 @@
 from torchx.specs.api import AppState, is_terminal
 
 from codeflare_sdk.cluster.cluster import Cluster, ClusterConfiguration
-from codeflare_sdk.job.jobs import JobDefinition
+from codeflare_sdk.job import RayJobClient
 
 import pytest
 
@@ -87,18 +87,25 @@ def run_mnist_raycluster_sdk(self):
 
         cluster.details()
 
-        jobdef = JobDefinition(
-            name="mnist",
-            script="./tests/e2e/mnist.py",
-            scheduler_args={"requirements": "./tests/e2e/mnist_pip_requirements.txt"},
+        auth_token = run_oc_command(["whoami", "--show-token=true"])
+        ray_dashboard = cluster.cluster_dashboard_uri()
+        header = {"Authorization": f"Bearer {auth_token}"}
+        client = RayJobClient(address=ray_dashboard, headers=header, verify=True)
+
+        # Submit the job
+        submission_id = client.submit_job(
+            entrypoint="python mnist.py",
+            runtime_env={
+                "working_dir": "./tests/e2e/",
+                "pip": "mnist_pip_requirements.txt",
+            },
         )
-        job = jobdef.submit(cluster)
-
+        print(f"Submitted job with ID: {submission_id}")
         done = False
         time = 0
         timeout = 900
         while not done:
-            status = job.status()
+            status = client.get_job_status(submission_id)
             if is_terminal(status.state):
                 break
             if not done:
@@ -108,11 +115,12 @@ def run_mnist_raycluster_sdk(self):
                 sleep(5)
                 time += 5
 
-        print(job.status())
-        self.assert_job_completion(status)
+        logs = client.get_job_logs(submission_id)
+        print(logs)
 
-        print(job.logs())
+        self.assert_job_completion(status)
 
+        client.delete_job(submission_id)
         cluster.down()
 
     # Assertions

tests/e2e/mnist_raycluster_sdk_test.py

@@ -4,27 +4,33 @@
 
 from torchx.specs.api import AppState, is_terminal
 
+from support import *
+
 from codeflare_sdk.cluster.cluster import get_cluster
-from codeflare_sdk.job.jobs import JobDefinition
+from codeflare_sdk.job import RayJobClient
 
 namespace = sys.argv[1]
 
 cluster = get_cluster("mnist", namespace)
 
 cluster.details()
 
-jobdef = JobDefinition(
-    name="mnist",
-    script="mnist.py",
-    scheduler_args={"requirements": "requirements.txt"},
-)
-job = jobdef.submit(cluster)
+auth_token = run_oc_command(["whoami", "--show-token=true"])
+ray_dashboard = cluster.cluster_dashboard_uri()
+header = {"Authorization": f"Bearer {auth_token}"}
+client = RayJobClient(address=ray_dashboard, headers=header, verify=True)
 
+# Submit the job
+submission_id = client.submit_job(
+    entrypoint="python mnist.py",
+    runtime_env={"working_dir": "/", "pip": "requirements.txt"},
+)
+print(f"Submitted job with ID: {submission_id}")
 done = False
 time = 0
 timeout = 900
 while not done:
-    status = job.status()
+    status = client.get_job_status(submission_id)
     if is_terminal(status.state):
         break
     if not done:
@@ -34,12 +40,13 @@
         sleep(5)
         time += 5
 
-print(f"Job has completed: {status.state}")
-
-print(job.logs())
+logs = client.get_job_logs(submission_id)
+print(logs)
 
+client.delete_job(submission_id)
 cluster.down()
 
+
 if not status.state == AppState.SUCCEEDED:
     exit(1)
 else:

tests/e2e/mnist_rayjob.py

@@ -59,10 +59,6 @@
     RayClusterStatus,
     CodeFlareClusterStatus,
 )
-from codeflare_sdk.job.jobs import (
-    JobDefinition,
-    Job,
-)
 from codeflare_sdk.utils.generate_cert import (
     generate_ca_cert,
     generate_tls_cert,
@@ -2487,24 +2483,6 @@ def test_wait_ready(mocker, capsys):
     )
 
 
-def test_jobdefinition_coverage(mocker):
-    mocker.patch("kubernetes.client.ApisApi.get_api_versions")
-    mocker.patch(
-        "kubernetes.client.CustomObjectsApi.get_cluster_custom_object",
-        return_value={"spec": {"domain": ""}},
-    )
-    abstract = JobDefinition()
-    cluster = createClusterWithConfig(mocker)
-    abstract._dry_run(cluster)
-    abstract.submit(cluster)
-
-
-def test_job_coverage():
-    abstract = Job()
-    abstract.status()
-    abstract.logs()
-
-
 def arg_check_side_effect(*args):
     assert args[0] == "fake-app-handle"