diff --git a/aweber_api/__init__.py b/aweber_api/__init__.py index 2799007..d036452 100644 --- a/aweber_api/__init__.py +++ b/aweber_api/__init__.py @@ -1,4 +1,4 @@ -from urlparse import parse_qs +from urllib.parse import parse_qs from aweber_api.base import ( ACCESS_TOKEN_URL, diff --git a/aweber_api/__init__.py.bak b/aweber_api/__init__.py.bak new file mode 100644 index 0000000..2799007 --- /dev/null +++ b/aweber_api/__init__.py.bak @@ -0,0 +1,173 @@ +from urlparse import parse_qs + +from aweber_api.base import ( + ACCESS_TOKEN_URL, + APIException, + API_BASE, + AUTHORIZE_URL, + AWeberBase, + REQUEST_TOKEN_URL, +) +from aweber_api.collection import AWeberCollection +from aweber_api.entry import AWeberEntry +from aweber_api.oauth import OAuthAdapter +from aweber_api.response import AWeberResponse + + +class AWeberAPI(AWeberBase): + """Base class for connecting to the AWeberAPI. + + Created with a consumer key and secret, then used to either generate + tokens for authorizing a user, or can be provided tokens and used to + access that user's resources. + + """ + + def __init__(self, consumer_key, consumer_secret): + self.adapter = OAuthAdapter(consumer_key, consumer_secret, API_BASE) + self.adapter.user = AWeberUser() + + @classmethod + def parse_authorization_code(cls, authorization_code): + """Exchange an authorization code for new api keys. + + Returns a tuple containing the new consumer key/secret and + access token key/secret. + + """ + # parse and validate authorization code + keys = cls._parse_and_validate_authorization_code(authorization_code) + consumer_key = keys[0] + consumer_secret = keys[1] + + # create an instance of AWeberAPI for getting the access token + instance = cls._create_new_instance(keys) + + # exchange request token for an access token + access_key, access_secret = instance.get_access_token() + + # return consumer key/secret and access token key/secret + return consumer_key, consumer_secret, access_key, access_secret + + @classmethod + def _parse_and_validate_authorization_code(cls, authorization_code): + """parse and validate authorization code.""" + keys = authorization_code.split('|') + if len(keys) < 5: + raise APIException('Invalid Authorization Code') + + return keys + + @classmethod + def _create_new_instance(cls, keys): + """Create an instance of AWeberAPI for getting the access token.""" + instance = cls(keys[0], keys[1]) + instance.user.request_token = keys[2] + instance.user.token_secret = keys[3] + instance.user.verifier = keys[4] + + return instance + + @property + def authorize_url(self): + """Return the authorize url. + + Potentially containing the request token parameter. + + """ + if self.user.request_token: + return "{0}?oauth_token={1}".format( + AUTHORIZE_URL, self.user.request_token) + + return AUTHORIZE_URL + + def get_request_token(self, callback_url): + """Get a new request token / token secret for the callback url. + + Returns request token / secret, and sets properties on the + AWeberUser object (self.user). + + """ + data = {'oauth_callback': callback_url} + response = self.adapter.request( + 'POST', REQUEST_TOKEN_URL, data) + self.user.request_token, self.user.token_secret = ( + self._parse_token_response(response)) + + return (self.user.request_token, self.user.token_secret) + + def get_access_token(self): + """Exchange request tokens for Access tokens. + + Gets an access token for the combination of + * request token + * token secret + * verifier + in the AWeberUser object at self.user. + + Updates the user object and returns the tokens. + + """ + data = {'oauth_verifier': self.user.verifier} + response = self.adapter.request( + 'POST', ACCESS_TOKEN_URL, data) + self.user.access_token, self.user.token_secret = ( + self._parse_token_response(response)) + + return (self.user.access_token, self.user.token_secret) + + def _parse_token_response(self, response): + """Parses token response. + + Return the token key and the token secret + + """ + if not isinstance(response, str): + raise TypeError('Expected response to be a string') + + data = parse_qs(response) + + if (data.get('oauth_token') is None) or ( + data.get('oauth_token_secret') is None): + raise ValueError('OAuth parameters not returned') + + return (data['oauth_token'][0], data['oauth_token_secret'][0]) + + def get_account(self, access_token=False, token_secret=False): + """Returns the AWeberEntry object for the account. + + Specified by the access_token and token_secret currently + in the self.user object. + + Optionally, access_token and token_secret can be provided to + replace the properties in self.user.access_token and + self.user.token_secret, respectively. + + """ + if access_token: + self.user.access_token = access_token + if token_secret: + self.user.token_secret = token_secret + + url = '/accounts' + response = self.adapter.request('GET', url) + accounts = self._read_response(url, response) + + return accounts[0] + + +class AWeberUser(object): + """Data storage object representing the user in the OAuth model. + + Has properties for request_token, token_secret, access_token, and + verifier. + + """ + request_token = None + token_secret = None + access_token = None + verifier = None + + def get_highest_priority_token(self): + """Return either the access token or the request token.""" + return self.access_token or self.request_token diff --git a/aweber_api/collection.py b/aweber_api/collection.py index b176e69..7fc4e30 100644 --- a/aweber_api/collection.py +++ b/aweber_api/collection.py @@ -1,6 +1,6 @@ from math import floor -from urlparse import parse_qs -from urllib import urlencode +from urllib.parse import parse_qs +from urllib.parse import urlencode from aweber_api.base import API_BASE from aweber_api.entry import AWeberEntry @@ -122,7 +122,7 @@ def __len__(self): def __iter__(self): return self - def next(self): + def __next__(self): """Get the next entry in the collection.""" if self._current < self.total_size: self._current += 1 diff --git a/aweber_api/collection.py.bak b/aweber_api/collection.py.bak new file mode 100644 index 0000000..b176e69 --- /dev/null +++ b/aweber_api/collection.py.bak @@ -0,0 +1,141 @@ +from math import floor +from urlparse import parse_qs +from urllib import urlencode + +from aweber_api.base import API_BASE +from aweber_api.entry import AWeberEntry +from aweber_api.response import AWeberResponse + + +class AWeberCollection(AWeberResponse): + """Represents a collection of similar objects. + + Encapsulates data that is found at the base URI's for a given object + type, ie: + /accounts + /accounts/XXX/lists + + Parses the data from the response and provides basic sequence like + operations, such as iteration and indexing to access the entries + that are contained in this collection. + + """ + + page_size = 100 + + def __init__(self, url, data, adapter): + self._entry_data = {} + self._current = 0 + + super(AWeberCollection, self).__init__(url, data, adapter) + self._key_entries(self._data) + + def get_by_id(self, id): + """Returns an entry from this collection. + + The Entry as found by its actual AWeber id, not its offset. + Will actually request the data from the API. + + """ + return self.load_from_url("{0}/{1}".format(self.url, id)) + + def _key_entries(self, response): + count = 0 + for entry in response['entries']: + self._entry_data[count + response['start']] = entry + count += 1 + + def _load_page_for_offset(self, offset): + page = self._get_page_params(offset) + response = self.adapter.request('GET', self.url, page) + self._key_entries(response) + + def _get_page_params(self, offset): + """Return the start and size of the paginated response.""" + next_link = self._data.get('next_collection_link', None) + if next_link is None: + """no more parameters in page!""" + raise StopIteration + + url, query = next_link.split('?') + query_parts = parse_qs(query) + self.page_size = int(query_parts['ws.size'][0]) + page_number = int(floor(offset / self.page_size)) + start = page_number * self.page_size + return {'ws.start': start, 'ws.size': self.page_size} + + def create(self, **kwargs): + """Method to create an item.""" + params = {'ws.op': 'create'} + params.update(kwargs) + + response = self.adapter.request( + 'POST', self.url, params, response='headers') + + resource_url = response['location'] + data = self.adapter.request('GET', resource_url) + return AWeberEntry(resource_url, data, self.adapter) + + def find(self, **kwargs): + """Method to request a collection.""" + params = {'ws.op': 'find'} + params.update(kwargs) + query_string = urlencode(params) + url = '{0.url}?{1}'.format(self, query_string) + data = self.adapter.request('GET', url) + + collection = AWeberCollection(url, data, self.adapter) + collection._data['total_size'] = self._get_total_size(url) + return collection + + def _get_total_size(self, uri, **kwargs): + """Get actual total size number from total_size_link.""" + total_size_uri = '{0}&ws.show=total_size'.format(uri) + return int(self.adapter.request('GET', total_size_uri)) + + def get_parent_entry(self): + """Return a collection's parent entry or None.""" + url_parts = self._partition_url() + if url_parts is None: + return None + + url = self._construct_parent_url(url_parts, 1) + + data = self.adapter.request('GET', url) + try: + entry = AWeberEntry(url, data, self.adapter) + + except TypeError: + return None + + return entry + + def _create_entry(self, offset): + """Add an entry to the collection""" + data = self._entry_data[offset] + url = data['self_link'].replace(API_BASE, '') + self._entries[offset] = AWeberEntry(url, data, self.adapter) + + def __len__(self): + return self.total_size + + def __iter__(self): + return self + + def next(self): + """Get the next entry in the collection.""" + if self._current < self.total_size: + self._current += 1 + return self[self._current - 1] + self._current = 0 + raise StopIteration + + def __getitem__(self, offset): + if offset < 0 or offset >= self._data['total_size']: + raise ValueError('Offset {0} does not exist'.format(offset)) + + if not offset in self._entries: + if not offset in self._entry_data: + self._load_page_for_offset(offset) + self._create_entry(offset) + return self._entries[offset] diff --git a/aweber_api/entry.py b/aweber_api/entry.py index 8e34faf..fe59df4 100644 --- a/aweber_api/entry.py +++ b/aweber_api/entry.py @@ -1,4 +1,4 @@ -from urllib import urlencode +from urllib.parse import urlencode import aweber_api from aweber_api.data_dict import DataDict diff --git a/aweber_api/entry.py.bak b/aweber_api/entry.py.bak new file mode 100644 index 0000000..8e34faf --- /dev/null +++ b/aweber_api/entry.py.bak @@ -0,0 +1,235 @@ +from urllib import urlencode + +import aweber_api +from aweber_api.data_dict import DataDict +from aweber_api.response import AWeberResponse + + +class AWeberEntry(AWeberResponse): + """Represents a single entry in the AWeber API data heirarchy. + + For example, single entries can be a: + * specific account + * list + * web form + + Built on data that is returned from an id-ed URI, such as: + * /accounts/XXXX + * /accounts/XXXX/lists/XXXX + + Can also be generated from the data in the entries array of a + collection object, which is identical to the data return from the + URI for that specific entry. + + Provides direct access to properties in the response, such as + self.id + + """ + + def __init__(self, url, data, adapter): + self._data = {} + self._diff = {} + super(AWeberEntry, self).__init__(url, data, adapter) + self._child_collections = {} + + def __setattr__(self, key, value): + if not key == '_data' and key in self._data: + self._data[key] = value + self._diff[key] = value + return value + return super(AWeberEntry, self).__setattr__(key, value) + + def delete(self): + """Invoke the API method to DELETE* this entry resource. + + * Note: + Not all entry resources are eligible to be DELETED, please + refer to the AWeber API Reference Documentation at + https://labs.aweber.com/docs/reference/1.0 for more + details on which entry resources may be deleted. + + """ + + self.adapter.request('DELETE', self.url, response='status') + return True + + def move(self, list_, **kwargs): + """Invoke the API method to Move an entry resource to a List. + + * Note: + Not all entry resources are eligible to be moved, please + refer to the AWeber API Reference Documentation at + https://labs.aweber.com/docs/reference/1.0 for more + details on which entry resources may be moved and if there + are any requirements for moving that resource. + + """ + params = {'ws.op': 'move', 'list_link': list_.self_link} + params.update(kwargs) + response = self.adapter.request( + 'POST', self.url, params, response='headers') + + new_resource = response['location'] + self._diff = {} + self._data = self.adapter.request('GET', new_resource) + return True + + def save(self): + self.adapter.request('PATCH', self.url, self._diff, response='status') + self._diff = {} + return True + + def get_activity(self): + """Invoke the API method to return all Subscriber activity. + + * Note: + This method only works on Subscriber Entry resources. + refer to the AWeber API Reference Documentation at + https://labs.aweber.com/docs/reference/1.0#subscriber + for more details on how to call this method. + + """ + self._method_for('subscriber') + params = {'ws.op': 'getActivity'} + query_string = urlencode(params) + url = '{0.url}?{1}'.format(self, query_string) + data = self.adapter.request('GET', url) + + collection = aweber_api.AWeberCollection(url, data, self.adapter) + collection._data['total_size'] = self._get_total_size(url) + return collection + + def findSubscribers(self, **kwargs): + """Invoke the API method to find all subscribers on all Lists. + + * Note: + This method only works on Account Entry resources and + requires access to subscriber information. please + refer to the AWeber API Reference Documentation at + https://labs.aweber.com/docs/reference/1.0#account + for more details on how to call this method. + + """ + self._method_for('account') + params = {'ws.op': 'findSubscribers'} + params.update(kwargs) + query_string = urlencode(params) + url = '{0.url}?{1}'.format(self, query_string) + + data = self.adapter.request('GET', url) + collection = aweber_api.AWeberCollection(url, data, self.adapter) + collection._data['total_size'] = self._get_total_size(url) + return collection + + def schedule_broadcast(self, bc_id, scheduled_for): + """Invoke the API method to schedule the given broadcast. + + * Note: + This method only works on List Entry resources and + requires send broadcast email permissions. Please + refer to the AWeber API Reference Documentation at + https://labs.aweber.com/docs/reference/1.0#broadcast_scheduler + for more details on how to call this method. + + """ + self._method_for('list') + body = {'scheduled_for': scheduled_for} + url = '{0}/broadcasts/{1}/schedule'.format(self.url, bc_id) + return self.adapter.request('POST', url, body, response='status') + + def get_broadcasts(self, status, **kwargs): + """Invoke the API method to retrieve broadcasts by status. + + * Note: + This method only works on List Entry resources. Please + refer to the AWeber API Reference Documentation at + https://labs.aweber.com/docs/reference/1.0#get_broadcasts + for more details on how to call this method. + + """ + self._method_for('list') + params = {'status': status} + params.update(kwargs) + query_string = urlencode(params) + url = '{0.url}/broadcasts?{1}'.format(self, query_string) + + data = self.adapter.request('GET', url) + collection = aweber_api.AWeberCollection(url, data, self.adapter) + collection._data['total_size'] = self._get_broadcast_count( + query_string) + return collection + + def cancel_broadcast(self, bc_id): + """Invoke the API method to cancel the given scheduled broadcast. + + * Note: + This method only works on List Entry resources and + requires send broadcast email permissions. Please refer + to the AWeber API Reference Documentation at + https://labs.aweber.com/docs/reference/1.0#cancel_broadcast + for more details on how to call this method. + + """ + self._method_for('list') + url = '{0}/broadcasts/{1}/cancel'.format(self.url, bc_id) + return self.adapter.request('POST', url, data={}, response='status') + + def _get_total_size(self, uri, **kwargs): + """Get actual total size number from total_size_link.""" + total_size_uri = '{0}&ws.show=total_size'.format(uri) + return int(self.adapter.request('GET', total_size_uri)) + + def _get_broadcast_count(self, query_string): + """Get actual total size number from total_size_link.""" + total_size_uri = '{0.url}/broadcasts/total?{1}'.format( + self, query_string) + return int(self.adapter.request('GET', total_size_uri)['total_size']) + + def get_parent_entry(self): + """Return the parent entry of this entry + + returns None if no parent exists. + + Example: + calling get_parent_entry on a SubscriberEntry will return the + List Entry that SubscriberEntry belongs to. For more + information on the AWeber API and how resources are arranged, + refer to: https://labs.aweber.com/docs/reference/1.0 + + """ + url_parts = self._partition_url() + if url_parts is None: + return None + + url = self._construct_parent_url(url_parts, 2) + + data = self.adapter.request('GET', url) + return AWeberEntry(url, data, self.adapter) + + def get_web_forms(self): + self._method_for('account') + data = self.adapter.request( + 'GET', "{0}?ws.op=getWebForms".format(self.url)) + return self._parseNamedOperation(data) + + def get_web_form_split_tests(self): + self._method_for('account') + data = self.adapter.request( + 'GET', "{0}?ws.op=getWebFormSplitTests".format(self.url)) + return self._parseNamedOperation(data) + + def _child_collection(self, attr): + if not attr in self._child_collections: + url = "{0}/{1}".format(self.url, attr) + self._child_collections[attr] = self.load_from_url(url) + return self._child_collections[attr] + + def __getattr__(self, attr): + if attr in self._data: + if isinstance(self._data[attr], dict): + return DataDict(self._data[attr], attr, self) + return self._data[attr] + elif attr in self.collections_map[self.type]: + return self._child_collection(attr) + else: + raise AttributeError(attr) diff --git a/aweber_api/oauth.py b/aweber_api/oauth.py index 2c3cbde..95b8e25 100644 --- a/aweber_api/oauth.py +++ b/aweber_api/oauth.py @@ -1,4 +1,4 @@ -from urllib import urlencode +from urllib.parse import urlencode import json import os @@ -85,7 +85,7 @@ def _get_client(self): return client def _prepare_request_body(self, method, url, data): - if method not in ['POST', 'GET', 'PATCH'] or len(data.keys()) == 0: + if method not in ['POST', 'GET', 'PATCH'] or len(list(data.keys())) == 0: return '' if method in ['POST', 'GET']: # WARNING: non-primative items in data must be json serialized. diff --git a/aweber_api/oauth.py.bak b/aweber_api/oauth.py.bak new file mode 100644 index 0000000..2c3cbde --- /dev/null +++ b/aweber_api/oauth.py.bak @@ -0,0 +1,97 @@ +from urllib import urlencode +import json +import os + +import oauth2 as oauth + +from aweber_api.base import APIException + + +class OAuthAdapter(object): + + def __init__(self, key, secret, base): + self.key = key + self.secret = secret + self.consumer = oauth.Consumer(key=self.key, secret=self.secret) + self.api_base = base + + def _parse(self, response): + try: + data = json.loads(response) + if not data or data == '': + return response + return data + except: + pass + return response + + def request(self, method, url, data={}, response='body'): + client = self._get_client() + url = self._expand_url(url) + body = self._prepare_request_body(method, url, data) + + content_type = 'application/json' + if method == 'GET' and body is not None and body is not '': + if '?' in url: + url = '{0}&{1}'.format(url, body) + else: + url = '{0}?{1}'.format(url, body) + + if method == 'POST': + content_type = 'application/x-www-form-urlencoded' + headers = {'Content-Type': content_type} + + resp, content = client.request( + url, method, body=body, headers=headers) + + if int(resp['status']) >= 400: + """ + API Service Errors: + + Please review the Exception that is raised it should indicate + what the error is. + + refer to https://labs.aweber.com/docs/troubleshooting for more + details. + """ + content = json.loads(content) + error = content.get('error', {}) + error_type = error.get('type') + error_msg = error.get('message') + raise APIException( + '{0}: {1}'.format(error_type, error_msg)) + + if response == 'body' and isinstance(content, str): + return self._parse(content) + if response == 'status': + return resp['status'] + if response == 'headers': + return resp + return None + + def _expand_url(self, url): + if not url[:4] == 'http': + return '{0}{1}'.format(self.api_base, url) + return url + + def _get_client(self): + token = self.user.get_highest_priority_token() + if token: + token = oauth.Token(token, self.user.token_secret) + client = oauth.Client(self.consumer, token=token) + else: + client = oauth.Client(self.consumer) + + return client + + def _prepare_request_body(self, method, url, data): + if method not in ['POST', 'GET', 'PATCH'] or len(data.keys()) == 0: + return '' + if method in ['POST', 'GET']: + # WARNING: non-primative items in data must be json serialized. + for key in data: + if type(data[key]) in [dict, list]: + data[key] = json.dumps(data[key]) + return urlencode(data) + if method == 'PATCH': + return json.dumps(data) diff --git a/tests/mock_adapter.py b/tests/mock_adapter.py index 4b03f43..a8577ab 100644 --- a/tests/mock_adapter.py +++ b/tests/mock_adapter.py @@ -1,7 +1,7 @@ import json import os -from urlparse import urlparse, parse_qs -from urllib import quote +from urllib.parse import urlparse, parse_qs +from urllib.parse import quote import mock diff --git a/tests/mock_adapter.py.bak b/tests/mock_adapter.py.bak new file mode 100644 index 0000000..4b03f43 --- /dev/null +++ b/tests/mock_adapter.py.bak @@ -0,0 +1,144 @@ +import json +import os +from urlparse import urlparse, parse_qs +from urllib import quote + +import mock + +from aweber_api import AWeberUser +from aweber_api import OAuthAdapter + +__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', '') diff --git a/tests/test_aweber_collection.py b/tests/test_aweber_collection.py index c5601fd..0091179 100644 --- a/tests/test_aweber_collection.py +++ b/tests/test_aweber_collection.py @@ -96,7 +96,7 @@ def setUp(self): def test_should_return_new_resource_entry_object(self): assert isinstance(self.resp, AWeberEntry) - assert self.resp.name == u'COLOR' + assert self.resp.name == 'COLOR' assert self.resp.is_subscriber_updateable == True assert self.resp.id == 2 assert self.resp.url == '/accounts/1/lists/303449/custom_fields/2' diff --git a/tests/test_aweber_collection.py.bak b/tests/test_aweber_collection.py.bak new file mode 100644 index 0000000..c5601fd --- /dev/null +++ b/tests/test_aweber_collection.py.bak @@ -0,0 +1,165 @@ +import json +from unittest import TestCase + +from aweber_api import AWeberAPI, AWeberCollection, AWeberEntry +from aweber_api.base import API_BASE, APIException +from mock_adapter import MockAdapter + + +class TestAWeberCollection(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + self.lists = self.aweber.load_from_url('/accounts/1/lists') + self.aweber.adapter.requests = [] + + def test_should_be_a_collection(self): + self.assertTrue(type(self.lists), AWeberCollection) + + def test_should_have_24_lists(self): + self.assertTrue(len(self.lists), 24) + + def test_should_be_able_get_each_via_offset(self): + for i in range(0, 23): + list = self.lists[i] + self.assertEqual(type(list), AWeberEntry) + self.assertEqual(list.type, 'list') + + def test_should_be_able_to_iterate_on_collection(self): + list_number = 0 + for list in self.lists: + self.assertEqual(type(list), AWeberEntry) + self.assertEqual(list.type, 'list') + list_number += 1 + self.assertEqual(list_number, 24) + + def test_should_support_get_by_id(self): + list = self.lists.get_by_id(303449) + self.assertEqual(type(list), AWeberEntry) + self.assertEqual(list.type, 'list') + self.assertEqual(list.id, 303449) + + def test_should_support_find_method(self): + base_url = '/accounts/1/lists/303449/subscribers' + subscriber_collection = self.aweber.load_from_url(base_url) + self.aweber.adapter.requests = [] + subscribers = subscriber_collection.find(email='joe@example.com') + request = self.aweber.adapter.requests[0] + + assert subscribers != False + assert isinstance(subscribers, AWeberCollection) + assert len(subscribers) == 1 + assert subscribers[0].self_link == \ + 'https://api.aweber.com/1.0/accounts/1/lists/303449/subscribers/50205517' + assert request['url'] == \ + '{0}?ws.op=find&email=joe%40example.com'.format(base_url) + + def test_find_should_handle_errors(self): + base_url = '/accounts/1/lists/303449/subscribers' + subscriber_collection = self.aweber.load_from_url(base_url) + self.assertRaises(APIException, subscriber_collection.find, name='joe') + + def test_should_create_entries_with_correct_url(self): + base_url = '/accounts/1' + account = self.aweber.load_from_url(base_url) + subscribers = account.findSubscribers(email='joe@example.com') + for subscriber in subscribers: + assert subscriber.url == subscriber.self_link.replace(API_BASE, '') + + +class TestWhenCreatingCustomFieldsFails(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + cf_url = '/accounts/1/lists/505454/custom_fields' + self.cf = self.aweber.load_from_url(cf_url) + self.aweber.adapter.requests = [] + + def test_should_raise_exception(self): + self.assertRaises(APIException, self.cf.create, name='Duplicate Name') + + +class TestCreatingCustomFields(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + cf_url = '/accounts/1/lists/303449/custom_fields' + self.cf = self.aweber.load_from_url(cf_url) + + self.aweber.adapter.requests = [] + self.resp = self.cf.create(name='Wedding Song') + self.create_req = self.aweber.adapter.requests[0] + self.get_req = self.aweber.adapter.requests[1] + + def test_should_return_new_resource_entry_object(self): + assert isinstance(self.resp, AWeberEntry) + assert self.resp.name == u'COLOR' + assert self.resp.is_subscriber_updateable == True + assert self.resp.id == 2 + assert self.resp.url == '/accounts/1/lists/303449/custom_fields/2' + + +class TestCreateMethod(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + url = '/accounts/1/lists/303449/any_collection' + self.any_collection = self.aweber.load_from_url(url) + + self.aweber.adapter.requests = [] + self.resp = self.any_collection.create( + a_string='Bob', a_dict={'Color': 'blue'}, a_list=['apple']) + self.create_req = self.aweber.adapter.requests[0] + self.get_req = self.aweber.adapter.requests[1] + + def test_should_make_request_with_correct_parameters(self): + expected_params = {'ws.op': 'create', 'a_string': 'Bob', + 'a_dict': json.dumps({'Color': 'blue'}), + 'a_list': json.dumps(['apple'])} + + self.assertEqual(self.create_req['data'], expected_params) + + def test_should_make_two_requests(self): + self.assertEqual(len(self.aweber.adapter.requests), 2) + + def test_should_have_requested_create_on_cf(self): + self.assertEqual(self.create_req['url'] , self.any_collection.url) + + def test_should_have_requested_create_with_post(self): + self.assertEqual(self.create_req['method'], 'POST') + + def test_should_refresh_created_resource(self): + self.assertEqual(self.get_req['method'], 'GET') + self.assertEqual(self.get_req['url'] , + '/accounts/1/lists/303449/any_collection/1') + + +class TestGettingParentEntry(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + self.lists = self.aweber.load_from_url('/accounts/1/lists') + self.accounts = self.aweber.load_from_url('/accounts') + self.custom_fields = self.aweber.load_from_url('/accounts/1/lists/303449/custom_fields') + + def test_should_be_able_get_parent_entry(self): + entry = self.lists.get_parent_entry() + + def test_lists_parent_should_be_account(self): + entry = self.lists.get_parent_entry() + self.assertEqual(type(entry), AWeberEntry) + self.assertEqual(entry.type, 'account') + + def test_custom_fields_parent_should_be_list(self): + entry = self.custom_fields.get_parent_entry() + self.assertEqual(type(entry), AWeberEntry) + self.assertEqual(entry.type, 'list') + + def test_accounts_parent_should_be_none(self): + entry = self.accounts.get_parent_entry() + self.assertEqual(entry, None) diff --git a/tests/test_aweber_entry.py b/tests/test_aweber_entry.py index eb73232..637010a 100644 --- a/tests/test_aweber_entry.py +++ b/tests/test_aweber_entry.py @@ -1,6 +1,6 @@ import re from unittest import TestCase -from urllib import urlencode +from urllib.parse import urlencode from aweber_api import AWeberAPI, AWeberCollection, AWeberEntry from aweber_api.base import APIException diff --git a/tests/test_aweber_entry.py.bak b/tests/test_aweber_entry.py.bak new file mode 100644 index 0000000..eb73232 --- /dev/null +++ b/tests/test_aweber_entry.py.bak @@ -0,0 +1,405 @@ +import re +from unittest import TestCase +from urllib import urlencode + +from aweber_api import AWeberAPI, AWeberCollection, AWeberEntry +from aweber_api.base import APIException +from mock_adapter import MockAdapter + + +class TestAWeberEntry(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + self.list = self.aweber.load_from_url('/accounts/1/lists/303449') + + def test_should_be_an_entry(self): + self.assertEqual(type(self.list), AWeberEntry) + self.assertEqual(self.list.type, 'list') + + def test_should_have_id(self): + self.assertEqual(self.list.id, 303449) + + def test_should_have_other_properties(self): + self.assertEqual(self.list.name, 'default303449') + + def test_should_have_child_collections(self): + campaigns = self.list.campaigns + self.assertEqual(type(campaigns), AWeberCollection) + + def test_findSubscribers_should_handle_errors(self): + account = self.aweber.load_from_url('/accounts/1') + self.assertRaises(APIException, account.findSubscribers, name='bob') + + +class AccountTestCase(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + self.account = self.aweber.load_from_url('/accounts/1') + + +class ListTestCase(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + self.list_ = self.aweber.load_from_url('/accounts/1/lists/303449') + + +class TestAWeberAccountEntry(AccountTestCase): + + def test_should_be_an_entry(self): + self.assertEqual(type(self.account), AWeberEntry) + self.assertEqual(self.account.type, 'account') + + +class TestAccountGetWebForms(AccountTestCase): + + def setUp(self): + super(TestAccountGetWebForms, self).setUp() + self.forms = self.account.get_web_forms() + + def test_should_be_a_list(self): + self.assertEqual(type(self.forms), list) + + def test_should_have_181_web_forms(self): + self.assertEqual(len(self.forms), 181) + + def test_each_should_be_entry(self): + for entry in self.forms: + self.assertEqual(type(entry), AWeberEntry) + self.assertEqual(entry.type, 'web_form') + + def test_each_should_have_correct_url(self): + url_regex = '/accounts\/1\/lists\/\d*/web_forms/\d*' + for entry in self.forms: + self.assertTrue(re.match(url_regex, entry.url)) + + +class TestAccountGetWebFormSplitTests(AccountTestCase): + + def setUp(self): + super(TestAccountGetWebFormSplitTests, self).setUp() + self.forms = self.account.get_web_form_split_tests() + + def test_should_be_a_list(self): + self.assertEqual(type(self.forms), list) + + def test_should_have_10_split_tests(self): + self.assertEqual(len(self.forms), 10) + + def test_each_should_be_entry(self): + for entry in self.forms: + self.assertEqual(type(entry), AWeberEntry) + self.assertEqual(entry.type, 'web_form_split_test') + + def test_each_should_have_correct_url(self): + url_regex = '/accounts\/1\/lists\/\d*/web_form_split_tests/\d*' + for entry in self.forms: + self.assertTrue(re.match(url_regex, entry.url)) + + +class TestAccountFindSubscribers(AccountTestCase): + + def test_should_support_find_method(self): + base_url = '/accounts/1' + account = self.aweber.load_from_url(base_url) + self.aweber.adapter.requests = [] + subscribers = account.findSubscribers(email='joe@example.com') + request = self.aweber.adapter.requests[0] + + assert subscribers != False + assert isinstance(subscribers, AWeberCollection) + assert len(subscribers) == 1 + assert subscribers[0].self_link == \ + 'https://api.aweber.com/1.0/accounts/1/lists/303449/subscribers/1' + + +class TestListScheduleBroadcast(ListTestCase): + + def setUp(self): + super(TestListScheduleBroadcast, self).setUp() + self.aweber.adapter.requests = [] + self.status = self.list_.schedule_broadcast( + bc_id=2, scheduled_for='2014-09-06 18:55:00') + self.request = self.aweber.adapter.requests[0] + + def test_should_return_status(self): + self.assertEqual(int(self.status), 201) + + def test_should_make_post_request(self): + self.assertEqual(self.request['method'], 'POST') + + def test_should_build_correct_url(self): + self.assertEqual(self.request['url'], + '/accounts/1/lists/303449/broadcasts/2/schedule' + ) + + def test_should_pass_scheduled_for_date(self): + self.assertEqual(self.request['data'], + {'scheduled_for': '2014-09-06 18:55:00'} + ) + + +class TestListScheduleBroadcastError(ListTestCase): + + def setUp(self): + super(TestListScheduleBroadcastError, self).setUp() + self.list_ = self.aweber.load_from_url('/accounts/1/lists/303449') + self.aweber.adapter.requests = [] + + def test_should_raise_exception_when_failing(self): + self.assertRaises( + APIException, + self.list_.schedule_broadcast, + bc_id=3, + scheduled_for='2014-09-06 18:55:00', + ) + + +class TestListCancelBroadcast(ListTestCase): + + def setUp(self): + super(TestListCancelBroadcast, self).setUp() + self.aweber.adapter.requests = [] + self.status = self.list_.cancel_broadcast(bc_id=2) + self.request = self.aweber.adapter.requests[0] + + def test_should_return_status(self): + self.assertEqual(int(self.status), 204) + + def test_should_make_post_request(self): + self.assertEqual(self.request['method'], 'POST') + + def test_should_build_correct_url(self): + self.assertEqual(self.request['url'], + '/accounts/1/lists/303449/broadcasts/2/cancel' + ) + + def test_should_pass_empty_date(self): + self.assertEqual(self.request['data'], {}) + + +class TestListCancelBroadcastError(ListTestCase): + + def setUp(self): + super(TestListCancelBroadcastError, self).setUp() + self.list_ = self.aweber.load_from_url('/accounts/1/lists/303449') + self.aweber.adapter.requests = [] + + def test_should_raise_exception_when_failing(self): + self.assertRaises( + APIException, + self.list_.cancel_broadcast, + bc_id=3, + ) + + +class TestListGetBroadcasts(ListTestCase): + + def setUp(self): + super(TestListGetBroadcasts, self).setUp() + self.aweber.adapter.requests = [] + self.broadcasts = self.list_.get_broadcasts(status='sent') + self.request = self.aweber.adapter.requests[0] + + def test_should_return_collection(self): + self.assertEqual(type(self.broadcasts), AWeberCollection) + + def test_should_make_get_request(self): + self.assertEqual(self.request['method'], 'GET') + + def test_should_build_correct_url(self): + self.assertEqual( + self.request['url'], + '/accounts/1/lists/303449/broadcasts?status=sent' + ) + + +class SubscriberTestCase(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + sub_url = '/accounts/1/lists/303449/subscribers/1' + self.subscriber = self.aweber.load_from_url(sub_url) + + +class TestGetAndSetData(SubscriberTestCase): + + def test_get_name(self): + self.assertEqual(self.subscriber.name, 'Joe Jones') + + def test_set_name(self): + self.subscriber.name = 'Randy Rhodes' + self.assertEqual(self.subscriber.name, 'Randy Rhodes') + + def test_get_custom_fields(self): + fields = self.subscriber.custom_fields + self.assertEqual(fields['Color'], 'blue') + + def test_set_custom_fields(self): + self.subscriber.custom_fields['Color'] = 'Red' + self.assertEqual(self.subscriber._data['custom_fields']['Color'], 'Red') + fields = self.subscriber.custom_fields + self.assertEqual(fields['Color'], 'Red') + + def test_should_be_able_get_activity(self): + activity = self.subscriber.get_activity() + + +class TestMovingSubscribers(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + subscriber_url = '/accounts/1/lists/303449/subscribers/1' + new_list_url = '/accounts/1/lists/505454' + self.subscriber = self.aweber.load_from_url(subscriber_url) + self.subscriber._diff['name'] = 'Joe Schmoe' + self.list = self.aweber.load_from_url(new_list_url) + self.move_subscriber() + + def move_subscriber(self, **kwargs): + self.aweber.adapter.requests = [] + self.resp = self.subscriber.move(self.list, **kwargs) + self.move_req = self.aweber.adapter.requests[0] + self.get_req = self.aweber.adapter.requests[1] + + def test_returned_true(self): + self.assertTrue(self.resp) + + def test_should_have_requested_move_with_post(self): + self.assertEqual(self.move_req['method'], 'POST') + + def test_should_have_requested_move_on_subscriber(self): + self.assertEqual(self.move_req['url'] , self.subscriber.url) + + def test_should_have_requested_move_with_correct_parameters(self): + expected_params = {'ws.op': 'move', 'list_link': self.list.self_link} + self.assertEqual(self.move_req['data'], expected_params) + + def test_should_make_two_requests(self): + self.assertEqual(len(self.aweber.adapter.requests), 2) + + def test_should_refresh_subscriber_resource(self): + self.assertEqual(self.get_req['method'], 'GET') + self.assertEqual(self.get_req['url'] , + '/accounts/1/lists/505454/subscribers/3') + + def test_should_reset_diff(self): + self.assertEqual(self.subscriber._diff, {}) + + def test_should_accept_last_followup_message_number_sent(self): + self.move_subscriber(last_followup_message_number_sent=999) + expected_params = {'ws.op': 'move', 'list_link': self.list.self_link, + 'last_followup_message_number_sent': 999} + + self.assertEqual(self.move_req['data'], expected_params) + +class TestSavingSubscriberData(SubscriberTestCase): + + def setUp(self): + super(TestSavingSubscriberData, self).setUp() + self.aweber.adapter.requests = [] + self.subscriber.name = 'Gary Oldman' + self.subscriber.custom_fields['Color'] = 'Red' + self.resp = self.subscriber.save() + self.req = self.aweber.adapter.requests[0] + + def test_returned_true(self): + self.assertTrue(self.resp) + + def test_should_make_request(self): + self.assertEqual(len(self.aweber.adapter.requests), 1) + + def test_should_have_requested_resource_url(self): + self.assertEqual(self.req['url'] , self.subscriber.url) + + def test_should_have_requested_with_patch(self): + self.assertEqual(self.req['method'], 'PATCH') + + def test_should_have_supplied_data(self): + self.assertEqual(self.req['data']['name'], 'Gary Oldman') + + def test_should_not_include_unchanged_data(self): + self.assertFalse('email' in self.req['data']) + + def test_should_given_all_custom_fields(self): + # Make changed, Model did not + self.assertEqual(self.req['data']['custom_fields']['Color'], 'Red') + self.assertEqual(self.req['data']['custom_fields']['Walruses'], '') + + +class TestSavingInvalidSubscriberData(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + sub_url = '/accounts/1/lists/303449/subscribers/2' + self.subscriber = self.aweber.load_from_url(sub_url) + self.subscriber.name = 'Gary Oldman' + self.subscriber.custom_fields['New Custom Field'] = 'Cookies' + + def test_save_failed(self): + self.assertRaises(APIException, self.subscriber.save) + + +class TestDeletingSubscriberData(SubscriberTestCase): + + def setUp(self): + super(TestDeletingSubscriberData, self).setUp() + self.aweber.adapter.requests = [] + self.response = self.subscriber.delete() + self.req = self.aweber.adapter.requests[0] + + def test_should_be_deleted(self): + self.assertTrue(self.response) + + def test_should_have_made_request(self): + self.assertEqual(len(self.aweber.adapter.requests), 1) + + def test_should_have_made_delete(self): + self.assertEqual(self.req['method'], 'DELETE') + + +class TestFailedSubscriberDelete(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + sub_url = '/accounts/1/lists/303449/subscribers/2' + self.subscriber = self.aweber.load_from_url(sub_url) + + def test_should_raise_exception_when_failing(self): + self.assertRaises(APIException, self.subscriber.delete) + + +class TestGettingParentEntry(TestCase): + + def setUp(self): + self.aweber = AWeberAPI('1', '2') + self.aweber.adapter = MockAdapter() + self.list = self.aweber.load_from_url('/accounts/1/lists/303449') + self.account = self.aweber.load_from_url('/accounts/1') + self.custom_field = self.aweber.load_from_url('/accounts/1/lists/303449/custom_fields/1') + + def test_should_be_able_get_parent_entry(self): + entry = self.list.get_parent_entry() + + def test_list_parent_should_be_account(self): + entry = self.list.get_parent_entry() + self.assertEqual(type(entry), AWeberEntry) + self.assertEqual(entry.type, 'account') + + def test_custom_field_parent_should_be_list(self): + entry = self.custom_field.get_parent_entry() + self.assertEqual(type(entry), AWeberEntry) + self.assertEqual(entry.type, 'list') + + def test_account_parent_should_be_none(self): + entry = self.account.get_parent_entry() + self.assertEqual(entry, None)