forked from aweber/AWeber-API-Python-Library
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmock_adapter.py
More file actions
147 lines (130 loc) · 6.17 KB
/
mock_adapter.py
File metadata and controls
147 lines (130 loc) · 6.17 KB
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
import json
import os
import mock
from aweber_api import AWeberUser, OAuthAdapter
try:
from urllib import quote
from urlparse import urlparse, parse_qs
except ImportError:
from urllib.parse import urlparse, parse_qs, quote
__all__ = ['MockAdapter']
responses = {
'GET' : {
'/accounts': ({}, 'accounts/page1'),
'/accounts/1': ({}, 'accounts/1'),
'/accounts/1?ws.op=findSubscribers&' \
'email=joe%40example.com': ({}, 'accounts/findSubscribers'),
'/accounts/1?ws.show=total_size&ws.op=findSubscribers&' \
'email=joe%40example.com': ({}, 'accounts/findSubscribers_ts'),
'/accounts/1?ws.op=getWebForms': ({}, 'accounts/webForms'),
'/accounts/1?ws.op=getWebFormSplitTests': ({}, 'accounts/webFormSplitTests'),
'/accounts/1/lists': ({}, 'lists/page1'),
'/accounts/1/lists?ws.start=20&ws.size=20': ({}, 'lists/page2'),
'/accounts/1/lists/303449': ({}, 'lists/303449'),
'/accounts/1/lists/505454': ({}, 'lists/505454'),
'/accounts/1/lists/303449/any_collection': ({}, 'any_collection/page1'),
'/accounts/1/lists/303449/any_collection/1': ({}, 'any_collection/1'),
'/accounts/1/lists/303449/campaigns': ({}, 'campaigns/303449'),
'/accounts/1/lists/303449/custom_fields': ({}, 'custom_fields/303449'),
'/accounts/1/lists/505454/custom_fields': ({}, 'custom_fields/505454'),
'/accounts/1/lists/303449/custom_fields/1': ({}, 'custom_fields/1'),
'/accounts/1/lists/303449/custom_fields/2': ({}, 'custom_fields/2'),
'/accounts/1/lists/303449/subscribers': ({}, 'subscribers/page1'),
'/accounts/1/lists/303449/subscribers/1': ({}, 'subscribers/1'),
'/accounts/1/lists/303449/subscribers/2': ({}, 'subscribers/2'),
'/accounts/1/lists/505454/subscribers/3': ({}, 'subscribers/3'),
'/accounts/1/lists/303449/subscribers/1?ws.op=getActivity': (
{}, 'subscribers/get_activity'),
'/accounts/1/lists/303449/subscribers/1?ws.show=total_size&ws.op=getActivity': (
{}, 'subscribers/get_activity_ts'),
'/accounts/1/lists/303449/subscribers?ws.op=find&name=joe': (
{'status': '400'}, 'error'),
'/accounts/1?ws.op=findSubscribers&name=bob': (
{'status': '400'}, 'error'),
'/accounts/1/lists/303449/subscribers?ws.op=find&' \
'email=joe%40example.com': ({}, 'subscribers/find'),
'/accounts/1/lists/303449/subscribers?ws.show=total_size&ws.op=find&' \
'email=joe%40example.com': ({}, 'subscribers/find_ts'),
'/accounts/1/lists/303449/broadcasts/total?status=sent': (
{'total_size': 10}, 'campaigns/303449'),
'/accounts/1/lists/303449/broadcasts?status=sent': (
{'total_size': 10}, 'campaigns/303449'),
},
'POST' : {
'/accounts/1/lists/303449/any_collection': ({
'status': '201',
'location': '/accounts/1/lists/303449/any_collection/1'}, None),
'/accounts/1/lists/303449/custom_fields': ({
'status': '201',
'location': '/accounts/1/lists/303449/custom_fields/2'}, None),
'/accounts/1/lists/505454/custom_fields': ({
'status': '400'}, 'custom_fields/error'),
'/accounts/1/lists/303449/subscribers/1': ({
'status': '201',
'location': '/accounts/1/lists/505454/subscribers/3'}, None),
'/accounts/1/lists/303449/broadcasts/2/schedule': ({
'status': '201',
'location': '/accounts/1/lists/303449/broadcasts/2/schedule'},
None
),
'/accounts/1/lists/303449/broadcasts/3/schedule': ({
'status': '400',
'location': '/accounts/1/lists/303449/broadcasts/3/schedule'},
'error'
),
'/accounts/1/lists/303449/broadcasts/2/cancel': ({
'status': '204',
'location': '/accounts/1/lists/303449/broadcasts/2/cancel'},
None
),
'/accounts/1/lists/303449/broadcasts/3/cancel': ({
'status': '400',
'location': '/accounts/1/lists/303449/broadcasts/3/cancel'},
'error'
),
},
'PATCH' : {
'/accounts/1/lists/303449/subscribers/1': ({'status': '209'}, None),
'/accounts/1/lists/303449/subscribers/2': ({'status': '400'}, 'error'),
},
'DELETE' : {
'/accounts/1/lists/303449/subscribers/1': ({'status': '200'}, None),
'/accounts/1/lists/303449/subscribers/2': ({'status': '400'}, 'error'),
}
}
def _sort_qs_for_url(url):
"""Sort query string parameters in desending order."""
parsed = urlparse(url)
if len(parsed.query) == 0:
return parsed.path
qs = parse_qs(parsed.query)
params = []
for key in reversed(sorted(qs.keys())):
params.append("{0}={1}".format(key, quote(qs[key][0])))
return "{0}?{1}".format(parsed.path, "&".join(params))
def request(self, url, method, **kwargs):
"""Return a tuple to simulate calling oauth2.Client.request."""
url = _sort_qs_for_url(url)
(headers, file) = responses[method][url]
if 'status' not in headers:
# assume 200 OK if not otherwise specified
headers['status'] = '200'
if file is None:
return (headers, '')
path = os.sep.join(__file__.split(os.sep)[:-1]+['data',''])
filename = "{0}{1}.json".format(path, file)
data = open(filename).read()
return (headers, data)
class MockAdapter(OAuthAdapter):
"""Mocked OAuthAdapter."""
requests = []
@mock.patch('oauth2.Client.request', request)
def request(self, method, url, data={}, response='body'):
"""Mock the oauth.Client.request method"""
url = _sort_qs_for_url(url)
req = super(MockAdapter, self).request(method, url, data, response)
self.requests.append({'method' : method, 'url' : url, 'data' : data})
return req
def __init__(self):
self.user = AWeberUser()
return super(MockAdapter, self).__init__('key', 'secret', '')