CoCalc -- test_systems.py

GitHub Repository: aws/aws-cli
Path: blob/develop/tests/unit/customizations/codedeploy/test_systems.py
²⁶³⁷ views
1
# Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
2
#
3
# Licensed under the Apache License, Version 2.0 (the "License"). You
4
# may not use this file except in compliance with the License. A copy of
5
# the License is located at
6
#
7
# http://aws.amazon.com/apache2.0/
8
#
9
# or in the "license" file accompanying this file. This file is
10
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11
# ANY KIND, either express or implied. See the License for the specific
12
# language governing permissions and limitations under the License.
13

14
import subprocess
15

16
from argparse import Namespace
17
from awscli.customizations.codedeploy.systems import Windows, Ubuntu, RHEL
18
from awscli.testutils import mock, unittest
19

20

21
class TestWindows(unittest.TestCase):
22
    def setUp(self):
23
        self.popen_patcher = mock.patch('subprocess.Popen')
24
        self.popen = self.popen_patcher.start()
25

26
        self.check_call_patcher = mock.patch('subprocess.check_call')
27
        self.check_call = self.check_call_patcher.start()
28

29
        self.open_patcher = mock.patch(
30
            'awscli.customizations.codedeploy.systems.open',
31
            mock.mock_open(), create=True
32
        )
33
        self.open = self.open_patcher.start()
34

35
        self.config_dir = r'C:\ProgramData\Amazon\CodeDeploy'
36
        self.config_file = 'conf.onpremises.yml'
37
        self.config_path = r'{0}\{1}'.format(self.config_dir, self.config_file)
38
        self.installer = 'codedeploy-agent.msi'
39
        self.bucket = 'bucket'
40
        self.key = 'key'
41
        self.region = 'us-east-1'
42

43
        self.body = 'install-script'
44
        self.reader = mock.MagicMock()
45
        self.reader.read.return_value = self.body
46
        self.s3 = mock.MagicMock()
47
        self.s3.get_object.return_value = {'Body': self.reader}
48

49
        self.session = mock.MagicMock()
50
        self.session.create_client.return_value = self.s3
51

52
        self.params = Namespace()
53
        self.params.session = self.session
54
        self.params.region = self.region
55
        self.params.bucket = self.bucket
56
        self.params.key = self.key
57

58
        self.windows = Windows(self.params)
59

60
    def tearDown(self):
61
        self.popen_patcher.stop()
62
        self.check_call_patcher.stop()
63
        self.open_patcher.stop()
64

65
    def test_config_dir(self):
66
        self.assertEqual(self.config_dir, self.windows.CONFIG_DIR)
67

68
    def test_config_file(self):
69
        self.assertEqual(self.config_file, self.windows.CONFIG_FILE)
70

71
    def test_config_path(self):
72
        self.assertEqual(self.config_path, self.windows.CONFIG_PATH)
73

74
    def test_installer(self):
75
        self.assertEqual(self.installer, self.windows.INSTALLER)
76

77
    def test_install(self):
78
        process = mock.MagicMock()
79
        process.communicate.side_effect = [('', ''), ('Running', '')]
80
        process.returncode = 0
81
        self.popen.return_value = process
82
        self.windows.install(self.params)
83
        self.popen.assert_has_calls([
84
            mock.call(
85
                [
86
                    'powershell.exe',
87
                    '-Command', 'Stop-Service',
88
                    '-Name', 'codedeployagent'
89
                ],
90
                stdout=subprocess.PIPE,
91
                stderr=subprocess.PIPE
92
            ),
93
            mock.call().communicate(),
94
            mock.call(
95
                [
96
                    'powershell.exe',
97
                    '-Command', 'Get-Service',
98
                    '-Name', 'codedeployagent'
99
                ],
100
                stdout=subprocess.PIPE,
101
                stderr=subprocess.PIPE
102
            ),
103
            mock.call().communicate()
104
        ])
105
        self.check_call.assert_has_calls([
106
            mock.call(
107
                [
108
                    r'.\{0}'.format(self.installer),
109
                    '/quiet',
110
                    '/l', r'.\codedeploy-agent-install-log.txt'
111
                ],
112
                shell=True
113
            ),
114
            mock.call([
115
                'powershell.exe',
116
                '-Command', 'Restart-Service',
117
                '-Name', 'codedeployagent'
118
            ])
119
        ])
120
        self.open.assert_called_with(self.installer, 'wb')
121
        self.open().write.assert_called_with(self.body)
