diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..860d102 --- /dev/null +++ b/.gitignore @@ -0,0 +1,164 @@ +# Byte-compiled / optimized / DLL files" +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +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 + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/latest/usage/project/#working-with-version-control +.pdm.toml +.pdm-python +.pdm-build/ + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +.idea/ + +development \ No newline at end of file diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..9a4c437 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.7.17 diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..9b38853 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,7 @@ +{ + "python.testing.pytestArgs": [ + "tests" + ], + "python.testing.unittestEnabled": false, + "python.testing.pytestEnabled": true +} \ No newline at end of file diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..3916f24 --- /dev/null +++ b/Makefile @@ -0,0 +1,40 @@ +# Makefile for building and uploading a Python package + +.PHONY: clean build upload docs new-version + +# Python binary to use +PYTHON := python3 + +# Target to run tests +test: + tox + +# Setup dev environment +setup-dev: + ./scripts/setup-dev.sh + +# Target to build the documentation (requires Sphinx) +docs: + sphinx-apidoc -o docs py_sat && cd docs && make html + +# Serve docs +serve-docs: + python -m http.server --directory docs/_build/html 32843 + +# Target to upload the package to PyPI (requires Twine) +upload: + twine upload dist/* + +# Target to clean up build artifacts +clean: + rm -rf build dist *.egg-info + +# Target to build the package (source distribution and wheel) +build: + $(PYTHON) setup.py sdist + + +# Target to upload the package to PyPI with new version +new-version: + make clean && make build && make docs && make upload + diff --git a/README.md b/README.md new file mode 100644 index 0000000..f75ad6c --- /dev/null +++ b/README.md @@ -0,0 +1,423 @@ +### Python SDK SAT +Python SDK for Tokopedia SAT endpoint. +This SDK can do end to end flow on SAT request, able to inquiry, order, get product list, and much more. Also able to automatically handle signature + +#### High Level Diagram +##### Inquiry Product +```mermaid +sequenceDiagram +Client ->> Server: Inquiry a client number (Inquiry) +Server -->> Client: Response the bills +Client ->> Server: Checkout an order (Checkout) +Server -->> Client: Response the an process order +Server -->> Client: Callback the final status order (Callback) + + loop check status every X times until final status +Client ->> Server: Check Status whenever receive an callback (CheckStatus) +Server -->> Client: Response the current order status + end +``` + +##### Non Inquiry Product +```mermaid +sequenceDiagram +Client ->> Server: Checkout an order (Checkout) +Server -->> Client: Response the an process order +Server -->> Client: Callback the final status order (Callback) + loop check status every X times until final status +Client ->> Server: Check Status whenever receive an callback (CheckStatus) +Server -->> Client: Response the current order status + end +``` + +#### Prerequisite +- Python >= 3.7 +- Tokopedia Account registered as Distributor SAT, please follow **DG B2B - Developer Guideline** +- Generate Pair Private Key & Public Key, you can follow **API Documentation Section 2.4 Digital Signature** +- Share your Public Key to [Tokopedia Dev Console](https://developer.tokopedia.com/console/apps) under your current apps. We recommend starting from testing apps. + + +#### Install Library +``` +pip install py-sat-sdk +``` + + +#### Init SDK +Below basic implementation to use the SDK. +```python +from py_sat import SATClientConfig, SATClient + +config = ( + SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE_KEY", # required + ) + # Below is optional parameter + .with_timeout(10) + # Public key is optional, used only for callback signature verification + .with_public_key("SAT_PUBLIC_KEY") + .with_is_debug(True) +) + +sat_client = SATClient(config) +response = sat_client.ping() +``` +Use **sat.WithSatBaseURL** to override SAT Base URL when you move to another environment. +Example production environment you can use https://b2b.tokopedia.com/api as the SAT Base URL. + +```python +from py_sat import SATClientConfig, SATClient + +config = ( + SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE_KEY", # required + ) + # Below is optional parameter + .with_timeout(10) + # Public key is optional, used only for callback signature verification + .with_public_key("SAT_PUBLIC_KEY") + # Override SAT Base URL + .with_sat_base_url("https://b2b.tokopedia.com/api") +) + +sat_client = SATClient(config) +response = sat_client.ping() +``` + +#### Ping +This method allows you to check SAT server health +```python +response = sat_client.ping() +``` + +#### Account +Account method used to check current your account balance. +This will help you to monitor your account balance. +You can utilize this function to prevent it from insufficient balance. +```python +response = sat_client.account() +``` + +#### Inquiry +Inquiry method mostly used to check a user bill for a product inquiry type +```python +from py_sat import SATClientConfig, SATClient +from py_sat.models import InquiryRequest, Field + +config = ( + SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE_KEY", # required + ) + # Below is optional parameter + .with_timeout(10) + # Public key is optional, used only for callback signature verification + .with_public_key("SAT_PUBLIC_KEY") + # Override SAT Base URL + .with_sat_base_url("https://b2b.tokopedia.com/api") +) + +sat_client = SATClient(config) + +response = sat_client.inquiry( + req=InquiryRequest( + product_code="pln-postpaid", + client_number="2121212", + fields=[Field(name="optional", value="optional")], + ) +) + +if response.is_success(): + # Handle success here + # Type InquiryBaseResponse + print(response.client_number) + print(response.client_name) +else: + # Handle error here + # Type ErrorResponse + print(response.get_error_messages()) +``` + +#### Checkout +Checkout allows your system to post the order to SAT server. It means the order will be processed, and your balance will be deducted. +The process will be asynchronous, so you required to implement Check Status to get the final order status. +```python +from py_sat import SATClientConfig, SATClient +from py_sat.models import OrderRequest, Field + +config = ( + SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE_KEY", # required + ) + # Below is optional parameter + .with_timeout(10) + # Public key is optional, used only for callback signature verification + .with_public_key("SAT_PUBLIC_KEY") + # Override SAT Base URL + .with_sat_base_url("https://b2b.tokopedia.com/api") +) + +sat_client = SATClient(config) + +req = OrderRequest( + id="unique_id", + product_code="pln-postpaid", + client_number="2121212", + amount=12500, + fields=[ + Field(name="optional", value="optional"), + ], +) +response = sat_client.checkout(req) + +if response.is_success(): + # Handle success here + # Type OrderDetail + print(response.id) + print(response.client_number) + print(response.client_name) +else: + # Handle error here + # Type ErrorResponse + print(response.get_error_messages()) +``` + +#### Check Status +Check Status will return the current order status and the detail order information. Please follow our API Doc to handle each error code. + +```python +from py_sat import SATClientConfig, SATClient +from py_sat.models import OrderRequest, Field + +config = ( + SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE_KEY", # required + ) + # Below is optional parameter + .with_timeout(10) + # Public key is optional, used only for callback signature verification + .with_public_key("SAT_PUBLIC_KEY") + # Override SAT Base URL + .with_sat_base_url("https://b2b.tokopedia.com/api") +) + +sat_client = SATClient(config) + +req = OrderRequest( + id="unique_id", + product_code="pln-postpaid", + client_number="2121212", + amount=12500, + fields=[ + Field(name="optional", value="optional"), + ], +) +response = sat_client.check_status(request_id) + +``` + +##### Handle Error Code From Order Status Failed +Order Status "Failed" always exposes error code. You can refer to our **API Documentation Section 4.8 Error Response** to handle each error code. +Below snipped code is the example of how you can handle the error code. +```python +from py_sat.models import OrderDetail + +response:OrderDetail = sat_client.check_status("unique_id") + +if response.is_success() and response.status == "Failed": + error_code = response.data.error_code + if error_code == "S00": + # do something + elif error_code == "P00": + # do something + elif error_code == "U00": + # do something +``` + + +#### List Product +List product will return all products that are available or specific product when you pass the product code on the parameter. +```python +from py_sat import SATClientConfig, SATClient + +config = ( + SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE_KEY", # required + ) + # Below is optional parameter + .with_timeout(10) + # Public key is optional, used only for callback signature verification + .with_public_key("SAT_PUBLIC_KEY") + # Override SAT Base URL + .with_sat_base_url("https://b2b.tokopedia.com/api") +) + +sat_client = SATClient(config) +response = sat_client.list_product() +``` +You can also filter the product by passing the product code on the parameter. +```python +from py_sat import SATClientConfig, SATClient + +config = ( + SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE_KEY", # required + ) + # Below is optional parameter + .with_timeout(10) + # Public key is optional, used only for callback signature verification + .with_public_key("SAT_PUBLIC_KEY") + # Override SAT Base URL + .with_sat_base_url("https://b2b.tokopedia.com/api") +) + +sat_client = SATClient(config) +response = sat_client.list_product("25k-xl") +``` + + +#### Callback +Client need to expose the Webhook using HTTP Server and implement the Handler using Callback interface. +Callback will help you to get the final status order real time based on the event via triggered from webhook. + +```python +from werkzeug.wrappers import Request, Response +from py_sat.models import OrderDetail +from py_sat import SATClientConfig, SATClient +import logging + +config = SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE_KEY", # required +).with_public_key("SAT_PUBLIC_KEY") + +sat_client = SATClient(config) + + +def handler(request: Request) -> Response: + try: + data = request.json + headers = dict(request.headers) + + def do_action(order_detail: OrderDetail): + assert isinstance(order_detail, OrderDetail) + assert order_detail.id == "1231231" + assert order_detail.status == "Success" + assert order_detail.product_code == "pln-prepaid-token-100k" + assert order_detail.sales_price == 102500 + assert order_detail.client_name == "User" + assert order_detail.client_number == "102111106111" + + # Do your action here + # For example, update your database + # or send a notification to the customer + logging.info("Order detail: %s", order_detail) + + sat_client.handle_callback( + sat_response_data=data, + sat_response_headers=headers, + do=do_action, + ) + + return Response(status=200, response=json.dumps({"message": "OK"})) + except Exception as e: + logging.exception("Error handling callback") + return Response( + status=500, + response=json.dumps({"error": str(e)}), + ) +``` + +### Handle Error +This SDK applied standard error payload that always provides error code, error detail, and http status. +Detail error handling each error code will be mentioned in our **API Documentation Section 4.8 Error Response**. +This is an example of how you can handle the error code. +```python +from py_sat.models import InquiryRequest +from py_sat import SATClient, SATClientConfig + +config = SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE", # required +).with_public_key("SAT_PUBLIC_KEY") + +sat_client = SATClient(config) + +response = sat_client.inquiry( + req=InquiryRequest( + product_code="not-found-product", + client_number="2121212", + ) +) + +if not response.is_success(): + error_code = response.get_error_codes() + if error_code == "S00": + # do something + elif error_code == "P00": + # do something + elif error_code == "U00": + # do something +``` +ResponseGeneralException is an error coming from non sat server, example: firewall, proxy, client http, etc. +You can parse the http response by yourself and handle it based on your need. +Most of the time you only need to use the http statusCode and handle it. + +```python +from py_sat.models import InquiryRequest +from py_sat.exceptions import ResponseGeneralException, GeneralException +from py_sat import SATClient, SATClientConfig + +config = SATClientConfig( + client_id="YOUR_CLIENT_ID", # required + client_secret="YOUR_CLIENT_SECRET", # required + private_key="YOUR_PRIVATE", # required +).with_public_key("SAT_PUBLIC_KEY") + +try: + sat_client = SATClient(config) + response = sat_client.inquiry( + req=InquiryRequest( + product_code="pln-postpaid", + client_number="2121212", + ) + ) +except ResponseGeneralException as e: + response = e.get_raw_response() + status_code = response.status_code + response_body = response.text +except GeneralException as e: + print(e) +except Exception as e: + print(e) +``` + +### Full Example +Please check on the tests folder to see the full implementation for each method. + + +### License +The MIT License (MIT) + +Copyright (c) 2024 Tokopedia + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/docs/Makefile b/docs/Makefile new file mode 100644 index 0000000..d4bb2cb --- /dev/null +++ b/docs/Makefile @@ -0,0 +1,20 @@ +# Minimal makefile for Sphinx documentation +# + +# You can set these variables from the command line, and also +# from the environment for the first two. +SPHINXOPTS ?= +SPHINXBUILD ?= sphinx-build +SOURCEDIR = . +BUILDDIR = _build + +# Put it first so that "make" without argument is like "make help". +help: + @$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O) + +.PHONY: help Makefile + +# Catch-all target: route all unknown targets to Sphinx using the new +# "make mode" option. $(O) is meant as a shortcut for $(SPHINXOPTS). +%: Makefile + @$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O) diff --git a/docs/conf.py b/docs/conf.py new file mode 100644 index 0000000..4ef4e38 --- /dev/null +++ b/docs/conf.py @@ -0,0 +1,35 @@ +# Configuration file for the Sphinx documentation builder. +# +# For the full list of built-in configuration values, see the documentation: +# https://www.sphinx-doc.org/en/master/usage/configuration.html + +# -- Project information ----------------------------------------------------- +# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information + +import os +import sys + +sys.path.insert(0, os.path.abspath("..")) + +project = "py_sat_sdk" +copyright = "2024, PT SAT" +author = "PT SAT" + +# -- General configuration --------------------------------------------------- +# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration + +extensions = [ + "sphinx.ext.autodoc", + "sphinx.ext.viewcode", + "sphinx.ext.napoleon", + "myst_parser", +] +templates_path = ["_templates"] +exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"] + + +# -- Options for HTML output ------------------------------------------------- +# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output + +html_theme = "sphinx_rtd_theme" +html_static_path = ["_static"] diff --git a/docs/index.rst b/docs/index.rst new file mode 100644 index 0000000..a5af66f --- /dev/null +++ b/docs/index.rst @@ -0,0 +1,29 @@ +.. py_sat documentation master file, created by + sphinx-quickstart on Mon Jun 10 11:06:08 2024. + You can adapt this file completely to your liking, but it should at least + contain the root `toctree` directive. + +Welcome to py_sat's documentation! +================================== + +.. toctree:: + :maxdepth: 2 + :caption: Contents: + + modules + + +Indices and tables +================== + +* :ref:`genindex` +* :ref:`modindex` +* :ref:`search` + + +README +================== + +.. include:: readme_link.md + :parser: myst_parser.sphinx_ + diff --git a/docs/make.bat b/docs/make.bat new file mode 100644 index 0000000..954237b --- /dev/null +++ b/docs/make.bat @@ -0,0 +1,35 @@ +@ECHO OFF + +pushd %~dp0 + +REM Command file for Sphinx documentation + +if "%SPHINXBUILD%" == "" ( + set SPHINXBUILD=sphinx-build +) +set SOURCEDIR=. +set BUILDDIR=_build + +%SPHINXBUILD% >NUL 2>NUL +if errorlevel 9009 ( + echo. + echo.The 'sphinx-build' command was not found. Make sure you have Sphinx + echo.installed, then set the SPHINXBUILD environment variable to point + echo.to the full path of the 'sphinx-build' executable. Alternatively you + echo.may add the Sphinx directory to PATH. + echo. + echo.If you don't have Sphinx installed, grab it from + echo.https://www.sphinx-doc.org/ + exit /b 1 +) + +if "%1" == "" goto help + +%SPHINXBUILD% -M %1 %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O% +goto end + +:help +%SPHINXBUILD% -M help %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O% + +:end +popd diff --git a/docs/modules.rst b/docs/modules.rst new file mode 100644 index 0000000..34871a1 --- /dev/null +++ b/docs/modules.rst @@ -0,0 +1,8 @@ +py_sat +====== + +.. toctree:: + :maxdepth: 4 + + py_sat + diff --git a/docs/py_sat.models.rst b/docs/py_sat.models.rst new file mode 100644 index 0000000..eea4221 --- /dev/null +++ b/docs/py_sat.models.rst @@ -0,0 +1,69 @@ +py\_sat.models package +====================== + +Submodules +---------- + +py\_sat.models.account module +----------------------------- + +.. automodule:: py_sat.models.account + :members: + :undoc-members: + :show-inheritance: + +py\_sat.models.base module +-------------------------- + +.. automodule:: py_sat.models.base + :members: + :undoc-members: + :show-inheritance: + +py\_sat.models.error module +--------------------------- + +.. automodule:: py_sat.models.error + :members: + :undoc-members: + :show-inheritance: + +py\_sat.models.inquiry module +----------------------------- + +.. automodule:: py_sat.models.inquiry + :members: + :undoc-members: + :show-inheritance: + +py\_sat.models.order module +--------------------------- + +.. automodule:: py_sat.models.order + :members: + :undoc-members: + :show-inheritance: + +py\_sat.models.ping module +-------------------------- + +.. automodule:: py_sat.models.ping + :members: + :undoc-members: + :show-inheritance: + +py\_sat.models.product module +----------------------------- + +.. automodule:: py_sat.models.product + :members: + :undoc-members: + :show-inheritance: + +Module contents +--------------- + +.. automodule:: py_sat.models + :members: + :undoc-members: + :show-inheritance: diff --git a/docs/py_sat.rst b/docs/py_sat.rst new file mode 100644 index 0000000..6a86696 --- /dev/null +++ b/docs/py_sat.rst @@ -0,0 +1,62 @@ +py\_sat package +=============== + +Subpackages +----------- + +.. toctree:: + :maxdepth: 4 + + py_sat.models + py_sat.signature + +Submodules +---------- + +py\_sat.client module +--------------------- + +.. automodule:: py_sat.client + :members: + :undoc-members: + :show-inheritance: + +py\_sat.constant module +----------------------- + +.. automodule:: py_sat.constant + :members: + :undoc-members: + :show-inheritance: + +py\_sat.exceptions module +------------------------- + +.. automodule:: py_sat.exceptions + :members: + :undoc-members: + :show-inheritance: + +py\_sat.http\_client module +--------------------------- + +.. automodule:: py_sat.http_client + :members: + :undoc-members: + :show-inheritance: + +py\_sat.utils module +-------------------- + +.. automodule:: py_sat.utils + :members: + :undoc-members: + :show-inheritance: + +Module contents +--------------- + +.. automodule:: py_sat + :members: + :undoc-members: + :show-inheritance: diff --git a/docs/py_sat.signature.rst b/docs/py_sat.signature.rst new file mode 100644 index 0000000..65a02b5 --- /dev/null +++ b/docs/py_sat.signature.rst @@ -0,0 +1,29 @@ +py\_sat.signature package +========================= + +Submodules +---------- + +py\_sat.signature.interface module +---------------------------------- + +.. automodule:: py_sat.signature.interface + :members: + :undoc-members: + :show-inheritance: + +py\_sat.signature.pss module +---------------------------- + +.. automodule:: py_sat.signature.pss + :members: + :undoc-members: + :show-inheritance: + +Module contents +--------------- + +.. automodule:: py_sat.signature + :members: + :undoc-members: + :show-inheritance: diff --git a/docs/readme_link.md b/docs/readme_link.md new file mode 100644 index 0000000..451beda --- /dev/null +++ b/docs/readme_link.md @@ -0,0 +1,2 @@ +```{include} ../README.md +``` diff --git a/docs/requirements.txt b/docs/requirements.txt new file mode 100644 index 0000000..1213c60 --- /dev/null +++ b/docs/requirements.txt @@ -0,0 +1,79 @@ +alabaster==0.7.13 +atomicwrites==1.4.1 +attrs==22.2.0 +Babel==2.11.0 +beautifulsoup4==4.12.3 +black==22.8.0 +bleach==4.1.0 +certifi==2024.2.2 +cffi==1.15.1 +charset-normalizer==2.0.12 +click==8.0.4 +colorama==0.4.5 +coverage==6.2 +dataclasses==0.8 +dataclasses-json==0.5.9 +decorator==5.1.1 +docutils==0.18.1 +furl==2.1.3 +idna==3.7 +imagesize==1.4.1 +importlib-metadata==4.8.3 +importlib-resources==5.4.0 +iniconfig==1.1.1 +Jinja2==3.0.3 +jwcrypto==1.5.2 +keyring==23.4.1 +Mako==1.1.6 +Markdown==3.3.7 +MarkupSafe==2.0.1 +marshmallow==3.14.1 +marshmallow-enum==1.5.1 +more-itertools==8.14.0 +mypy-extensions==1.0.0 +oauthlib==3.2.2 +orderedmultidict==1.0.1 +packaging==21.3 +pathspec==0.9.0 +pipdeptree==2.2.1 +pkginfo==1.10.0 +platformdirs==2.4.0 +pluggy==1.0.0 +py==1.11.0 +py-sat==1.0.0 +pycparser==2.21 +pycryptodome==3.11.0 +Pygments==2.14.0 +pyparsing==3.1.2 +pytest==7.0.1 +pytest-dotenv==0.5.2 +pytest-httpserver==1.0.5 +python-dateutil==2.9.0.post0 +python-dotenv==0.20.0 +pytz==2024.1 +readme-renderer==34.0 +requests==2.27.1 +requests-oauthlib==2.0.0 +requests-toolbelt==1.0.0 +rfc3986==1.5.0 +scrapeasy==0.12 +six==1.16.0 +snowballstemmer==2.2.0 +soupsieve==2.3.2.post1 +Sphinx==5.3.0 +sphinxcontrib-applehelp==1.0.2 +sphinxcontrib-devhelp==1.0.2 +sphinxcontrib-htmlhelp==2.0.0 +sphinxcontrib-jsmath==1.0.1 +sphinxcontrib-qthelp==1.0.3 +sphinxcontrib-serializinghtml==1.1.5 +tomli==1.2.3 +tqdm==4.64.1 +typed-ast==1.5.5 +typing-inspect==0.9.0 +typing_extensions==4.1.1 +urllib3==1.26.18 +validators==0.20.0 +webencodings==0.5.1 +Werkzeug==2.0.3 +zipp==3.6.0 diff --git a/py_sat/LICENSE b/py_sat/LICENSE new file mode 100644 index 0000000..aaaece0 --- /dev/null +++ b/py_sat/LICENSE @@ -0,0 +1,9 @@ +The MIT License (MIT) + +Copyright (c) 2024 Tokopedia + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/py_sat/__init__.py b/py_sat/__init__.py new file mode 100644 index 0000000..db12517 --- /dev/null +++ b/py_sat/__init__.py @@ -0,0 +1 @@ +from .client import SATClient, SATClientConfig diff --git a/py_sat/client.py b/py_sat/client.py new file mode 100644 index 0000000..2f5126e --- /dev/null +++ b/py_sat/client.py @@ -0,0 +1,373 @@ +""" +client package contains the main class to interact with the SAT service. +""" + +import json +import logging +from typing import Any, Callable, Dict, Optional, Union + +import requests +from requests.exceptions import HTTPError + +from py_sat.constant import (ACCESS_TOKEN_URL, ACCOUNT_PATH, CHECK_STATUS_PATH, + CHECKOUT_PATH, INQUIRY_PATH, PING_PATH, + PLAYGROUND_SAT_BASE_URL, PRODUCT_LIST_PATH) +from py_sat.exceptions import (GeneralException, InvalidInputException, + ResponseGeneralException, + UnauthenticatedException) +from py_sat.http_client import HTTPClient +from py_sat.models import (Account, ErrorResponse, InquiryRequest, + InquiryResponse, OrderDetail, OrderRequest, + PartnerProduct, PingResponse, ProductListResponse) +from py_sat.signature import Signature, SignatureType +from py_sat.utils import (generate_json_api_request, + parse_json_api_list_response, + parse_json_api_response) + + +class SATClientConfig: + """ + SATClientConfig is a configuration class to initialize the SATClient + """ + + # Required + client_id: str + client_secret: str + private_key: str + + # Optional + public_key: Optional[str] + padding_type: SignatureType + is_debug: bool + sat_base_url: str + access_token_base_url: str + timeout: int + logger: logging.Logger + + def __init__( + self, + client_id: str, + client_secret: str, + private_key: str, + ): + if not client_id or not isinstance(client_id, str): + raise InvalidInputException("Client ID are required and must be a string") + + if not client_secret or not isinstance(client_secret, str): + raise InvalidInputException( + "Client Secret are required and must be a string" + ) + + if not private_key or not isinstance(private_key, str): + raise InvalidInputException("Private key is required and must be a string") + + self.client_id = client_id + self.client_secret = client_secret + self.private_key = private_key + + self._set_default_value() + + def _set_default_value(self): + logger: logging.Logger + logger = logging.getLogger(__name__) + logger.setLevel(logging.DEBUG) + self.logger = logger + + self.public_key = None + self.padding_type = SignatureType.PSS + self.is_debug = False + self.sat_base_url = PLAYGROUND_SAT_BASE_URL + self.access_token_base_url = ACCESS_TOKEN_URL + self.timeout = 30 + + def with_logger(self, logger: logging.Logger): + self.logger = logger + return self + + def with_public_key(self, public_key: str): + self.public_key = public_key + return self + + def with_padding_type(self, padding_type: SignatureType): + self.padding_type = padding_type + return self + + def with_is_debug(self, is_debug: bool): + self.is_debug = is_debug + return self + + def with_sat_base_url(self, sat_base_url: str): + self.sat_base_url = sat_base_url + return self + + def with_timeout(self, timeout: int): + self.timeout = timeout + return self + + def with_access_token_base_url(self, access_token_base_url: str): + self.access_token_base_url = access_token_base_url + return self + + +class SATClient: + """ + SATClient is a main class for SAT SDK, this class contains all the method to interact with SAT service + """ + + _config: SATClientConfig + signature: Signature + _logger: logging.Logger + _http_client: HTTPClient + + def __init__(self, config: SATClientConfig): + self._config = config + self.signature = Signature( + config.private_key, config.public_key, config.padding_type + ) + self._logger = config.logger + self._http_client = HTTPClient( + base_url=config.sat_base_url, + oauth_base_url=config.access_token_base_url, + client_id=config.client_id, + client_secret=config.client_secret, + logger=config.logger, + ) + + def ping(self) -> Union[PingResponse, ErrorResponse]: + """ + Ping is a method to check the SAT server health + :return: PingResponse or ErrorResponse + :raise ResponseGeneralException: if there are unexpected error when hitting SAT (403 forbidden, network error) + :raise GeneralException: if there is an unexpected exception when pinging + """ + try: + url = f"{self._config.sat_base_url}{PING_PATH}" + + req = requests.Request(method="GET", url=url) + response = self._http_client.send_request(req) + response.raise_for_status() + + json_response = response.json() + return PingResponse.from_dict(json_response).with_raw_response(response) + except HTTPError as exc: + return self._handle_http_error(exc) + except Exception as exc: + self._logger.error(f"Error when pinging: {exc}") + raise GeneralException(exc) + + def inquiry(self, req: InquiryRequest) -> Union[InquiryResponse, ErrorResponse]: + """ + Inquiry is a method to get user bills based on client number and product code + :param req: InquiryRequest + :return: InquiryBaseResponse or ErrorResponse + :raise ResponseGeneralException: if there are unexpected error when hitting SAT (403 forbidden, network error) + :raise GeneralException: if there is an unexpected exception when inquiring + """ + try: + url = f"{self._config.sat_base_url}{INQUIRY_PATH}" + + body = generate_json_api_request(req.to_dict()) + http_req = requests.Request(method="POST", url=url, json=body) + response = self._http_client.send_request(http_req) + response.raise_for_status() + + json_response = response.json() + data = parse_json_api_response(json_response) + + return InquiryResponse.from_dict(data).with_raw_response(response) + except HTTPError as exc: + return self._handle_http_error(exc) + except Exception as exc: + self._logger.error(f"Error when inquiry: {exc}") + raise GeneralException(exc) + + def checkout(self, req: OrderRequest) -> Union[OrderDetail, ErrorResponse]: + """ + Checkout is a method to do payment an order based on client number, product code and request id. + Request ID should use unique identifier for each transaction + :param req: OrderRequest + :return: OrderDetail or ErrorResponse + :raise ResponseGeneralException: if there are unexpected error when hitting SAT (403 forbidden, network error) + :raise GeneralException: if there is an unexpected exception when checking out + """ + try: + url = f"{self._config.sat_base_url}{CHECKOUT_PATH}" + + body = generate_json_api_request(req.to_dict()) + body_str = json.dumps(body) + signature = self.signature.sign(body_str) + + http_req = requests.Request( + method="POST", + url=url, + json=body, + headers={"signature": signature}, + ) + response = self._http_client.send_request(http_req) + response.raise_for_status() + + json_response = response.json() + data = parse_json_api_response(json_response) + + return OrderDetail.from_dict(data).with_raw_response(response) + except HTTPError as exc: + return self._handle_http_error(exc) + except Exception as exc: + self._logger.error(f"Error when checkout: {exc}") + raise GeneralException(exc) + + def check_status(self, request_id: str) -> Union[OrderDetail, ErrorResponse]: + """ + CheckStatus is a method to check the final status of an order. + request id is must be filled + :param request_id: request id to check the status + :return: OrderDetail or ErrorResponse + :raise ResponseGeneralException: if there are unexpected error when hitting SAT (403 forbidden, network error) + :raise GeneralException: if there is an unexpected exception when checking status + """ + try: + url = f"{self._config.sat_base_url}{CHECK_STATUS_PATH.format(request_id=request_id)}" + + http_req = requests.Request( + method="GET", + url=url, + ) + response = self._http_client.send_request(http_req) + response.raise_for_status() + + json_response = response.json() + data = parse_json_api_response(json_response) + + return OrderDetail.from_dict(data).with_raw_response(response) + except HTTPError as exc: + return self._handle_http_error(exc) + except Exception as exc: + self._logger.error(f"Error when check status: {exc}") + raise GeneralException(exc) + + def list_product( + self, code: Optional[str] = None + ) -> Union[ProductListResponse, ErrorResponse]: + """ + ListProduct is a method to get all the product list enabled on your credentials. + you can also specify the product code, to get only one product detail. + specify product code will be very beneficial to sync product status on your engine + it will come with low bandwidth and fast response + :param code: product code to filter the product list + :return: ProductListResponse or ErrorResponse + :raise ResponseGeneralException: if there are unexpected error when hitting SAT (403 forbidden, network error) + :raise GeneralException: if there is an unexpected exception when listing product + """ + try: + url = f"{self._config.sat_base_url}{PRODUCT_LIST_PATH}" + + http_req = requests.Request( + method="GET", + url=url, + params={"product_code": code}, + ) + response = self._http_client.send_request(http_req) + response.raise_for_status() + + json_response = response.json() + data = parse_json_api_list_response(json_response) + products = [ + PartnerProduct.from_dict(item).with_raw_response(response) + for item in data + ] + + return ProductListResponse(products=products).with_raw_response(response) + except HTTPError as exc: + return self._handle_http_error(exc) + except Exception as exc: + self._logger.error(f"Error when list product: {exc}") + raise GeneralException(exc) + + def account(self) -> Union[Account, ErrorResponse]: + """ + Account is a method to check account balance + :return: Account or ErrorResponse + :raise ResponseGeneralException: if there are unexpected error when hitting SAT (403 forbidden, network error) + :raise GeneralException: if there is an unexpected exception when checking account + """ + + try: + url = f"{self._config.sat_base_url}{ACCOUNT_PATH}" + + http_req = requests.Request( + method="GET", + url=url, + ) + response = self._http_client.send_request(http_req) + response.raise_for_status() + + json_response = response.json() + data = parse_json_api_response(json_response) + + return Account.from_dict(data).with_raw_response(response) + except HTTPError as exc: + return self._handle_http_error(exc) + except Exception as exc: + self._logger.error(f"Error when checking balance: {exc}") + raise GeneralException(exc) + + def handle_callback( + self, + sat_response_data: Dict[str, Any], + sat_response_headers: Dict[str, Any], + do: Callable[[OrderDetail], None], + ): + """ + HandleCallback is method http.HandlerFunc to handle callback request from SAT + you can customize the implementation based on this interface Callback + :param sat_response_data: JSON API response data from SAT webhook, must be a dictionary + :param sat_response_headers: HTTP headers from SAT webhook, must be a dictionary + :param do: Your custom function to handle the callback, must be a function with OrderDetail as + parameter and return void + :raise InvalidInputException: if the header does not contain the signature + :raise UnauthenticatedException: if the signature is not valid + """ + signature: Optional[str] = sat_response_headers.get( + "signature" + ) or sat_response_headers.get("Signature") + if signature is None: + raise InvalidInputException( + "Signature is not present in the header, please check the request" + ) + + verify = self.signature.verify(json.dumps(sat_response_data), signature) + if not verify: + raise UnauthenticatedException("Signature is not valid") + + data = parse_json_api_response(sat_response_data) + order_detail = OrderDetail.from_dict(data) + + do(order_detail) + + def get_http_client(self) -> HTTPClient: + """ + GetHTTTPClient will return http client which already wrapped to support oauth2 + for custom integration to SAT Service + :return: HTTPClient + """ + return self._http_client + + def get_signature(self) -> Signature: + """ + GetSignature will return signature object which already wrapped to support RSA signature + for custom integration to SAT Service + :return: Signature + """ + return self.signature + + def _handle_http_error(self, exc: HTTPError): + try: + status = exc.response.status_code + message = exc.response.text + self._logger.debug(f"HTTP Error: {status}, {message}") + + data = exc.response.json() + resp = ErrorResponse.from_dict(data).with_raw_response(exc.response) + return resp + except Exception as e: + raise ResponseGeneralException(exc) diff --git a/py_sat/constant.py b/py_sat/constant.py new file mode 100644 index 0000000..2463e8c --- /dev/null +++ b/py_sat/constant.py @@ -0,0 +1,19 @@ +# Base URL +ACCESS_TOKEN_URL = "https://accounts.tokopedia.com/token" +PLAYGROUND_SAT_BASE_URL = "https://b2b-playground.tokopedia.com/api" + +# Path +PING_PATH = "/ping" +INQUIRY_PATH = "/v2/inquiry" +CHECKOUT_PATH = "/v2/order" +CHECK_STATUS_PATH = "/v2/order/{request_id}" +PRODUCT_LIST_PATH = "/v2/product-list" +ACCOUNT_PATH = "/v2/account" + +SIGNATURE_HEADER_KEY = "signature" + +SDK_NAME = "py_sat" +SDK_VERSION = "v1.0.0" +SDK_LABEL = f"{SDK_NAME}@{SDK_VERSION}" + +DATE_TIME_FORMAT = "%a, %d %b %Y %H:%M:%S GMT" diff --git a/py_sat/exceptions.py b/py_sat/exceptions.py new file mode 100644 index 0000000..f503fdf --- /dev/null +++ b/py_sat/exceptions.py @@ -0,0 +1,38 @@ +from requests import HTTPError, Request, Response + + +class InvalidInputException(Exception): + pass + + +class UnauthenticatedException(Exception): + pass + + +class SignatureErrorException(Exception): + pass + + +class ResponseGeneralException(Exception): + _raw_request: Request + _raw_response: Response + + def __init__(self, exc: HTTPError): + if isinstance(exc, HTTPError): + self._raw_response = exc.response + self._raw_request = exc.request + + super().__init__(exc) + + def get_raw_response(self) -> Response: + return self._raw_response + + +class GeneralException(Exception): + def __init__( + self, + original_exception, + message="An error occurred", + ): + self.message = f"{message}: {original_exception}" + super().__init__(self.message) diff --git a/py_sat/http_client.py b/py_sat/http_client.py new file mode 100644 index 0000000..fb644d7 --- /dev/null +++ b/py_sat/http_client.py @@ -0,0 +1,86 @@ +import logging +from datetime import datetime, timezone + +import requests +from oauthlib.oauth2 import BackendApplicationClient +from requests_oauthlib import OAuth2Session + +from py_sat.constant import DATE_TIME_FORMAT, SDK_LABEL + + +class HTTPClient: + _base_url: str + _oauth_base_url: str + _access_token: str + _client_id: str + _client_secret: str + _session: requests.Session + _auth: OAuth2Session + _logger: logging.Logger + _is_debug: bool + + def __init__( + self, + base_url: str, + oauth_base_url: str, + client_id: str, + client_secret: str, + logger: logging.Logger, + is_debug: bool = False, + ): + self._base_url = base_url + self._oauth_base_url = oauth_base_url + self._client_id = client_id + self._client_secret = client_secret + self._session = self._prepare_session() + self._logger = logger + self._is_debug = is_debug + + self._auth = OAuth2Session( + client=BackendApplicationClient(client_id), + client_id=client_id, + token_updater=lambda token: self._save_token(token), + auto_refresh_url=self._oauth_base_url, + auto_refresh_kwargs={ + "client_id": client_id, + "client_secret": client_secret, + }, + ) + self._access_token = self.get_access_token() + + def _save_token(self, token): + self._access_token = token + + @staticmethod + def _prepare_session(): + session = requests.Session() + session.headers.update({}) + + return session + + def get_access_token(self): + token = self._auth.fetch_token( + token_url=self._oauth_base_url, + client_id=self._client_id, + client_secret=self._client_secret, + ) + + return token.get("access_token") + + def send_request(self, request: requests.Request) -> requests.Response: + """Sends a prepared Request object and returns the response.""" + request.headers["Date"] = datetime.now(timezone.utc).strftime(DATE_TIME_FORMAT) + request.headers["Content-Type"] = "application/json" + request.headers["Accept"] = "application/json" + request.headers["X-Sat-Sdk-Version"] = SDK_LABEL + request.headers["Authorization"] = f"Bearer {self._access_token}" + + prepared_request = self._auth.prepare_request(request) + + response = self._session.send(prepared_request) + + self._logger.info( + f"Request: {response.request.url}, {response.request.method} {response.request.body} {response.request.headers}" + ) + + return response diff --git a/py_sat/models/__init__.py b/py_sat/models/__init__.py new file mode 100644 index 0000000..6640aa3 --- /dev/null +++ b/py_sat/models/__init__.py @@ -0,0 +1,8 @@ +from py_sat.models.account import Account +from py_sat.models.base import Field +from py_sat.models.error import ErrorObject, ErrorResponse +from py_sat.models.inquiry import InquiryRequest, InquiryResponse +from py_sat.models.order import OrderDetail, OrderRequest +from py_sat.models.ping import PingResponse +from py_sat.models.product import (PartnerProduct, ProductListResponse, + ProductStatus) diff --git a/py_sat/models/account.py b/py_sat/models/account.py new file mode 100644 index 0000000..ca0f774 --- /dev/null +++ b/py_sat/models/account.py @@ -0,0 +1,16 @@ +from dataclasses import dataclass, field + +from dataclasses_json import Undefined, dataclass_json + +from py_sat.models.base import BaseResponse + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class Account(BaseResponse): + """ + Account is a schema related with account entity + """ + + id: int + saldo: int diff --git a/py_sat/models/base.py b/py_sat/models/base.py new file mode 100644 index 0000000..f1bb187 --- /dev/null +++ b/py_sat/models/base.py @@ -0,0 +1,53 @@ +from dataclasses import dataclass +from datetime import datetime +from typing import Optional + +import dateutil.parser +import requests +from dataclasses_json import (DataClassJsonMixin, Undefined, config, + dataclass_json) + + +class BaseResponse(DataClassJsonMixin): + _raw_response: requests.Response + + def get_raw_response(self) -> requests.Response: + return self._raw_response + + def with_raw_response(self, raw_response: requests.Response): + self._raw_response = raw_response + return self + + def is_success(self): + return 200 <= self._raw_response.status_code < 300 + + +class BaseRequest(DataClassJsonMixin): + pass + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class Field: + name: str + value: str + + +def datetime_encoder(input: Optional[datetime]) -> str: + if input is None: + return None + + return input.isoformat() + + +def datetime_decoder(input: Optional[str]) -> Optional[datetime]: + if input is None: + return None + + return dateutil.parser.isoparse(input) + + +datetime_config = config( + encoder=datetime_encoder, + decoder=datetime_decoder, +) diff --git a/py_sat/models/error.py b/py_sat/models/error.py new file mode 100644 index 0000000..9bf81c4 --- /dev/null +++ b/py_sat/models/error.py @@ -0,0 +1,75 @@ +from dataclasses import dataclass, field +from typing import List + +from dataclasses_json import Undefined, dataclass_json + +from py_sat.models.base import BaseResponse + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class ErrorObject: + """ + Error object, represent error detail from SAT API + """ + + id: str = field(default="") + title: str = field(default="") + detail: str = field(default="") + status: str = field(default="") + code: str = field(default="") + meta: dict = field(default_factory=dict) + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class ErrorResponse(BaseResponse): + """ + Error response object, represent error response from SAT API + """ + + errors: List[ErrorObject] = field(default_factory=list) + + def get_error_messages(self) -> str: + """ + Get error messages from errors + :return: string of error messages, separated by new line + """ + if len(self.errors) <= 0: + return "" + + messages = [ + f"{error.status} - {error.code} - {error.detail}" for error in self.errors + ] + return "\n".join(messages) + + def get_error_codes(self) -> str: + """ + Get error codes from errors + :return: string of error codes, separated by comma + """ + if len(self.errors) <= 0: + return "" + + return ", ".join([error.code for error in self.errors]) + + def get_error_statuses(self) -> str: + """ + Get error statuses from errors + :return: string of error statuses, separated by comma + """ + if len(self.errors) <= 0: + return "" + + return ", ".join([error.status for error in self.errors]) + + def get_error_details(self) -> str: + """ + Get error details from errors + :return: string of error details, separated by comma + """ + + if len(self.errors) <= 0: + return "" + + return ", ".join([error.detail for error in self.errors]) diff --git a/py_sat/models/inquiry.py b/py_sat/models/inquiry.py new file mode 100644 index 0000000..b8eb239 --- /dev/null +++ b/py_sat/models/inquiry.py @@ -0,0 +1,44 @@ +from dataclasses import dataclass, field +from typing import List, Optional + +from dataclasses_json import Undefined, dataclass_json + +from py_sat.models.base import BaseRequest, BaseResponse, Field + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class InquiryRequest(BaseRequest): + """ + InquiryRequest to hold inquiry request data + """ + + product_code: str + client_number: str + amount: Optional[int] = field(default=None) + id: Optional[str] = field(default=None) + fields: Optional[List[Field]] = field(default=None) + downline_id: Optional[str] = field(default=None) + type: str = field(default="inquiry", init=False) + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class InquiryResponse(BaseResponse): + """ + InquiryResponse to hold inquiry response + """ + + id: str = field(default="") + product_code: str = field(default="") + sales_price: float = field(default=0.0) + fields: List[Field] = field(default_factory=list) + inquiry_result: List[Field] = field(default_factory=list) + base_price: float = field(default=0.0) + admin_fee: float = field(default=0.0) + client_name: str = field(default="") + client_number: str = field(default="") + meter_id: str = field(default="") + ref_id: str = field(default="") + max_payment: int = field(default=0) + min_payment: int = field(default=0) diff --git a/py_sat/models/order.py b/py_sat/models/order.py new file mode 100644 index 0000000..7f8f08e --- /dev/null +++ b/py_sat/models/order.py @@ -0,0 +1,44 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import List, Optional + +from dataclasses_json import Undefined, config, dataclass_json + +from py_sat.models.base import (BaseRequest, BaseResponse, Field, + datetime_config) + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class OrderRequest(BaseRequest): + id: str = field(default="") + product_code: str = field(default="") + client_number: str = field(default="") + amount: float = field(default=0.0) + fields: Optional[List[Field]] = field(default=None) + downline_id: Optional[str] = field(default=None) + + type: str = field(default="order", init=False) + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class OrderDetail(BaseResponse): + id: str = field(default="") + fields: Optional[List[Field]] = field(default=None) + fulfillment_result: List[Field] = field(default_factory=list) + fulfilled_at: Optional[datetime] = field( + default=None, + metadata=datetime_config, + ) + error_code: str = field(default="") + error_detail: str = field(default="") + product_code: str = field(default="") + status: str = field(default="") + partner_fee: int = field(default=0) + sales_price: int = field(default=0) + admin_fee: int = field(default=0) + client_name: str = field(default="") + client_number: str = field(default="") + voucher_code: str = field(default="") + serial_number: str = field(default="") diff --git a/py_sat/models/ping.py b/py_sat/models/ping.py new file mode 100644 index 0000000..7fe612a --- /dev/null +++ b/py_sat/models/ping.py @@ -0,0 +1,13 @@ +from dataclasses import dataclass + +from dataclasses_json import Undefined, dataclass_json + +from py_sat.models.base import BaseResponse + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class PingResponse(BaseResponse): + buildhash: str + sandbox: bool + status: str diff --git a/py_sat/models/product.py b/py_sat/models/product.py new file mode 100644 index 0000000..90a166c --- /dev/null +++ b/py_sat/models/product.py @@ -0,0 +1,31 @@ +import enum +from dataclasses import dataclass, field +from typing import List + +from dataclasses_json import Undefined, dataclass_json + +from py_sat.models.base import BaseResponse + + +class ProductStatus(enum.Enum): + Active = 1 + Inactive = 2 + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class PartnerProduct(BaseResponse): + id: str = field(default="") + name: str = field(default="") + operator_name: str = field(default="") + category_name: str = field(default="") + is_inquiry: bool = field(default=False) + sales_price: int = field(default=0) + status: ProductStatus = field(default=ProductStatus.Active) + client_number: str = field(default="") + + +@dataclass_json(undefined=Undefined.EXCLUDE) +@dataclass +class ProductListResponse(BaseResponse): + products: List[PartnerProduct] = field(default_factory=list) diff --git a/py_sat/signature/__init__.py b/py_sat/signature/__init__.py new file mode 100644 index 0000000..1838ba6 --- /dev/null +++ b/py_sat/signature/__init__.py @@ -0,0 +1,115 @@ +from enum import Enum +from typing import Optional + +from Crypto.PublicKey import RSA + +from py_sat.exceptions import InvalidInputException, SignatureErrorException +from py_sat.signature.interface import SignatureAlgorithm +from py_sat.signature.pss import PSSPaddingAlgorithm + + +class SignatureType(Enum): + PSS = "PSS" + + +class Signature: + """ + Signature to hold that signature needs, and contain parsed public and private key + """ + + _private_key: Optional[RSA.RsaKey] + _public_key: Optional[RSA.RsaKey] + _algorithm: SignatureAlgorithm + + def __init__( + self, + private_key_str: Optional[str], + public_key_str: Optional[str], + padding_type: SignatureType, + ): + if not padding_type: + raise InvalidInputException("Padding type is required") + + self._private_key = self._parse_rsa_private_key_from_pem_str(private_key_str) + self._public_key = self._parse_public_key(public_key_str) + self._algorithm = self.__decide_padding_algorithm(padding_type) + + def verify(self, msg: str, signature: str) -> bool: + """ + Verify the signature of the message, return True if the signature is valid, False otherwise + + :param msg: message to verify, must be a string + :param signature: signature to verify, must be a string + :return: True if the signature is valid, False otherwise + :raise InvalidInputException: if the message or signature is not a string or public key is not set + :raise SignatureErrorException: if there is an error verifying the signature + """ + if not isinstance(msg, str) or not isinstance(signature, str): + raise InvalidInputException("Message and signature must be strings") + + if not self._public_key: + raise InvalidInputException("Public key not set") + + try: + return self._algorithm.verify(self._public_key, msg, signature) + except Exception as exc: + raise SignatureErrorException(f"Error verifying signature: {exc}") + + def sign(self, msg: str) -> str: + """ + Sign the message and return the signature + + :param msg: to sign, must be a string + :return: the signature as a string + :raises InvalidInputException: if the message is not a string or private key is not set + :raises SignatureErrorException: if there is an error signing the message + """ + if not isinstance(msg, str): + raise InvalidInputException("Message must be a string") + + if not self._private_key: + raise InvalidInputException("Private key not set") + + try: + return self._algorithm.sign(self._private_key, msg) + except Exception as exc: + raise SignatureErrorException(f"Error signing message: {exc}") + + @staticmethod + def __decide_padding_algorithm(padding_type: SignatureType) -> SignatureAlgorithm: + if padding_type == SignatureType.PSS: + return PSSPaddingAlgorithm() + else: + raise InvalidInputException(f"Unknown padding type: {padding_type}") + + @staticmethod + def _parse_rsa_private_key_from_pem_str( + private_key_pem: Optional[str], + ) -> Optional[RSA.RsaKey]: + try: + """Parses an RSA private key from a PEM-encoded string.""" + if not private_key_pem: + return None + + private_key = RSA.import_key(private_key_pem.encode()) + if not isinstance(private_key, RSA.RsaKey): + raise InvalidInputException("Key is not a valid RSA public key") + + return private_key + except (ValueError, IndexError, TypeError) as exc: + raise InvalidInputException(f"Invalid RSA private key PEM: {exc}") + + @staticmethod + def _parse_public_key(public_key_pem: Optional[str]) -> Optional[RSA.RsaKey]: + """Parses an RSA public key from a PEM-encoded string.""" + try: + if not public_key_pem: + return None + + public_key = RSA.import_key(public_key_pem.encode()) + + if not isinstance(public_key, RSA.RsaKey): + raise InvalidInputException("Key is not a valid RSA public key") + return public_key + except (ValueError, IndexError, TypeError) as exc: + raise InvalidInputException(f"Invalid RSA public key PEM: {exc}") diff --git a/py_sat/signature/interface.py b/py_sat/signature/interface.py new file mode 100644 index 0000000..12ca6a1 --- /dev/null +++ b/py_sat/signature/interface.py @@ -0,0 +1,18 @@ +from abc import ABC, abstractmethod +from typing import Union + +from Crypto.PublicKey import RSA + + +class SignatureAlgorithm(ABC): + @abstractmethod + def verify( + self, public_key: RSA.RsaKey, msg: Union[str, bytes], signature: str + ) -> bool: + """Verifies a signature""" + pass + + @abstractmethod + def sign(self, private_key: RSA.RsaKey, msg: Union[str, bytes]) -> str: + """Signs a message""" + pass diff --git a/py_sat/signature/pss.py b/py_sat/signature/pss.py new file mode 100644 index 0000000..ff68554 --- /dev/null +++ b/py_sat/signature/pss.py @@ -0,0 +1,66 @@ +import base64 +from typing import Union + +from Crypto.Hash import SHA256 +from Crypto.PublicKey import RSA +from Crypto.Signature import pss + +from py_sat.signature.interface import SignatureAlgorithm + + +class PSSPaddingAlgorithm(SignatureAlgorithm): + def verify( + self, public_key: RSA, msg: Union[str, bytes], signature_base64: str + ) -> bool: + """Verifies a PKCS1v15 signature using SHA-256. + + Args: + public_key: The RSA public key used for verification. + msg: The message (either a string or bytes) to verify. + signature_base64: The base64-encoded signature to verify. + + Returns: + True if the signature is valid, False otherwise. + """ + + if not signature_base64.strip(): + raise ValueError("Signature is empty") + + if isinstance(msg, str): + msg = msg.encode("utf-8") + + try: + signature_bytes = base64.b64decode(signature_base64) + except ValueError: + raise ValueError("Invalid base64 signature") + + hash_message = SHA256.new(msg) + salt = int(public_key.n.bit_length() / 8) - hash_message.digest_size - 2 + verifier = pss.new(public_key, salt_bytes=salt) + + try: + verifier.verify(hash_message, signature_bytes) + return True + except (ValueError, TypeError): + return False + + def sign(self, private_key: RSA, msg: Union[str, bytes]) -> str: + """Signs a message using PSS padding and RSA algorithm. + + Args: + private_key: The RSA private key used for signing. + msg: The message (either a string or bytes) to sign. + + Returns: + The base64-encoded signature as a string. + """ + if isinstance(msg, str): + msg = msg.encode("utf-8") + + hash = SHA256.new(msg) + salt = int(private_key.n.bit_length() / 8) - hash.digest_size - 2 + signature = pss.new(private_key, salt_bytes=salt).sign(hash) + + base64_signature = base64.b64encode(signature) + encoded_signature = str(base64_signature, "utf-8") + return encoded_signature diff --git a/py_sat/utils.py b/py_sat/utils.py new file mode 100644 index 0000000..824e5e7 --- /dev/null +++ b/py_sat/utils.py @@ -0,0 +1,73 @@ +from typing import Any, Dict, List + + +def parse_json_api_response(response: dict) -> Dict[str, Any]: + """ + Parse JSON API response to dictionary + + :param response: JSON API response + :return: aggregate dictionary of id, type, and attributes + """ + data = response.get("data", {}) + return __parse_json_api_dict(data) + + +def parse_json_api_list_response(response: dict) -> List[Dict[str, Any]]: + """ + Parse JSON API list response to list of dictionary + + :param response: JSON API response + :return: aggregate dictionary of id, type, and attributes + """ + data = response.get("data", []) + + return [__parse_json_api_dict(item) for item in data] + + +def __parse_json_api_dict(data): + id = data.get("id", "") + type = data.get("type", "") + attributes = data.get("attributes", {}) + return {"id": id, "type": type, **attributes} + + +def generate_json_api_request(request: dict) -> Dict[str, Any]: + """ + Generate JSON API request from dictionary + + :param request: normal dictionary + :return: JSON API request format + """ + id = extract_id(request) + type = extract_type(request) + + data = {} + if id is not None: + data["id"] = id + if type is not None: + data["type"] = type + + # Omit none values + attribute = {k: v for k, v in request.items() if v is not None} + data["attributes"] = attribute + + return {"data": data} + + +def extract_type(request): + type = None + if "type" in request: + type = request.get("type", None) + del request["type"] + return type + + +def extract_id(request): + id = None + if "id" in request: + id = request.get("id", None) + del request["id"] + if "request_id" in request: + id = request.get("request_id", None) + del request["request_id"] + return id diff --git a/scripts/setup-dev.sh b/scripts/setup-dev.sh new file mode 100755 index 0000000..1f6bfee --- /dev/null +++ b/scripts/setup-dev.sh @@ -0,0 +1,27 @@ +#!/usr/bin/env bash + +echo "Installing development dependency" +pip install -U tox +pip install -U sphinx myst-parser sphinx-rtd-theme twine + + +# This python versions is used for testing +PYTHON_VERSIONS=("3.7" "3.8" "3.9" "3.10") +for version in "${PYTHON_VERSIONS[@]}"; do + if ! pyenv versions --bare | grep -q "^${version}"; then + echo "Python ${version} is not installed. Installing now..." + pyenv install "${version}" + else + echo "Python ${version} is already installed." + fi +done + +echo "Python installation check completed." + +echo "Installing development dependencies" +tox -e py38 --devenv .venv +echo "Finished installing development dependencies." + + +source .venv/bin/activate + diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..8ec4e05 --- /dev/null +++ b/setup.py @@ -0,0 +1,37 @@ +# read the contents of your README file +from pathlib import Path + +from setuptools import find_packages, setup + +this_directory = Path(__file__).parent +long_description = (this_directory / "README.md").read_text() + +setup( + name="py_sat_sdk", + packages=find_packages(), + version="1.0.6", + description="Python SDK for Tokopedia SAT API", + author="SAT Team", + install_requires=[ + "requests>=2.27.1", + "requests-oauthlib>=2.0.0", + "pycryptodome>=3.11.0", + "dataclasses-json>=0.5.9", + "python-dateutil>=2.9.0.post0", + ], + extras_require={ + "test": [ + "pytest>=4.4.1", + "pytest-sugar", + "pytest-dotenv>=0.5.2", + "pytest-httpserver>=1.0.5", + ], + }, + test_suite="tests", + classifiers=[ + "Development Status :: 5 - Production/Stable", + "Intended Audience :: Developers", + ], + long_description=long_description, + long_description_content_type="text/markdown", +) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..690d995 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,158 @@ +import enum +import os +import random +import string + +import pytest +from pytest_httpserver import HTTPServer + +from py_sat import SATClient, SATClientConfig +from py_sat.constant import ACCESS_TOKEN_URL, PLAYGROUND_SAT_BASE_URL + +TESTING_PRIVATE_KEY = """-----BEGIN PRIVATE KEY----- +MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDyR0kXD0bu1nl8 +nZP+GLI8bSVFbk5yKTu99LlLevTTFLx6sIXfabgKPHIpwr0xGf99yobD1ZNZ276x +ffnMAeILNA5XsvaMnPpVB4kNoqlDaQdd4ICelKQwt90QD9CIGptNLL2wtUgEn1g9 +BV6k5xcT8L9Dw/ZqpMCnfBeGRsWJd84LBfN4lLe5k9MXxE4MLUfTC1xxLmO9C9xw +d92aucyRPkv8n+B9dOlYS8C29huTbregl2rEF32dMyYG1qmVH7ufjM4CX9KdNKA7 +hwnJExqrPvhAtj92Ar0Z5JnPfm8SUjQQhFeySeqSHS+kVEsd/AhrsqMsdUSt9ou2 +xJTJmiL9AgMBAAECggEANHOO5Q1jZassn3gs9T6K/c6CWmr0VD5NhwUfhXIP5U/Q +sz4aqYDFfX/TFmvo0iPG/oh1TxninfpXKS11Aj/pHFRPg5iEzHHitzxbpUZRHz0y +gVYsekiDWGHB2+uUkZazBwz33zUL66ZEr+dE823tPt2oxsa6xyE2bTwOCr2xH96S +eE6280gJ1LtgGKD4TFsmXSkgJIDyF2rT6ZkPeUR36N0zWiSRQbskLaTqwJSpT6mY +hdZRQBvH/6X2JJgOmWWVizE1ChdOVB7rN3tb7NBSbwf5TWNGsNdhQQoTPxAtGiX8 +O+jXilCIjLSqudIDv20jCa8spFuXwpIeBhR/7394GQKBgQD6+g4yZSybGMnMgNM6 +OrL1GnJbxtnuMENMi7e2NgeLxKmpWE3cPqTNe9uWNxXYexDgDLNtAfjw7+UZxO6f +e3cPXuyVsr2xTkEscLBNDXrwQV3phnfwpxX++8Flv0xP34TocWAIKXPcQj8h2SYr +6/zhY40YnRsUgsR0x4eAlqJHBQKBgQD3IKl15jjzHEyqVvWuAjsw3LO8rA96i7vg +WDV5LBrsJgE11M7iurwZxZ71STPVzWK5UzxsZBCJgK41NIvgVx6QIuU0HXw9taXY +7PrAwNcpx7yKksXPYsTGhKumLvFQBY/R9qP2RcFN5WQjTSBxp0B5QMDHNz01dPi6 +l0TfjKO9mQKBgQCZRRJcdmsaQLYkfNwCaIyXoNIL+FFo4/KFkaHc1fwfwDd4ouPR +yDPvBV/hybw+m1F/8mG1BYpY4bhA14J+xPC941OKTEEKQecNU7hnJf9ZMCJBFgyz +W+bT9D10fLIG6VMKfQqPkXkfHxnc+vcTxaeGobwuNuutx/pf8uZugg+SXQKBgQC3 +qjOnpxHeRMMJuhVfXNMm7nA6odnjJuTbyFL9mnTr2xb9LgsQYN4ZfVE1VVFL7hgY +Si9XE0tjFhri+gmXEshpMTYNdHh42H7I6N830FpY99Q9XPXcurgqHkIAAVVhNrD7 +yAV1q8QNo5W30sNxFG+Lbj+YD4rTJvsQmgoa5shuyQKBgQCDJuBgjNgyHqWrKEEt +76OJDiXI4mXDg3N6oCjsP/ZsP7mhkmUsokDS1paSnQxtt0tbft+fpbQSFtDGkRKr +OM5SBQhJEa//gFAofgYt+Fo6wcj1NSgBeLspsYx/avRdj9drVGyxOKJX0zsWQemU +Gf6bGumDG6hy0wu4aWDjSvTmww== +-----END PRIVATE KEY----- +""" + +TESTING_PUBLIC_KEY = """-----BEGIN PUBLIC KEY----- +MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA8kdJFw9G7tZ5fJ2T/hiy +PG0lRW5Ocik7vfS5S3r00xS8erCF32m4CjxyKcK9MRn/fcqGw9WTWdu+sX35zAHi +CzQOV7L2jJz6VQeJDaKpQ2kHXeCAnpSkMLfdEA/QiBqbTSy9sLVIBJ9YPQVepOcX +E/C/Q8P2aqTAp3wXhkbFiXfOCwXzeJS3uZPTF8RODC1H0wtccS5jvQvccHfdmrnM +kT5L/J/gfXTpWEvAtvYbk263oJdqxBd9nTMmBtaplR+7n4zOAl/SnTSgO4cJyRMa +qz74QLY/dgK9GeSZz35vElI0EIRXsknqkh0vpFRLHfwIa7KjLHVErfaLtsSUyZoi +/QIDAQAB +-----END PUBLIC KEY----- +""" + + +class TestEnvironment(str, enum.Enum): + __test__ = False + LOCAL = "local" + SANDBOX = "sandbox" + + +@pytest.fixture(scope="session") +def env(): + env = os.getenv("PY_SAT_TEST_ENV", "local") + return TestEnvironment[env.upper()] + + +def __construct_local_test_config(make_httpserver: HTTPServer): + base_url = make_httpserver.url_for("") + base_url_without_trailing_slash = base_url[:-1] + + access_token_base_url = make_httpserver.url_for("token") + + config = ( + SATClientConfig( + client_id="client_id", + client_secret="client_secret", + private_key=TESTING_PRIVATE_KEY, + ) + .with_timeout(15) + .with_public_key(TESTING_PUBLIC_KEY) + .with_sat_base_url(base_url_without_trailing_slash) + .with_access_token_base_url(access_token_base_url) + .with_is_debug(True) + ) + # Allow insecure transport for testing + os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1" + make_httpserver.expect_request("/token", method="POST").respond_with_json( + response_json={ + "access_token": "testingToken", + "token_type": "bearer", + "expires_in": 3600, + } + ) + return config + + +def __construct_sandbox_test_config(): + client_id = os.getenv("PY_SAT_TEST_CLIENT_ID") + client_secret = os.getenv("PY_SAT_TEST_CLIENT_SECRET") + if not client_id or not client_secret: + raise ValueError("Missing client id or client secret for sandbox testing") + + private_key = os.getenv("PY_SAT_TEST_PRIVATE_KEY") + private_key = bytes(private_key, "utf-8").decode("unicode_escape") + public_key = os.getenv("PY_SAT_TEST_PUBLIC_KEY") + public_key = bytes(public_key, "utf-8").decode("unicode_escape") + if not private_key or not public_key: + raise ValueError("Missing private key or public key for sandbox testing") + + config = ( + SATClientConfig( + client_id=client_id, + client_secret=client_secret, + private_key=private_key, + ) + .with_timeout(10) + .with_public_key(public_key) + .with_is_debug(True) + ) + + return config + + +@pytest.fixture(scope="session") +def sat_config(make_httpserver: HTTPServer, env: TestEnvironment): + if env == TestEnvironment.LOCAL: + return __construct_local_test_config(make_httpserver) + elif env == TestEnvironment.SANDBOX: + return __construct_sandbox_test_config() + else: + raise ValueError(f"Invalid environment: {env}") + + +@pytest.fixture(scope="session") +def sat_client(sat_config: SATClientConfig): + sat_client = SATClient(sat_config) + + return sat_client + + +class TestUtil: + @staticmethod + def generate_random_string(length: int) -> str: + return "".join( + random.choice(string.ascii_uppercase + string.digits) for _ in range(length) + ) + + @staticmethod + def generate_random_number(length: int) -> str: + return "".join(random.choice(string.digits) for _ in range(length)) + + @staticmethod + def generate_client_number() -> str: + return TestUtil.generate_random_number(5) + "1111" + + +@pytest.fixture(scope="session") +def util(): + return TestUtil() diff --git a/tests/test_account.py b/tests/test_account.py new file mode 100644 index 0000000..af59b16 --- /dev/null +++ b/tests/test_account.py @@ -0,0 +1,49 @@ +""" +Test account endpoint + +This example shows how to use the account endpoint to get the account balance from the SAT server. + +For more information, please refer to "Account Balance" section in the SAT documentation. +""" + +from pytest_httpserver import HTTPServer + +from py_sat import SATClient +from py_sat.constant import ACCOUNT_PATH + + +def test_account_success( + sat_client: SATClient, + make_httpserver: HTTPServer, +): + """ + Example of account endpoint success + + :param sat_client: + :param make_httpserver: + """ + make_httpserver.expect_request( + ACCOUNT_PATH, + method="GET", + headers={ + "authorization": "Bearer testingToken", + }, + ).respond_with_json( + response_json={ + "data": { + "type": "account", + "id": "2203", + "attributes": { + "saldo": 50000000, + }, + } + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.account() + assert response.is_success() + assert response.get_raw_response().status_code == 200 + assert response.id == 2203 + assert response.saldo == 50000000 diff --git a/tests/test_callback.py b/tests/test_callback.py new file mode 100644 index 0000000..bf768ac --- /dev/null +++ b/tests/test_callback.py @@ -0,0 +1,207 @@ +""" +This file is used to test the callback mechanism from SAT + +For partners that need the callback mechanism from SAT +follow these callback requirements below: +1. Your service should be using SSL (HTTPS) +2. Your service should whitelist our IP (52.74.35.225/32) to hit your service +3. Your service should allow method POST +4. Your service should allow Content-type application/vnd.api+json + +For more detail, see "Callback for Partner" section in the documentation +""" + +import json +import logging + +import requests +from pytest_httpserver import HTTPServer +from werkzeug.wrappers import Request, Response + +from py_sat import SATClient +from py_sat.models import OrderDetail +from py_sat.utils import parse_json_api_response + + +def test_callback(make_httpserver: HTTPServer, sat_client: SATClient): + """ + Example of how to handle the callback from SAT using SAT Python SDK + """ + make_httpserver.expect_request( + "/callback", + method="POST", + headers={ + "content-type": "application/vnd.api+json", + }, + ).respond_with_handler(create_handler(sat_client)) + + # Simulate the callback from SAT + body = { + "data": { + "type": "order", + "id": "1231231", + "attributes": { + "admin_fee": 0, + "client_name": "User", + "client_number": "102111106111", + "error_code": "", + "error_detail": "", + "fields": None, + "fulfilled_at": "2020-12-09T10:48:45Z", + "fulfillment_result": [], + "partner_fee": 0, + "product_code": "pln-prepaid-token-100k", + "sales_price": 102500, + "serial_number": "5196 15840828 2085 4701", + "status": "Success", + "voucher_code": "5196 1584 0828 2085 4701", + }, + } + } + signature = sat_client.signature.sign(json.dumps(body)) + headers = { + "content-type": "application/vnd.api+json", + "signature": signature, + } + response = requests.post( + make_httpserver.url_for("/callback"), + json=body, + headers=headers, + ) + + assert response.json() == {"message": "OK"} + assert response.status_code == 200 + + +def create_handler(sat_client: SATClient): + def handler(request: Request) -> Response: + try: + data = request.json + headers = dict(request.headers) + + def do_action(order_detail: OrderDetail): + assert isinstance(order_detail, OrderDetail) + assert order_detail.id == "1231231" + assert order_detail.status == "Success" + assert order_detail.product_code == "pln-prepaid-token-100k" + assert order_detail.sales_price == 102500 + assert order_detail.client_name == "User" + assert order_detail.client_number == "102111106111" + + # Do your action here + # For example, update your database + # or send a notification to the customer + logging.info("Order detail: %s", order_detail) + + sat_client.handle_callback( + sat_response_data=data, + sat_response_headers=headers, + do=do_action, + ) + + return Response(status=200, response=json.dumps({"message": "OK"})) + except Exception as e: + logging.exception("Error handling callback") + return Response( + status=500, + response=json.dumps({"error": str(e)}), + ) + + return handler + + +def test_callback_signature(make_httpserver: HTTPServer, sat_client: SATClient): + """ + Example of how to handle the callback from SAT without using SATClient handle_callback function + but only using the signature verification + """ + make_httpserver.expect_request( + "/callback", + method="POST", + headers={ + "content-type": "application/vnd.api+json", + }, + ).respond_with_handler(create_signature_only_handler(sat_client)) + + # Simulate the callback from SAT + body = { + "data": { + "type": "order", + "id": "1231231", + "attributes": { + "admin_fee": 0, + "client_name": "User", + "client_number": "102111106111", + "error_code": "", + "error_detail": "", + "fields": None, + "fulfilled_at": "2020-12-09T10:48:45Z", + "fulfillment_result": [], + "partner_fee": 0, + "product_code": "pln-prepaid-token-100k", + "sales_price": 102500, + "serial_number": "5196 15840828 2085 4701", + "status": "Success", + "voucher_code": "5196 1584 0828 2085 4701", + }, + } + } + signature = sat_client.signature.sign(json.dumps(body)) + headers = { + "content-type": "application/vnd.api+json", + "signature": signature, + } + response = requests.post( + make_httpserver.url_for("/callback"), + json=body, + headers=headers, + ) + + assert response.json() == {"message": "OK"} + assert response.status_code == 200 + + +def create_signature_only_handler(sat_client: SATClient): + def handler(request: Request) -> Response: + try: + data = request.json + headers = dict(request.headers) + + # Custom logic here + logging.info("Data: %s", data) + logging.info("Headers: %s", headers) + + # Do signature verification + signature = headers.get("signature") or headers.get("Signature") + if not signature: + return Response( + status=400, + response=json.dumps({"error": "Signature is required"}), + ) + + if not sat_client.signature.verify(json.dumps(data), signature): + return Response( + status=400, + response=json.dumps({"error": "Invalid signature"}), + ) + + input = parse_json_api_response(data) + order_detail = OrderDetail.from_dict(input) + assert isinstance(order_detail, OrderDetail) + assert order_detail.id == "1231231" + assert order_detail.status == "Success" + assert order_detail.product_code == "pln-prepaid-token-100k" + # Do your action here + # For example, update your database + # or send a notification to the customer + logging.info("Order detail: %s", order_detail) + + return Response(status=200, response=json.dumps({"message": "OK"})) + except Exception as e: + logging.exception("Error handling callback") + return Response( + status=500, + response=json.dumps({"error": str(e)}), + ) + + return handler diff --git a/tests/test_check_status.py b/tests/test_check_status.py new file mode 100644 index 0000000..be14ea7 --- /dev/null +++ b/tests/test_check_status.py @@ -0,0 +1,363 @@ +""" +This module contains the tests for the check_status method in the SATClient class. + +This example shows how to use the check_status method to check the status of a transaction from the SAT server. + +For more information, please refer to "Check Status" section in the SAT documentation. +""" + +import datetime +import json +import time + +from conftest import TestUtil +from pytest_httpserver import HTTPServer + +from py_sat import SATClient +from py_sat.constant import CHECK_STATUS_PATH, CHECKOUT_PATH, SDK_LABEL +from py_sat.models import (ErrorObject, ErrorResponse, Field, OrderDetail, + OrderRequest) +from py_sat.utils import generate_json_api_request + + +def test_check_status_success( + make_httpserver: HTTPServer, + sat_client: SATClient, + util: TestUtil, +): + """ + Test check status success + + :param make_httpserver: + :param sat_client: + :param util: + """ + random_string = "PYSAT" + util.generate_random_string(8) + random_client_id = util.generate_client_number() + + req = OrderRequest( + id=random_string, + product_code="speedy-indihome", + client_number=random_client_id, + amount=3500, + ) + body = generate_json_api_request(req.to_dict()) + body_str = json.dumps(body) + signature = sat_client.signature.sign(body_str) + + make_httpserver.expect_request( + CHECKOUT_PATH, + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "type": "order", + "id": random_string, + "attributes": { + "product_code": "speedy-indihome", + "client_number": random_client_id, + "amount": 3500, + }, + } + }, + ).respond_with_json( + response_json={ + "data": { + "type": "order", + "id": random_string, + "attributes": { + "client_number": random_client_id, + "error_code": "", + "error_detail": "", + "fields": None, + "fulfilled_at": None, + "partner_fee": 2000, + "product_code": "speedy-indihome", + "sales_price": 3500, + "serial_number": "", + "status": "Pending", + }, + } + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + assert sat_client.signature.verify(body_str, signature) + + response = sat_client.checkout(req) + + assert response.is_success() + assert response.to_json() + assert response.to_dict() + assert response.get_raw_response().status_code == 200 + assert response == OrderDetail( + partner_fee=2000, + product_code="speedy-indihome", + client_number=random_client_id, + sales_price=3500, + status="Pending", + id=random_string, + ) + + make_httpserver.expect_request( + CHECK_STATUS_PATH.format(request_id=random_string), + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + ).respond_with_json( + response_json={ + "data": { + "type": "order", + "id": random_string, + "attributes": { + "admin_fee": 2500, + "client_name": "Tokopedia User Default", + "client_number": random_client_id, + "error_code": "", + "error_detail": "", + "fields": None, + "fulfilled_at": datetime.datetime.now( + tz=datetime.timezone.utc + ).isoformat(), + "fulfillment_result": [ + {"name": "Nomor Referensi", "value": "174298636"}, + {"name": "Nama Pelanggan", "value": "Tokopedia User Default"}, + {"name": "Nomor Pelanggan", "value": "611981111"}, + {"name": "Jumlah Tagihan", "value": "1"}, + {"name": "Periode Bayar", "value": "Maret 2022"}, + {"name": "Total Tagihan", "value": "Rp 1.000"}, + {"name": "Biaya Admin", "value": "Rp 2.500"}, + {"name": "Total Bayar", "value": "Rp 3.500"}, + ], + "partner_fee": 2000, + "product_code": "speedy-indihome", + "sales_price": 3500, + "serial_number": "174298636", + "status": "Success", + "voucher_code": "", + }, + } + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + times = 1 + while True: + time.sleep(1) + response = sat_client.check_status(random_string) + if response.status != "Pending" or times > 3: + break + + times += 1 + + assert response.is_success() + assert response.to_json() + assert response.to_dict() + assert response.get_raw_response().status_code == 200 + assert response.product_code == "speedy-indihome" + assert response.client_number == random_client_id + assert response.status == "Success" + assert response.id == random_string + assert response.fulfilled_at is not None + assert datetime.datetime.now( + datetime.timezone.utc + ) - response.fulfilled_at < datetime.timedelta(seconds=10) + + +def test_check_status_failed( + make_httpserver: HTTPServer, + sat_client: SATClient, + util: TestUtil, +): + """ + Test check status failed + + :param make_httpserver: + :param sat_client: + :param util: + """ + random_string = "PYSAT" + util.generate_random_string(8) + from py_sat.models import Field + + req = OrderRequest( + id=random_string, + product_code="pln-postpaid", + client_number="2121212", + amount=12500, + fields=[ + Field(name="optional", value="optional"), + ], + ) + body = generate_json_api_request(req.to_dict()) + body_str = json.dumps(body) + signature = sat_client.signature.sign(body_str) + + make_httpserver.expect_request( + CHECKOUT_PATH, + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "attributes": { + "amount": 12500, + "client_number": "2121212", + "fields": [{"name": "optional", "value": "optional"}], + "product_code": "pln-postpaid", + }, + "id": random_string, + "type": "order", + } + }, + ).respond_with_json( + response_json={ + "data": { + "type": "order", + "id": random_string, + "attributes": { + "client_number": "2121212", + "error_code": "", + "error_detail": "", + "fields": [{"name": "optional", "value": "optional"}], + "fulfilled_at": None, + "partner_fee": 1000, + "product_code": "pln-postpaid", + "sales_price": 12500, + "serial_number": "", + "status": "Pending", + }, + } + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + assert sat_client.signature.verify(body_str, signature) + + response = sat_client.checkout(req) + + assert response.is_success() + assert response.get_raw_response().status_code == 200 + assert response == OrderDetail( + client_number="2121212", + fields=[ + Field(name="optional", value="optional"), + ], + partner_fee=1000, + product_code="pln-postpaid", + sales_price=12500, + status="Pending", + id=random_string, + ) + + make_httpserver.expect_request( + CHECK_STATUS_PATH.format(request_id=random_string), + method="GET", + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + ).respond_with_json( + response_json={ + "data": { + "type": "order", + "id": random_string, + "attributes": { + "client_number": "2121212", + "error_code": "S02", + "error_detail": "Product is not available", + "fields": [{"name": "optional", "value": "optional"}], + "fulfilled_at": None, + "partner_fee": 1000, + "product_code": "pln-postpaid", + "sales_price": 12500, + "serial_number": "", + "status": "Failed", + }, + } + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + times = 1 + while True: + time.sleep(1) + response = sat_client.check_status(random_string) + if response.status != "Pending" or times > 3: + break + + times += 1 + + assert response.is_success() + assert response.get_raw_response().status_code == 200 + assert response == OrderDetail( + client_number="2121212", + fields=[ + Field(name="optional", value="optional"), + ], + partner_fee=1000, + product_code="pln-postpaid", + sales_price=12500, + status="Failed", + error_code="S02", + error_detail="Product is not available", + id=random_string, + ) + + +def test_check_status_not_found( + make_httpserver: HTTPServer, + sat_client: SATClient, + util: TestUtil, +): + """ + Test check status not found + + :param make_httpserver: + :param sat_client: + :param util: + """ + random_string = "PYSAT" + util.generate_random_string(8) + + make_httpserver.expect_request( + CHECK_STATUS_PATH.format(request_id=random_string), + method="GET", + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + ).respond_with_json( + response_json={ + "errors": [ + {"code": "P02", "detail": "Transaction is not found", "status": "400"} + ] + }, + status=400, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.check_status(random_string) + + assert not response.is_success() + assert response.get_raw_response().status_code == 400 + assert response == ErrorResponse( + errors=[ + ErrorObject( + detail="Transaction is not found", + status="400", + code="P02", + ) + ] + ) diff --git a/tests/test_checkout.py b/tests/test_checkout.py new file mode 100644 index 0000000..6da5741 --- /dev/null +++ b/tests/test_checkout.py @@ -0,0 +1,307 @@ +""" +Test checkout endpoint + +This example shows how to use the checkout endpoint to create an order in the SAT server. + +For more information, please refer to "Checkout" section in the SAT documentation. +""" + +import json + +from conftest import TestUtil +from pytest_httpserver import HTTPServer +from pytest_httpserver.httpserver import HandlerType + +from py_sat import SATClient +from py_sat.constant import CHECKOUT_PATH, SDK_LABEL +from py_sat.models import (ErrorObject, ErrorResponse, Field, OrderDetail, + OrderRequest) +from py_sat.utils import generate_json_api_request + + +def test_checkout( + make_httpserver: HTTPServer, + sat_client: SATClient, + util: TestUtil, +): + """ + Example of checkout endpoint success + + :param make_httpserver: + :param sat_client: + :param util: + """ + random_string = "PYSAT" + util.generate_random_string(8) + req = OrderRequest( + id=random_string, + product_code="pln-postpaid", + client_number="2121212", + amount=12500, + fields=[ + Field(name="optional", value="optional"), + ], + ) + body = generate_json_api_request(req.to_dict()) + body_str = json.dumps(body) + signature = sat_client.signature.sign(body_str) + + make_httpserver.expect_request( + CHECKOUT_PATH, + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "attributes": { + "amount": 12500, + "client_number": "2121212", + "fields": [{"name": "optional", "value": "optional"}], + "product_code": "pln-postpaid", + }, + "id": random_string, + "type": "order", + } + }, + ).respond_with_json( + response_json={ + "data": { + "type": "order", + "id": random_string, + "attributes": { + "client_number": "2121212", + "error_code": "", + "error_detail": "", + "fields": [{"name": "optional", "value": "optional"}], + "fulfilled_at": None, + "partner_fee": 1000, + "product_code": "pln-postpaid", + "sales_price": 12500, + "serial_number": "", + "status": "Pending", + }, + } + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + assert sat_client.signature.verify(body_str, signature) + + response = sat_client.checkout(req) + + assert response.to_json(indent=4) + assert response.is_success() + assert response.get_raw_response().status_code == 200 + assert response == OrderDetail( + client_number="2121212", + fields=[ + Field(name="optional", value="optional"), + ], + partner_fee=1000, + product_code="pln-postpaid", + sales_price=12500, + status="Pending", + id=random_string, + ) + + +def test_checkout_product_not_found( + make_httpserver: HTTPServer, + sat_client: SATClient, + util: TestUtil, +): + """ + Example of checkout endpoint product not found + + :param make_httpserver: + :param sat_client: + :param util: + """ + random_string = "PYSAT" + util.generate_random_string(8) + req = OrderRequest( + id=random_string, + product_code="non-exist-product", + client_number="102111496000", + amount=12500, + fields=[ + Field(name="optional", value="optional"), + ], + ) + body = generate_json_api_request(req.to_dict()) + body_str = json.dumps(body) + signature = sat_client.signature.sign(body_str) + + make_httpserver.expect_request( + CHECKOUT_PATH, + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "id": random_string, + "type": "order", + "attributes": { + "product_code": "non-exist-product", + "client_number": "102111496000", + "amount": 12500, + "fields": [{"name": "optional", "value": "optional"}], + }, + } + }, + ).respond_with_json( + response_json={ + "errors": [{"code": "P04", "detail": "Product not found", "status": "400"}] + }, + status=400, + headers={"Content-Type": "application/json"}, + ) + + assert sat_client.signature.verify(body_str, signature) + + response = sat_client.checkout(req) + + assert not response.is_success() + assert response.get_raw_response().status_code == 400 + assert response == ErrorResponse( + errors=[ErrorObject(code="P04", detail="Product not found", status="400")] + ) + + +def test_checkout_duplicate_request_id( + make_httpserver: HTTPServer, + sat_client: SATClient, + util: TestUtil, +): + """ + Example of checkout endpoint duplicate request id + + :param make_httpserver: + :param sat_client: + :param util: + """ + random_string = "PYSAT" + util.generate_random_string(8) + req = OrderRequest( + id=random_string, + product_code="pln-postpaid", + client_number="2121212", + amount=12500, + fields=[ + Field(name="optional", value="optional"), + ], + ) + body = generate_json_api_request(req.to_dict()) + body_str = json.dumps(body) + signature = sat_client.signature.sign(body_str) + + make_httpserver.expect_request( + CHECKOUT_PATH, + handler_type=HandlerType.ONESHOT, + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "attributes": { + "amount": 12500, + "client_number": "2121212", + "fields": [{"name": "optional", "value": "optional"}], + "product_code": "pln-postpaid", + }, + "id": random_string, + "type": "order", + } + }, + ).respond_with_json( + response_json={ + "data": { + "type": "order", + "id": random_string, + "attributes": { + "client_number": "2121212", + "error_code": "", + "error_detail": "", + "fields": [{"name": "optional", "value": "optional"}], + "fulfilled_at": None, + "partner_fee": 1000, + "product_code": "pln-postpaid", + "sales_price": 12500, + "serial_number": "", + "status": "Pending", + }, + } + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + make_httpserver.expect_request( + CHECKOUT_PATH, + handler_type=HandlerType.ONESHOT, + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "attributes": { + "amount": 12500, + "client_number": "2121212", + "fields": [{"name": "optional", "value": "optional"}], + "product_code": "pln-postpaid", + }, + "id": random_string, + "type": "order", + } + }, + ).respond_with_json( + response_json={ + "errors": [ + { + "code": "P03", + "detail": "Duplicate Request ID, please check your system", + "status": "400", + } + ] + }, + status=400, + headers={"Content-Type": "application/json"}, + ) + + assert sat_client.signature.verify(body_str, signature) + + response = sat_client.checkout(req) + + assert response.is_success() + assert response.get_raw_response().status_code == 200 + assert response == OrderDetail( + client_number="2121212", + fields=[ + Field(name="optional", value="optional"), + ], + partner_fee=1000, + product_code="pln-postpaid", + sales_price=12500, + status="Pending", + id=random_string, + ) + + response = sat_client.checkout(req) + + assert not response.is_success() + assert response.get_raw_response().status_code == 400 + assert response == ErrorResponse( + errors=[ + ErrorObject( + code="P03", + detail="Duplicate Request ID, please check your system", + status="400", + ) + ] + ) diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..2c3fe97 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,81 @@ +""" +Test cases for SATClient class. + +This example shows how to use the SATClient class. +""" + +import pytest +from conftest import TESTING_PRIVATE_KEY + +from py_sat import SATClient, SATClientConfig +from py_sat.exceptions import InvalidInputException + + +def test_sat_client_required_param(sat_client: SATClient): + """ + Test SATClient class with required parameters. + :param sat_client: + :return: + """ + assert sat_client is not None + assert isinstance(sat_client, SATClient) + + +def test_sat_client_error_input(): + """ + Test SATClient class with error input. + :return: + """ + with pytest.raises(InvalidInputException): + config = SATClientConfig( + client_id="client_id", + client_secret="client_secret", + private_key=None, + ) + http_client = SATClient(config) + + +def test_sat_client_error_padding_input(): + """ + Test SATClient class with error padding input. + :return: + """ + with pytest.raises(InvalidInputException): + config = SATClientConfig( + client_id="client_id", + client_secret="client_secret", + private_key=TESTING_PRIVATE_KEY, + ).with_padding_type("test") + http_client = SATClient(config) + + +def test_sat_client_invalid_private_key(): + """ + Test SATClient class with invalid private key. + :return: + """ + with pytest.raises(InvalidInputException) as exc_info: + config = SATClientConfig( + client_id="client_id", + client_secret="client_secret", + private_key="invalid_private_key", + ) + _ = SATClient(config) + + assert "Invalid RSA private key" in str(exc_info.value) + + +def test_sat_client_invalid_public_key(): + """ + Test SATClient class with invalid public key. + :return: + """ + with pytest.raises(InvalidInputException) as exc_info: + config = SATClientConfig( + client_id="client_id", + client_secret="client_secret", + private_key=TESTING_PRIVATE_KEY, + ).with_public_key("invalid_public_key") + _ = SATClient(config) + + assert "Invalid RSA public key" in str(exc_info.value) diff --git a/tests/test_inquiry.py b/tests/test_inquiry.py new file mode 100644 index 0000000..75f08b1 --- /dev/null +++ b/tests/test_inquiry.py @@ -0,0 +1,330 @@ +""" +Test Inquiry API + +This example shows how to use the inquiry endpoint to get the inquiry result from the SAT server. + +For more information, please refer to "Inquiry" section in the SAT documentation. +""" + +from conftest import TestEnvironment +from pytest_httpserver import HTTPServer +from pytest_httpserver.httpserver import HandlerType + +from py_sat import SATClient +from py_sat.constant import INQUIRY_PATH, SDK_LABEL +from py_sat.models import (ErrorObject, ErrorResponse, InquiryRequest, + InquiryResponse) +from py_sat.models.inquiry import Field + + +def test_inquiry( + make_httpserver: HTTPServer, + sat_client: SATClient, +): + """ + Example of inquiry endpoint success + + :param make_httpserver: + :param sat_client: + """ + make_httpserver.expect_request( + INQUIRY_PATH, + method="POST", + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "type": "inquiry", + "attributes": { + "product_code": "pln-postpaid", + "client_number": "2121212", + "fields": [{"name": "optional", "value": "optional"}], + }, + } + }, + handler_type=HandlerType.ONESHOT, + ).respond_with_json( + response_json={ + "data": { + "type": "inquiry", + "id": "2121212", + "attributes": { + "admin_fee": 2500, + "base_price": 25000, + "client_name": "TOKOPXXXX UXX", + "client_number": "2121212", + "fields": [{"name": "optional", "value": "optional"}], + "inquiry_result": [ + {"name": "ID Pelanggan", "value": "2121212"}, + {"name": "Nama", "value": "TOKOPXXXX UXX"}, + {"name": "Total Bayar", "value": "Rp 27.500"}, + {"name": "IDPEL", "value": "2121212"}, + {"name": "NAMA", "value": "Tokopedia User Default"}, + {"name": "TOTAL TAGIHAN", "value": "1 BULAN"}, + {"name": "BL/TH", "value": "MAR20"}, + {"name": "RP TAG PLN", "value": "Rp 25.000"}, + {"name": "ADMIN BANK", "value": "Rp 2.500"}, + {"name": "TOTAL BAYAR", "value": "Rp 27.500"}, + ], + "meter_id": "2121212", + "product_code": "pln-postpaid", + "sales_price": 27500, + }, + } + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.inquiry( + req=InquiryRequest( + product_code="pln-postpaid", + client_number="2121212", + fields=[Field(name="optional", value="optional")], + ) + ) + + assert response.is_success() + assert response == InquiryResponse( + id="2121212", + product_code="pln-postpaid", + sales_price=27500, + fields=[Field(name="optional", value="optional")], + inquiry_result=[ + Field(name="ID Pelanggan", value="2121212"), + Field(name="Nama", value="TOKOPXXXX UXX"), + Field(name="Total Bayar", value="Rp 27.500"), + Field(name="IDPEL", value="2121212"), + Field(name="NAMA", value="Tokopedia User Default"), + Field(name="TOTAL TAGIHAN", value="1 BULAN"), + Field(name="BL/TH", value="MAR20"), + Field(name="RP TAG PLN", value="Rp 25.000"), + Field(name="ADMIN BANK", value="Rp 2.500"), + Field(name="TOTAL BAYAR", value="Rp 27.500"), + ], + base_price=25000, + admin_fee=2500, + client_name="TOKOPXXXX UXX", + client_number="2121212", + meter_id="2121212", + ref_id="", + max_payment=0, + min_payment=0, + ) + + +def test_inquiry_product_not_found( + make_httpserver: HTTPServer, + sat_client: SATClient, + env: TestEnvironment, +): + """ + Example of inquiry endpoint product not found + + :param make_httpserver: + :param sat_client: + :param env: + """ + make_httpserver.expect_request( + INQUIRY_PATH, + method="POST", + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "type": "inquiry", + "attributes": { + "product_code": "not-found-product", + "client_number": "2121212", + }, + } + }, + handler_type=HandlerType.ONESHOT, + ).respond_with_json( + response_json={ + "errors": [{"detail": "Product not found", "status": "400", "code": "P04"}] + }, + status=400, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.inquiry( + req=InquiryRequest( + product_code="not-found-product", + client_number="2121212", + ) + ) + + assert not response.is_success() + assert response.get_raw_response().status_code == 400 + assert response == ErrorResponse( + errors=[ErrorObject(detail="Product not found", status="400", code="P04")] + ) + assert response.get_error_messages() == "400 - P04 - Product not found" + assert response.get_error_codes() == "P04" + assert response.get_error_statuses() == "400" + + +def test_inquiry_s00( + make_httpserver: HTTPServer, + sat_client: SATClient, + env: TestEnvironment, +): + """ + Example of inquiry endpoint internal server error + + :param make_httpserver: + :param sat_client: + :param env: + """ + make_httpserver.expect_request( + INQUIRY_PATH, + method="POST", + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "type": "inquiry", + "attributes": { + "product_code": "speedy-indihome", + "client_number": "102111496000", + }, + } + }, + handler_type=HandlerType.ONESHOT, + ).respond_with_json( + response_json={ + "errors": [ + {"code": "S00", "detail": "Internal Server Error", "status": "500"} + ] + }, + status=500, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.inquiry( + req=InquiryRequest( + product_code="speedy-indihome", + client_number="102111496000", + ) + ) + + assert not response.is_success() + assert response.get_raw_response().status_code == 500 + assert response == ErrorResponse( + errors=[ErrorObject(detail="Internal Server Error", status="500", code="S00")] + ) + assert response.get_error_messages() == "500 - S00 - Internal Server Error" + assert response.get_error_codes() == "S00" + assert response.get_error_statuses() == "500" + assert response.get_error_details() == "Internal Server Error" + + +def test_inquiry_success_downline_id( + make_httpserver: HTTPServer, + sat_client: SATClient, +): + """ + Example of inquiry endpoint success + + :param make_httpserver: + :param sat_client: + """ + make_httpserver.expect_request( + INQUIRY_PATH, + method="POST", + headers={ + "content-type": "application/json", + "authorization": "Bearer testingToken", + "X-Sat-Sdk-Version": SDK_LABEL, + }, + json={ + "data": { + "type": "inquiry", + "attributes": { + "product_code": "pln-postpaid", + "client_number": "111111111111", + "fields": [{"name": "optional", "value": "optional"}], + "downline_id": "client-123", + }, + } + }, + handler_type=HandlerType.ONESHOT, + ).respond_with_json( + response_json={ + "data": { + "type": "inquiry", + "id": "2121212", + "attributes": { + "admin_fee": 2500, + "base_price": 25000, + "client_name": "TOKOPXXXX UXX", + "client_number": "2121212", + "fields": [{"name": "optional", "value": "optional"}], + "inquiry_result": [ + {"name": "ID Pelanggan", "value": "2121212"}, + {"name": "Nama", "value": "TOKOPXXXX UXX"}, + {"name": "Total Bayar", "value": "Rp 27.500"}, + {"name": "IDPEL", "value": "2121212"}, + {"name": "NAMA", "value": "Tokopedia User Default"}, + {"name": "TOTAL TAGIHAN", "value": "1 BULAN"}, + {"name": "BL/TH", "value": "MAR20"}, + {"name": "RP TAG PLN", "value": "Rp 25.000"}, + {"name": "ADMIN BANK", "value": "Rp 2.500"}, + {"name": "TOTAL BAYAR", "value": "Rp 27.500"}, + ], + "meter_id": "2121212", + "product_code": "pln-postpaid", + "sales_price": 27500, + }, + } + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.inquiry( + req=InquiryRequest( + product_code="pln-postpaid", + client_number="111111111111", + fields=[Field(name="optional", value="optional")], + downline_id="client-123", + ) + ) + + assert response.is_success() + assert response == InquiryResponse( + id="2121212", + product_code="pln-postpaid", + sales_price=27500, + fields=[Field(name="optional", value="optional")], + inquiry_result=[ + Field(name="ID Pelanggan", value="2121212"), + Field(name="Nama", value="TOKOPXXXX UXX"), + Field(name="Total Bayar", value="Rp 27.500"), + Field(name="IDPEL", value="2121212"), + Field(name="NAMA", value="Tokopedia User Default"), + Field(name="TOTAL TAGIHAN", value="1 BULAN"), + Field(name="BL/TH", value="MAR20"), + Field(name="RP TAG PLN", value="Rp 25.000"), + Field(name="ADMIN BANK", value="Rp 2.500"), + Field(name="TOTAL BAYAR", value="Rp 27.500"), + ], + base_price=25000, + admin_fee=2500, + client_name="TOKOPXXXX UXX", + client_number="2121212", + meter_id="2121212", + ref_id="", + max_payment=0, + min_payment=0, + ) diff --git a/tests/test_ping.py b/tests/test_ping.py new file mode 100644 index 0000000..576275d --- /dev/null +++ b/tests/test_ping.py @@ -0,0 +1,43 @@ +""" +Test the ping method of the SATClient class. + +This example shows how to use the ping method to get the ping result from the SAT server. + +For more information, please refer to "Ping" section in the SAT documentation. +""" + +from conftest import TestEnvironment +from pytest_httpserver import HTTPServer + +from py_sat import SATClient +from py_sat.constant import PING_PATH + + +def test_ping( + make_httpserver: HTTPServer, + sat_client: SATClient, + env: TestEnvironment, +): + """ + Example of ping endpoint success + + :param make_httpserver: + :param sat_client: + :param env: + """ + make_httpserver.expect_request(PING_PATH).respond_with_json( + response_json={"buildhash": "b05b97a", "sandbox": True, "status": "ok"}, + status=200, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.ping() + + if env == TestEnvironment.LOCAL: + assert response.buildhash == "b05b97a" + assert response.sandbox is True + assert response.status == "ok" + + if env == TestEnvironment.SANDBOX: + assert response.sandbox is True + assert response.status == "ok" diff --git a/tests/test_product_list.py b/tests/test_product_list.py new file mode 100644 index 0000000..c079156 --- /dev/null +++ b/tests/test_product_list.py @@ -0,0 +1,193 @@ +""" +Example of using product list endpoint + +This example shows how to use the product list endpoint to get a list of products available in the SAT server. + +For more information, please refer to "Web Services Details" section in the SAT documentation. +""" + +from pytest_httpserver import HTTPServer +from pytest_httpserver.httpserver import HandlerType + +from py_sat import SATClient +from py_sat.constant import PRODUCT_LIST_PATH +from py_sat.models import ErrorObject, ErrorResponse, ProductStatus + + +def test_product_list_success( + sat_client: SATClient, + make_httpserver: HTTPServer, +): + """ + Example of product list endpoint success + + :param sat_client: + :param make_httpserver: + """ + make_httpserver.expect_request( + PRODUCT_LIST_PATH, + method="GET", + headers={ + "authorization": "Bearer testingToken", + }, + handler_type=HandlerType.ONESHOT, + ).respond_with_json( + response_json={ + "data": [ + { + "attributes": { + "is_inquiry": False, + "price": 24913, + "product_name": "XL 25,000", + "status": 1, + }, + "id": "25k-xl", + "type": "product", + }, + { + "attributes": { + "is_inquiry": False, + "price": 25100, + "product_name": "Three 25,000", + "status": 1, + }, + "id": "25k-three", + "type": "product", + }, + { + "attributes": { + "is_inquiry": False, + "price": 0, + "product_name": "test-product", + "status": 1, + }, + "id": "test-1", + "type": "product", + }, + { + "attributes": { + "is_inquiry": False, + "price": 0, + "product_name": "", + "status": 1, + }, + "id": "test2", + "type": "product", + }, + { + "attributes": { + "is_inquiry": False, + "price": 0, + "product_name": "", + "status": 1, + }, + "id": "test4", + "type": "product", + }, + { + "attributes": { + "is_inquiry": True, + "price": 4797124, + "product_name": "Bank DBS KTA Bank DBS", + "status": 1, + }, + "id": "kta-bank-dbs", + "type": "product", + }, + ] + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.list_product() + assert response.is_success() + assert response.get_raw_response().status_code == 200 + assert len(response.products) > 3 + for product in response.products: + assert product.id + assert product.status == ProductStatus.Active + + +def test_product_list_one_item( + sat_client: SATClient, + make_httpserver: HTTPServer, +): + """ + Example of product list endpoint with one item + + :param sat_client: + :param make_httpserver: + """ + make_httpserver.expect_request( + PRODUCT_LIST_PATH, + method="GET", + headers={ + "authorization": "Bearer testingToken", + }, + handler_type=HandlerType.ONESHOT, + ).respond_with_json( + response_json={ + "data": [ + { + "attributes": { + "is_inquiry": False, + "price": 24913, + "product_name": "XL 25,000", + "status": 1, + }, + "id": "25k-xl", + "type": "product", + } + ] + }, + status=200, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.list_product("25k-xl") + assert response.is_success() + assert response.get_raw_response().status_code == 200 + assert len(response.products) == 1 + for product in response.products: + assert product.id + assert product.status == ProductStatus.Active + + +def test_product_not_exists( + sat_client: SATClient, + make_httpserver: HTTPServer, +): + """ + Example of product list endpoint with product not exists + + :param sat_client: + :param make_httpserver: + """ + make_httpserver.expect_request( + PRODUCT_LIST_PATH, + method="GET", + headers={ + "authorization": "Bearer testingToken", + }, + handler_type=HandlerType.ONESHOT, + ).respond_with_json( + response_json={ + "errors": [{"code": "P04", "detail": "Product not found", "status": "400"}] + }, + status=400, + headers={"Content-Type": "application/json"}, + ) + + response = sat_client.list_product("not-exists-product") + assert not response.is_success() + assert response.get_raw_response().status_code == 400 + assert response == ErrorResponse( + errors=[ + ErrorObject( + code="P04", + detail="Product not found", + status="400", + ) + ] + ) diff --git a/tests/test_signature.py b/tests/test_signature.py new file mode 100644 index 0000000..1019bef --- /dev/null +++ b/tests/test_signature.py @@ -0,0 +1,44 @@ +""" +Test signature module. + +This example shows how to use the signature module to sign and verify a message. + +For more information, please refer to "Signature" section in the SAT documentation. +""" + +import pytest + +from py_sat import SATClient, SATClientConfig + + +@pytest.mark.parametrize( + "test_input, test_message, verified", + [ + ("Hello, World!", "signature", False), + ("Hello, World!", "Hello, World! ", False), + ( + '{"test_message": "Hello, World!!"}', + '{"test_message": "Hello, World!!"}', + True, + ), + ( + '{"test_message": "Hello, World!!"}', + '{"test_message": "Hello, Not World!!"}', + False, + ), + ], +) +def test_signature(test_input, test_message, verified, sat_client: SATClient): + """ + Test signature module. + + :param test_input: + :param test_message: + :param verified: + :param sat_client: + """ + signature = sat_client.signature.sign(test_input) + assert signature + + is_verified = sat_client.signature.verify(test_message, signature) + assert verified == is_verified diff --git a/tox.ini b/tox.ini new file mode 100644 index 0000000..06ab4e8 --- /dev/null +++ b/tox.ini @@ -0,0 +1,27 @@ +[tox] +require= + tox>4 +envlist = + lint + py37 + py38 + py39 + py310 +isolated_build = True +skip_missing_interpreters = False + +[testenv] +pip_version = pip==22.0.4 +deps = + .[test] +commands = + pytest {posargs: tests} + +[testenv:lint] +description = Run black to check code formatting +deps = + black + isort +commands = + black . + isort . \ No newline at end of file