Read procfile

catfish/project/__init__.py

@@ -1,14 +1,29 @@
-from dataclasses import dataclass
+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"]
 
 
 @dataclass
 class Project:
     root: Path
+    processes: List[Process] = field(default_factory=list)
 
     def __post_init__(self):
         self.root = Path(self.root)
         assert self.exists()
+        self.processes = self.read_processes()
+
+    def read_processes(self):
+        for location in PROCFILE_LOCATIONS:
+            procfile_path = self.root.joinpath(location)
+            if procfile_path.exists():
+                with procfile_path.open() as f:
+                    return list(parse_procfile_processes(f.readlines()))
+        return []
 
     def exists(self):
-        return self.root.exists()
+        return self.root.exists() and self.root.is_dir()

catfish/project/procfile.py

@@ -0,0 +1,22 @@
+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/README.md

@@ -0,0 +1,4 @@
+# Example Project
+
+
+This is an example project for use with catfish. Primarily for testing.

example/etc/environments/development/Procfile

@@ -0,0 +1 @@
+web: python -m http.server $PORT

tests/test_project/test_project.py

@@ -13,3 +13,9 @@ class ProjectTestCase(BaseTestCase):
     def test_cant_create_nonexistent_directory(self):
         with self.assertRaises(AssertionError):
             Project("/nonexistent")
+
+    def test_read_processes(self):
+        self.assertEqual(len(self.project.processes), 1)
+        process = self.project.processes[0]
+        self.assertEqual(process.name, "web")
+        self.assertEqual(process.command, "python -m http.server $PORT")