Make method private and add constant

sky/serve/constants.py

@@ -75,3 +75,6 @@
 
 # Initial version of service.
 INITIAL_VERSION = 1
+
+# Pattern used for extracting replica ID from replica launch or log files.
+REPLICA_ID_PATTERN = r'.*replica_(\d+)(_launch)?\.log'

sky/serve/serve_utils.py

@@ -628,9 +628,8 @@ def _get_replica_status() -> serve_state.ReplicaStatus:
     return ''
 
 
-def extract_replica_id_from_launch_log_file_name(file_name: str) -> int:
-    pattern = r'.*replica_(\d+)_launch.log'
-    match = re.search(pattern, file_name)
+def _extract_replica_id_from_launch_log_file_name(file_name: str) -> int:
+    match = re.search(serve_constants.REPLICA_ID_PATTERN, file_name)
     if match:
         return int(match.group(1))
     raise ValueError(f'Failed to get replica id from file name: {file_name}')
@@ -640,7 +639,7 @@ def has_valid_replica_id(file_name: str,
                          target_replica_id: Optional[int]) -> bool:
     if target_replica_id is None:
         return True
-    replica_id = extract_replica_id_from_launch_log_file_name(file_name)
+    replica_id = _extract_replica_id_from_launch_log_file_name(file_name)
     return replica_id == target_replica_id
 
 
@@ -668,7 +667,7 @@ def prepare_replica_logs_for_download(service_name: str, timestamp: str,
         has_valid_replica_id(file, target_replica_id)
     ]
     for launch_log_file in launch_log_files:
-        replica_id = extract_replica_id_from_launch_log_file_name(
+        replica_id = _extract_replica_id_from_launch_log_file_name(
             launch_log_file)
         replica_info = serve_state.get_replica_info_from_id(
             service_name, replica_id)

tests/unit_tests/test_serve_utils.py

@@ -5,15 +5,24 @@
 
 class TestExtractReplicaIdFromLaunchLogFileName:
 
-    def test_extract(self):
-        assert 1 == serve_utils.extract_replica_id_from_launch_log_file_name(
+    def test_extract_when_launch_log_file(self):
+        assert 1 == serve_utils._extract_replica_id_from_launch_log_file_name(
             'replica_1_launch.log')
 
+    def test_extract_when_log_file(self):
+        assert 1 == serve_utils._extract_replica_id_from_launch_log_file_name(
+            'replica_1.log')
+
     def test_when_path_expanded(self):
-        assert 1 == serve_utils.extract_replica_id_from_launch_log_file_name(
+        assert 1 == serve_utils._extract_replica_id_from_launch_log_file_name(
             '/Users/firstlast/.sky/serve/vicuna/replica_1_launch.log')
 
-    def test_extract_when_no_match(self):
+    def test_extract_when_no_match_because_invalid_prefix(self):
         with pytest.raises(ValueError):
-            serve_utils.extract_replica_id_from_launch_log_file_name(
+            serve_utils._extract_replica_id_from_launch_log_file_name(
                 '/Users/firstlast/.sky/serve/vicuna/bad_prefix_1_launch.log')
+
+    def test_extract_when_no_match_because_invalid_suffix(self):
+        with pytest.raises(ValueError):
+            serve_utils._extract_replica_id_from_launch_log_file_name(
+                '/Users/firstlast/.sky/serve/vicuna/replica_1_invalid.log')