Add idents to processes

catfish/project/__init__.py

@@ -1,16 +1,31 @@
+import re
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import List
 
-from .procfile import Process, parse_procfile_processes
-
 PROCFILE_LOCATIONS = ["Procfile", "etc/environments/development/Procfile"]
+PROCFILE_LINE = re.compile(r"^([A-Za-z0-9_]+):\s*(.+)$")
+
+
+class DuplicateProcessException(Exception):
+    pass
+
+
+def parse_procfile_processes(project, procfile_lines):
+    seen_names = []
+    for line in procfile_lines:
+        m = PROCFILE_LINE.match(line)
+        if m:
+            if m.group(1) in seen_names:
+                raise DuplicateProcessException(m.group(1))
+            seen_names.append(m.group(1))
+            yield Process(project, m.group(1), m.group(2))
 
 
 @dataclass
 class Project:
     root: Path
-    processes: List[Process] = field(default_factory=list)
+    processes: List["Process"] = field(default_factory=list)
 
     def __post_init__(self):
         self.root = Path(self.root)
@@ -22,7 +37,7 @@ class Project:
             procfile_path = self.root.joinpath(location)
             if procfile_path.exists():
                 with procfile_path.open() as f:
-                    return list(parse_procfile_processes(f.readlines()))
+                    return list(parse_procfile_processes(self, f.readlines()))
         return []
 
     def exists(self):
@@ -31,3 +46,20 @@ class Project:
     @property
     def name(self):
         return self.root.name
+
+    def get_process(self, name):
+        try:
+            return next(proc for proc in self.processes if proc.name == name)
+        except StopIteration:
+            return None
+
+
+@dataclass
+class Process:
+    project: Project
+    name: str
+    command: str
+
+    @property
+    def ident(self):
+        return self.project.name + ":" + self.name

catfish/project/procfile.py

@@ -1,22 +0,0 @@
-import re
-from typing import NamedTuple
-
-Process = NamedTuple("Process", [("name", str), ("command", str)])
-
-
-class DuplicateProcessException(Exception):
-    pass
-
-
-PROCFILE_LINE = re.compile(r"^([A-Za-z0-9_]+):\s*(.+)$")
-
-
-def parse_procfile_processes(procfile_lines):
-    seen_names = []
-    for line in procfile_lines:
-        m = PROCFILE_LINE.match(line)
-        if m:
-            if m.group(1) in seen_names:
-                raise DuplicateProcessException(m.group(1))
-            yield Process(m.group(1), m.group(2))
-            seen_names.append(m.group(1))

example/etc/environments/development/Procfile

@@ -1 +1,2 @@
 web: python -m http.server $PORT
+bg: python src/dummy_program.py

tests/__init__.py

@@ -21,8 +21,8 @@ from catfish.utils.sockets import create_base_socket_dir, delete_base_socket_dir
 
 class BaseTestCase(TestCase):
     TESTS_DIR = Path(os.path.dirname(__file__))
-    DUMMY_EXE = TESTS_DIR.joinpath("dummy_program.py")
     EXAMPLE_DIR = TESTS_DIR.parent.joinpath("example")
+    DUMMY_EXE = EXAMPLE_DIR.joinpath("src", "dummy_program.py")
 
     def setUp(self):
         create_base_socket_dir()

tests/test_project/test_project.py

@@ -1,4 +1,4 @@
-from catfish.project import Project
+from catfish.project import DuplicateProcessException, Project, parse_procfile_processes
 from tests import BaseTestCase
 
 
@@ -15,10 +15,37 @@ class ProjectTestCase(BaseTestCase):
             Project("/nonexistent")
 
     def test_read_processes(self):
-        self.assertEqual(len(self.project.processes), 1)
+        self.assertEqual(len(self.project.processes), 2)
-        process = self.project.processes[0]
+        web_process = self.project.processes[0]
-        self.assertEqual(process.name, "web")
+        self.assertEqual(web_process.name, "web")
-        self.assertEqual(process.command, "python -m http.server $PORT")
+        self.assertEqual(web_process.command, "python -m http.server $PORT")
+        self.assertEqual(web_process.project, self.project)
+
+        bg_process = self.project.processes[1]
+        self.assertEqual(bg_process.name, "bg")
+        self.assertEqual(bg_process.command, "python src/dummy_program.py")
+        self.assertEqual(bg_process.project, self.project)
+
+    def test_get_process(self):
+        self.assertEqual(self.project.get_process("web").name, "web")
+        self.assertEqual(self.project.get_process("bg").name, "bg")
+        self.assertIsNone(self.project.get_process("nonexistent"))
 
     def test_name(self):
         self.assertEqual(self.project.name, "example")
+
+
+class ProcessTestCase(BaseTestCase):
+    def setUp(self):
+        super().setUp()
+        self.project = Project(self.EXAMPLE_DIR)
+        self.process = self.project.get_process("web")
+
+    def test_process_ident(self):
+        self.assertEqual(self.process.ident, "example:web")
+
+    def test_duplicate_procfile(self):
+        with self.assertRaises(DuplicateProcessException) as e:
+            list(parse_procfile_processes(self.project, ["web: 123.py", "web: 456.py"]))
+
+        self.assertEqual(str(e.exception), "web")