Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
aws
GitHub Repository: aws/aws-cli
 Path: blob/develop/tests/unit/customizations/configservice/test_getstatus.py
1569 views
1
# Copyright 2014 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
from awscli.compat import StringIO

14
from awscli.testutils import mock, unittest

15
from awscli.customizations.configservice.getstatus import GetStatusCommand

16


17


18
class TestGetStatusCommand(unittest.TestCase):

19
    def setUp(self):

20
        self.session = mock.Mock()

21


22
        # Setup the config client mock.

23
        self.config_client = mock.Mock()

24
        self.session.create_client.return_value = self.config_client

25


26
        # Create some handles to control the description outputs.

27
        self.recorder_status = []

28
        self.channel_status = []

29


30
        # Set the output handles to the client.

31
        self.config_client.describe_configuration_recorder_status.\

32
            return_value = {'ConfigurationRecordersStatus':

33
                            self.recorder_status}

34
        self.config_client.describe_delivery_channel_status.\

35
            return_value = {'DeliveryChannelsStatus': self.channel_status}

36


37
        self.parsed_args = mock.Mock()

38
        self.parsed_globals = mock.Mock()

39
        self.cmd = GetStatusCommand(self.session)

40


41
    def _make_delivery_channel_status(self, name, stream_delivery_status,

42
                                      history_delivery_status,

43
                                      snapshot_delivery_status):

44
        status = {

45
            'name': 'default',

46
            'configStreamDeliveryInfo': stream_delivery_status,

47
            'configHistoryDeliveryInfo': history_delivery_status,

48
            'configSnapshotDeliveryInfo': snapshot_delivery_status

49
        }

50
        return status

51


52
    def test_create_client(self):

53
        # Set values for the parsed globals.

54
        self.parsed_globals.verify_ssl = True

55
        self.parsed_globals.region = 'us-east-1'

56
        self.parsed_globals.endpoint_url = 'https://foo.com'

57


58
        # Ensure that the client was built with the proper arguments.

59
        self.cmd._run_main(self.parsed_args, self.parsed_globals)

60
        self.session.create_client.assert_called_with(

61
            'config',

62
            verify=self.parsed_globals.verify_ssl,

63
            region_name=self.parsed_globals.region,

64
            endpoint_url=self.parsed_globals.endpoint_url

65
        )

66


67
    def test_configuration_recorder_success(self):

68
        status = {'name': 'default', 'recording': True,

69
                  'lastStatus': 'SUCCESS'}

70
        self.recorder_status.append(status)

71


72
        expected_output = (

73
            'Configuration Recorders:\n\n'

74
            'name: default\n'

75
            'recorder: ON\n'

76
            'last status: SUCCESS\n\n'

77
            'Delivery Channels:\n\n'

78
        )

79


80
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

81
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

82
            self.assertEqual(expected_output, mock_stdout.getvalue())

83


84
    def test_configuration_recorder_fail(self):

85
        status = {'name': 'default', 'recording': True,

86
                  'lastStatus': 'FAILURE', 'lastErrorCode': '500',

87
                  'lastErrorMessage': 'This is the error'}

88
        self.recorder_status.append(status)

89


90
        expected_output = (

91
            'Configuration Recorders:\n\n'

92
            'name: default\n'

93
            'recorder: ON\n'

94
            'last status: FAILURE\n'

95
            'error code: 500\n'

96
            'message: This is the error\n\n'

97
            'Delivery Channels:\n\n'

98
        )

99


100
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

101
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

102
            self.assertEqual(expected_output, mock_stdout.getvalue())

103


104
    def test_configuration_recorder_off(self):

105
        status = {'name': 'default', 'recording': False}

106
        self.recorder_status.append(status)

107


108
        expected_output = (

109
            'Configuration Recorders:\n\n'

110
            'name: default\n'

111
            'recorder: OFF\n\n'

112
            'Delivery Channels:\n\n'

113
        )

114


115
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

116
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

117
            self.assertEqual(expected_output, mock_stdout.getvalue())

118


119
    def test_multiple_configuration_recorders(self):

120
        status = {'name': 'default', 'recording': True,

121
                  'lastStatus': 'SUCCESS'}

122
        self.recorder_status.append(status)

123


124
        status = {'name': 'default', 'recording': True,

125
                  'lastStatus': 'FAILURE', 'lastErrorCode': '500',

126
                  'lastErrorMessage': 'This is the error'}

