[release test] remove sdk property in AnyscaleJobManager (ray-project#60277)

just save the sdk as a private member instead

Signed-off-by: Lonnie Liu <95255098+aslonnie@users.noreply.github.com>
Signed-off-by: 400Ping <jiekaichang@apache.org>

Author: aslonnie

release/ray_release/job_manager/anyscale_job_manager.py

@@ -38,6 +38,7 @@ def __init__(self, cluster_manager: ClusterManager):
         self.start_time = None
         self.counter = 0
         self.cluster_manager = cluster_manager
+        self._sdk = cluster_manager.sdk
         self._last_job_result = None
         self._last_logs = None
         self.cluster_startup_timeout = 600
@@ -59,8 +60,6 @@ def _run_job(
             f"Executing {cmd_to_run} with {env_vars_for_job} via Anyscale job submit"
         )
 
-        anyscale_client = self.sdk
-
         runtime_env = {
             "env_vars": env_vars_for_job,
         }
@@ -82,7 +81,7 @@ def _run_job(
                     max_retries=0,
                 ),
             )
-            job_response = anyscale_client.create_job(job_request)
+            job_response = self._sdk.create_job(job_request)
         except Exception as e:
             raise JobStartupFailed(
                 "Error starting job with name "
@@ -96,10 +95,6 @@ def _run_job(
         logger.info(f"Link to job: " f"{format_link(self.job_url)}")
         return
 
-    @property
-    def sdk(self):
-        return self.cluster_manager.sdk
-
     @property
     def last_job_result(self):
         return self._last_job_result
@@ -131,9 +126,8 @@ def in_progress(self) -> bool:
         return self.last_job_result and self.last_job_status not in terminal_state
 
     def _get_job_status_with_retry(self):
-        anyscale_client = self.cluster_manager.sdk
         return exponential_backoff_retry(
-            lambda: anyscale_client.get_production_job(self.job_id),
+            lambda: self._sdk.get_production_job(self.job_id),
             retry_exceptions=Exception,
             initial_retry_delay_s=1,
             max_retries=3,
@@ -144,7 +138,7 @@ def _terminate_job(self, raise_exceptions: bool = False):
             return
         logger.info(f"Terminating job {self.job_id}...")
         try:
-            self.sdk.terminate_job(self.job_id)
+            self._sdk.terminate_job(self.job_id)
             logger.info(f"Job {self.job_id} terminated!")
         except Exception:
             msg = f"Couldn't terminate job {self.job_id}!"

release/ray_release/tests/test_anyscale_job_manager.py

@@ -2,21 +2,34 @@
 
 import pytest
 
+from ray_release.anyscale_util import Anyscale
+from ray_release.cluster_manager.cluster_manager import ClusterManager
 from ray_release.job_manager.anyscale_job_manager import AnyscaleJobManager
+from ray_release.test import Test
 
 
 class FakeJobResult:
     def __init__(self, _id: str):
         self.id = _id
 
 
+class FakeSDK(Anyscale):
+    def project_name_by_id(self, project_id: str) -> str:
+        return "fake_project_name"
+
+
 def test_get_last_logs_long_running_job():
     """Test calling get_last_logs() on long-running jobs.
 
     When the job is running longer than 4 hours, get_last_logs() should skip
     downloading the logs and return None.
     """
-    anyscale_job_manager = AnyscaleJobManager(cluster_manager=None)
+    fake_test = Test(name="fake_test")
+    fake_sdk = FakeSDK()
+    cluster_manager = ClusterManager(
+        test=fake_test, project_id="fake_project_id", sdk=fake_sdk
+    )
+    anyscale_job_manager = AnyscaleJobManager(cluster_manager=cluster_manager)
     anyscale_job_manager._duration = 4 * 3_600 + 1
     anyscale_job_manager._last_job_result = FakeJobResult(_id="foo")
     assert anyscale_job_manager.get_last_logs() is None