122

123
    def test_uninstall(self):
124
        process = mock.MagicMock()
125
        process.communicate.side_effect = [('', ''), ('', '')]
126
        process.returncode = 0
127
        self.popen.return_value = process
128
        self.windows.uninstall(self.params)
129
        self.popen.assert_has_calls([
130
            mock.call(
131
                [
132
                    'powershell.exe',
133
                    '-Command', 'Stop-Service',
134
                    '-Name', 'codedeployagent'
135
                ],
136
                stdout=subprocess.PIPE,
137
                stderr=subprocess.PIPE
138
            ),
139
            mock.call().communicate(),
140
            mock.call(
141
                [
142
                    'wmic',
143
                    'product', 'where', 'name="CodeDeploy Host Agent"',
144
                    'call', 'uninstall', '/nointeractive'
145
                ],
146
                stdout=subprocess.PIPE,
147
                stderr=subprocess.PIPE
148
            ),
149
            mock.call().communicate()
150
        ])
151

152

153
class TestLinux(unittest.TestCase):
154
    def setUp(self):
155
        self.popen_patcher = mock.patch('subprocess.Popen')
156
        self.popen = self.popen_patcher.start()
157

158
        self.check_call_patcher = mock.patch('subprocess.check_call')
159
        self.check_call = self.check_call_patcher.start()
160

161
        self.open_patcher = mock.patch(
162
            'awscli.customizations.codedeploy.systems.open',
163
            mock.mock_open(), create=True
164
        )
165
        self.open = self.open_patcher.start()
166

167
        self.environ_patcher = mock.patch('os.environ')
168
        self.environ = self.environ_patcher.start()
169
        self.environ.copy.return_value = dict()
170

171
        self.config_dir = '/etc/codedeploy-agent/conf'
172
        self.config_file = 'codedeploy.onpremises.yml'
173
        self.config_path = '{0}/{1}'.format(self.config_dir, self.config_file)
174
        self.installer = 'install'
175
        self.bucket = 'bucket'
176
        self.key = 'key'
177
        self.region = 'us-east-1'
178

179
        self.access_key_id = 'ACCESSKEYID'
180
        self.secret_access_key = 'SECRETACCESSKEY'
181
        self.session_token = 'SESSION_TOKEN'
182
        self.credentials = mock.MagicMock()
183
        self.credentials.access_key = self.access_key_id
184
        self.credentials.secret_key = self.secret_access_key
185
        self.credentials.token = self.session_token
186

187
        self.environment = dict({
188
            'AWS_REGION': self.region,
189
            'AWS_ACCESS_KEY_ID': self.access_key_id,
190
            'AWS_SECRET_ACCESS_KEY': self.secret_access_key,
191
            'AWS_SESSION_TOKEN': self.session_token
192
        })
193

194
        self.body = 'install-script'
195
        self.reader = mock.MagicMock()
196
        self.reader.read.return_value = self.body
197
        self.s3 = mock.MagicMock()
198
        self.s3.get_object.return_value = {'Body': self.reader}
199

200
        self.session = mock.MagicMock()
201
        self.session.create_client.return_value = self.s3
202
        self.session.get_credentials.return_value = self.credentials
203

204
        self.params = Namespace()
205
        self.params.session = self.session
206
        self.params.region = self.region
207
        self.params.bucket = self.bucket
208
        self.params.key = self.key
209

210
    def tearDown(self):
211
        self.popen_patcher.stop()
212
        self.check_call_patcher.stop()
213
        self.open_patcher.stop()
214
        self.environ_patcher.stop()
215

216

217
class TestUbuntu(TestLinux):
218
    def setUp(self):
219
        super(self.__class__, self).setUp()
220
        self.ubuntu = Ubuntu(self.params)
221

222
    def test_config_dir(self):
223
        self.assertEqual(self.config_dir, self.ubuntu.CONFIG_DIR)
224

225
    def test_config_file(self):
226
        self.assertEqual(self.config_file, self.ubuntu.CONFIG_FILE)
227

228
    def test_config_path(self):
229
        self.assertEqual(self.config_path, self.ubuntu.CONFIG_PATH)
230

231
    def test_installer(self):
232
        self.assertEqual(self.installer, self.ubuntu.INSTALLER)
233

234
    @mock.patch('os.geteuid', create=True)
235
    def test_validate_administrator_throws(self, geteuid):
236
        geteuid.return_value = 1
237
        with self.assertRaisesRegex(
238
                RuntimeError, 'You must run this command as sudo.'):
