.

xmlrpcserver/reqs/request_xml_service.py

@@ -0,0 +1,109 @@
+import xml.etree.ElementTree as ET
+from xmlrpcserver.reqs.messages import Messages
+
+
+class RequestXmlService:
+    REQUEST_NODE_NAME: str = 'request'
+    HEADER_NODE_NAME: str = 'header'
+    RESULT_NODE_NAME: str = 'result'
+    REQUEST_ID_ATTRIBUTE_NAME: str = 'parcel_id'
+    RESPONSE_ID_ATTRIBUTE_NAME: str = 'reply_parcel_id'
+    SOURCE_REQUEST_ID_ATTRIBUTE_NAME: str = 'source_parcel_id'
+    RESULT_CODE_ATTRIBUTE_NAME: str = 'result_code'
+    RESULT_MESSAGE_ATTRIBUTE_NAME: str = 'result_message'
+    HEADER_XPATH: str = HEADER_NODE_NAME
+
+    def get_request_uuid(self, document: ET.Element) -> str:
+        return self.get_header_attribute(document, self.REQUEST_ID_ATTRIBUTE_NAME)
+
+    def get_response_uuid(self, document: ET.Element) -> str:
+        return self.get_header_attribute(document, self.RESPONSE_ID_ATTRIBUTE_NAME)
+
+    def get_source_request_uuid(self, document: ET.Element) -> str:
+        return self.get_header_attribute(document, self.SOURCE_REQUEST_ID_ATTRIBUTE_NAME)
+
+    def get_header_node(self, document: ET.Element) -> ET.Element:
+        return document.find(self.HEADER_XPATH)
+
+    def get_header_attribute(self, document: ET.Element, attribute_name: str) -> str:
+        header = self.get_header_node(document)
+        if header is None:
+            raise Exception(Messages.MISSING_TAG.value)
+        result = header.get(attribute_name)
+        return result.strip() if result else ''
+
+    def get_request_document(self, request_uuid: str, response_uuid) -> ET.Element:
+        document = ET.Element(self.REQUEST_NODE_NAME)
+        header = ET.SubElement(document, self.HEADER_NODE_NAME)
+        header.set(self.REQUEST_ID_ATTRIBUTE_NAME, request_uuid.strip())
+        if response_uuid:
+            header.set(self.RESPONSE_ID_ATTRIBUTE_NAME, response_uuid.strip())
+        return document
+
+    def get_common_response_xml(self, request_uuid: str, response_uuid: str, result_code: int, result_message: str) -> ET.Element:
+        document = self.get_request_document(response_uuid, request_uuid)
+
+        result = ET.SubElement(document, self.RESULT_NODE_NAME)
+        result.set(self.RESULT_CODE_ATTRIBUTE_NAME, str(result_code))
+        result.set(self.RESULT_MESSAGE_ATTRIBUTE_NAME, result_message)
+
+        return document
+
+    def set_result(self, document: ET.Element, result_code: int, result_message: str):
+        result = document.find(self.RESULT_NODE_NAME)
+        if result is None:
+            result = ET.SubElement(document, self.RESULT_NODE_NAME)
+        result.set(self.RESULT_CODE_ATTRIBUTE_NAME, str(result_code))
+        result.set(self.RESULT_MESSAGE_ATTRIBUTE_NAME, result_message)
+
+    def set_request_uuid(self, document: ET.Element, request_uuid: str):
+        header = self.get_header_node(document)
+        header.set(self.REQUEST_ID_ATTRIBUTE_NAME, request_uuid.strip())
+
+    def set_source_request_uuid(self, document: ET.Element, source_request_uuid: str):
+        header = self.get_header_node(document)
+        header.set(self.SOURCE_REQUEST_ID_ATTRIBUTE_NAME, source_request_uuid.strip())
+
+    def get_common_request_xml(self, request_uuid: str) -> ET.Element:
+        return self.get_request_document(request_uuid, None)
+
+    def get_response_xml(self, document: ET.Element, request_uuid, response_uuid: str, result_code: int, result_message: str) -> ET.Element:
+        header = document.find(self.HEADER_NODE_NAME)
+        header.set(self.REQUEST_ID_ATTRIBUTE_NAME, response_uuid.strip())
+        header.set(self.RESPONSE_ID_ATTRIBUTE_NAME, request_uuid.strip() if request_uuid else None)
+        result = document.find(self.RESULT_NODE_NAME)
+        result.set(self.RESULT_CODE_ATTRIBUTE_NAME, str(result_code))
+        result.set(self.RESULT_MESSAGE_ATTRIBUTE_NAME, result_message)
+
+    def get_element(self, document: ET.Element, node: ET.Element, name: str) -> ET.Element:
+        for child in node:
+            if child.tag.casefold() == name.casefold():
+                return child
+        return ET.SubElement(document, name)
+
+    def validate_common_request_xml(self, xml: str):
+        if not xml:
+            raise Exception(Messages.NO_REQUEST_DATA_PROVIDED.name)
+        try:
+            document = ET.parse(xml)
+        except:
+            raise Exception(Messages.INVALID_REQUEST_DATA)
+        self.validate_common_request_xml(document)
+
+    def validate_common_request_xml(self, document):
+        if not document:
+            raise Exception(Messages.NO_REQUEST_DATA_PROVIDED.name)
+        if document.tag != self.REQUEST_NODE_NAME:
+            raise Exception(Messages.MISSING_TAG.name, self.REQUEST_NODE_NAME)
+        try:
+            request_uuid = self.get_header_attribute(document, self.REQUEST_ID_ATTRIBUTE_NAME)
+        except:
+            raise Exception(Messages.INVALID_REQUEST_DATA)
+        if not request_uuid:
+            raise Exception(Messages.MISSING_ATTRIBUTE, self.REQUEST_ID_ATTRIBUTE_NAME, self.HEADER_NODE_NAME)
+
+    def get_result_code(self, document: ET.Element):
+        result = self.get_element(document, document, self.RESULT_NODE_NAME)
+        if not result:
+            return None
+        return result.get(self.RESULT_CODE_ATTRIBUTE_NAME)