← Back to branch summary

~abentley/juju-ci-tools/client-from-config-4

« back to all changes in this revision

Viewing changes to tests/test_timeout.py

  • Committer: Aaron Bentley
  • Date: 2014-02-24 17:18:29 UTC
  • mto: This revision was merged to the branch mainline in revision 252.
  • Revision ID: aaron.bentley@canonical.com-20140224171829-sz644yhoygu7m9dm
Use tags to identify and shut down instances.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_timeout.py
1
 
from argparse import Namespace
2
 
from contextlib import contextmanager
3
 
import datetime
4
 
import random
5
 
from signal import SIGTERM
6
 
from unittest import TestCase
7
 
 
8
 
from mock import (
9
 
    call,
10
 
    patch,
11
 
    )
12
 
 
13
 
from timeout import (
14
 
    main,
15
 
    parse_args,
16
 
    run_command,
17
 
    signals,
18
 
    )
19
 
from tests import parse_error
20
 
 
21
 
 
22
 
class TestParseArgs(TestCase):
23
 
 
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'])
28
 
 
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'])
33
 
 
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'])
38
 
 
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')
44
 
 
45
 
 
46
 
class TestMain(TestCase):
47
 
 
48
 
    def test_main(self):
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'])
53
 
 
54
 
 
55
 
class TestRunCommand(TestCase):
56
 
 
57
 
    @contextmanager
58
 
    def patch_po(self):
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
62
 
 
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']),
68
 
                             123)
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)
72
 
 
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']),
77
 
                             123)
78
 
        self.assertEqual(
79
 
            poll_mock.mock_calls, [call(), call(), call()])
80
 
        self.assertEqual(
81
 
            sleep_mock.mock_calls, [call(0.1), call(0.1)])
82
 
 
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']),
92
 
                                 124)
93
 
        self.assertEqual(
94
 
            poll_mock.mock_calls, [call(), call()])
95
 
        self.assertEqual(
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()