Skip to content

PaypalProvider: allow to make also get requests #439

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 32 additions & 18 deletions payments/paypal/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,20 +114,21 @@ def _get_links(self, payment):
return extra_data.get("links", {})

@authorize
def post(self, payment, *args, **kwargs):
def http_request(self, payment, *args, http_method=None, **kwargs):
kwargs["headers"] = {
"Content-Type": "application/json",
"Authorization": self.access_token,
}
if "data" in kwargs:
kwargs["data"] = json.dumps(kwargs["data"])
response = requests.post(*args, **kwargs)
response = http_method(*args, **kwargs)
try:
data = response.json()
except ValueError:
data = {}
if 400 <= response.status_code <= 500:
self.set_error_data(payment, data)
if payment is not None:
self.set_error_data(payment, data)
logger.debug(data)
message = "Paypal error"
if response.status_code == 400:
Expand All @@ -139,29 +140,41 @@ def post(self, payment, *args, **kwargs):
message = error_data.get("message", message)
else:
logger.warning(message, extra={"status_code": response.status_code})
payment.change_status(PaymentStatus.ERROR, message)
if payment is not None:
payment.change_status(PaymentStatus.ERROR, message)
raise PaymentError(message)
self.set_response_data(payment, data)
if payment is not None:
self.set_response_data(payment, data)
return data

def post(self, payment, *args, **kwargs):
"""Perform a POST request to the PayPal API."""
return self.http_request(payment, *args, http_method=requests.post, **kwargs)

def get(self, payment, *args, **kwargs):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please add a short docstring for this, so folks can find it and use it.

"""Perform a GET request to the PayPal API."""
return self.http_request(payment, *args, http_method=requests.get, **kwargs)

def get_last_response(self, payment, is_auth=False):
extra_data = json.loads(payment.extra_data or "{}")
if is_auth:
return extra_data.get("auth_response", {})
return extra_data.get("response", {})

def get_access_token(self, payment):
last_auth_response = self.get_last_response(payment, is_auth=True)
created = payment.created
now = timezone.now()
if (
"access_token" in last_auth_response
and "expires_in" in last_auth_response
and (created + timedelta(seconds=last_auth_response["expires_in"])) > now
):
return "{} {}".format(
last_auth_response["token_type"], last_auth_response["access_token"]
)
if payment is not None:
last_auth_response = self.get_last_response(payment, is_auth=True)
created = payment.created
now = timezone.now()
if (
"access_token" in last_auth_response
and "expires_in" in last_auth_response
and (created + timedelta(seconds=last_auth_response["expires_in"]))
> now
):
return "{} {}".format(
last_auth_response["token_type"], last_auth_response["access_token"]
)
headers = {"Accept": "application/json", "Accept-Language": "en_US"}
post = {"grant_type": "client_credentials"}
response = requests.post(
Expand All @@ -172,8 +185,9 @@ def get_access_token(self, payment):
)
response.raise_for_status()
data = response.json()
last_auth_response.update(data)
self.set_response_data(payment, last_auth_response, is_auth=True)
if payment is not None:
last_auth_response.update(data)
self.set_response_data(payment, last_auth_response, is_auth=True)
return "{} {}".format(data["token_type"], data["access_token"])

def get_transactions_items(self, payment):
Expand Down
42 changes: 42 additions & 0 deletions payments/paypal/test_paypal.py
Original file line number Diff line number Diff line change
Expand Up @@ -387,6 +387,48 @@ def test_provider_renews_access_token(self, mocked_post):
payment_response = json.loads(self.payment.extra_data)["auth_response"]
self.assertEqual(payment_response["access_token"], new_token)

def test_provider_get_with_none_payment(self):
test_url = "http://example.com/api/test"
expected_get_response_data = {"status": "success"}
expected_token = "test_access_token"
expected_token_type = "Bearer"

with patch("requests.post") as mocked_post, patch("requests.get") as mocked_get:
# Mock for token acquisition
token_response_mock = MagicMock()
token_response_mock.json.return_value = {
"access_token": expected_token,
"token_type": expected_token_type,
"expires_in": 3600,
}
token_response_mock.status_code = 200

# Mock for the actual GET request
get_response_mock = MagicMock()
get_response_mock.json.return_value = expected_get_response_data
get_response_mock.status_code = 200

mocked_post.return_value = token_response_mock
mocked_get.return_value = get_response_mock

response_data = self.provider.get(None, test_url)

mocked_post.assert_called_once_with(
self.provider.oauth2_url,
data={"grant_type": "client_credentials"},
headers={"Accept": "application/json", "Accept-Language": "en_US"},
auth=(self.provider.client_id, self.provider.secret),
)

mocked_get.assert_called_once_with(
test_url,
headers={
"Content-Type": "application/json",
"Authorization": f"{expected_token_type} {expected_token}",
},
)
self.assertEqual(response_data, expected_get_response_data)


class TestPaypalCardProvider(TestCase):
def setUp(self):
Expand Down
Loading