import logging
from typing import Type
from lxml import etree
from lxml.builder import E, ElementMaker
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_cores.SymXchange.models import (
AdministrativeCredentials,
CardCredentials,
Credentials,
DeviceInformation,
UserNumberCredentials,
namespaces,
)
EMPTY_MOCK_RESPONSE = (
'<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">'
"<soap:Body>"
"</soap:Body>"
"</soap:Envelope>"
)
[docs]
class SymXchangeBaseQuery(BaseQuery):
def __init__(
self,
logger: logging.Logger,
credentials: Type[Credentials],
device_info: DeviceInformation,
mock_response_string: str = None,
mock_failure=False,
):
# SOAP headers
self.headers = []
self._credentials = credentials
self._device_info = device_info
self._xml = None
self.mock_response_string = (
mock_response_string if mock_response_string else EMPTY_MOCK_RESPONSE
)
super().__init__(logger, mock_failure)
@property
def service(self) -> str:
raise NotImplementedError()
@property
def device_info(self) -> DeviceInformation:
return self._device_info
[docs]
def create_envelope(self, body: list, nsmap: dict = None) -> ElementMaker:
soap_map = nsmap or {} | {
"soap": namespaces.SOAPENV,
"com": namespaces.SYMXCHANGE_COMMON,
}
soap = ElementMaker(namespace=namespaces.SOAPENV, nsmap=soap_map)
if not isinstance(body, list):
body = [body]
return soap.Envelope(soap.Header(*self.headers), soap.Body(*body))
[docs]
def create_device_info(self) -> ElementMaker:
return E.DeviceInformation(
"",
{
"DeviceType": self.device_info.device_type,
"DeviceNumber": self.device_info.device_number,
},
)
[docs]
def create_credentials(self) -> ElementMaker:
def credentials_element(credentials: Type[Credentials]) -> ElementMaker:
match credentials: # noqa: E999
case CardCredentials() as var:
return E.CardCredentials(E.CardNumber(var.card_number))
case AdministrativeCredentials() as var:
return E.AdministrativeCredentials(E.Password(var.password))
case UserNumberCredentials() as var:
return E.UserNumberCredentials(
E.UserNumber(str(var.user_number)), E.Password(var.password)
)
case _:
raise NotImplementedError("Credentials type not implemented")
return E.Credentials(credentials_element(self._credentials))
[docs]
def serialize(self, envelope) -> str:
# 1. Wrapping SOAP Envelope in <PassThru service="{wsdl_name}">...</PassThru>
# 2. Base64 encoding the inner message
# Example: <PassThru service="account">eG1s</PassThru>
self._xml = etree.tostring(envelope, encoding="unicode")
encoded_xml = self.encode_pass_thru_data(self._xml)
return f'<PassThru service="{self.service}">{encoded_xml}</PassThru>'
def __str__(self) -> str:
return self._xml
[docs]
def mock_response(self):
return self.mock_response_string