Functional one-time payments and tests

.gitignore

@@ -81,6 +81,7 @@ target/
 
 # Jupyter Notebook
 .ipynb_checkpoints
+*.ipynb
 
 # IPython
 profile_default/
@@ -89,7 +90,7 @@ ipython_config.py
 # pyenv
 #   For a library or package, you might want to ignore these files since the code is
 #   intended to run in multiple environments; otherwise, check them in:
-# .python-version
+.python-version
 
 # pipenv
 #   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.

pyproject.toml

@@ -12,8 +12,11 @@ requires-python = ">=3.10"
 dependencies = [
     "httpx (>=0.28.1,<0.29.0)",
     "pydantic (>=2.10.6,<3.0.0)",
+    "pydantic-settings (>=2.10.6,<3.0.0)",
     "cryptography (>=45.0.4,<46.0.0)",
-    "http-message-signatures (>=0.6.1,<0.7.0)"
+    "http-message-signatures (>=0.6.1,<0.7.0)",
+    "http-sf (>=1.0.4,<1.1.0)",
+    "python-ulid[pydantic] (>=3.1.0,<4.0.0)"
 ]
 
 [build-system]
@@ -28,4 +31,6 @@ pyflakes = "^3.2.0"
 datamodel-code-generator = "^0.28.1"
 pytest = "^8.3.4"
 pyyaml = "^6.0.2"
+jupyterlab = "^4.5.0"
+selenium = "^4.38.0"
 

src/open_payments_sdk/api/auth.py

@@ -1,147 +1,105 @@
 """
 Grants Module
 """
+
 from logging import Logger
 
 from open_payments_sdk.gnap_utils.security import SecurityBase
 from open_payments_sdk.http import HttpClient
 from open_payments_sdk.models.auth import AccessToken, Grant
-from open_payments_sdk.models.auth import (GrantContinueResponse, GrantRequest,
-                                           InteractRef)
+from open_payments_sdk.models.auth import GrantContinueResponse, GrantRequest, InteractRef
 from open_payments_sdk.utils.utils import get_default_covered_components, get_default_headers
 
 
-
 class Grants(SecurityBase):
     """
     Class to handle Grants in the sdk
     """
-    def __init__(self, keyid: str, private_key: str ,logger: Logger,http_client: HttpClient):
-        super().__init__(keyid=keyid, private_key=private_key,logger=logger)
+
+    def __init__(self, keyid: str, private_key: str, logger: Logger, http_client: HttpClient):
+        super().__init__(keyid=keyid, private_key=private_key, logger=logger)
         self.logger = logger
         self.http_client = http_client
 
     def post_grant_request(
-            self,
-            grant_request: GrantRequest,
-            auth_server_endpoint: str,
-        ) -> Grant:
+        self,
+        grant_request: GrantRequest,
+        auth_server_endpoint: str,
+    ) -> Grant:
         """
         Grant Request
         """
         data = grant_request.model_dump(exclude_unset=True, mode="json")
 
-        req_headers = {
-            **get_default_headers()
-        }
+        req_headers = {**get_default_headers()}
         request = self.http_client.build_request(
-            method="POST",
-            url=auth_server_endpoint,
-            json=data,
-            headers=req_headers
+            method="POST", url=auth_server_endpoint, json=data, headers=req_headers
         )
         request = self.set_content_digest(request=request)
-        request = self.sign_request(request,("content-type","content-digest","content-length",*get_default_covered_components()))
+        request = self.sign_request(
+            request, ("content-type", "content-digest", "content-length", *get_default_covered_components())
+        )
         response = self.http_client.send(request=request)
         return Grant.model_validate(response.json())
 
     def post_grant_continuation_request(
-            self,
-            interact_ref: InteractRef,
-            continue_uri: str,
-            access_token: str
-        ) -> GrantContinueResponse:
+        self, interact_ref: InteractRef, continue_uri: str, access_token: str
+    ) -> GrantContinueResponse:
         """
         Continue Grant Request
         """
         data = interact_ref.model_dump(exclude_unset=True, mode="json")
