Spaces:
Running
Running
| import unittest | |
| from unittest.mock import MagicMock, Mock, patch | |
| import start | |
| class StartupTest(unittest.TestCase): | |
| def test_waits_for_analysis_health_before_accepting_requests(self): | |
| process = Mock() | |
| process.poll.return_value = None | |
| response = MagicMock() | |
| response.__enter__.return_value.status = 200 | |
| response.__enter__.return_value.read.return_value = b'{"ok": true}' | |
| with patch.object(start, "urlopen", side_effect=[OSError("not listening"), response]) as probe, \ | |
| patch.object(start.time, "sleep") as sleep: | |
| self.assertTrue(start.wait_for_analysis(process, lambda: False)) | |
| self.assertEqual(probe.call_count, 2) | |
| sleep.assert_called_once_with(0.5) | |
| def test_exits_without_starting_node_if_analysis_fails(self): | |
| analysis = Mock() | |
| analysis.poll.return_value = 3 | |
| analysis.returncode = 3 | |
| with patch.object(start.subprocess, "Popen", return_value=analysis) as spawn, \ | |
| patch.object(start, "load_dotenv"), patch.object(start.signal, "signal"): | |
| self.assertEqual(start.main(), 1) | |
| spawn.assert_called_once() | |
| def test_times_out_if_analysis_never_becomes_ready(self): | |
| analysis = Mock() | |
| analysis.poll.return_value = None | |
| with patch.object(start, "urlopen", side_effect=OSError("not listening")), \ | |
| patch.object(start.time, "monotonic", side_effect=[0, 2]), \ | |
| patch.object(start.time, "sleep") as sleep: | |
| self.assertFalse(start.wait_for_analysis(analysis, lambda: False, timeout_seconds=1)) | |
| sleep.assert_not_called() | |
| def test_starts_node_after_analysis_is_ready(self): | |
| analysis = Mock() | |
| analysis.poll.return_value = None | |
| node = Mock() | |
| node.poll.return_value = 2 | |
| node.returncode = 2 | |
| with patch.object(start.subprocess, "Popen", side_effect=[analysis, node]) as spawn, \ | |
| patch.object(start, "load_dotenv"), patch.object(start.signal, "signal"), \ | |
| patch.object(start, "wait_for_analysis", side_effect=lambda *_: spawn.call_count == 1) as readiness: | |
| self.assertEqual(start.main(), 2) | |
| self.assertEqual(spawn.call_count, 2) | |
| self.assertEqual(spawn.call_args_list[1].args[0], ["node", "dist/server.js"]) | |
| readiness.assert_called_once() | |
| if __name__ == "__main__": | |
| unittest.main() | |