← Back to branch summary

~oddbloke/cloud-init/fix-gce-az

« back to all changes in this revision

Viewing changes to tests/unittests/test_rh_subscription.py

  • Committer: Daniel Watkins
  • Date: 2015-07-22 12:06:34 UTC
  • Revision ID: daniel.watkins@canonical.com-20150722120634-wsg8rwzcaanhc2pn
Make full data source available to code that handles mirror selection.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/unittests/test_rh_subscription.py
 
1
from cloudinit import util
 
2
from cloudinit.config import cc_rh_subscription
 
3
import logging
 
4
import mock
 
5
import unittest
 
6
 
 
7
 
 
8
class GoodTests(unittest.TestCase):
 
9
    def setUp(self):
 
10
        super(GoodTests, self).setUp()
 
11
        self.name = "cc_rh_subscription"
 
12
        self.cloud_init = None
 
13
        self.log = logging.getLogger("good_tests")
 
14
        self.args = []
 
15
        self.handle = cc_rh_subscription.handle
 
16
        self.SM = cc_rh_subscription.SubscriptionManager
 
17
 
 
18
        self.config = {'rh_subscription':
 
19
                       {'username': 'scooby@do.com',
 
20
                        'password': 'scooby-snacks'
 
21
                        }}
 
22
        self.config_full = {'rh_subscription':
 
23
                            {'username': 'scooby@do.com',
 
24
                             'password': 'scooby-snacks',
 
25
                             'auto-attach': True,
 
26
                             'service-level': 'self-support',
 
27
                             'add-pool': ['pool1', 'pool2', 'pool3'],
 
28
                             'enable-repo': ['repo1', 'repo2', 'repo3'],
 
29
                             'disable-repo': ['repo4', 'repo5']
 
30
                             }}
 
31
 
 
32
    def test_already_registered(self):
 
33
        '''
 
34
        Emulates a system that is already registered. Ensure it gets
 
35
        a non-ProcessExecution error from is_registered()
 
36
        '''
 
37
        with mock.patch.object(cc_rh_subscription.SubscriptionManager,
 
38
                               '_sub_man_cli') as mockobj:
 
39
            self.SM.log_success = mock.MagicMock()
 
40
            self.handle(self.name, self.config, self.cloud_init,
 
41
                        self.log, self.args)
 
42
            self.assertEqual(self.SM.log_success.call_count, 1)
 
43
            self.assertEqual(mockobj.call_count, 1)
 
44
 
 
45
    def test_simple_registration(self):
 
46
        '''
 
47
        Simple registration with username and password
 
48
        '''
 
49
        self.SM.log_success = mock.MagicMock()
 
50
        reg = "The system has been registered with ID:" \
 
51
              " 12345678-abde-abcde-1234-1234567890abc"
 
52
        self.SM._sub_man_cli = mock.MagicMock(
 
53
            side_effect=[util.ProcessExecutionError, (reg, 'bar')])
 
54
        self.handle(self.name, self.config, self.cloud_init,
 
55
                    self.log, self.args)
 
56
        self.assertIn(mock.call(['identity']),
 
57
                      self.SM._sub_man_cli.call_args_list)
 
58
        self.assertIn(mock.call(['register', '--username=scooby@do.com',
 
59
                                 '--password=scooby-snacks'],
 
60
                                logstring_val=True),
 
61
                      self.SM._sub_man_cli.call_args_list)
 
62
 
 
63
        self.assertEqual(self.SM.log_success.call_count, 1)
 
64
        self.assertEqual(self.SM._sub_man_cli.call_count, 2)
 
65
 
 
66
    def test_full_registration(self):
 
67
        '''
 
68
        Registration with auto-attach, service-level, adding pools,
 
69
        and enabling and disabling yum repos
 
70
        '''
 
71
        call_lists = []
 
72
        call_lists.append(['attach', '--pool=pool1', '--pool=pool3'])
 
73
        call_lists.append(['repos', '--enable=repo2', '--enable=repo3',
 
74
                           '--disable=repo5'])
 
75
        call_lists.append(['attach', '--auto', '--servicelevel=self-support'])
 
76
        self.SM.log_success = mock.MagicMock()
 
77
        reg = "The system has been registered with ID:" \
 
78
              " 12345678-abde-abcde-1234-1234567890abc"
 
79
        self.SM._sub_man_cli = mock.MagicMock(
 
80
            side_effect=[util.ProcessExecutionError, (reg, 'bar'),
 
81
                         ('Service level set to: self-support', ''),
 
82
                         ('pool1\npool3\n', ''), ('pool2\n', ''), ('', ''),
 
83
                         ('Repo ID: repo1\nRepo ID: repo5\n', ''),
 
84
                         ('Repo ID: repo2\nRepo ID: repo3\nRepo ID: '
 
85
                          'repo4', ''),
 
86
                         ('', '')])
 
87
        self.handle(self.name, self.config_full, self.cloud_init,
 
88
                    self.log, self.args)
 
89
        for call in call_lists:
 
90
            self.assertIn(mock.call(call), self.SM._sub_man_cli.call_args_list)
 
91
        self.assertEqual(self.SM.log_success.call_count, 1)
 
92
        self.assertEqual(self.SM._sub_man_cli.call_count, 9)
 
93
 
 
94
 
 
95
class TestBadInput(unittest.TestCase):
 
96
    name = "cc_rh_subscription"
 
