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"]) def test_port(self): self.assertIsNotNone(self.process.port)