chore: Add UPO

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
 

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()
 
 

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

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"))))

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()