Add check if API is running on process check

src/tribler/core/utilities/process_manager/process.py

@@ -9,6 +9,7 @@
 from typing import Any, List, Optional, TYPE_CHECKING
 
 import psutil
+import requests
 
 from tribler.core.utilities.process_manager import sql_scripts
 from tribler.core.utilities.process_manager.utils import with_retry
@@ -17,6 +18,8 @@
 if TYPE_CHECKING:
     from tribler.core.utilities.process_manager import ProcessManager
 
+API_CHECK_TIMEOUT_IN_SECONDS = 5
+
 
 class ProcessKind(Enum):
     GUI = 'gui'
@@ -177,8 +180,29 @@ def is_running(self):
             # The same PID value was reused for a new process, so the previous process is not running anymore
             return False
 
+        # Check if Core API is running if port is known
+        if self.kind == ProcessKind.Core and self.api_port is not None:
+            return self.is_api_running()
+
         return True
 
+    def is_api_running(self) -> bool:
+        """
+        Checks if REST API is running at the API port.
+
+        This is done by checking the Docs endpoint which is always available
+        and does not require API_KEY to access.
+        """
+        if not self.api_port:
+            return False
+
+        try:
+            docs_url = f"http://localhost:{self.api_port}/docs"
+            _ = requests.get(docs_url, timeout=API_CHECK_TIMEOUT_IN_SECONDS)
+            return True
+        except requests.exceptions.RequestException:
+            return False
+
     def set_api_port(self, api_port: int):
         self.api_port = api_port
         self.save()

src/tribler/core/utilities/process_manager/tests/test_process.py

@@ -4,12 +4,23 @@
 
 import psutil
 import pytest
+import requests.exceptions
 
 from tribler.core.utilities.process_manager.manager import ProcessManager, logger
 from tribler.core.utilities.process_manager.process import ProcessKind, TriblerProcess
 
 
-def test_tribler_process():
+@patch('requests.get', side_effect=requests.exceptions.RequestException)
+def test_tribler_process_without_running_api(mock_get: Mock):
+    p = TriblerProcess.current_process(ProcessKind.Core, 123, manager=Mock())
+    p.api_port = 456
+
+    assert not p.is_api_running()
+    assert not p.is_running()
+
+
+@patch('requests.get', return_value=Mock(status_code=200))
+def test_tribler_process_with_running_api(mock_get: Mock):
     p = TriblerProcess.current_process(ProcessKind.Core, 123, manager=Mock())
     assert p.is_current_process()
     assert p.is_running()