import random import boto3 from moto.events import mock_events RULES = [ {'Name': 'test1', 'ScheduleExpression': 'rate(5 minutes)'}, {'Name': 'test2', 'ScheduleExpression': 'rate(1 minute)'}, {'Name': 'test3', 'EventPattern': '{"source": ["test-source"]}'} ] TARGETS = { 'test-target-1': { 'Id': 'test-target-1', 'Arn': 'arn:aws:lambda:us-west-2:111111111111:function:test-function-1', 'Rules': ['test1', 'test2'] }, 'test-target-2': { 'Id': 'test-target-2', 'Arn': 'arn:aws:lambda:us-west-2:111111111111:function:test-function-2', 'Rules': ['test1', 'test3'] }, 'test-target-3': { 'Id': 'test-target-3', 'Arn': 'arn:aws:lambda:us-west-2:111111111111:function:test-function-3', 'Rules': ['test1', 'test2'] }, 'test-target-4': { 'Id': 'test-target-4', 'Arn': 'arn:aws:lambda:us-west-2:111111111111:function:test-function-4', 'Rules': ['test1', 'test3'] }, 'test-target-5': { 'Id': 'test-target-5', 'Arn': 'arn:aws:lambda:us-west-2:111111111111:function:test-function-5', 'Rules': ['test1', 'test2'] }, 'test-target-6': { 'Id': 'test-target-6', 'Arn': 'arn:aws:lambda:us-west-2:111111111111:function:test-function-6', 'Rules': ['test1', 'test3'] } } def get_random_rule(): return RULES[random.randint(0, len(RULES) - 1)] def generate_environment(): client = boto3.client('events', 'us-west-2') for rule in RULES: client.put_rule( Name=rule['Name'], ScheduleExpression=rule.get('ScheduleExpression', ''), EventPattern=rule.get('EventPattern', '') ) targets = [] 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) return client @mock_events def test_list_rules(): client = generate_environment() response = client.list_rules() assert(response is not None) assert(len(response['Rules']) > 0) @mock_events def test_describe_rule(): rule_name = get_random_rule()['Name'] client = generate_environment() response = client.describe_rule(Name=rule_name) assert(response is not None) assert(response.get('Name') == rule_name) assert(response.get('Arn') is not None) @mock_events def test_enable_disable_rule(): rule_name = get_random_rule()['Name'] client = generate_environment() # Rules should start out enabled in these tests. rule = client.describe_rule(Name=rule_name) assert(rule['State'] == 'ENABLED') client.disable_rule(Name=rule_name) rule = client.describe_rule(Name=rule_name) assert(rule['State'] == 'DISABLED') client.enable_rule(Name=rule_name) rule = client.describe_rule(Name=rule_name) assert(rule['State'] == 'ENABLED') @mock_events def test_list_rule_names_by_target(): test_1_target = TARGETS['test-target-1'] test_2_target = TARGETS['test-target-2'] client = generate_environment() rules = client.list_rule_names_by_target(TargetArn=test_1_target['Arn']) assert(len(rules['RuleNames']) == len(test_1_target['Rules'])) for rule in rules['RuleNames']: assert(rule in test_1_target['Rules']) rules = client.list_rule_names_by_target(TargetArn=test_2_target['Arn']) assert(len(rules['RuleNames']) == len(test_2_target['Rules'])) for rule in rules['RuleNames']: assert(rule in test_2_target['Rules']) @mock_events def test_list_rules(): client = generate_environment() rules = client.list_rules() assert(len(rules['Rules']) == len(RULES)) @mock_events def test_list_targets_by_rule(): rule_name = get_random_rule()['Name'] client = generate_environment() targets = client.list_targets_by_rule(Rule=rule_name) expected_targets = [] for target in TARGETS: if rule_name in TARGETS[target].get('Rules'): expected_targets.append(target) assert(len(targets['Targets']) == len(expected_targets)) @mock_events def test_remove_targets(): rule_name = get_random_rule()['Name'] client = generate_environment() targets = client.list_targets_by_rule(Rule=rule_name)['Targets'] targets_before = len(targets) assert(targets_before > 0) client.remove_targets(Rule=rule_name, Ids=[targets[0]['Id']]) targets = client.list_targets_by_rule(Rule=rule_name)['Targets'] targets_after = len(targets) assert(targets_before - 1 == targets_after) if __name__ == '__main__': test_list_rules() test_describe_rule() test_enable_disable_rule() test_list_rule_names_by_target() test_list_rules() test_list_targets_by_rule() test_remove_targets()