127
        self.recorder_status.append(status)

128


129
        status = {'name': 'default', 'recording': False}

130
        self.recorder_status.append(status)

131


132
        expected_output = (

133
            'Configuration Recorders:\n\n'

134
            'name: default\n'

135
            'recorder: ON\n'

136
            'last status: SUCCESS\n\n'

137
            'name: default\n'

138
            'recorder: ON\n'

139
            'last status: FAILURE\n'

140
            'error code: 500\n'

141
            'message: This is the error\n\n'

142
            'name: default\n'

143
            'recorder: OFF\n\n'

144
            'Delivery Channels:\n\n'

145
        )

146
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

147
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

148
            self.assertEqual(expected_output, mock_stdout.getvalue())

149


150
    def test_delivery_channel_success_single_delivery_info(self):

151
        name = 'default'

152
        success = {'lastStatus': 'SUCCESS'}

153


154
        stream_delivery_status = success

155
        history_delivery_status = {}

156
        snapshot_delivery_status = {}

157


158
        status = self._make_delivery_channel_status(

159
            name, stream_delivery_status=stream_delivery_status,

160
            history_delivery_status=history_delivery_status,

161
            snapshot_delivery_status=snapshot_delivery_status

162
        )

163
        self.channel_status.append(status)

164


165
        expected_output = (

166
            'Configuration Recorders:\n\n'

167
            'Delivery Channels:\n\n'

168
            'name: default\n'

169
            'last stream delivery status: SUCCESS\n\n'

170
        )

171


172
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

173
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

174
            self.assertEqual(expected_output, mock_stdout.getvalue())

175


176
    def test_delivery_channel_success_multiple_delivery_info(self):

177
        name = 'default'

178
        success = {'lastStatus': 'SUCCESS'}

179


180
        stream_delivery_status = success

181
        history_delivery_status = success

182
        snapshot_delivery_status = success

183


184
        status = self._make_delivery_channel_status(

185
            name, stream_delivery_status=stream_delivery_status,

186
            history_delivery_status=history_delivery_status,

187
            snapshot_delivery_status=snapshot_delivery_status

188
        )

189
        self.channel_status.append(status)

190


191
        expected_output = (

192
            'Configuration Recorders:\n\n'

193
            'Delivery Channels:\n\n'

194
            'name: default\n'

195
            'last stream delivery status: SUCCESS\n'

196
            'last history delivery status: SUCCESS\n'

197
            'last snapshot delivery status: SUCCESS\n\n'

198
        )

199


200
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

201
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

202
            self.assertEqual(expected_output, mock_stdout.getvalue())

203


204
    def test_delivery_channel_fail_single_delivery_info(self):

205
        name = 'default'

206
        failure = {'lastStatus': 'FAILURE', 'lastErrorCode': '500',

207
                   'lastErrorMessage': 'This is the error'}

208


209
        stream_delivery_status = failure

210
        history_delivery_status = {}

211
        snapshot_delivery_status = {}

212


213
        status = self._make_delivery_channel_status(

214
            name, stream_delivery_status=stream_delivery_status,

215
            history_delivery_status=history_delivery_status,

216
            snapshot_delivery_status=snapshot_delivery_status

217
        )

218
        self.channel_status.append(status)

219


220
        expected_output = (

221
            'Configuration Recorders:\n\n'

222
            'Delivery Channels:\n\n'

223
            'name: default\n'

224
            'last stream delivery status: FAILURE\n'

225
            'error code: 500\n'

226
            'message: This is the error\n\n'

227
        )

228


229
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

230
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

231
            self.assertEqual(expected_output, mock_stdout.getvalue())

232


233
    def test_delivery_channel_mixed_multiple_delivery_info(self):

234
        name = 'default'

235
        success = {'lastStatus': 'SUCCESS'}

236
        failure = {'lastStatus': 'FAILURE', 'lastErrorCode': '500',

237
                   'lastErrorMessage': 'This is the error'}

238


239
        stream_delivery_status = failure

240
        history_delivery_status = success

241
        snapshot_delivery_status = success

242


243
        status = self._make_delivery_channel_status(

244
            name, stream_delivery_status=stream_delivery_status,

245
            history_delivery_status=history_delivery_status,

246
            snapshot_delivery_status=snapshot_delivery_status

247
        )

248
        self.channel_status.append(status)

249


