diff --git a/README.md b/README.md index feec78c..e2d7d78 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,27 @@ Run all e2e/functional/current tests (ksef) $ TESTS_MARKERS="init_signed and functional and not ignore" ./run_tests.sh ``` +## Example + +See [tests/test_e2e.py](tests/test_e2e.py) for detailed examples of usage. + +```Python +from ksef_utils.server import KSEFServer, KSEFService +from ksef_utils.config import TestConfig, DemoConfig, ProdConfig + +config = TestConfig() +server = KSEFServer(config) +service = KSEFService(service) + +invoice_data = { + # ... + # see tests/conftest.py for example json invoice +} + +session_token = service.init_signed() +response_send_invoice = service.send_invoice(**invoice_data) +``` + ## OpenAPI ``` diff --git a/docs/index.rst b/docs/index.rst index b41da88..9aa707c 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -3,9 +3,11 @@ You can adapt this file completely to your liking, but it should at least contain the root `toctree` directive. -Welcome to KSeF-utils's documentation! +KSeF-utils ====================================== +Python client for `KSeF `_ system. + .. toctree:: :maxdepth: 2 :caption: Contents: @@ -14,6 +16,10 @@ Welcome to KSeF-utils's documentation! :target: https://ksef-utils.readthedocs.io/en/latest/?badge=latest :alt: Documentation Status +* `Krajowy System e-Faktur (KSeF) `_ +* `ksef-utils @ GitHub `_ +* `ksef-utils @ PyPI `_ + Indices and tables ================== diff --git a/ksef_utils/server.py b/ksef_utils/server.py index aa0a148..1405d83 100644 --- a/ksef_utils/server.py +++ b/ksef_utils/server.py @@ -1,8 +1,8 @@ from time import sleep from datetime import datetime -import json -import base64 -import hashlib +from json import dumps +from base64 import b64encode +from hashlib import sha256 import requests from ksef_utils.utils import ( render_template, @@ -198,7 +198,20 @@ def send_invoice(self, data: dict, session_token: str): "SessionToken": session_token, "Content-Type": "application/json", } - response = requests.put(url, data=json.dumps(data), headers=headers) + response = requests.put(url, data=dumps(data), headers=headers) + return response + + def post_payment_identifier( + self, session_token, ksef_reference_list: list[str] + ): + data = {"ksefReferenceNumberList": ksef_reference_list} + headers = { + "accept": "application/json", + "SessionToken": session_token, + "Content-Type": "application/json", + } + url = f"{self.config.URL}/api/online/Payment/Identifier/Request" + response = requests.post(url, data=data, headers=headers) return response @@ -210,7 +223,7 @@ def init_session(self): response_json = self.server.authorization_challenge() challenge = response_json.get("challenge") if not challenge: - print(json.dumps(response_json, indent=4)) + print(dumps(response_json, indent=4)) raise Exception( response_json.get("exception").get("exceptionDetailList") ) @@ -232,7 +245,7 @@ def init_signed(self): response_json = self.server.authorization_challenge() challenge = response_json.get("challenge") if not challenge: - print(json.dumps(response_json, indent=4)) + print(dumps(response_json, indent=4)) raise Exception( response_json.get("exception").get("exceptionDetailList") ) @@ -259,7 +272,7 @@ def wait_until_logged(self): logged = False while not logged: response = self.server.get_status(self.init_token) - print(json.dumps(response.json(), indent=4)) + print(dumps(response.json(), indent=4)) if response.json().get("processingCode") == 310: logged = True sleep(1) @@ -273,13 +286,13 @@ def send_invoice(self, **kwargs): "hashSHA": { "algorithm": "SHA-256", "encoding": "Base64", - "value": base64.b64encode( - hashlib.sha256(invoice_encoded).digest() - ).decode("utf-8"), + "value": b64encode(sha256(invoice_encoded).digest()).decode( + "utf-8" + ), }, } invoice_payload = { - "invoiceBody": base64.b64encode(invoice_encoded).decode("utf-8"), + "invoiceBody": b64encode(invoice_encoded).decode("utf-8"), "type": "plain", } data = {"invoiceHash": invoice_hash, "invoicePayload": invoice_payload} @@ -290,7 +303,7 @@ def wait_until_invoice(self, reference_number): while not invoice_status: response = self.get_invoice_status(reference_number) print(response.status_code) - print(json.dumps(response.json(), indent=4)) + print(dumps(response.json(), indent=4)) invoice_status = response.json().get("invoiceStatus") if not invoice_status.get("ksefReferenceNumber"): invoice_status = {} @@ -330,6 +343,10 @@ def wait_until_token(self, element_reference_number): sleep(1) return response.json() + def post_payment_identifier(self): + response = self.server.post_payment_identifier(self.init_token, []) + return response.json() + class KSEFUtils: @staticmethod diff --git a/tests/conftest.py b/tests/conftest.py index a5de80e..6960a8b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,5 @@ import logging +from datetime import datetime from os import getenv from pytest import fixture from ksef_utils.server import KSEFServer, KSEFService @@ -42,3 +43,67 @@ def debug_requests(): debug_requests() + + +@fixture +def invoice_data(config): + creation_date = datetime.now(config.TIMEZONE).isoformat( + timespec="milliseconds" + ) + return { + "vendor1": { + "nip": config.KSEF_NIP, + "name": "KSeF-utils Kraków", + "address1": "ul. Github 11/12", + "address2": "40-100 Kraków", + "country_code": "PL", + "contact": { + "email": "test.ksef-utils@test.speedwell.pl", + "phone": 12121212121212, + }, + "bank_account": { + "nr": "PL11111111111111111111111", + "swift": "KSEFTEST", + "bank_name": "KSeF Bank", + "description": "Business account", + "type": 1, + }, + }, + "vendor2": { + "nip": 2222222239, + "name": "KSeF-utils Kraków", + "address1": "ul. Github 11/12", + "address2": "40-100 Kraków", + "country_code": "PL", + }, + "invoice": { + "creation_date": creation_date, + "currency": "PLN", + "type": "VAT", + "number": "FV-2023/12/10", + "location": "Kraków", + "date_of_sale": "2023-12-11", + "services": [ + { + "number": "1", + "gross_value": "", + "net_price": "100", + "net_value": "200", + "quantity": "2", + "title": "Service", + "vat": "23", + "vat_value": "", + "unit": "szt", + } + ], + "total_gross_value": "246", + "total_value": "200", + "total_vat_value": "46", + "payment": { + "due_date": "2023-12-20", + "description": "10 days", + "form": 6, + }, + "footer_note": "Tests: https://github.com/pprzetacznik/ksef-utils", + }, + } diff --git a/tests/features/e2e.feature b/tests/features/e2e.feature index 28babe1..571fc1c 100644 --- a/tests/features/e2e.feature +++ b/tests/features/e2e.feature @@ -6,3 +6,4 @@ Feature: KSeF Web API Given signed in using cert When generate token Then sign in using token + Then send an invoice diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 765e9d7..780353f 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -34,3 +34,16 @@ def then_sign_in_token(config, service, testing_context): config.KSEF_TOKEN = testing_context.get("authorisationToken") response = service.init_session() print(response) + + +@then("send an invoice") +def then_sign_in_token(config, service, invoice_data): + response_send_invoice = service.send_invoice(**invoice_data) + print(response_send_invoice.status_code) + print(dumps(response_send_invoice.json(), indent=4)) + reference_number = response_send_invoice.json().get( + "elementReferenceNumber" + ) + invoice_status = service.wait_until_invoice(reference_number) + invoice = service.get_invoice(invoice_status.get("ksefReferenceNumber")) + print(invoice) diff --git a/tests/test_ksef.py b/tests/test_ksef.py index 63de201..c95e874 100644 --- a/tests/test_ksef.py +++ b/tests/test_ksef.py @@ -26,7 +26,6 @@ def test_get_session(service, server, config): print(dumps(response.json(), indent=4)) -@mark.current @mark.functional def test_auth(config, service): response = service.init_session() @@ -41,70 +40,6 @@ def test_init_session(config, service): print(dumps(response, indent=4)) -@fixture -def invoice_data(config): - creation_date = datetime.now(config.TIMEZONE).isoformat( - timespec="milliseconds" - ) - return { - "vendor1": { - "nip": config.KSEF_NIP, - "name": "KSeF-utils Kraków", - "address1": "ul. Github 11/12", - "address2": "40-100 Kraków", - "country_code": "PL", - "contact": { - "email": "test.ksef-utils@test.speedwell.pl", - "phone": 12121212121212, - }, - "bank_account": { - "nr": "PL11111111111111111111111", - "swift": "KSEFTEST", - "bank_name": "KSeF Bank", - "description": "Business account", - "type": 1, - }, - }, - "vendor2": { - "nip": 2222222239, - "name": "KSeF-utils Kraków", - "address1": "ul. Github 11/12", - "address2": "40-100 Kraków", - "country_code": "PL", - }, - "invoice": { - "creation_date": creation_date, - "currency": "PLN", - "type": "VAT", - "number": "FV-2023/12/10", - "location": "Kraków", - "date_of_sale": "2023-12-11", - "services": [ - { - "number": "1", - "gross_value": "", - "net_price": "100", - "net_value": "200", - "quantity": "2", - "title": "Service", - "vat": "23", - "vat_value": "", - "unit": "szt", - } - ], - "total_gross_value": "246", - "total_value": "200", - "total_vat_value": "46", - "payment": { - "due_date": "2023-12-20", - "description": "10 days", - "form": 6, - }, - "footer_note": "Tests: https://github.com/pprzetacznik/ksef-utils", - }, - } - - @mark.functional def test_send_invoice(service, invoice_data): response = service.init_session() @@ -138,7 +73,6 @@ def test_send_invoice_signed(service, invoice_data): print(dumps(response, indent=4)) -# @mark.current def test_init_sign_request(service, invoice_data): session_token = service.init_signed() print(f"session_token: {session_token}") @@ -151,3 +85,10 @@ def test_init_sign_request(service, invoice_data): invoice_status = service.wait_until_invoice(reference_number) invoice = service.get_invoice(invoice_status.get("ksefReferenceNumber")) print(invoice) + + +@mark.current +def test_payment_identifier(config, service): + service.init_signed() + response = service.post_payment_identifier() + print(dumps(response, indent=4))