From 038aaec78fb568ef5c5da83cc24a4e8fcbe3390d Mon Sep 17 00:00:00 2001 From: Piotr Przetacznik Date: Thu, 4 Jan 2024 23:47:35 +0100 Subject: [PATCH] chore: Add UPO --- README.md | 2 +- ksef_utils/server.py | 28 ++++++++++++++++++++-------- tests/features/e2e.feature | 2 ++ tests/test_e2e.py | 23 +++++++++++++++++++++-- tests/test_ksef.py | 20 ++++---------------- 5 files changed, 48 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index e2d7d78..c98c47d 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ [![PyPI - Python Version](https://img.shields.io/pypi/pyversions/ksef-utils)](https://pypi.org/project/ksef-utils/) [![Documentation Status](https://readthedocs.org/projects/ksef-utils/badge/?version=latest)](https://ksef-utils.readthedocs.io/en/latest/?badge=latest) -This project contains utilities and example requests that can be helpful when integrating with Polish central invoicing system called [Krajowy System e-Faktur (KSeF)](https://www.podatki.gov.pl/ksef/). +This project contains Python utilities and example requests that can be helpful when integrating with Polish central invoicing system called [Krajowy System e-Faktur (KSeF)](https://www.podatki.gov.pl/ksef/). ## Installation diff --git a/ksef_utils/server.py b/ksef_utils/server.py index 1405d83..981d4e8 100644 --- a/ksef_utils/server.py +++ b/ksef_utils/server.py @@ -54,10 +54,7 @@ def get_status(self, session_token): def get_upo(self, session_token, reference_number): headers = { - # "accept": "application/json", - "accept": "application/vnd.v3+json", - "SessionToken": session_token, - "Content-Type": "application/json", + "accept": "application/json", } response = requests.get( f"{self.config.URL}/api/common/Status/{reference_number}", @@ -211,7 +208,7 @@ def post_payment_identifier( "Content-Type": "application/json", } url = f"{self.config.URL}/api/online/Payment/Identifier/Request" - response = requests.post(url, data=data, headers=headers) + response = requests.post(url, json=data, headers=headers) return response @@ -309,7 +306,7 @@ def wait_until_invoice(self, reference_number): invoice_status = {} if not invoice_status: sleep(1) - return invoice_status + return response.json() def get_invoice_status(self, reference_number: str): return self.server.get_invoice_status( @@ -324,6 +321,19 @@ def get_upo(self, reference_number: str) -> str: response = self.server.get_upo(self.init_token, reference_number) return response.json() + def wait_until_upo( + self, reference_number: str, max_retries: int = 60, interval: str = 2 + ) -> str: + processing_code = 310 + while processing_code != 200 and max_retries > 0: + response = self.get_upo(reference_number) + processing_code = response.get("processingCode") + print(dumps(response, indent=4)) + if processing_code != 200: + sleep(interval) + max_retries -= 1 + return response + def generate_token(self): response = self.server.generate_token(self.init_token) return response.json() @@ -343,8 +353,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, []) + def post_payment_identifier(self, ksef_reference_list: list[str] = None): + response = self.server.post_payment_identifier( + self.init_token, ksef_reference_list + ) return response.json() diff --git a/tests/features/e2e.feature b/tests/features/e2e.feature index 571fc1c..1b5caaf 100644 --- a/tests/features/e2e.feature +++ b/tests/features/e2e.feature @@ -7,3 +7,5 @@ Feature: KSeF Web API When generate token Then sign in using token Then send an invoice + Then terminate session + Then get upo diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 780353f..1ec079e 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -1,6 +1,8 @@ from json import dumps +from base64 import b64decode from pytest import fixture from pytest_bdd import scenario, given, when, then +from ksef_utils.utils import format_xml @scenario("e2e.feature", "End to end") @@ -37,13 +39,30 @@ def then_sign_in_token(config, service, testing_context): @then("send an invoice") -def then_sign_in_token(config, service, invoice_data): +def then_send_invoice(config, service, invoice_data, testing_context): response_send_invoice = service.send_invoice(**invoice_data) print(response_send_invoice.status_code) print(dumps(response_send_invoice.json(), indent=4)) + testing_context["invoice_response"] = response_send_invoice.json() reference_number = response_send_invoice.json().get( "elementReferenceNumber" ) - invoice_status = service.wait_until_invoice(reference_number) + response = service.wait_until_invoice(reference_number) + invoice_status = response.get("invoiceStatus") invoice = service.get_invoice(invoice_status.get("ksefReferenceNumber")) print(invoice) + testing_context["invoice_response"] = response + + +@then("terminate session") +def then_terminate_session(service): + response = service.session_terminate() + print(dumps(response, indent=4)) + + +@then("get upo") +def then_get_upo(config, service, invoice_data, testing_context): + response = service.wait_until_upo( + testing_context["invoice_response"].get("referenceNumber") + ) + print(format_xml(b64decode(response.get("upo")))) diff --git a/tests/test_ksef.py b/tests/test_ksef.py index c95e874..904ba58 100644 --- a/tests/test_ksef.py +++ b/tests/test_ksef.py @@ -50,7 +50,8 @@ def test_send_invoice(service, invoice_data): reference_number = response_send_invoice.json().get( "elementReferenceNumber" ) - invoice_status = service.wait_until_invoice(reference_number) + response = service.wait_until_invoice(reference_number) + invoice_status = response.get("invoiceStatus") invoice = service.get_invoice(invoice_status.get("ksefReferenceNumber")) print(invoice) @@ -66,27 +67,14 @@ def test_send_invoice_signed(service, invoice_data): reference_number = response_send_invoice.json().get( "elementReferenceNumber" ) - invoice_status = service.wait_until_invoice(reference_number) + response = service.wait_until_invoice(reference_number) + invoice_status = response.get("invoiceStatus") invoice = service.get_invoice(invoice_status.get("ksefReferenceNumber")) print(invoice) response = service.session_terminate() print(dumps(response, indent=4)) -def test_init_sign_request(service, invoice_data): - session_token = service.init_signed() - print(f"session_token: {session_token}") - 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) - - @mark.current def test_payment_identifier(config, service): service.init_signed()