97
    cloud_init = None
 
98
    log = logging.getLogger("bad_tests")
 
99
    args = []
 
100
    SM = cc_rh_subscription.SubscriptionManager
 
101
    reg = "The system has been registered with ID:" \
 
102
          " 12345678-abde-abcde-1234-1234567890abc"
 
103
 
 
104
    config_no_password = {'rh_subscription':
 
105
                          {'username': 'scooby@do.com'
 
106
                           }}
 
107
 
 
108
    config_no_key = {'rh_subscription':
 
109
                     {'activation-key': '1234abcde',
 
110
                      }}
 
111
 
 
112
    config_service = {'rh_subscription':
 
113
                      {'username': 'scooby@do.com',
 
114
                       'password': 'scooby-snacks',
 
115
                       'service-level': 'self-support'
 
116
                       }}
 
117
 
 
118
    config_badpool = {'rh_subscription':
 
119
                      {'username': 'scooby@do.com',
 
120
                       'password': 'scooby-snacks',
 
121
                       'add-pool': 'not_a_list'
 
122
                       }}
 
123
    config_badrepo = {'rh_subscription':
 
124
                      {'username': 'scooby@do.com',
 
125
                       'password': 'scooby-snacks',
 
126
                       'enable-repo': 'not_a_list'
 
127
                       }}
 
128
    config_badkey = {'rh_subscription':
 
129
                     {'activation_key': 'abcdef1234',
 
130
                      'org': '123',
 
131
                      }}
 
132
 
 
133
    def setUp(self):
 
134
        super(TestBadInput, self).setUp()
 
135
        self.handle = cc_rh_subscription.handle
 
136
 
 
137
    def test_no_password(self):
 
138
        '''
 
139
        Attempt to register without the password key/value
 
140
        '''
 
141
        self.input_is_missing_data(self.config_no_password)
 
142
 
 
143
    def test_no_org(self):
 
144
        '''
 
145
        Attempt to register without the org key/value
 
146
        '''
 
147
        self.input_is_missing_data(self.config_no_key)
 
148
 
 
149
    def test_service_level_without_auto(self):
 
150
        '''
 
151
        Attempt to register using service-level without the auto-attach key
 
152
        '''
 
153
        self.SM.log_warn = mock.MagicMock()
 
154
        self.SM._sub_man_cli = mock.MagicMock(
 
155
            side_effect=[util.ProcessExecutionError, (self.reg, 'bar')])
 
156
        self.handle(self.name, self.config_service, self.cloud_init,
 
157
                    self.log, self.args)
 
158
        self.assertEqual(self.SM._sub_man_cli.call_count, 1)
 
159
        self.assertEqual(self.SM.log_warn.call_count, 2)
 
160
 
 
161
    def test_pool_not_a_list(self):
 
162
        '''
 
163
        Register with pools that are not in the format of a list
 
164
        '''
 
165
        self.SM.log_warn = mock.MagicMock()
 
166
        self.SM._sub_man_cli = mock.MagicMock(
 
167
            side_effect=[util.ProcessExecutionError, (self.reg, 'bar')])
 
168
        self.handle(self.name, self.config_badpool, self.cloud_init,
 
169
                    self.log, self.args)
 
170
        self.assertEqual(self.SM._sub_man_cli.call_count, 2)
 
171
        self.assertEqual(self.SM.log_warn.call_count, 2)
 
172
 
 
173
    def test_repo_not_a_list(self):
 
174
        '''
 
175
        Register with repos that are not in the format of a list
 
176
        '''
 
177
        self.SM.log_warn = mock.MagicMock()
 
178
        self.SM._sub_man_cli = mock.MagicMock(
 
179
            side_effect=[util.ProcessExecutionError, (self.reg, 'bar')])
 
180
        self.handle(self.name, self.config_badrepo, self.cloud_init,
 
181
                    self.log, self.args)
 
182
        self.assertEqual(self.SM.log_warn.call_count, 3)
 
183
        self.assertEqual(self.SM._sub_man_cli.call_count, 2)
 
184
 
 
185
    def test_bad_key_value(self):
 
186
        '''
 
187
        Attempt to register with a key that we don't know
 
188
        '''
 
189
        self.SM.log_warn = mock.MagicMock()
 
190
        self.SM._sub_man_cli = mock.MagicMock(
 
191
            side_effect=[util.ProcessExecutionError, (self.reg, 'bar')])
 
192
        self.handle(self.name, self.config_badkey, self.cloud_init,
 
193
                    self.log, self.args)
 
194
        self.assertEqual(self.SM.log_warn.call_count, 2)
 
195
        self.assertEqual(self.SM._sub_man_cli.call_count, 1)
 
196
 
 
197
    def input_is_missing_data(self, config):
 
198
        '''
 
199
        Helper def for tests that having missing information
 
200
        '''
 
201
        self.SM.log_warn = mock.MagicMock()
 
202
        self.SM._sub_man_cli = mock.MagicMock(
 
203
            side_effect=[util.ProcessExecutionError])
 
204
        self.handle(self.name, config, self.cloud_init,
 
205
                    self.log, self.args)
 
206
        self.SM._sub_man_cli.assert_called_with(['identity'])
 
207
        self.assertEqual(self.SM.log_warn.call_count, 4)
 
208
        self.assertEqual(self.SM._sub_man_cli.call_count, 1)