Removed OrderedDicts for 2.6 and dict.iteritems() calls for 3.3+ compatibility.

This commit is contained in:
Justin Wiley 2016-12-01 19:10:59 -08:00 committed by Justin Wiley
parent db0b494b4f
commit 6c85a85e0d
2 changed files with 17 additions and 22 deletions

View File

@ -40,9 +40,14 @@ class Rule(object):
class EventsBackend(BaseBackend):
def __init__(self):
self.rules = OrderedDict()
self.rules = {}
# This array tracks the order in which the rules have been added, since 2.6 doesn't have OrderedDicts
self.rules_order = []
self.next_tokens = {}
def _get_rule_by_index(self, i):
return self.rules.get(self.rules_order[i])
def _gen_next_token(self, index):
token = os.urandom(128).encode('base64')
self.next_tokens[token] = index
@ -65,6 +70,7 @@ class EventsBackend(BaseBackend):
return start_index, end_index, new_next_token
def delete_rule(self, name):
self.rules_order.pop(self.rules_order.index(name))
return self.rules.pop(name) is not None
def describe_rule(self, name):
@ -84,19 +90,14 @@ class EventsBackend(BaseBackend):
return False
def generate_presigned_url(self):
pass
def list_rule_names_by_target(self, target_arn, next_token=None, limit=None):
rules_array = self.rules.values()
matching_rules = []
return_obj = {}
start_index, end_index, new_next_token = self._process_token_and_limits(len(rules_array), next_token, limit)
start_index, end_index, new_next_token = self._process_token_and_limits(len(self.rules), next_token, limit)
for i in range(start_index, end_index):
rule = rules_array[i]
rule = self._get_rule_by_index(i)
for target in rule.targets:
if rule.targets[target]['Arn'] == target_arn:
matching_rules.append(rule.name)
@ -108,8 +109,6 @@ class EventsBackend(BaseBackend):
return return_obj
def list_rules(self, prefix=None, next_token=None, limit=None):
rules_array = self.rules.values()
match_string = '.*'
if prefix is not None:
match_string = '^' + prefix + match_string
@ -119,10 +118,10 @@ class EventsBackend(BaseBackend):
matching_rules = []
return_obj = {}
start_index, end_index, new_next_token = self._process_token_and_limits(len(rules_array), next_token, limit)
start_index, end_index, new_next_token = self._process_token_and_limits(len(self.rules), next_token, limit)
for i in range(start_index, end_index):
rule = rules_array[i]
rule = self._get_rule_by_index(i)
if match_regex.match(rule.name):
matching_rules.append(rule)
@ -150,14 +149,10 @@ class EventsBackend(BaseBackend):
return return_obj
def put_events(self):
# For the purposes of this mock, there is no backend action for putting an event.
# Response module will deal with replying.
pass
def put_rule(self, name, **kwargs):
rule = Rule(name, **kwargs)
self.rules[rule.name] = rule
self.rules_order.append(rule.name)
return rule.arn
def put_targets(self, name, targets):

View File

@ -61,9 +61,9 @@ def generate_environment():
)
targets = []
for target, target_attr in TARGETS.iteritems():
if rule['Name'] in target_attr.get('Rules'):
targets.append({'Id': target, 'Arn': target_attr['Arn']})
for target in TARGETS:
if rule['Name'] in TARGETS[target].get('Rules'):
targets.append({'Id': target, 'Arn': TARGETS[target]['Arn']})
client.put_targets(Rule=rule['Name'], Targets=targets)
@ -140,8 +140,8 @@ def test_list_targets_by_rule():
targets = client.list_targets_by_rule(Rule=rule_name)
expected_targets = []
for target, attrs in TARGETS.iteritems():
if rule_name in attrs.get('Rules'):
for target in TARGETS:
if rule_name in TARGETS[target].get('Rules'):
expected_targets.append(target)
assert(len(targets['Targets']) == len(expected_targets))