← 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-07-05 14:15:26 UTC
  • mfrom: (1468.1.6 client-from-config-3)
  • Revision ID: aaron.bentley@canonical.com-20160705141526-v1gveg82a4swb3fb
Merged client-from-config-3 into client-from-config-4.

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()