1
from argparse import Namespace
2
from contextlib import contextmanager
5
from signal import SIGTERM
6
from unittest import TestCase
19
from tests import parse_error
22
class TestParseArgs(TestCase):
24
def test_parse_args(self):
25
args, command = parse_args(['500', 'foo', 'bar'])
26
self.assertEqual(args, Namespace(duration=500.0, signal='TERM'))
27
self.assertEqual(command, ['foo', 'bar'])
29
def test_parse_args_command_options(self):
30
args, command = parse_args(['500', 'foo', '--bar'])
31
self.assertEqual(args, Namespace(duration=500.0, signal='TERM'))
32
self.assertEqual(command, ['foo', '--bar'])
34
def test_parse_args_signal(self):
35
args, command = parse_args(['500', '--', 'foo', '--signal'])
36
self.assertEqual(args, Namespace(duration=500.0, signal='TERM'))
37
self.assertEqual(command, ['foo', '--signal'])
39
def test_parse_args_signal_novalue(self):
40
with parse_error(self) as stderr:
41
args, command = parse_args(['500', 'foo', '--signal'])
42
self.assertRegexpMatches(
43
stderr.getvalue(), 'argument --signal: expected one argument')
46
class TestMain(TestCase):
49
signal_name, signal_value = random.choice(signals.items())
50
with patch('timeout.run_command', autospec=True) as rc_mock:
51
main(['500', '--signal', signal_name, 'foo', 'bar'])
52
rc_mock.assert_called_once_with(500, signal_value, ['foo', 'bar'])
55
class TestRunCommand(TestCase):
59
with patch('subprocess.Popen', autospec=True) as po_mock:
60
with patch('time.sleep') as sleep_mock:
61
yield po_mock, po_mock.return_value.poll, sleep_mock
63
def test_run_and_poll(self):
64
with self.patch_po() as (po_mock, poll_mock, sleep_mock):
65
poll_mock = po_mock.return_value.poll
66
poll_mock.return_value = 123
67
self.assertEqual(run_command(57.5, SIGTERM, ['ls', 'foo']),
69
po_mock.assert_called_once_with(['ls', 'foo'], creationflags=0)
70
poll_mock.assert_called_once_with()
71
self.assertEqual(sleep_mock.call_count, 0)
73
def test_multiple_polls(self):
74
with self.patch_po() as (po_mock, poll_mock, sleep_mock):
75
poll_mock.side_effect = [None, None, 123, 124]
76
self.assertEqual(run_command(57.5, SIGTERM, ['ls', 'foo']),
79
poll_mock.mock_calls, [call(), call(), call()])
81
sleep_mock.mock_calls, [call(0.1), call(0.1)])
83
def test_duration_elapsed(self):
84
start = datetime.datetime(2015, 1, 1)
85
middle = start + datetime.timedelta(seconds=57.4)
86
end = start + datetime.timedelta(seconds=57.6)
87
with self.patch_po() as (po_mock, poll_mock, sleep_mock):
88
poll_mock.side_effect = [None, None, None, None]
89
with patch('utility.until_timeout.now') as utn_mock:
90
utn_mock.side_effect = [start, middle, end, end]
91
self.assertEqual(run_command(57.5, SIGTERM, ['ls', 'foo']),
94
poll_mock.mock_calls, [call(), call()])
96
sleep_mock.mock_calls, [call(0.1), call(0.1)])
97
self.assertEqual(utn_mock.mock_calls, [call(), call(), call()])
98
po_mock.return_value.send_signal.assert_called_once_with(SIGTERM)
99
po_mock.return_value.wait.assert_called_once_with()