moto/tests/test_events/test_events.py

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_delete_rule():
    client = generate_environment()

    client.delete_rule(Name=RULES[0]['Name'])
    rules = client.list_rules()
    assert(len(rules['Rules']) == len(RULES) - 1)


@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()