-        req_headers = {
-            **get_default_headers(),
-            **self.get_auth_header(access_token=access_token)
-        }
-        request = self.http_client.build_request(
-            method="POST",
-            url=continue_uri,
-            json=data,
-            headers=req_headers
-        )
+        req_headers = {**get_default_headers(), **self.get_auth_header(access_token=access_token)}
+        request = self.http_client.build_request(method="POST", url=continue_uri, json=data, headers=req_headers)
         request = self.set_content_digest(request=request)
-        request = self.sign_request(request,("content-type","content-digest","content-length","authorization",*get_default_covered_components()))
+        request = self.sign_request(
+            request,
+            ("content-type", "content-digest", "content-length", "authorization", *get_default_covered_components()),
+        )
         response = self.http_client.send(request=request)
         return GrantContinueResponse.model_validate(response.json())
 
-    def delete_grant(
-            self,
-            req_id: str,
-            auth_server_endpoint: str,
-            access_token: str
-        ) -> None:
+    def delete_grant(self, req_id: str, auth_server_endpoint: str, access_token: str) -> None:
         """
         Delete Grant
         """
         base_url = auth_server_endpoint.rstrip("/")
         url = f"{base_url}/continue/{req_id}"
-        req_headers = {
-            **self.get_auth_header(access_token=access_token)
-        }
-        request = self.http_client.build_request(
-            method="DELETE",
-            url=url,
-            headers=req_headers
-        )
-        request = self.sign_request(request,("authorization",*get_default_covered_components()))
+        req_headers = {**self.get_auth_header(access_token=access_token)}
+        request = self.http_client.build_request(method="DELETE", url=url, headers=req_headers)
+        request = self.sign_request(request, ("authorization", *get_default_covered_components()))
         self.http_client.send(request=request)
 
+
 class AccessTokens(SecurityBase):
     """
     Access Token Class
     """
-    def __init__(self, keyid: str , private_key: str,logger: Logger, http_client: HttpClient ):
+
+    def __init__(self, keyid: str, private_key: str, logger: Logger, http_client: HttpClient):
         super().__init__(keyid=keyid, private_key=private_key, logger=logger)
         self.http_client = http_client
 
-    def post_rotate_access_token(
-            self,
-            token_id: str,
-            auth_server_endpoint: str,
-            access_token: str
-        ) -> AccessToken:
+    def post_rotate_access_token(self, token_id: str, auth_server_endpoint: str, access_token: str) -> AccessToken:
         """
         Rotate Access Token
         """
         base_url = auth_server_endpoint.rstrip("/")
         url = f"{base_url}/token/{token_id}"
-        req_headers = {
-            **self.get_auth_header(access_token=access_token)
-        }
-        request = self.http_client.build_request(
-            method="POST",
-            url=url,
-            headers=req_headers
-        )
-        request = self.sign_request(request,("authorization",*get_default_covered_components()))
+        req_headers = {**self.get_auth_header(access_token=access_token)}
+        request = self.http_client.build_request(method="POST", url=url, headers=req_headers)
+        request = self.sign_request(request, ("authorization", *get_default_covered_components()))
         response = self.http_client.send(request=request)
         return AccessToken.model_validate(response.json())
 
-    def delete_access_token(
-            self,
-            token_id: str,
-            auth_server_endpoint: str,
-            access_token: str
-        ) -> None:
+    def delete_access_token(self, token_id: str, auth_server_endpoint: str, access_token: str) -> None:
         """
         Delete Access Token
         """
         base_url = auth_server_endpoint.rstrip("/")
         url = f"{base_url}/token/{token_id}"
-        req_headers = {
-            **self.get_auth_header(access_token=access_token)
-        }
+        req_headers = {**self.get_auth_header(access_token=access_token)}
 
-        request = self.http_client.build_request(
-            method="DELETE",
-            url=url,
-            headers=req_headers
-        )
-        request = self.sign_request(request,("authorization",*get_default_covered_components()))
-        self.http_client.send(request=request)
+        request = self.http_client.build_request(method="DELETE", url=url, headers=req_headers)
+        request = self.sign_request(request, ("authorization", *get_default_covered_components()))
+        self.http_client.send(request=request)