Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Basic merchant functionality implemented

Currently able to look up resources from the API using an Active Resource
pattern and can create connect urls.
  • Loading branch information...
commit c7e8c937d5a734e746e3c1439a39a4c5b9f84b8d 1 parent 48b6b68
@alexjg alexjg authored
View
1  .gitignore
@@ -1 +1,2 @@
*.pyc
+*.sw*
View
4 gocardless/__init__.py
@@ -1,4 +1,8 @@
from .client import Client
environment = 'production'
+client = None
+
+def set_details(details):
+ client = Client(details)
View
193 gocardless/client.py
@@ -1,9 +1,18 @@
+import base64
+import datetime
import json
+import logging
+import os
+import urllib
import gocardless
+import urlbuilder
+from .utils import generate_signature, to_query
from .request import Request
-from .exceptions import ClientError
+from .exceptions import ClientError, SignatureError
+from .resources import Merchant, Subscription, Bill, PreAuthorization, User
+logger = logging.getLogger(__name__)
class Client(object):
@@ -18,7 +27,8 @@ class Client(object):
@classmethod
def get_base_url(cls):
- """Return the correct base URL for the current environment. If one has
+ """
+ Return the correct base URL for the current environment. If one has
been manually set, default to that.
"""
return cls.base_url or cls.BASE_URLS[gocardless.environment]
@@ -30,25 +40,39 @@ def __init__(self, **account_details):
if 'app_secret' not in account_details:
raise ValueError('You must provide an app_secret')
+ if "token" not in account_details:
+ raise ValueError("You must provide an access token")
+
self._app_id = account_details['app_id']
self._app_secret = account_details['app_secret']
self._access_token = account_details.get('token')
self._merchant_id = account_details.get('merchant_id')
def api_get(self, path, **kwargs):
- """Issue an GET request to the API server.
+ """
+ Issue an GET request to the API server.
:param path: the path that will be added to the API prefix
- :param params: query string parameters
"""
return self._request('get', Client.API_PATH + path, **kwargs)
+ def api_post(self, path, data, **kwargs):
+ """Issue a PUT request to the API server
+
+ :param path: The path that will be added to the API prefix
+ :param data: The data to post to the url.
+ """
+ self.set_payload(data)
+ return self._request('post', Client.API_PATH + path, **kwargs)
+
def _request(self, method, path, **kwargs):
- """Send a request to the GoCardless API servers.
+ """
+ Send a request to the GoCardless API servers.
:param method: the HTTP method to use (e.g. +:get+, +:post+)
:param path: the path fragment of the URL
"""
+ logger.debug("Executing request to path {0}".format(path))
request_url = Client.get_base_url() + path
request = Request(method, request_url)
@@ -63,29 +87,160 @@ def _request(self, method, path, **kwargs):
return request.perform()
def merchant(self):
- """Returns the current Merchant's details.
-
"""
- return self.api_get('/merchants/%s' % self._merchant_id)
+ Returns the current Merchant's details.
+ """
+ return Merchant(self.api_get('/merchants/%s' % self._merchant_id), self)
def users(self):
- """Index a merchant's customers
+ """
+ Index a merchant's customers
"""
return self.api_get('/merchants/%s/users' % self._merchant_id)
-
-
- def subscriptions(self):
- """Returns all subscriptions for a merchant."""
- return self.api_get('/merchants/%s/subscriptions/' % self._merchant_id)
+ def user(self, id):
+ """
+ Find a user by id
+ """
+ return User.find_with_client(id, self)
- def get_subscription(self, id):
- """Returns a single subscription
+ def pre_authorization(self, id):
"""
- return self.api_get('/subscriptions/%s' % (id))
+ Find a pre authorization with id `id`
+ """
+ return PreAuthorization.find_with_client(id, self)
- def cancel_subscription(self, id):
- """Cancels a subscription given an id"""
+ def subscription(self, id):
+ """
+ Returns a single subscription
+ """
+ return Subscription.find_with_client(id, self)
+
+ def bill(self, id):
+ """
+ Find a bill with id `id`
+ """
+ return Bill.find_with_client(id, self)
+
+ def new_subscription_url(self, amount, interval_length, interval_unit,
+ name=None, description=None, interval_count=None, start_at=None,
+ expires_at=None, redirect_uri=None, cancel_uri=None, state=None):
+ """Generate a url for creating a new subscription
+
+ :param amount: The amount to charge each time
+ :param interval_length: The length of time between each charge, this
+ is an integer, the units are specified by interval_unit.
+ :param interval_unit: The unit to measure the interval length, must
+ be one of "day' or "week"
+ :param name: The name to give the suvscription
+ :param description: The description of the subscription
+ :param interval_count: The Calculates expires_at based on the number
+ of intervals you would like to collect. If both interval_count and
+ expires_at are specified the expires_at parameter will take
+ precedence
+ :param expires_at: When the subscription expires, should be a datetime
+ object.
+ :param starts_at: When the subscription starts, should be a datetime
+ object
+ :param redirect_uri: URI to redirect to after the authorization process
+ :param cancel_uri: URI to redirect the user to if they cancel
+ authorization
+ :param state: String which will be passed to the merchant on
+ redirect.
+ """
+ params = urlbuilder.SubscriptionParams(amount, self._merchant_id,
+ interval_length, interval_unit, name=name,
+ description=description, interval_count=interval_count,
+ expires_at=expires_at, start_at=start_at)
+ builder = urlbuilder.UrlBuilder(self, redirect_uri=redirect_uri,
+ cancel_uri=cancel_uri, state=state)
+ return builder.build_and_sign(params)
+
+
+ def new_bill_url(self, amount, name=None, description=None,
+ redirect_uri=None, cancel_uri=None, state=None):
+ """Generate a url for creating a new bill
+
+ :param amount: The amount to bill the customer
+ :param name: The name of the bill
+ :param description: The description of the bill
+ :param redirect_uri: URI to redirect to after the authorization process
+ :param cancel_uri: URI to redirect the user to if they cancel
+ authorization
+ :param state: String which will be passed to the merchant on
+ redirect.
+
+ """
+ params = urlbuilder.BillParams(amount, self._merchant_id, name=name,
+ description=description)
+ builder = urlbuilder.UrlBuilder(self, redirect_uri=redirect_uri,
+ cancel_uri=cancel_uri, state=state)
+ return builder.build_and_sign(params)
+
+ def new_preauthorization_url(self,max_amount, interval_length,\
+ interval_unit, expires_at=None, name=None, description=None,\
+ interval_count=None, calendar_intervals=None,
+ redirect_uri=None, cancel_uri=None, state=None):
+ """Get a url for creating new pre_authorizations
+
+ :param max_amount: A float which is the maximum amount for this
+ pre_authorization
+ :param interval_length: The length of this pre_authorization
+ :param interval_unit: The units in which the interval_length
+ is measured, must be one of
+ - "day"
+ - "month"
+ :param expires_at: The date that this pre_authorization will
+ expire, must be a datetime object which is in the future.
+ :param name: A short string which is the name of the pre_authorization
+ :param description: A longer string describing what the
+ pre_authorization is for.
+ :param interval_count: calculates expires_at based on the number of
+ payment intervals you would like the resource to have. Must be a
+ positive integer greater than 0. If you specify both an interval_count
+ and an expires_at argument then the expires_at argument will take
+ precedence.
+ :param calendar_intervals: Describes whether the interval resource
+ should be aligned with calendar weeks or months, default is False
+ :param redirect_uri: URI to redirect to after the authorization process
+ :param cancel_uri: URI to redirect the user to if they cancel
+ authorization
+ :param state: String which will be passed to the merchant on
+ redirect.
+
+ """
+ params = urlbuilder.PreAuthorizationParams(max_amount, self._merchant_id, \
+ interval_length, interval_unit, expires_at=expires_at, name=name, description=description,\
+ interval_count=interval_count, calendar_intervals=calendar_intervals)
+ builder = urlbuilder.UrlBuilder(self, redirect_uri=redirect_uri, cancel_uri=cancel_uri, state=state)
+ return builder.build_and_sign(params)
+
+ def confirm_resource(self, params):
+ """Confirm a payment
+
+ This send a post request to the confirmation URI for a payment.
+ params should contain these elements from the request
+ - resource_uri
+ - resource_id
+ - resource_type
+ - signature
+ - state (if any)
+ """
+ keys = ["resource_uri", "resource_id", "resource_type", "state"]
+ to_check = dict([[k,v] for k,v in params.items() if k in keys])
+ signature = generate_signature(to_check, self._app_secret)
+ if not signature == params["signature"]:
+ raise SignatureError("Invalid signature when confirming resource")
+ auth_string = base64.b64encode("{0}:{1}".format(
+ self._app_id, self._app_secret))
+ to_post = {
+ "resource_id":params["resource_id"],
+ "resource_type":params["resource_type"]
+ }
+ self.api_post(params["resource_uri"], to_post, auth=auth_string)
+
+
+
View
3  gocardless/exceptions.py
@@ -2,3 +2,6 @@
class ClientError(Exception):
pass
+class SignatureError(Exception):
+ pass
+
View
34 gocardless/merchant.py
@@ -0,0 +1,34 @@
+
+class Merchant(object):
+
+ def __init__(self, client, data):
+ self.id = data["merchant_id"]
+ self.client = client
+ self.endpoint = "/merchants/{0}".format(self.id)
+
+ def subscriptions(self):
+ """
+ Return all the subscriptions for this merchant.
+ """
+ path = "/merchants/{0}/subscriptions".format(self.id)
+ return self.client.subscriptions()
+
+ def subscription(self, subscription_id):
+ """
+ Return the subscription with id `subscription_id` or `None`
+ """
+ return self.client.subscription(subscription_id)
+
+ def pre_authorizations(self):
+ """
+ Return all the pre-authorizations for this merchant
+ """
+ return self.client.pre_authorizations()
+
+ def pre_authorization(self, pre_authorization_id):
+ """
+ Return the pre_authorization with `pre_authorization_id` or `None`
+ """
+
+
+
View
74 gocardless/resources.py
@@ -0,0 +1,74 @@
+import datetime
+import json
+import logging
+import re
+import sys
+import types
+
+import utils
+import gocardless
+from gocardless.exceptions import ClientError
+
+
+class Resource(object):
+
+ def __init__(self, attrs, client):
+ self.id = attrs["id"]
+ self.client = client
+ if "sub_resource_uris" in attrs:
+ for name, uri in attrs.pop("sub_resource_uris").items():
+ path = re.sub(".*/api/v1", "", uri)
+ module = sys.modules[self.__module__]
+ sub_klass = getattr(module, utils.singularize(utils.camelize(name)))
+ def create_get_resource_func(the_path, the_klass):
+ """
+ In python functions close over their environment so in
+ order to create the correct closure we need a function
+ creator, see
+ http://stackoverflow.com/questions/233673/\
+ lexical-closures-in-python/235764#235764
+ """
+ def get_resources(inst):
+ data = inst.client.api_get(the_path)
+ return [the_klass(attrs, self.client) for attrs in data]
+ return get_resources
+ res_func = create_get_resource_func(path, sub_klass)
+ func_name = "get_{0}".format(name)
+ res_func.name = func_name
+ setattr(self, func_name, types.MethodType(res_func, self, self.__class__))
+ self.created_at = datetime.datetime.strptime(attrs.pop("created_at"), "%Y-%m-%dT%H:%M:%SZ")
+ for key, value in attrs.items():
+ setattr(self, key, value)
+
+ def get_endpoint(self):
+ return self.endpoint.replace(":id", self.id)
+
+ @classmethod
+ def find_with_client(cls, id, client):
+ path = cls.endpoint.replace(":id", id)
+ return cls(client.api_get(path), client)
+
+ @classmethod
+ def find(cls, id):
+ if not gocardless.client:
+ raise ClientError("You must set your account details first")
+ return cls.find_with_client(id, gocardless.client)
+
+
+
+class Merchant(Resource):
+ endpoint = "/merchants/:id"
+
+class Subscription(Resource):
+ endpoint = "/subscriptions/:id"
+
+class PreAuthorization(Resource):
+ endpoint = "/pre_authorizations/:id"
+
+class Bill(Resource):
+ endpoint = "/bills/:id"
+
+class User(Resource):
+ endpoint = "/users/:id"
+
+
View
187 gocardless/urlbuilder.py
@@ -0,0 +1,187 @@
+import base64
+import datetime
+import os
+import urlparse
+import utils
+
+class UrlBuilder(object):
+
+ def __init__(self, client):
+ self.client = client
+
+ def build_and_sign(self, params, state=None, redirect_uri=None,
+ cancel_uri=None):
+ param_dict = {}
+ param_dict[utils.singularize(params.resource_name)] = params.to_dict().copy()
+ if state:
+ param_dict["state"] = state
+ if redirect_uri:
+ param_dict["redirect_uri"] = redirect_uri
+ if cancel_uri:
+ param_dict["cancel_uri"] = cancel_uri
+ param_dict["client_id"] = self.client._app_id
+ param_dict["timestamp"] = datetime.datetime.now().isoformat()[:-7] + "Z"
+ param_dict["nonce"] = base64.b64encode(os.urandom(40))
+
+ signature = utils.generate_signature(param_dict, self.client._app_secret)
+ param_dict["signature"] = signature
+ url = self.client.get_base_url() + "/connect/" + params.resource_name + \
+ "/new?" + utils.to_query(param_dict)
+ return url
+
+class BasicParams(object):
+
+ def __init__(self, amount, merchant_id, name=None, description=None):
+ if not amount > 0:
+ raise ValueError("amount must be positive, value passed was"
+ " {0}".format(amount))
+ self.amount = amount
+ self.merchant_id = merchant_id
+
+ if name:
+ self.name = name
+
+ if description:
+ self.description = description
+ self.attrnames = ["amount", "name", "description", "merchant_id"]
+
+ def to_dict(self):
+ result = {}
+ for attrname in self.attrnames:
+ val = getattr(self, attrname, None)
+ if val:
+ result[attrname] = val
+ return result
+
+
+class PreAuthorizationParams(object):
+
+ def __init__(self,max_amount, merchant_id, interval_length,\
+ interval_unit, expires_at=None, name=None, description=None,\
+ interval_count=None, calendar_intervals=None):
+
+ self.merchant_id = merchant_id
+ self.resource_name = "pre_authorizations"
+
+ if not max_amount > 0:
+ raise ValueError("""max_amount must be
+ positive value passed was {0}""".format(max_amount))
+ self.max_amount = max_amount
+
+ if not interval_length > 0:
+ raise ValueError("interval_length must be positive, value "
+ "passed was {0}".format(interval_length))
+ self.interval_length = interval_length
+
+ valid_units = ["month", "day"]
+ if interval_unit not in valid_units:
+ raise ValueError("interval_unit must be one of {0},"
+ "value passed was {1}".format(valid_units, interval_unit))
+ self.interval_unit = interval_unit
+
+ if expires_at:
+ if (expires_at - datetime.datetime.now()).total_seconds() < 0:
+ raise ValueError("expires_at must be in the future, date "
+ "passed was {0}".format(expires_at.isoformat()))
+ self.expires_at = expires_at
+ else:
+ self.expires_at = None
+
+ if interval_count:
+ if interval_count < 0:
+ raise ValueError("interval_count must be positive "
+ "value passed was {0}".format(interval_count))
+ self.interval_count = interval_count
+ else:
+ self.interval_count = None
+
+ self.name = name if name else None
+ self.description = description if description else None
+ self.calendar_intervals = calendar_intervals if calendar_intervals\
+ else None
+
+ def to_dict(self):
+ result = {}
+ attrnames = ["merchant_id", "name", "description", \
+ "interval_count", "interval_unit", "interval_length", \
+ "max_amount", "calendar_intervals", "expires_at", \
+ ]
+ for attrname in attrnames:
+ val = getattr(self, attrname, None)
+ if val:
+ result[attrname] = val
+ return result
+
+
+class BillParams(BasicParams):
+
+ def __init__(self, amount, merchant_id, name=None, description=None):
+ BasicParams.__init__(self, amount, merchant_id, name=name, description=description)
+ self.resource_name = "bills"
+
+
+
+class SubscriptionParams(BasicParams):
+
+ def __init__(self, amount, merchant_id, interval_length, interval_unit,
+ name=None, description=None,start_at=None, expires_at=None,
+ interval_count=None):
+ BasicParams.__init__(self, amount, merchant_id, description=description, name=name)
+ self.resource_name = "subscriptions"
+ self.merchant_id = merchant_id
+
+ if not interval_length > 0:
+ raise ValueError("interval_length must be positive, value "
+ "passed was {0}".format(interval_length))
+ self.interval_length = interval_length
+
+ valid_units = ["month", "day"]
+ if interval_unit not in valid_units:
+ raise ValueError("interval_unit must be one of {0},"
+ "value passed was {1}".format(valid_units, interval_unit))
+ self.interval_unit = interval_unit
+
+ if expires_at:
+ self.check_date_in_future(expires_at, "expires_at")
+ self.expires_at = expires_at
+
+ if start_at:
+ self.check_date_in_future(start_at, "start_at")
+ self.start_at = start_at
+
+ if expires_at and start_at:
+ if (expires_at - start_at).total_seconds() < 0:
+ raise ValueError("start_at must be before expires_at")
+
+ if interval_count:
+ if interval_count < 0:
+ raise ValueError("interval_count must be positive "
+ "value passed was {0}".format(interval_count))
+ self.interval_count = interval_count
+
+ self.name = name if name else None
+ self.description = description if description else None
+
+ self.attrnames.extend(["description", "interval_count",
+ "interval_unit", "interval_length", "expires_at",
+ "start_at"])
+
+ def check_date_in_future(self, date, argname):
+ if (date - datetime.datetime.now()).total_seconds() < 0:
+ raise ValueError("{0} must be in the future, date passed was"
+ "{1}".format(argname, date.isoformat()))
+
+
+ def to_dict(self):
+ result = {}
+ for attrname in self.attrnames:
+ val = getattr(self, attrname, None)
+ if val:
+ if attrname in ["start_at", "expires_at"]:
+ result[attrname] = val.isoformat()[:-7] + "Z"
+ else:
+ result[attrname] = val
+ return result
+
+
+
View
11 gocardless/utils.py
@@ -1,11 +1,11 @@
import urllib
import hashlib
import hmac
+import re
def percent_encode(string):
return urllib.quote(string.encode('utf-8'), '~')
-
def to_query(obj, ns=None):
if isinstance(obj, dict):
pairs = sum((to_query(v, u"{0}[{1}]".format(ns, k) if ns else k)
@@ -25,3 +25,12 @@ def generate_signature(data, secret):
digest of the data.
"""
return hmac.new(secret, to_query(data), hashlib.sha256).hexdigest()
+
+def camelize(to_uncamel):
+ result = []
+ for word in re.split("_", to_uncamel):
+ result.append(word[0].upper() + word[1:])
+ return "".join(result)
+
+def singularize(to_sing):
+ return re.sub("s$", "", to_sing)
View
308 test/test_client.py
@@ -0,0 +1,308 @@
+import base64
+import datetime
+import json
+import unittest
+import mock
+from mock import patch
+import sys
+import urlparse
+
+import gocardless
+from gocardless.client import Client
+from gocardless import utils
+from gocardless import urlbuilder
+from gocardless.exceptions import SignatureError
+from test_resources import create_mock_attrs
+
+merchant_json = json.loads("""{
+ "created_at": "2011-11-18T17:07:09Z",
+ "description": null,
+ "id": "WOQRUJU9OH2HH1",
+ "name": "Tom's Delicious Chicken Shop",
+ "first_name": "Tom",
+ "last_name": "Blomfield",
+ "email": "tom@gocardless.com",
+ "uri": "https://gocardless.com/api/v1/merchants/WOQRUJU9OH2HH1",
+ "balance": "12.00",
+ "pending_balance": "0.00",
+ "next_payout_date": "2011-11-25T17: 07: 09Z",
+ "next_payout_amount": "12.00",
+ "sub_resource_uris": {
+ "users": "https://gocardless.com/api/v1/merchants/WOQRUJU9OH2HH1/users",
+ "bills": "https://gocardless.com/api/v1/merchants/WOQRUJU9OH2HH1/bills",
+ "pre_authorizations": "https://gocardless.com/api/v1/merchants/WOQRUJU9OH2HH1/pre_authorizations",
+ "subscriptions": "https://gocardless.com/api/v1/merchants/WOQRUJU9OH2HH1/subscriptions"
+ }
+}
+""")
+
+subscription_json = json.loads("""
+{
+ "amount":"44.0",
+ "interval_length":1,
+ "interval_unit":"month",
+ "created_at":"2011-09-12T13:51:30Z",
+ "currency":"GBP",
+ "name":"London Gym Membership",
+ "description":"Entitles you to use all of the gyms around London",
+ "expires_at":null,
+ "next_interval_start":"2011-10-12T13:51:30Z",
+ "id": "AJKH638A99",
+ "merchant_id":"WOQRUJU9OH2HH1",
+ "status":"active",
+ "user_id":"HJEH638AJD",
+ "uri":"https://gocardless.com/api/v1/subscriptions/1580",
+ "sub_resource_uris":{
+ "bills":"https://gocardless.com/api/v1/merchants/WOQRUJU9OH2HH1/bills?source_id=1580"
+ }
+}
+""")
+
+mock_account_details = {
+ 'app_id': 'id01',
+ 'app_secret': 'sec01',
+ 'token': 'tok01',
+ 'merchant_id': merchant_json["id"],
+ }
+
+class ClientTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.account_details = mock_account_details.copy()
+ self.client = Client(**self.account_details)
+
+ def test_base_url_returns_the_correct_url_for_production(self):
+ gocardless.environment = 'production'
+ self.assertEqual(Client.get_base_url(), 'https://gocardless.com')
+
+ def test_base_url_returns_the_correct_url_for_sandbox(self):
+ gocardless.environment = 'sandbox'
+ self.assertEqual(Client.get_base_url(), 'https://sandbox.gocardless.com')
+
+ def test_base_url_returns_the_correct_url_when_set_manually(self):
+ Client.base_url = 'https://abc.gocardless.com'
+ self.assertEqual(Client.get_base_url(), 'https://abc.gocardless.com')
+
+ def test_app_id_required(self):
+ self.account_details.pop('app_id')
+ with self.assertRaises(ValueError):
+ Client(**self.account_details)
+
+ def test_app_secret_required(self):
+ self.account_details.pop('app_secret')
+ with self.assertRaises(ValueError):
+ Client(**self.account_details)
+
+ def test_get_merchant(self):
+ with patch.object(self.client, 'api_get'):
+ self.client.api_get.return_value = merchant_json
+ merchant = self.client.merchant()
+ self.assertEqual(merchant.id, self.account_details["merchant_id"])
+
+ def test_get_subscription(self):
+ self.get_resource_tester("subscription", subscription_json)
+
+ def test_get_user(self):
+ self.get_resource_tester("user", create_mock_attrs({}))
+
+ def test_get_pre_authorization(self):
+ self.get_resource_tester("pre_authorization", create_mock_attrs({}))
+
+ def test_get_bill(self):
+ self.get_resource_tester("bill", create_mock_attrs({}))
+
+ def get_resource_tester(self, resource_name, resource_fixture):
+ expected_klass = getattr(sys.modules["gocardless.resources"], utils.camelize(resource_name))
+ with patch.object(self.client, 'api_get'):
+ self.client.api_get.return_value = resource_fixture
+ obj = getattr(self.client, resource_name)("1")
+ self.assertEqual(resource_fixture["id"], obj.id)
+ self.assertIsInstance(obj, expected_klass)
+
+class ConfirmResourceTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.client = Client(**mock_account_details)
+ self.params = {
+ "resource_uri":"http://aresource",
+ "resource_id":"1",
+ "resource_type":"subscription",
+ }
+
+ def test_incorrect_signature_raises(self):
+ self.params["signature"] = "asignature"
+ with self.assertRaises(SignatureError):
+ self.client.confirm_resource(self.params)
+
+ def test_resource_posts(self):
+ self.params["signature"] = utils.generate_signature(self.params,
+ mock_account_details["app_secret"])
+ with patch.object(self.client, 'api_post') as mock_post:
+ expected_data = {
+ "resource_type":self.params["resource_type"],
+ "resource_id":self.params["resource_id"]
+ }
+ expected_auth = base64.b64encode("{0}:{1}".format(
+ mock_account_details["app_id"],
+ mock_account_details["app_secret"]))
+ self.client.confirm_resource(self.params)
+ mock_post.assert_called_with(self.params["resource_uri"],
+ expected_data, auth=expected_auth)
+
+
+class UrlBuilderTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.app_secret = "12345"
+ self.merchant_id = "123"
+ self.app_id = "234234"
+ mock_client = mock.Mock()
+ mock_client.merchant_id = self.merchant_id
+ mock_client._app_secret = self.app_secret
+ mock_client._app_id = self.app_id
+ mock_client.get_base_url.return_value = "https://gocardless.com"
+ self.urlbuilder = urlbuilder.UrlBuilder(mock_client)
+
+ def make_mock_params(self, paramdict):
+ mock_params = mock.Mock()
+ if not "resource_name" in paramdict:
+ mock_params.resource_name = "aresource"
+ else:
+ mock_params.resource_name = paramdict.pop("resource_name")
+ mock_params.to_dict.return_value = paramdict
+ return mock_params
+
+ def get_url_params(self, url):
+ param_dict = urlparse.parse_qs(urlparse.urlparse(url).query)
+ return dict([[k,v[0]] for k,v in param_dict.items()])
+
+ def test_urlbuilder_url_contains_correct_parameters(self):
+ params = self.make_mock_params({"resource_name": "bill",
+ "amount":20.0,
+ "merchant_id":"merchid"})
+ url = self.urlbuilder.build_and_sign(params)
+ urlparams = self.get_url_params(url)
+ for k,v in params.to_dict().items():
+ if k == "resource_name":
+ continue
+ self.assertEqual(urlparams["bill[{0}]".format(k)], str(v))
+
+ def test_resource_name_is_singularized_in_url(self):
+ params = self.make_mock_params({"resource_name":"bills", \
+ "amount":20.0})
+ url = self.urlbuilder.build_and_sign(params)
+ urlparams = self.get_url_params(url)
+ self.assertTrue(urlparams.has_key("bill[amount]"))
+
+
+ def test_add_merchant_id_to_limit(self):
+ params = self.make_mock_params({"resource_name": "bill",
+ "merchant_id":self.merchant_id})
+ url = self.urlbuilder.build_and_sign(params)
+ urlparams = self.get_url_params(url)
+ self.assertEqual(urlparams["bill[merchant_id]"], self.merchant_id)
+
+ def test_url_contains_state(self):
+ params = self.make_mock_params({})
+ url = self.urlbuilder.build_and_sign(params, state="somestate")
+ urlparams = self.get_url_params(url)
+ self.assertEqual(urlparams["state"], "somestate")
+
+ def test_url_contains_redirect(self):
+ params = self.make_mock_params({})
+ url = self.urlbuilder.build_and_sign(params, redirect_uri="http://somesuchplace.com")
+ urlparams = self.get_url_params(url)
+ self.assertEqual(urlparams["redirect_uri"], "http://somesuchplace.com")
+
+ def test_url_contains_cancel(self):
+ params = self.make_mock_params({})
+ url = self.urlbuilder.build_and_sign(params,
+ cancel_uri="http://cancel")
+ urlparams = self.get_url_params(url)
+ self.assertEqual(urlparams["cancel_uri"], "http://cancel")
+
+ def test_url_contains_nonce(self):
+ params = self.make_mock_params({"somekey":"someval"})
+ url = self.urlbuilder.build_and_sign(params)
+ urlparams = self.get_url_params(url)
+ self.assertIsNotNone(urlparams["nonce"])
+
+ def test_url_nonce_is_random(self):
+ params = self.make_mock_params({"somekey":"somval"})
+ url1 = self.urlbuilder.build_and_sign(params)
+ url2 = self.urlbuilder.build_and_sign(params)
+ self.assertNotEqual(self.get_url_params(url1)["nonce"],\
+ self.get_url_params(url2)["nonce"])
+
+ def test_url_contains_client_id(self):
+ params = self.make_mock_params({"somekey":"someval"})
+ url = self.urlbuilder.build_and_sign(params)
+ urlparams = self.get_url_params(url)
+ self.assertEqual(urlparams["client_id"], self.app_id)
+
+ def test_url_contains_resource_name(self):
+ params = self.make_mock_params({"resource_name" : "pre_authorizations"})
+ url = self.urlbuilder.build_and_sign(params)
+ path = urlparse.urlparse(url).path
+ self.assertEqual(path, "/connect/pre_authorizations/new")
+
+ def test_url_contains_timestamp(self):
+ testdate = datetime.datetime.strptime("2010-01-01:0800", "%Y-%m-%d:%H%M")
+ with patch('datetime.datetime'):
+ datetime.datetime.now.return_value = testdate
+ params = self.make_mock_params({"somekey":"somval"})
+ url = self.urlbuilder.build_and_sign(params)
+ urlparams = self.get_url_params(url)
+ self.assertEqual(urlparams["timestamp"], testdate.isoformat()[:-7] + "Z")
+
+
+
+class Matcher(object):
+ """Object for comparing objects with an arbitrary comparison function
+
+ This is used as a matcher for testing properties of arguments in mocks
+ see http://www.voidspace.org.uk/python/mock/examples.html#matching-any-argument-in-assertions
+ """
+ def __init__(self, func):
+ self.func = func
+ def __eq__(self, other):
+ return self.func(other)
+
+class ClientUrlBuilderTestCase(unittest.TestCase):
+ """Integration test for the Client <-> UrlBuilder relationship
+
+ Tests that the url building methods on the client correctly
+ call methods on the urlbuilder class
+ """
+
+ def urlbuilder_argument_check(self, method, expected_type, *args):
+ mock_inst = mock.Mock(urlbuilder.UrlBuilder)
+ with patch('gocardless.urlbuilder.UrlBuilder') as mock_builder:
+ mock_inst.build_and_sign.return_value = "http://someurl"
+ mock_builder.return_value = mock_inst
+ c = Client(**mock_account_details)
+ getattr(c, method)(*args)
+ matcher = Matcher(lambda x: type(x) == expected_type)
+ mock_inst.build_and_sign.assert_called_with(matcher)
+
+ def test_new_preauth_calls_urlbuilder(self):
+ self.urlbuilder_argument_check("new_preauthorization_url",
+ urlbuilder.PreAuthorizationParams,
+ 3, 7, "day")
+
+ def test_new_bill_calls_urlbuilder(self):
+ self.urlbuilder_argument_check("new_bill_url",
+ urlbuilder.BillParams,
+ 4)
+
+ def test_new_subscription_calls_urlbuilder(self):
+ self.urlbuilder_argument_check("new_subscription_url",
+ urlbuilder.SubscriptionParams,
+ 10, 10, "day")
+
+
+
+
+
+
+
View
200 test/test_params.py
@@ -0,0 +1,200 @@
+import datetime
+import mock
+import unittest
+import urlparse
+
+import gocardless
+from gocardless import utils, urlbuilder
+
+class ExpiringLimitTestCase(object):
+ """superclass factoring out tests for expiring limit param objects"""
+
+ def test_interval_length_is_positive(self):
+ pars = self.create_params(10, "1321230", 1, "day")
+ with self.assertRaises(ValueError):
+ pars = self.create_params(10, "1123210", -1, "day")
+
+ def test_interval_unit_is_valid(self):
+ pars = self.create_params(10, 10, "11235432", "day")
+ with self.assertRaises(ValueError):
+ pars = self.create_params(10, 10, "1432233123", "invalid")
+
+ def future_date_tester(self, argname):
+ invalid_date = datetime.datetime.now() - datetime.timedelta(100)
+ valid_date = datetime.datetime.now() + datetime.timedelta(2000)
+ par1 = self.create_params(10, 10, "23423421", "day", **{argname:valid_date})
+ with self.assertRaises(ValueError):
+ par1 = self.create_params(10, 10, "2342341", "day",
+ **{argname:invalid_date})
+
+ def test_expires_at_in_future(self):
+ self.future_date_tester("expires_at")
+
+ def test_interval_count_positive(self):
+ with self.assertRaises(ValueError):
+ self.create_params(10, 10, "merchid", "day", interval_count=-1)
+
+class PreAuthParamsTestCase(ExpiringLimitTestCase, unittest.TestCase):
+
+ def default_args_construct(self, extra_options):
+ """
+ For testing optional arguments, builds the param object with valid
+ required arguments and adds optionl arguments as keywords from
+ `extra_options`
+
+ :param extra_options: Extra optional keyword arguments to pass to
+ the constructor.
+ """
+ return urlbuilder.\
+ PreAuthorizationParams(12, "3456", 6, "month", **extra_options)
+
+ def create_params(self, *args, **kwargs):
+ return urlbuilder.PreAuthorizationParams(*args, **kwargs)
+
+ def test_max_amount_is_positive(self):
+ self.assertRaises(ValueError, \
+ urlbuilder.PreAuthorizationParams, -1, "1232532", 4, "month")
+
+ def test_interval_length_is_a_positive_integer(self):
+ self.assertRaises(ValueError, \
+ urlbuilder.PreAuthorizationParams, 12, "!2343", -3, "month")
+
+ def test_interval_unit_is_one_of_accepted(self):
+ for unit_type in ["month", "day"]:
+ pa = urlbuilder.PreAuthorizationParams(12, "1234", 3, unit_type)
+ self.assertRaises(ValueError, \
+ urlbuilder.PreAuthorizationParams, 21,"1234", 4, "soem other unit")
+
+ def test_expires_at_is_later_than_now(self):
+ earlier = datetime.datetime.now() - datetime.timedelta(1)
+ self.assertRaises(ValueError, self.default_args_construct, \
+ {"expires_at":earlier})
+
+ def test_interval_count_is_postive_integer(self):
+ self.assertRaises(ValueError, self.default_args_construct, \
+ {"interval_count":-1})
+
+
+class PreAuthParamsToDictTestCase(unittest.TestCase):
+ def setUp(self):
+ self.all_params = {
+ "max_amount":12,
+ "interval_unit":"day",
+ "interval_length":10,
+ "merchant_id":"1234435",
+ "name":"aname",
+ "description":"adesc",
+ "interval_count":123,
+ "expires_at":datetime.datetime.strptime("2020-01-01", "%Y-%m-%d"),
+ "calendar_intervals":True
+ }
+ self.required_keys = [
+ "max_amount", "interval_unit", "interval_length", "merchant_id"]
+
+ def create_from_params_dict(self, in_params):
+ params = in_params.copy()
+ pa = urlbuilder.PreAuthorizationParams(params.pop("max_amount"), \
+ params.pop("merchant_id"), \
+ params.pop("interval_length"), \
+ params.pop("interval_unit"),\
+ **params)
+ return pa
+
+ def assert_inverse(self, keys):
+ params = dict([[k,v] for k,v in self.all_params.items() \
+ if k in keys])
+ pa = self.create_from_params_dict(params)
+ self.assertEqual(params, pa.to_dict())
+
+ def test_to_dict_all_params(self):
+ self.assert_inverse(self.all_params.keys())
+
+ def test_to_dict_only_required(self):
+ self.assert_inverse(self.required_keys)
+
+
+class BillParamsTestCase(unittest.TestCase):
+
+ def create_params(self, *args, **kwargs):
+ return urlbuilder.BillParams(*args, **kwargs)
+
+ def test_amount_is_positive(self):
+ params = self.create_params(10, "merchid")
+ with self.assertRaises(ValueError):
+ par2 = self.create_params(-1, "merchid")
+
+ def test_to_dict_required(self):
+ pars = self.create_params(10, "merchid")
+ res = pars.to_dict()
+ expected = {"amount":10, "merchant_id":"merchid"}
+ self.assertEqual(res, expected)
+
+ def test_to_dict_optional(self):
+ pars = self.create_params(10, "merchid", name="aname", description="adesc")
+ res = pars.to_dict()
+ expected = {"amount":10,
+ "name":"aname",
+ "description":"adesc",
+ "merchant_id":"merchid"
+ }
+ self.assertEqual(res, expected)
+
+ def test_resource_name_is_bills(self):
+ pars = urlbuilder.BillParams(10, "merchid")
+ self.assertEqual(pars.resource_name, "bills")
+
+
+class SubscriptionParamsTestCase(ExpiringLimitTestCase, unittest.TestCase):
+
+ def create_params(self, *args, **kwargs):
+ return urlbuilder.SubscriptionParams(*args, **kwargs)
+
+ def test_start_at_in_future(self):
+ valid_date = datetime.datetime.now() + datetime.timedelta(200)
+ invalid_date = datetime.datetime.now() - datetime.timedelta(100)
+ par1 = self.create_params(10,"merchid", 10, "day", start_at=valid_date)
+ with self.assertRaises(ValueError):
+ par2 = self.create_params(10, "merchid", 10, "day",
+ start_at=invalid_date)
+
+ def test_expires_at_after_start_at(self):
+ date1 = datetime.datetime.now() + datetime.timedelta(100)
+ date2 = datetime.datetime.now() + datetime.timedelta(200)
+ par1 = self.create_params(10, "merchid", 10, "day",
+ expires_at=date2, start_at=date1)
+ with self.assertRaises(ValueError):
+ par2 = self.create_params(10, "merchid", 10, "day",
+ expires_at=date1, start_at=date2)
+
+ def test_to_dict_only_required(self):
+ expected = {
+ "merchant_id":"merchid",
+ "amount":10,
+ "interval_length":10,
+ "interval_unit":"day"}
+ pars = self.create_params(10, "merchid", 10, "day")
+ self.assertEqual(expected, pars.to_dict())
+
+ def test_to_dict_all(self):
+ start_at = datetime.datetime.now() + datetime.timedelta(1000)
+ expires_at =datetime.datetime.now() + datetime.timedelta(2000)
+ expected = {
+ "merchant_id":"merchid",
+ "amount":10,
+ "interval_length":10,
+ "interval_unit":"day",
+ "interval_count":5,
+ "start_at":start_at.isoformat()[:-7] + "Z",
+ "expires_at":expires_at.isoformat()[:-7] + "Z",
+ "name":"aname",
+ "description":"adesc",
+ }
+ par = self.create_params(10, "merchid", 10, "day", start_at=start_at,
+ expires_at=expires_at, interval_count=5, name="aname",
+ description="adesc")
+ self.assertEqual(expected, par.to_dict())
+
+
+
+
+
View
62 test/test_request.py
@@ -0,0 +1,62 @@
+import unittest
+import mock
+
+#from gocardless import request
+import gocardless.request
+
+
+class RequestTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.request = gocardless.request.Request('get', 'http://test.com')
+
+ def test_valid_method_allows_valid_methods(self):
+ for method in ('get', 'post', 'put'):
+ self.assertTrue(self.request._valid_method('get'))
+
+ def test_valid_method_disallows_invalid_methods(self):
+ self.assertFalse(self.request._valid_method('fake_method'))
+
+ def test_use_bearer_auth_sets_auth_header(self):
+ self.request.use_bearer_auth('token')
+ self.assertEqual(self.request._opts['headers']['Authorization'],
+ 'bearer token')
+
+ def test_use_http_auth_sets_auth_details_in_opts(self):
+ self.request.use_http_auth('user', 'pass')
+ self.assertEqual(self.request._opts['auth'], ('user', 'pass'))
+
+ def test_set_payload_ignores_null_payloads(self):
+ self.request.set_payload(None)
+ self.assertTrue('Content-Type' not in self.request._opts['headers'])
+ self.assertTrue('data' not in self.request._opts)
+
+ def test_set_payload_sets_content_type(self):
+ self.request.set_payload({'a': 'b'})
+ self.assertEqual(self.request._opts['headers']['Content-Type'],
+ 'application/json')
+
+ def test_set_payload_encodes_payload(self):
+ self.request.set_payload({'a': 'b'})
+ self.assertEqual(self.request._opts['data'], '{"a": "b"}')
+
+ @mock.patch('gocardless.request.requests')
+ def test_perform_calls_get_for_gets(self, mock_requests):
+ mock_requests.get.return_value.content = '{"a": "b"}'
+ self.request.perform()
+ mock_requests.get.assert_called_once_with(mock.ANY, headers=mock.ANY)
+
+ @mock.patch('gocardless.request.requests')
+ def test_perform_calls_post_for_posts(self, mock_requests):
+ mock_requests.post.return_value.content = '{"a": "b"}'
+ self.request._method = 'post'
+ self.request.perform()
+ mock_requests.post.assert_called_once_with(mock.ANY, headers=mock.ANY)
+
+ @mock.patch('gocardless.request.requests.get')
+ def test_perform_decodes_json(self, mock_get):
+ response = mock.Mock()
+ response.content = '{"a": "b"}'
+ mock_get.return_value = response
+ self.assertEqual(self.request.perform(), {'a': 'b'})
+
View
116 test/test_resources.py
@@ -0,0 +1,116 @@
+import copy
+import datetime
+import json
+import mock
+from mock import patch
+import unittest
+
+import gocardless
+from gocardless.resources import Resource
+
+class TestResource(Resource):
+ endpoint = "/testendpoint/:id"
+
+ def __init__(self, attrs, client):
+ attrs = create_mock_attrs(attrs)
+ Resource.__init__(self, attrs, client)
+
+class TestSubResource(Resource):
+ endpoint = "/subresource/:id"
+
+class OtherTestSubResource(Resource):
+ endpoint = "/subresource2/:id"
+
+def create_mock_attrs(to_merge):
+ """
+ Creats an attribute set for creating a resource from,
+ includes the basic created, modified and id keys. Merges
+ that with to_merge
+ """
+ attrs = {
+ "created_at":"2012-04-18T17:53:12Z",
+ "id":"1"}
+ attrs.update(to_merge)
+ return attrs
+
+class ResourceTestCase(unittest.TestCase):
+
+ def test_endpoint_declared_by_class(self):
+ resource = TestResource({"id":"1"}, None)
+ self.assertEqual(resource.get_endpoint(), "/testendpoint/1")
+
+ def test_resource_attributes(self):
+ attrs = {"key1":"one",
+ "key2":"two",
+ "key3":"three",
+ "id":"1"}
+ res = TestResource(attrs.copy(), None)
+ for key, value in attrs.items():
+ self.assertEqual(getattr(res, key), value)
+
+ def test_resource_created_at_modified_at_are_dates(self):
+ created = datetime.datetime.strptime('2012-04-18T17:53:12Z',\
+ "%Y-%m-%dT%H:%M:%SZ")
+ attrs = create_mock_attrs({"created_at":'2012-04-18T17:53:12Z',
+ "id":"1"})
+ res = TestResource(attrs, None)
+ self.assertEqual(res.created_at, created)
+
+class ResourceSubresourceTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.resource = TestResource({"sub_resource_uris":
+ {"test_sub_resources":
+ "https://gocardless.com/api/v1/merchants/WOQRUJU9OH2HH1/bills?\
+ source_id=1580",
+ "other_test_sub_resources": "aurl"},
+ "id":"1"},
+ None)
+
+ def test_resource_lists_subresources(self):
+ self.assertTrue(hasattr(self.resource, "get_test_sub_resources"))
+ self.assertTrue(callable(getattr(self.resource, "get_test_sub_resources")))
+
+ def test_resource_subresource_returns_subresource_instances(self):
+ mock_return = map(create_mock_attrs, [{"id":1},{"id":2}])
+ mock_client = mock.Mock()
+ mock_client.api_get.return_value = mock_return
+ self.resource.client = mock_client
+ result = self.resource.get_test_sub_resources()
+ for res in result:
+ self.assertIsInstance(res, TestSubResource)
+ self.assertEqual(set([1,2]), set([item.id for item in result]))
+
+ def test_resource_is_correct_instance(self):
+ """
+ Expose an issue where the closure which creates sub_resource functions
+ in the `Resource` constructor does not close over the class name
+ correctly and thus every sub_resource function ends up referencing
+ the same class.
+ """
+ mock_client = mock.Mock()
+ mock_client.api_get.return_value = [create_mock_attrs({"id":"1"})]
+ self.resource.client = mock_client
+ result = self.resource.get_test_sub_resources()
+ self.assertIsInstance(result[0], TestSubResource)
+
+
+class FindResourceTestCase(unittest.TestCase):
+
+ def test_find_resource_by_id_with_client(self):
+ client = mock.Mock()
+ client.api_get.return_value = {"id":"1"}
+ resource = TestResource.find_with_client("1", client)
+ self.assertEqual(resource.id, "1")
+
+ def test_find_resource_without_details_throws_clienterror(self):
+ self.assertRaises(gocardless.exceptions.ClientError, TestResource.find, 1)
+
+ @patch('gocardless.client')
+ def test_find_resource_without_client(self, mock_client):
+ mock_client.api_get.return_value = {"id":"1"}
+ self.assertEqual(TestResource.find("1").id, "1")
+
+
+
+
View
68 test/test_utils.py
@@ -0,0 +1,68 @@
+# coding: utf-8
+
+import unittest
+
+
+from gocardless import utils
+
+
+class PercentEncodeTestCase(unittest.TestCase):
+
+ def test_works_with_empty_strings(self):
+ self.assertEqual(utils.percent_encode(u""), u"")
+
+ def test_doesnt_encode_lowercase_alpha_characters(self):
+ self.assertEqual(utils.percent_encode(u"abcxyz"), u"abcxyz")
+
+ def test_doesnt_encode_uppercase_alpha_characters(self):
+ self.assertEqual(utils.percent_encode(u"ABCXYZ"), u"ABCXYZ")
+
+ def test_doesnt_encode_digits(self):
+ self.assertEqual(utils.percent_encode(u"1234567890"), u"1234567890")
+
+ def test_doesnt_encode_unreserved_non_alphanum_chars(self):
+ self.assertEqual(utils.percent_encode(u"-._~"), u"-._~")
+
+ def test_encodes_non_ascii_alpha_characters(self):
+ self.assertEqual(utils.percent_encode(u"å"), u"%C3%A5")
+
+ def test_encodes_reserved_ascii_characters(self):
+ self.assertEqual(utils.percent_encode(u" !\"#$%&'()"),
+ u"%20%21%22%23%24%25%26%27%28%29")
+ self.assertEqual(utils.percent_encode(u"*+,/{|}:;"),
+ u"%2A%2B%2C%2F%7B%7C%7D%3A%3B")
+ self.assertEqual(utils.percent_encode(u"<=>?@[\\]^`"),
+ u"%3C%3D%3E%3F%40%5B%5C%5D%5E%60")
+
+ def test_encodes_other_non_ascii_characters(self):
+ self.assertEqual(utils.percent_encode(u"支払い"),
+ u"%E6%94%AF%E6%89%95%E3%81%84")
+
+
+class SignatureTestCase(unittest.TestCase):
+ def setUp(self):
+ self.secret = '5PUZmVMmukNwiHc7V/TJvFHRQZWZumIpCnfZKrVYGpuAdkCcEfv3LIDSrsJ+xOVH'
+ self.api_key = ''
+ self.client_id = '4jqkF9tirkr3zfWCgEKxLDy3UmF1sWpHPVm8X69yiB7Lqb63usVOPzrm0jEepc5R'
+
+ def test_hmac(self):
+ # make sure our signature function
+ # works correctly
+ sig = utils.generate_signature({"foo": "bar", "example": [1, "a"]},self.secret)
+ self.assertEqual(sig, '5a9447aef2ebd0e12d80d80c836858c6f9c13219f615ef5d135da408bcad453d')
+
+
+class CamelizeTestCase(unittest.TestCase):
+ def test_camelize_multi_word(self):
+ teststr = "camelize_this_please"
+ expected = "CamelizeThisPlease"
+ self.assertEqual(expected, utils.camelize(teststr))
+
+
+class SingularizeTestCase(unittest.TestCase):
+ def test_singularize(self):
+ to_singularize = "PreAuthorisations"
+ expected = "PreAuthorisation"
+ self.assertEqual(expected, utils.singularize(to_singularize))
+
+
Please sign in to comment.
Something went wrong with that request. Please try again.