250
        expected_output = (

251
            'Configuration Recorders:\n\n'

252
            'Delivery Channels:\n\n'

253
            'name: default\n'

254
            'last stream delivery status: FAILURE\n'

255
            'error code: 500\n'

256
            'message: This is the error\n'

257
            'last history delivery status: SUCCESS\n'

258
            'last snapshot delivery status: SUCCESS\n\n'

259
        )

260


261
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

262
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

263
            self.assertEqual(expected_output, mock_stdout.getvalue())

264


265
    def test_multiple_delivery_channels(self):

266
        name = 'default'

267
        success = {'lastStatus': 'SUCCESS'}

268
        failure = {'lastStatus': 'FAILURE', 'lastErrorCode': '500',

269
                   'lastErrorMessage': 'This is the error'}

270


271
        stream_delivery_status = failure

272
        history_delivery_status = success

273
        snapshot_delivery_status = success

274


275
        status = self._make_delivery_channel_status(

276
            name, stream_delivery_status=stream_delivery_status,

277
            history_delivery_status=history_delivery_status,

278
            snapshot_delivery_status=snapshot_delivery_status

279
        )

280
        self.channel_status.append(status)

281
        self.channel_status.append(status)

282


283
        expected_output = (

284
            'Configuration Recorders:\n\n'

285
            'Delivery Channels:\n\n'

286
            'name: default\n'

287
            'last stream delivery status: FAILURE\n'

288
            'error code: 500\n'

289
            'message: This is the error\n'

290
            'last history delivery status: SUCCESS\n'

291
            'last snapshot delivery status: SUCCESS\n\n'

292
            'name: default\n'

293
            'last stream delivery status: FAILURE\n'

294
            'error code: 500\n'

295
            'message: This is the error\n'

296
            'last history delivery status: SUCCESS\n'

297
            'last snapshot delivery status: SUCCESS\n\n'

298
        )

299


300
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

301
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

302
            self.assertEqual(expected_output, mock_stdout.getvalue())

303


304
    def test_full_get_status(self):

305
        # Create the configuration recorder statuses.

306
        status = {'name': 'default', 'recording': True,

307
                  'lastStatus': 'SUCCESS'}

308
        self.recorder_status.append(status)

309


310
        status = {'name': 'default', 'recording': True,

311
                  'lastStatus': 'FAILURE', 'lastErrorCode': '500',

312
                  'lastErrorMessage': 'This is the error'}

313
        self.recorder_status.append(status)

314


315
        status = {'name': 'default', 'recording': False}

316
        self.recorder_status.append(status)

317


318
        # Create the delivery channel statuses.

319
        name = 'default'

320
        success = {'lastStatus': 'SUCCESS'}

321
        failure = {'lastStatus': 'FAILURE', 'lastErrorCode': '500',

322
                   'lastErrorMessage': 'This is the error'}

323


324
        stream_delivery_status = failure

325
        history_delivery_status = success

326
        snapshot_delivery_status = success

327


328
        status = self._make_delivery_channel_status(

329
            name, stream_delivery_status=stream_delivery_status,

330
            history_delivery_status=history_delivery_status,

331
            snapshot_delivery_status=snapshot_delivery_status

332
        )

333
        self.channel_status.append(status)

334
        self.channel_status.append(status)

335


336
        expected_output = (

337
            'Configuration Recorders:\n\n'

338
            'name: default\n'

339
            'recorder: ON\n'

340
            'last status: SUCCESS\n\n'

341
            'name: default\n'

342
            'recorder: ON\n'

343
            'last status: FAILURE\n'

344
            'error code: 500\n'

345
            'message: This is the error\n\n'

346
            'name: default\n'

347
            'recorder: OFF\n\n'

348
            'Delivery Channels:\n\n'

349
            'name: default\n'

350
            'last stream delivery status: FAILURE\n'

351
            'error code: 500\n'

352
            'message: This is the error\n'

353
            'last history delivery status: SUCCESS\n'

354
            'last snapshot delivery status: SUCCESS\n\n'

355
            'name: default\n'

356
            'last stream delivery status: FAILURE\n'

357
            'error code: 500\n'

358
            'message: This is the error\n'

359
            'last history delivery status: SUCCESS\n'

360
            'last snapshot delivery status: SUCCESS\n\n'

361
        )

362


363
        with mock.patch('sys.stdout', StringIO()) as mock_stdout:

364
            self.cmd._run_main(self.parsed_args, self.parsed_globals)

365
            self.assertEqual(expected_output, mock_stdout.getvalue())

366


367