Add function to process to detect if it's running

catfish/project/__init__.py

@@ -6,6 +6,8 @@ from typing import Dict, List
 
 from dotenv import dotenv_values
 
+from catfish.utils.processes import get_running_process_for
+
 PROCFILE_LOCATIONS = ["Procfile", "etc/environments/development/Procfile"]
 ENVFILE_LOCATIONS = [".env", "etc/environments/development/env"]
 PROCFILE_LINE = re.compile(r"^([A-Za-z0-9_]+):\s*(.+)$")
@@ -91,3 +93,10 @@ class Process:
     @property
     def logs_socket(self):
         return f"{self.project.name}-{self.name}.sock"
+
+    def get_running_process(self):
+        return get_running_process_for(self)
+
+    @property
+    def is_running(self):
+        return self.get_running_process() is not None

catfish/utils/processes.py

@@ -3,8 +3,6 @@ from typing import List
 
 import psutil
 
-from catfish.project import Process
-
 CURRENT_PROCESS = psutil.Process()
 
 
@@ -51,7 +49,7 @@ def wait_for_process_terminate(pid: int):
         time.sleep(0.1)
 
 
-def find_running_process_for(process: Process):
+def get_running_process_for(process):
     for proc in get_root_process().children(recursive=True):
         try:
             if proc.environ().get("CATFISH_IDENT") == process.ident:

tests/test_project/test_project.py

@@ -62,3 +62,6 @@ class ProcessTestCase(BaseTestCase):
             list(parse_procfile_processes(self.project, ["web: 123.py", "web: 456.py"]))
 
         self.assertEqual(str(e.exception), "web")
+
+    def test_running_process(self):
+        self.assertFalse(self.process.is_running)

tests/test_worker/test_server.py

@@ -5,7 +5,7 @@ import psutil
 
 from catfish.project import Project
 from catfish.utils.processes import (
-    find_running_process_for,
+    get_running_process_for,
     is_process_running,
     terminate_processes,
     wait_for_process_terminate,
@@ -33,7 +33,8 @@ class ProcessWorkerTestCase(BaseWorkerTestCase):
             {"path": str(self.project.root), "process": str(self.process.name)},
         )
         self.assertTrue(is_process_running(response["pid"]))
-        self.assertEqual(find_running_process_for(self.process).pid, response["pid"])
+        self.assertEqual(get_running_process_for(self.process).pid, response["pid"])
+        self.assertTrue(self.process.is_running)
 
     def test_additional_environment(self):
         response = send_to_server(
@@ -72,7 +73,7 @@ class ProcessWorkerTestCase(BaseWorkerTestCase):
         psutil.Process(initial_pid).send_signal(signal.SIGHUP)
         wait_for_process_terminate(initial_pid)
         time.sleep(2)
-        new_process = find_running_process_for(self.process)
+        new_process = get_running_process_for(self.process)
         self.assertNotEqual(new_process.pid, initial_pid)
         self.assertFalse(is_process_running(initial_pid))
         self.assertTrue(is_process_running(new_process.pid))
@@ -84,7 +85,7 @@ class ProcessWorkerTestCase(BaseWorkerTestCase):
         initial_pid = response["pid"]
         wait_for_process_terminate(initial_pid)
         time.sleep(2)
-        new_process = find_running_process_for(self.project.get_process("die"))
+        new_process = get_running_process_for(self.project.get_process("die"))
         self.assertNotEqual(new_process.pid, initial_pid)
         self.assertFalse(is_process_running(initial_pid))
         self.assertTrue(is_process_running(new_process.pid))
@@ -96,7 +97,7 @@ class ProcessWorkerTestCase(BaseWorkerTestCase):
         initial_pid = response["pid"]
         wait_for_process_terminate(initial_pid)
         time.sleep(2)
-        new_process = find_running_process_for(self.project.get_process("exit"))
+        new_process = get_running_process_for(self.project.get_process("exit"))
         self.assertNotEqual(new_process.pid, initial_pid)
         self.assertFalse(is_process_running(initial_pid))
         self.assertTrue(is_process_running(new_process.pid))
@@ -110,7 +111,7 @@ class ProcessWorkerTestCase(BaseWorkerTestCase):
         terminate_processes([psutil.Process(initial_pid)])
         wait_for_process_terminate(initial_pid)
         time.sleep(2)
-        self.assertIsNone(find_running_process_for(self.process))
+        self.assertIsNone(get_running_process_for(self.process))
         self.assertFalse(is_process_running(initial_pid))