94 lines
3.4 KiB
Python
94 lines
3.4 KiB
Python
from click.termui import _ansi_colors as ansi_colors # type: ignore
|
|
|
|
from catfish.project import DuplicateProcessException, Project, parse_procfile_processes
|
|
from catfish.worker.server import PayloadType, send_to_server
|
|
from tests import BaseTestCase, BaseWorkerTestCase
|
|
|
|
|
|
class ProjectTestCase(BaseTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.project = Project(self.EXAMPLE_DIR)
|
|
|
|
def test_exists(self):
|
|
self.assertTrue(self.project.exists())
|
|
|
|
def test_cant_create_nonexistent_directory(self):
|
|
with self.assertRaises(AssertionError):
|
|
Project("/nonexistent")
|
|
|
|
def test_read_processes(self):
|
|
self.assertEqual(len(self.project.processes), 4)
|
|
web_process = self.project.processes[0]
|
|
self.assertEqual(web_process.name, "web")
|
|
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)
|
|
|
|
bg_process = self.project.processes[2]
|
|
self.assertEqual(bg_process.name, "die")
|
|
self.assertEqual(bg_process.command, "python src/die_soon.py")
|
|
self.assertEqual(bg_process.project, self.project)
|
|
|
|
bg_process = self.project.processes[3]
|
|
self.assertEqual(bg_process.name, "exit")
|
|
self.assertEqual(bg_process.command, "python src/exit_soon.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")
|
|
|
|
def test_read_environment(self):
|
|
self.assertEqual(self.project.env["FOO"], "bar")
|
|
|
|
|
|
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")
|
|
|
|
def test_running_process(self):
|
|
self.assertFalse(self.process.is_running)
|
|
|
|
def test_unique_colour(self):
|
|
unique_colours = {proc.colour for proc in self.project.processes}
|
|
self.assertEqual(len(unique_colours), len(self.project.processes))
|
|
for colour in unique_colours:
|
|
self.assertIn(colour, ansi_colors)
|
|
|
|
|
|
class RunningProcessTestCase(BaseWorkerTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.project = Project(self.EXAMPLE_DIR)
|
|
self.process = self.project.get_process("web")
|
|
self.response = send_to_server(
|
|
PayloadType.PROCESS,
|
|
{"path": str(self.project.root), "process": str(self.process.name)},
|
|
)
|
|
|
|
def test_running_process(self):
|
|
self.assertTrue(self.process.is_running)
|
|
|
|
def test_pid(self):
|
|
self.assertEqual(self.process.pid, self.response["pid"])
|