1
- import io
2
- import json
3
1
from dataclasses import dataclass
4
2
5
- import botocore
6
- import botocore .client
7
3
import pytest
8
4
9
5
@@ -23,60 +19,3 @@ class LambdaContext:
23
19
client_context = None
24
20
25
21
return LambdaContext ()
26
-
27
-
28
- class BotoMockerFixture :
29
- # Botocore
30
- _make_api_call = botocore .client .BaseClient ._make_api_call
31
-
32
- def __init__ (self , mocker ):
33
- self .mocker = mocker
34
-
35
- def patch (self , new ):
36
- self .mocker .patch ('botocore.client.BaseClient._make_api_call' , new = new )
37
-
38
- @staticmethod
39
- def build_make_api_call (service_table ):
40
- def make_api_call (self , operation_name , kwarg ):
41
- service_name = type (self ).__name__ .lower ()
42
-
43
- operation_table = service_table .get (service_name )
44
- if operation_table is not None and operation_name in operation_table :
45
- operation = operation_table .get (operation_name )
46
- if isinstance (operation , Exception ):
47
- raise operation
48
- return operation (self , operation_name , kwarg ) if callable (operation ) else operation
49
- return BotoMockerFixture ._make_api_call (self , operation_name , kwarg )
50
-
51
- return make_api_call
52
-
53
- @staticmethod
54
- def build_lambda_invoke_handler (response_table ):
55
- def handle_lambda_invoke (self , operation_name , kwarg ):
56
- function_name = kwarg .get ('FunctionName' )
57
-
58
- response = response_table .get (function_name )
59
- if response is not None :
60
- payload = response .get ('Payload' )
61
- if isinstance (payload , Exception ):
62
- if 'FunctionError' in response :
63
- payload = {'errorMessage' : str (payload ), 'errorType' : type (payload ).__name__ }
64
- else :
65
- raise payload
66
- if payload :
67
- payload = json .dumps (payload )
68
- payload = payload .encode ()
69
- return response | {
70
- 'Payload' : botocore .response .StreamingBody (io .BytesIO (payload ), len (payload ))
71
- }
72
- return BotoMockerFixture ._make_api_call (self , operation_name , kwarg )
73
-
74
- return handle_lambda_invoke
75
-
76
-
77
- # For all scopes.
78
- boto_mocker = pytest .fixture ()(lambda mocker : BotoMockerFixture (mocker ))
79
- class_boto_mocker = pytest .fixture (scope = 'class' )(lambda class_mocker : BotoMockerFixture (class_mocker ))
80
- module_boto_mocker = pytest .fixture (scope = 'module' )(lambda module_mocker : BotoMockerFixture (module_mocker ))
81
- package_boto_mocker = pytest .fixture (scope = 'package' )(lambda package_mocker : BotoMockerFixture (package_mocker ))
82
- session_boto_mocker = pytest .fixture (scope = 'session' )(lambda session_mocker : BotoMockerFixture (session_mocker ))
0 commit comments