← Back to branch summary

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

« back to all changes in this revision

Viewing changes to tests/test_concurrently.py

  • Committer: Aaron Bentley
  • Date: 2016-03-18 14:47:06 UTC
  • mto: This revision was merged to the branch mainline in revision 1324.
  • Revision ID: aaron.bentley@canonical.com-20160318144706-z7wy9c21m3psi6g5
Introduce set_model_name, update tests, check controller on bootstrap.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_concurrently.py
1
 
import logging
2
 
from mock import (
3
 
    Mock,
4
 
    patch,
5
 
    )
6
 
import os
7
 
 
8
 
import concurrently
9
 
from tests import (
10
 
    TestCase,
11
 
)
12
 
from utility import temp_dir
13
 
 
14
 
 
15
 
class ConcurrentlyTest(TestCase):
16
 
 
17
 
    log_level = logging.ERROR
18
 
 
19
 
    def test_main(self):
20
 
        with patch('concurrently.run_all') as r_mock:
21
 
            with patch('concurrently.summarise_tasks',
22
 
                       return_value=0) as s_mock:
23
 
                returncode = concurrently.main(
24
 
                    ['-v', '-l', '.', 'one=foo a b', 'two=bar c'])
25
 
        self.assertEqual(0, returncode)
26
 
        task_one = concurrently.Task('one=foo a b')
27
 
        task_two = concurrently.Task('two=bar c')
28
 
        r_mock.assert_called_once_with([task_one, task_two])
29
 
        s_mock.assert_called_once_with([task_one, task_two])
30
 
 
31
 
    @patch('sys.stderr')
32
 
    def test_main_error(self, se_mock):
33
 
        with patch('concurrently.run_all', side_effect=ValueError('bad')):
34
 
            returncode = concurrently.main(['-v', 'one=foo a b', 'two=bar c'])
35
 
        self.assertEqual(253, returncode)
36
 
        self.assertIn('ERROR bad', self.log_stream.getvalue())
37
 
 
38
 
    def test_parse_args(self):
39
 
        args = concurrently.parse_args(
40
 
            ['-v', '-l', '~/logs', 'one=foo a b', 'two=bar c'])
41
 
        self.assertEqual(logging.DEBUG, args.verbose)
42
 
        self.assertEqual(os.path.expanduser('~/logs'), args.log_dir)
43
 
        self.assertEqual(
44
 
            ['one=foo a b', 'two=bar c'], args.tasks)
45
 
 
46
 
    def test_summarise_tasks(self):
47
 
        task_one = concurrently.Task('one=foo a')
48
 
        task_one.returncode = 0
49
 
        task_two = concurrently.Task('two=bar b')
50
 
        task_two.returncode = 0
51
 
        tasks = [task_one, task_two]
52
 
        self.assertEqual(0, concurrently.summarise_tasks(tasks))
53
 
        task_one.returncode = 1
54
 
        self.assertEqual(1, concurrently.summarise_tasks(tasks))
55
 
        task_two.returncode = 3
56
 
        self.assertEqual(4, concurrently.summarise_tasks(tasks))
57
 
 
58
 
    def test_run_all(self):
59
 
        task_one = concurrently.Task('one=foo a')
60
 
        task_two = concurrently.Task('two=bar b')
61
 
        mutable_tasks = [task_one, task_two]
62
 
        with patch.object(task_one, 'finish', autospec=True) as f1_mock:
63
 
            with patch.object(task_one, 'start', autospec=True) as s1_mock:
64
 
                with patch.object(task_two, 'finish',
65
 
                                  autospec=True,) as f2_mock:
66
 
                    with patch.object(task_two, 'start',
67
 
                                      autospec=True) as s2_mock:
68
 
                        concurrently.run_all(mutable_tasks)
69
 
        s1_mock.assert_called_once_with()
70
 
        f1_mock.assert_called_once_with()
71
 
        s2_mock.assert_called_once_with()
72
 
        f2_mock.assert_called_once_with()
73
 
        self.assertEqual([], mutable_tasks)
74
 
 
75
 
 
76
 
class TaskTest(TestCase):
77
 
 
78
 
    def test_init(self):
79
 
        with temp_dir() as base:
80
 
            task = concurrently.Task('one=foo a b c', log_dir=base)
81
 
            self.assertEqual('one', task.name)
82
 
            self.assertEqual('foo a b c', task.commandline)
83
 
            self.assertEqual(['foo', 'a', 'b', 'c'], task.command)
84
 
            self.assertEqual(
85
 
                os.path.join(base, 'one-out.log'), task.out_log_name)
86
 
            self.assertEqual(
87
 
                os.path.join(base, 'one-err.log'), task.err_log_name)
88
 
            self.assertIsNone(task.returncode)
89
 
            self.assertIsNone(task.proc)
90
 
 
91
 
    def test_start(self):
92
 
        with temp_dir() as base:
93
 
            task = concurrently.Task('one=foo a', log_dir=base)
94
 
            with patch('subprocess.Popen',
95
 
                       autospec=True, return_value='proc') as p_mock:
96
 
                with task.start() as proc:
97
 
                    self.assertEqual('proc', proc)
98
 
                    self.assertEqual('proc', task.proc)
99
 
                    self.assertEqual(1, p_mock.call_count)
100
 
                    args, kwargs = p_mock.call_args
101
 
                    self.assertEqual((['foo', 'a'], ), args)
102
 
                    kwargs['stdout'].write('out')
103
 
                    kwargs['stderr'].write('err')
104
 
            self.assertIs(
105
 
                True,
106
 
                os.path.exists(os.path.join(base, 'one-out.log')))
107
 
            self.assertIs(
108
 
                True,
109
 
                os.path.exists(os.path.join(base, 'one-out.log')))
110
 
 
111
 
    def test_finish(self):
112
 
        task = concurrently.Task('one=foo a')
113
 
        task.proc = Mock(spec=['wait'])
114
 
        task.finish()
115
 
        task.proc.wait.assert_called_once_with()