1import logging 2import random 3import string 4 5import salt.config 6import salt.loader 7import salt.modules.boto_cloudwatch_event as boto_cloudwatch_event 8from tests.support.mixins import LoaderModuleMockMixin 9from tests.support.mock import MagicMock, patch 10from tests.support.unit import TestCase, skipIf 11 12# pylint: disable=import-error,no-name-in-module,unused-import 13try: 14 import boto 15 import boto3 16 from botocore.exceptions import ClientError 17 from botocore import __version__ as found_botocore_version 18 19 HAS_BOTO = True 20except ImportError: 21 HAS_BOTO = False 22 23# pylint: enable=import-error,no-name-in-module,unused-import 24log = logging.getLogger(__name__) 25 26 27def _has_required_boto(): 28 """ 29 Returns True/False boolean depending on if Boto is installed and correct 30 version. 31 """ 32 if not HAS_BOTO: 33 return False 34 else: 35 return True 36 37 38if _has_required_boto(): 39 region = "us-east-1" 40 access_key = "GKTADJGHEIQSXMKKRBJ08H" 41 secret_key = "askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs" 42 conn_parameters = { 43 "region": region, 44 "key": access_key, 45 "keyid": secret_key, 46 "profile": {}, 47 } 48 error_message = ( 49 "An error occurred (101) when calling the {0} operation: Test-defined error" 50 ) 51 not_found_error = ClientError( 52 { 53 "Error": { 54 "Code": "ResourceNotFoundException", 55 "Message": "Test-defined error", 56 } 57 }, 58 "msg", 59 ) 60 error_content = {"Error": {"Code": 101, "Message": "Test-defined error"}} 61 rule_name = "test_thing_type" 62 rule_desc = "test_thing_type_desc" 63 rule_sched = "rate(20 min)" 64 rule_arn = "arn:::::rule/arn" 65 rule_ret = dict( 66 Arn=rule_arn, 67 Description=rule_desc, 68 EventPattern=None, 69 Name=rule_name, 70 RoleArn=None, 71 ScheduleExpression=rule_sched, 72 State="ENABLED", 73 ) 74 create_rule_ret = dict( 75 Name=rule_name, 76 ) 77 target_ret = dict( 78 Id="target1", 79 ) 80 81 82class BotoCloudWatchEventTestCaseBase(TestCase, LoaderModuleMockMixin): 83 conn = None 84 85 def setup_loader_modules(self): 86 self.opts = opts = salt.config.DEFAULT_MINION_OPTS.copy() 87 utils = salt.loader.utils( 88 opts, whitelist=["boto3", "args", "systemd", "path", "platform"], context={} 89 ) 90 return {boto_cloudwatch_event: {"__utils__": utils}} 91 92 def setUp(self): 93 super().setUp() 94 boto_cloudwatch_event.__init__(self.opts) 95 del self.opts 96 97 # Set up MagicMock to replace the boto3 session 98 # connections keep getting cached from prior tests, can't find the 99 # correct context object to clear it. So randomize the cache key, to prevent any 100 # cache hits 101 conn_parameters["key"] = "".join( 102 random.choice(string.ascii_lowercase + string.digits) for _ in range(50) 103 ) 104 105 self.patcher = patch("boto3.session.Session") 106 self.addCleanup(self.patcher.stop) 107 self.addCleanup(delattr, self, "patcher") 108 mock_session = self.patcher.start() 109 110 session_instance = mock_session.return_value 111 self.conn = MagicMock() 112 self.addCleanup(delattr, self, "conn") 113 session_instance.client.return_value = self.conn 114 115 116class BotoCloudWatchEventTestCaseMixin: 117 pass 118 119 120@skipIf(HAS_BOTO is False, "The boto module must be installed.") 121class BotoCloudWatchEventTestCase( 122 BotoCloudWatchEventTestCaseBase, BotoCloudWatchEventTestCaseMixin 123): 124 """ 125 TestCase for salt.modules.boto_cloudwatch_event module 126 """ 127 128 def test_that_when_checking_if_a_rule_exists_and_a_rule_exists_the_rule_exists_method_returns_true( 129 self, 130 ): 131 """ 132 Tests checking event existence when the event already exists 133 """ 134 self.conn.list_rules.return_value = {"Rules": [rule_ret]} 135 result = boto_cloudwatch_event.exists(Name=rule_name, **conn_parameters) 136 137 self.assertTrue(result["exists"]) 138 139 def test_that_when_checking_if_a_rule_exists_and_a_rule_does_not_exist_the_exists_method_returns_false( 140 self, 141 ): 142 """ 143 Tests checking rule existence when the rule does not exist 144 """ 145 self.conn.list_rules.return_value = {"Rules": []} 146 result = boto_cloudwatch_event.exists(Name=rule_name, **conn_parameters) 147 148 self.assertFalse(result["exists"]) 149 150 def test_that_when_checking_if_a_rule_exists_and_boto3_returns_an_error_the_rule_exists_method_returns_error( 151 self, 152 ): 153 """ 154 Tests checking rule existence when boto returns an error 155 """ 156 self.conn.list_rules.side_effect = ClientError(error_content, "list_rules") 157 result = boto_cloudwatch_event.exists(Name=rule_name, **conn_parameters) 158 159 self.assertEqual( 160 result.get("error", {}).get("message"), error_message.format("list_rules") 161 ) 162 163 def test_that_when_describing_rule_and_rule_exists_the_describe_rule_method_returns_rule( 164 self, 165 ): 166 """ 167 Tests describe rule for an existing rule 168 """ 169 self.conn.describe_rule.return_value = rule_ret 170 result = boto_cloudwatch_event.describe(Name=rule_name, **conn_parameters) 171 172 self.assertEqual(result.get("rule"), rule_ret) 173 174 def test_that_when_describing_rule_and_rule_does_not_exists_the_describe_method_returns_none( 175 self, 176 ): 177 """ 178 Tests describe rule for an non existent rule 179 """ 180 self.conn.describe_rule.side_effect = not_found_error 181 result = boto_cloudwatch_event.describe(Name=rule_name, **conn_parameters) 182 183 self.assertNotEqual(result.get("error"), None) 184 185 def test_that_when_describing_rule_and_boto3_returns_error_the_describe_method_returns_error( 186 self, 187 ): 188 self.conn.describe_rule.side_effect = ClientError( 189 error_content, "describe_rule" 190 ) 191 result = boto_cloudwatch_event.describe(Name=rule_name, **conn_parameters) 192 193 self.assertEqual( 194 result.get("error", {}).get("message"), 195 error_message.format("describe_rule"), 196 ) 197 198 def test_that_when_creating_a_rule_succeeds_the_create_rule_method_returns_true( 199 self, 200 ): 201 """ 202 tests True when rule created 203 """ 204 self.conn.put_rule.return_value = create_rule_ret 205 result = boto_cloudwatch_event.create_or_update( 206 Name=rule_name, 207 Description=rule_desc, 208 ScheduleExpression=rule_sched, 209 **conn_parameters 210 ) 211 self.assertTrue(result["created"]) 212 213 def test_that_when_creating_a_rule_fails_the_create_method_returns_error(self): 214 """ 215 tests False when rule not created 216 """ 217 self.conn.put_rule.side_effect = ClientError(error_content, "put_rule") 218 result = boto_cloudwatch_event.create_or_update( 219 Name=rule_name, 220 Description=rule_desc, 221 ScheduleExpression=rule_sched, 222 **conn_parameters 223 ) 224 self.assertEqual( 225 result.get("error", {}).get("message"), error_message.format("put_rule") 226 ) 227 228 def test_that_when_deleting_a_rule_succeeds_the_delete_method_returns_true(self): 229 """ 230 tests True when delete rule succeeds 231 """ 232 self.conn.delete_rule.return_value = {} 233 result = boto_cloudwatch_event.delete(Name=rule_name, **conn_parameters) 234 235 self.assertTrue(result.get("deleted")) 236 self.assertEqual(result.get("error"), None) 237 238 def test_that_when_deleting_a_rule_fails_the_delete_method_returns_error(self): 239 """ 240 tests False when delete rule fails 241 """ 242 self.conn.delete_rule.side_effect = ClientError(error_content, "delete_rule") 243 result = boto_cloudwatch_event.delete(Name=rule_name, **conn_parameters) 244 self.assertFalse(result.get("deleted")) 245 self.assertEqual( 246 result.get("error", {}).get("message"), error_message.format("delete_rule") 247 ) 248 249 def test_that_when_listing_targets_and_rule_exists_the_list_targets_method_returns_targets( 250 self, 251 ): 252 """ 253 Tests listing targets for an existing rule 254 """ 255 self.conn.list_targets_by_rule.return_value = {"Targets": [target_ret]} 256 result = boto_cloudwatch_event.list_targets(Rule=rule_name, **conn_parameters) 257 258 self.assertEqual(result.get("targets"), [target_ret]) 259 260 def test_that_when_listing_targets_and_rule_does_not_exist_the_list_targets_method_returns_error( 261 self, 262 ): 263 """ 264 Tests list targets for an non existent rule 265 """ 266 self.conn.list_targets_by_rule.side_effect = not_found_error 267 result = boto_cloudwatch_event.list_targets(Rule=rule_name, **conn_parameters) 268 269 self.assertNotEqual(result.get("error"), None) 270 271 def test_that_when_putting_targets_succeeds_the_put_target_method_returns_no_failures( 272 self, 273 ): 274 """ 275 tests None when targets added 276 """ 277 self.conn.put_targets.return_value = {"FailedEntryCount": 0} 278 result = boto_cloudwatch_event.put_targets( 279 Rule=rule_name, Targets=[], **conn_parameters 280 ) 281 self.assertIsNone(result["failures"]) 282 283 def test_that_when_putting_targets_fails_the_put_targets_method_returns_error(self): 284 """ 285 tests False when thing type not created 286 """ 287 self.conn.put_targets.side_effect = ClientError(error_content, "put_targets") 288 result = boto_cloudwatch_event.put_targets( 289 Rule=rule_name, Targets=[], **conn_parameters 290 ) 291 self.assertEqual( 292 result.get("error", {}).get("message"), error_message.format("put_targets") 293 ) 294 295 def test_that_when_removing_targets_succeeds_the_remove_targets_method_returns_true( 296 self, 297 ): 298 """ 299 tests True when remove targets succeeds 300 """ 301 self.conn.remove_targets.return_value = {"FailedEntryCount": 0} 302 result = boto_cloudwatch_event.remove_targets( 303 Rule=rule_name, Ids=[], **conn_parameters 304 ) 305 306 self.assertIsNone(result["failures"]) 307 self.assertEqual(result.get("error"), None) 308 309 def test_that_when_removing_targets_fails_the_remove_targets_method_returns_error( 310 self, 311 ): 312 """ 313 tests False when remove targets fails 314 """ 315 self.conn.remove_targets.side_effect = ClientError( 316 error_content, "remove_targets" 317 ) 318 result = boto_cloudwatch_event.remove_targets( 319 Rule=rule_name, Ids=[], **conn_parameters 320 ) 321 self.assertEqual( 322 result.get("error", {}).get("message"), 323 error_message.format("remove_targets"), 324 ) 325