Spaces:
Running
Running
File size: 2,437 Bytes
5208038 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | 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()
|