import logging
import uuid
from typing import Optional
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_cores.Symitar import data_helpers
EMPTY_MOCK_RESPONSE = """<data>
<version></version>
<sym></sym>
<ISMEMOMODE></ISMEMOMODE>
</data>"""
[docs]
class SymitarBaseQuery(BaseQuery):
def __init__(
self,
logger: logging.Logger,
customer_id: str,
unit_number: str,
device_type: str,
card_prefix: str,
repgen_name: str,
procedure: str,
data_attrs: dict = None,
guid: Optional[str] = None,
mock_response_string: str = None,
data_file_name: str = "",
rg_params: dict = None,
purge_account_cache: bool = False,
mock_failure=False,
use_symxchange=False,
):
"""
Symitar query object for building Poweron based SymConnect messages.
:param logger: Logger
:param customer_id: Symitar account number
:param unit_number: SymConnect UnitNumber
:param device_type: SymConnect DeviceType
:param card_prefix: SymConnect CardPrefix
:param repgen_name: Regpen/PowerOn name
:param procedure: RGSTATE
:param data_attrs: Key/Value pairs for letter file
:param guid: Unique indentifer for SymConnect message
:param mock_response_string: The response to emulate what Symitar would respond with
:param data_file_name: Name of the letter file that will be created if using data_attrs
:param rg_params: JRGUSERCHR1 through JRGUSERCHR5 parameters
:param purge_account_cache: Will tell the adapter to dump it's cache for this account
:param mock_failure: Simulate a Core failure
"""
self.customer_id = customer_id
self.ind_number = str(self.customer_id).zfill(10)
self.unit_number = unit_number
self.device_type = device_type
self.card_prefix = card_prefix
self.repgen_name = repgen_name
self.procedure = procedure
self.guid = guid
self.data_attrs = data_attrs if data_attrs else {}
self.mock_response_string = (
mock_response_string if mock_response_string else EMPTY_MOCK_RESPONSE
)
self.data_file_name = data_file_name
self.rg_params = rg_params if rg_params else {}
self.purge_account_cache = purge_account_cache
self.use_symxchange = use_symxchange
super().__init__(logger, mock_failure=mock_failure)
[docs]
def build(self):
if not self.guid:
self.guid = str(uuid.uuid4())
call = (
"RG~{guid!s}~A{unit_number!s}~B{device_type!s}~DCARD"
"~F{card_prefix!s}{ind_number!s}~G{repgen_name!s}"
"~JRGSESSION=10~JRGSTATE={procedure}"
).format(
guid=self.guid,
unit_number=self.unit_number,
device_type=self.device_type,
card_prefix=self.card_prefix,
ind_number=self.ind_number,
repgen_name=self.repgen_name,
procedure=self.procedure,
)
# Possible Poweron/Repgen parameters are in the format of RGUSERCHR# and RGUSERNUM#
# with # being 1 through 5
for key in sorted(self.rg_params.keys()):
upper_key = key.upper()
if "RGUSERCHR" in upper_key or "RGUSERNUM" in upper_key:
value = str(self.rg_params[key])
upper_key = upper_key.replace("JRGUSER", "RGUSER")
if value is not None and value.upper() != "NONE":
call += "~J{}={}".format(upper_key, str(value))
# PURGEACCOUNTCACHE is not part of the SymConnect protocol but a configuration
# for the Symitar adapter. Setting it will cause the adapter to dump it's cache
# for account number used within ~F{CARD_PREFIX}{10_DIGIT_ACCOUNT_NUMBER}
if self.purge_account_cache:
call += "~PURGEACCOUNTCACHE"
# RGUSERCHR1=DATAFILENAME={LETTER_FILE_NAME} is also not part of the SymConnect protocol but a
# mechanism for the Symitar adapter to create a letter file within the Symitar core.
# The file content will be anything after the first \n
payload = []
if self.data_file_name and self.data_attrs:
call = "{}~JRGUSERCHR1=DATAFILENAME={}".format(call, self.data_file_name)
for key in sorted(self.data_attrs.keys()):
value = self.data_attrs[key]
if "member_" not in key.lower():
if value is None or value.upper() == "NONE":
value = ""
payload.append(str(key) + "=" + str(value))
call += "\n" + "\n".join(payload)
if self.use_symxchange:
call = data_helpers.make_symxchange_xml(call)
return call
[docs]
def mock_response(self):
return self.mock_response_string
[docs]
def build_mock_response(response: str) -> str:
"""A helper method for constructing a valid SymConnect response from a string.
>>> build_mock_response("<data><first_name>John</first_name></data>")
'RSRG~12345~K0~JRGLINE=<data><first_name>John</first_name></data>~JRGDATATYPE=9'
:param response: A string that you want to convert to a SymConnect response
:return: A valid SymConnect response
"""
lines = response.splitlines()
response = "RSRG~12345~K0"
for line in lines:
line = line.strip()
if line:
response += f"~JRGLINE={line}"
response += "~JRGDATATYPE=9"
return response