forked from zappa/Zappa
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathutils.py
75 lines (55 loc) · 2.05 KB
/
utils.py
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
import placebo
import boto3
import os
import functools
from contextlib import contextmanager
from mock import patch, MagicMock
try:
file
except NameError: # builtin 'file' was removed in Python 3
from io import IOBase as file
PLACEBO_DIR = os.path.join(os.path.dirname(__file__), "placebo")
def placebo_session(function):
"""
Decorator to help do testing with placebo.
Simply wrap the function you want to test and make sure to add
a "session" argument so the decorator can pass the placebo session.
Accepts the following environment variables to configure placebo:
PLACEBO_MODE: set to "record" to record AWS calls and save them
PLACEBO_PROFILE: optionally set an AWS credential profile to record with
"""
@functools.wraps(function)
def wrapper(*args, **kwargs):
session_kwargs = {
"region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
}
profile_name = os.environ.get("PLACEBO_PROFILE", None)
if profile_name:
session_kwargs["profile_name"] = profile_name
session = boto3.Session(**session_kwargs)
self = args[0]
prefix = self.__class__.__name__ + "." + function.__name__
record_dir = os.path.join(PLACEBO_DIR, prefix)
if not os.path.exists(record_dir):
os.makedirs(record_dir)
pill = placebo.attach(session, data_path=record_dir)
if os.environ.get("PLACEBO_MODE") == "record":
pill.record()
else:
pill.playback()
kwargs["session"] = session
return function(*args, **kwargs)
return wrapper
@contextmanager
def patch_open():
"""Patch open() to allow mocking both open() itself and the file that is
yielded.
Yields the mock for "open" and "file", respectively."""
mock_open = MagicMock(spec=open)
mock_file = MagicMock(spec=file)
@contextmanager
def stub_open(*args, **kwargs):
mock_open(*args, **kwargs)
yield mock_file
with patch("__builtin__.open", stub_open):
yield mock_open, mock_file