summaryrefslogtreecommitdiff
path: root/tests/contrib/awslambda/conftest.py
blob: c1a481215d39f7a988676650f960a7395e78fdb9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

import pytest
from raven.contrib.awslambda import LambdaClient
import uuid
import time


class MockClient(LambdaClient):
    def __init__(self, *args, **kwargs):
        self.events = []
        super(MockClient, self).__init__(*args, **kwargs)

    def send(self, **kwargs):
        self.events.append(kwargs)

    def is_enabled(self, **kwargs):
        return True


class LambdaIndentityStub(object):
    def __init__(self, id=1, pool_id=1):
        self.cognito_identity_id = id
        self.cognito_identity_pool_id = pool_id

    def __getitem__(self, item):
        return getattr(self, item)

    def get(self, name, default=None):
        return getattr(self, name, default)


class LambdaContextStub(object):

    def __init__(self, function_name, memory_limit_in_mb=128, timeout=300, function_version='$LATEST'):
        self.function_name = function_name
        self.memory_limit_in_mb = memory_limit_in_mb
        self.timeout = timeout
        self.function_version = function_version
        self.timeout = timeout
        self.invoked_function_arn = 'invoked_function_arn'
        self.log_group_name = 'log_group_name'
        self.log_stream_name = 'log_stream_name'
        self.identity = LambdaIndentityStub(id=0, pool_id=0)
        self.client_context = None
        self.aws_request_id = str(uuid.uuid4())
        self.start_time = time.time() * 1000

    def __getitem__(self, item, default):
        return getattr(self, item, default)

    def get(self, name, default=None):
        return getattr(self, name, default)

    def get_remaining_time_in_millis(self):
        return max(self.timeout * 1000 - int((time.time() * 1000) - self.start_time), 0)


class LambdaEventStub(object):
    def __init__(self, body=None, headers=None, http_method='GET', path='/test', query_string=None):
        self.body = body
        self.headers = headers
        self.httpMethod = http_method
        self.isBase64Encoded = False
        self.path = path
        self.queryStringParameters = query_string
        self.resource = path
        self.stageVariables = None
        self.requestContext = {
            'accountId': '0000000',
            'apiId': 'AAAAAAAA',
            'httpMethod': http_method,
            'identity': LambdaIndentityStub(),
            'path': path,
            'requestId': 'test-request',
            'resourceId': 'bbzeyv',
            'resourcePath': '/test',
            'stage': 'test-stage'
        }

    def __getitem__(self, name):
        return getattr(self, name)

    def get(self, name, default=None):
        return getattr(self, name, default)


@pytest.fixture
def lambda_env(monkeypatch):
    monkeypatch.setenv('SENTRY_DSN', 'http://public:secret@example.com/1')
    monkeypatch.setenv('AWS_LAMBDA_FUNCTION_NAME', 'test_func')
    monkeypatch.setenv('AWS_LAMBDA_FUNCTION_VERSION', '$LATEST')
    monkeypatch.setenv('SENTRY_RELEASE', '$LATEST')
    monkeypatch.setenv('SENTRY_ENVIRONMENT', 'testing')


@pytest.fixture
def mock_client():
    return MockClient

@pytest.fixture
def lambda_event():
    return LambdaEventStub

@pytest.fixture
def lambda_context():
    return LambdaContextStub