239
            self.ubuntu.validate_administrator()
240

241
    def test_install(self):
242
        process = mock.MagicMock()
243
        process.communicate.return_value = ('', '')
244
        process.returncode = 0
245
        self.popen.return_value = process
246
        self.ubuntu.install(self.params)
247
        self.popen.assert_has_calls([
248
            mock.call(
249
                ['service', 'codedeploy-agent', 'stop'],
250
                stdout=subprocess.PIPE,
251
                stderr=subprocess.PIPE
252
            ),
253
            mock.call().communicate()
254
        ])
255
        self.check_call.assert_has_calls([
256
            mock.call(['apt-get', '-y', 'update']),
257
            mock.call(['apt-get', '-y', 'install', 'ruby2.0']),
258
            mock.call(['chmod', '+x', './{0}'.format(self.installer)]),
259
            mock.call(
260
                ['./{0}'.format(self.installer), 'auto'],
261
                env=self.environment
262
            )
263
        ])
264
        self.open.assert_called_with(self.installer, 'wb')
265
        self.open().write.assert_called_with(self.body)
266

267
    def test_uninstall(self):
268
        process = mock.MagicMock()
269
        process.communicate.return_value = ('', '')
270
        process.returncode = 0
271
        self.popen.return_value = process
272
        self.ubuntu.uninstall(self.params)
273
        self.popen.assert_has_calls([
274
            mock.call(
275
                ['service', 'codedeploy-agent', 'stop'],
276
                stdout=subprocess.PIPE,
277
                stderr=subprocess.PIPE
278
            ),
279
            mock.call().communicate()
280
        ])
281
        self.check_call.assert_has_calls([
282
            mock.call(['dpkg', '-r', 'codedeploy-agent'])
283
        ])
284

285
class TestRHEL(TestLinux):
286
    def setUp(self):
287
        super(self.__class__, self).setUp()
288
        self.rhel = RHEL(self.params)
289

290
    def test_config_dir(self):
291
        self.assertEqual(self.config_dir, self.rhel.CONFIG_DIR)
292

293
    def test_config_file(self):
294
        self.assertEqual(self.config_file, self.rhel.CONFIG_FILE)
295

296
    def test_config_path(self):
297
        self.assertEqual(self.config_path, self.rhel.CONFIG_PATH)
298

299
    def test_installer(self):
300
        self.assertEqual(self.installer, self.rhel.INSTALLER)
301

302
    @mock.patch('os.geteuid', create=True)
303
    def test_validate_administrator_throws(self, geteuid):
304
        geteuid.return_value = 1
305
        with self.assertRaisesRegex(
306
                RuntimeError, 'You must run this command as sudo.'):
307
            self.rhel.validate_administrator()
308

309
    def test_install(self):
310
        process = mock.MagicMock()
311
        process.communicate.return_value = ('', '')
312
        process.returncode = 0
313
        self.popen.return_value = process
314
        self.rhel.install(self.params)
315
        self.popen.assert_has_calls([
316
            mock.call(
317
                ['service', 'codedeploy-agent', 'stop'],
318
                stdout=subprocess.PIPE,
319
                stderr=subprocess.PIPE
320
            ),
321
            mock.call().communicate()
322
        ])
323
        self.check_call.assert_has_calls([
324
            mock.call(['yum', '-y', 'install', 'ruby']),
325
            mock.call(['chmod', '+x', './{0}'.format(self.installer)]),
326
            mock.call(
327
                ['./{0}'.format(self.installer), 'auto'],
328
                env=self.environment
329
            )
330
        ])
331
        self.open.assert_called_with(self.installer, 'wb')
332
        self.open().write.assert_called_with(self.body)
333

334
    def test_uninstall(self):
335
        process = mock.MagicMock()
336
        process.communicate.return_value = ('', '')
337
        process.returncode = 0
338
        self.popen.return_value = process
339
        self.rhel.uninstall(self.params)
340
        self.popen.assert_has_calls([
341
            mock.call(
342
                ['service', 'codedeploy-agent', 'stop'],
343
                stdout=subprocess.PIPE,
344
                stderr=subprocess.PIPE
345
            ),
346
            mock.call().communicate()
347
        ])
348
        self.check_call.assert_has_calls([
349
            mock.call(['yum', '-y', 'erase', 'codedeploy-agent'])
350
        ])
351

352
if __name__ == "__main__":
353
    unittest.main()
354

355
Product

Resources

Company
