Commit 6c50713a authored by Jérome Perrin's avatar Jérome Perrin Committed by Gabriel Monnerat

stripe: support creating checkout session with all dicts and lists serialized

```py
kw = {...}
kw['metadata[key]'] = 'value'
connector.createSession(**kw)
```

but it would be more beautiful to be able to write it like this:

```py
kw = {...}
kw['metadata] = {'key': 'value'}
connector.createSession(**kw)
```

This would need change in the way the request data is serialised in
createSession.
parent f269e6ca
Pipeline #25079 passed with stage
in 0 seconds
......@@ -26,10 +26,7 @@
#
##############################################################################
from six.moves import urllib
from copy import deepcopy
import requests
import six
......@@ -60,23 +57,16 @@ class StripeConnector(XMLObject):
PropertySheet.DublinCore
)
def buildLineKey(self, prefix, key_list):
for key in key_list:
prefix += "[%s]" % key
return prefix
def buildLine(self, data_dict, prefix, key_list, value):
if not isinstance(value, dict):
data_dict[self.buildLineKey(prefix, key_list)] = value
def serializeSessionParameter(self, request_data, key, value):
if isinstance(value, list) or isinstance(value, dict):
iterator = six.iteritems(value) if isinstance(value, dict) else enumerate(value)
for subkey, subvalue in iterator:
self.serializeSessionParameter(
request_data,
"{}[{}]".format(key, subkey),
subvalue)
else:
for subkey, subvalue in six.iteritems(value):
self.buildLine(data_dict, prefix, key_list + [subkey,], subvalue)
def buildLineItemList(self, prefix, line_item_list):
data_dict = {}
for key, value in enumerate(line_item_list):
self.buildLine(data_dict, prefix, [key,], value)
return data_dict
request_data[key] = value
def createSession(self, data, **kw):
"""
......@@ -85,12 +75,6 @@ class StripeConnector(XMLObject):
data is a dict, see https://stripe.com/docs/api/checkout/sessions/create
"""
end_point = "checkout/sessions"
# copy data, not to mutate caller's data
request_data = deepcopy(data)
if "mode" not in request_data:
request_data["mode"] = "payment"
header_dict = {
'Content-Type': 'application/x-www-form-urlencoded',
}
......@@ -98,8 +82,13 @@ class StripeConnector(XMLObject):
if not url_string.endswith("/"):
url_string += "/"
url_string += end_point
line_item_list = request_data.pop("line_items")
request_data.update(self.buildLineItemList("line_items", line_item_list))
request_data = {}
for key, value in six.iteritems(data):
self.serializeSessionParameter(request_data, key, value)
if "mode" not in request_data:
request_data["mode"] = "payment"
response = requests.post(
url_string,
headers=header_dict,
......
......@@ -45,19 +45,20 @@ class TestStripePaymentSession(ERP5TypeTestCase):
self.connector_reference = "abc"
self.session_url = "https://mock:8080/checkout/sessions"
self.default_item_line = {
"price_data": {
"currency": "eur",
"unit_amount": "100",
"product_data": {
"name": "First Line",
}
},
"quantity": 1
}
self.data = {
"success_url": "http://success",
"cancel_url": "http://cancel",
"line_items": [{
"price_data": {
"currency": "eur",
"unit_amount": "100",
"product_data": {
"name": "First Line",
}
},
"quantity": 1
}, {
"line_items": [self.default_item_line, {
"price_data": {
"currency": "eur",
"unit_amount": "200",
......@@ -220,6 +221,157 @@ class TestStripePaymentSession(ERP5TypeTestCase):
response = connector.createSession(data=self.data.copy())
self.assertEqual(response["id"], "123")
def test_api_create_session_with_metadata(self):
connector = self._create_connector()
def _request_callback(request):
self.assertEqual(
'application/x-www-form-urlencoded', request.headers['Content-Type'],
request.headers)
body = parse_qs(request.body)
self.assertEqual(
body, {
"success_url": ["http://success"],
"cancel_url": ["http://cancel"],
"line_items[0][price_data][currency]": ["eur"],
"line_items[0][price_data][unit_amount]": ["100"],
"line_items[0][price_data][product_data][name]": ["First Line"],
"line_items[0][quantity]": ["1"],
"metadata[key]": ["value"],
"mode": ["payment"],
})
return (
200, {
'content-type': 'application/json'
},
json.dumps(
{
"id": "123",
"status": "open",
"object": "checkout.session"
}))
with responses.RequestsMock() as rsps:
rsps.add_callback(
responses.POST,
self.session_url,
_request_callback,
)
response = connector.createSession(
data={
"success_url": "http://success",
"cancel_url": "http://cancel",
"line_items": [
self.default_item_line
],
"metadata": {
"key": "value",
},
"mode": "payment"
})
self.assertEqual(response["id"], "123")
def test_api_create_session_with_any_hash(self):
""" Test if dicts and lists are serialized properly
"""
connector = self._create_connector()
def _request_callback(request):
self.assertEqual(
'application/x-www-form-urlencoded', request.headers['Content-Type'],
request.headers)
body = parse_qs(request.body)
self.assertEqual(
body, {
"automatic_tax[enabled]": ['true'],
"automatic_tax[status]": ['complete'],
"success_url": ["http://success"],
"cancel_url": ["http://cancel"],
'custom_text[shipping_text][message]': ['Rue XV'],
'custom_text[submit][message]': ['Rue XV'],
'customer_details[address]': ['Rue XV'],
'customer_details[email]': ['text@text.com'],
'customer_details[name]': ['My name'],
'customer_details[phone]': ['2199909'],
'customer_details[tax_exempt]': ['tax'],
'customer_details[tax_ids][0][type]': ['eu_vat'],
'customer_details[tax_ids][0][value]': ['33'],
'customer_details[tax_ids][1][type]': ['br_cnpj'],
'customer_details[tax_ids][1][value]': ['33'],
"line_items[0][price_data][currency]": ["eur"],
"line_items[0][price_data][unit_amount]": ["100"],
"line_items[0][price_data][product_data][name]": ["First Line"],
"line_items[0][quantity]": ["1"],
"metadata[key]": ["value"],
"metadata[key1]": ["value1"],
"metadata[key2]": ["value2"],
"mode": ["payment"],
"payment_method_options[acss_debit][current]": ['usd']
})
return (
200, {
'content-type': 'application/json'
},
json.dumps(
{
"id": "123",
"status": "open",
"object": "checkout.session"
}))
with responses.RequestsMock() as rsps:
rsps.add_callback(
responses.POST,
self.session_url,
_request_callback,
)
response = connector.createSession(
data={
"success_url": "http://success",
"cancel_url": "http://cancel",
"line_items": [
self.default_item_line
],
"metadata": {
"key": "value",
"key1": "value1",
"key2": "value2",
},
"automatic_tax": {
"enabled": "true",
"status": "complete"
},
"custom_text": {
"shipping_text": {
"message": "Rue XV",
},
"submit": {
"message": "Rue XV",
}
},
"customer_details": {
"address": "Rue XV",
"email": "text@text.com",
"name": "My name",
"phone": "2199909",
"tax_exempt": "tax",
"tax_ids": [{
"type": "eu_vat",
"value": "33"
}, {
"type": "br_cnpj",
"value": "33"
}]
},
"payment_method_options": {
"acss_debit": {
"current": "usd"
}
},
"mode": "payment"
})
self.assertEqual(response["id"], "123")
def test_create_stripe_payment_session_open(self):
connector = self._create_connector()
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment