200 lines
5.3 KiB
Python
200 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from xml.dom.minidom import getDOMImplementation, parseString
|
|
|
|
from ..exceptions import exception_classes, OMResponseException
|
|
from ..types import cast_dict_to_childtype
|
|
|
|
|
|
class Message:
|
|
"""
|
|
Base message class
|
|
|
|
:param name: Name of the message
|
|
:param attrs: Message attributes
|
|
:param childs: Message children
|
|
"""
|
|
|
|
# Fields defined by the base type class
|
|
BASE_FIELDS = {}
|
|
# Fields defined by subclasses
|
|
FIELDS = {}
|
|
# Child types
|
|
CHILDS = {}
|
|
# Fields dicts consist of the field name as name and the field type as value
|
|
# Use None if the field type is unknown, any type is allowed then
|
|
|
|
|
|
class Childs:
|
|
"""
|
|
Contains message childs
|
|
"""
|
|
|
|
CHILDS = {}
|
|
|
|
def __init__(self, child_types, child_dict):
|
|
self.CHILDS = child_types
|
|
self._child_dict = child_dict
|
|
|
|
def __getattr__(self, name):
|
|
if name in self.CHILDS.keys():
|
|
return self._child_dict.get(name)
|
|
else:
|
|
raise AttributeError()
|
|
|
|
def __setattr__(self, name, value):
|
|
if name in self.CHILDS.keys():
|
|
if self.CHILDS[name] is not None and type(value) != self.CHILDS[name]:
|
|
raise TypeError()
|
|
self._child_dict[name] = value
|
|
else:
|
|
object.__setattr__(self, name, value)
|
|
|
|
|
|
def __init__(self, name=None, attrs={}, childs={}):
|
|
self.name = name
|
|
if not self.name:
|
|
self.name = self.__class__.__name__
|
|
self._attrs = attrs
|
|
self._childs = childs
|
|
self.childs = self.Childs(self.CHILDS, self._childs)
|
|
|
|
def __getattr__(self, name):
|
|
fields = self.FIELDS | self.BASE_FIELDS
|
|
if name in fields.keys():
|
|
return self._attrs.get(name)
|
|
else:
|
|
raise AttributeError()
|
|
|
|
def __setattr__(self, name, value):
|
|
fields = self.FIELDS | self.BASE_FIELDS
|
|
if name in fields.keys():
|
|
if fields[name] is not None and type(value) != fields[name]:
|
|
raise TypeError()
|
|
self._attrs[name] = value
|
|
else:
|
|
object.__setattr__(self, name, value)
|
|
|
|
def __repr__(self):
|
|
return "{}({}, {}, {})".format(self.__class__.__name__, self.name, repr(self._attrs), repr(self._childs))
|
|
|
|
|
|
class Request(Message):
|
|
"""
|
|
Request message type class
|
|
"""
|
|
|
|
BASE_FIELDS = {
|
|
"seq": int,
|
|
}
|
|
|
|
|
|
class Response(Message):
|
|
"""
|
|
Response message type class
|
|
"""
|
|
|
|
BASE_FIELDS = {
|
|
"seq": int,
|
|
"errCode": None,
|
|
"info": None,
|
|
"bad": None,
|
|
"maxLen": None,
|
|
}
|
|
|
|
def raise_on_error(self):
|
|
"""
|
|
Raises an exception if the response contains an error.
|
|
|
|
Usage::
|
|
|
|
>>> try:
|
|
>>> r.raise_on_error()
|
|
>>> except mitel_ommclient2.exceptions.EAuth as e:
|
|
>>> print("We don't care about authentication!")
|
|
|
|
See children of :class:`mitel_ommclient2.exceptions.OMResponseException` for all possible exceptions.
|
|
"""
|
|
|
|
if self.errCode is not None:
|
|
raise exception_classes.get(self.errCode, OMResponseException)(response=self)
|
|
|
|
|
|
REQUEST_TYPES = {}
|
|
RESPONSE_TYPES = {}
|
|
|
|
def request_type(c):
|
|
REQUEST_TYPES[c.__name__] = c
|
|
return c
|
|
|
|
def response_type(c):
|
|
RESPONSE_TYPES[c.__name__] = c
|
|
return c
|
|
|
|
from .getaccount import GetAccount, GetAccountResp
|
|
from .getppdev import GetPPDev, GetPPDevResp
|
|
from .getppuser import GetPPUser, GetPPUserResp
|
|
from .open import Open, OpenResp
|
|
from .ping import Ping, PingResp
|
|
|
|
def construct(request):
|
|
"""
|
|
Builds the XML message DOM and returns as string
|
|
"""
|
|
impl = getDOMImplementation()
|
|
message = impl.createDocument(None, request.name, None)
|
|
root = message.documentElement
|
|
|
|
for k, v in request._attrs.items():
|
|
root.setAttribute(str(k), str(v))
|
|
|
|
|
|
for k, v in request._childs.items():
|
|
child = message.createElement(k)
|
|
if v is not None:
|
|
for c_k, c_v in v.items():
|
|
child.setAttribute(str(c_k), str(c_v))
|
|
root.appendChild(child)
|
|
return root.toxml()
|
|
|
|
def parse(message):
|
|
message = parseString(message)
|
|
root = message.documentElement
|
|
|
|
name = root.tagName
|
|
attrs = {}
|
|
childs = {}
|
|
|
|
response_type = RESPONSE_TYPES.get(name)
|
|
fields = response_type.FIELDS | response_type.BASE_FIELDS
|
|
|
|
for i in range(0, root.attributes.length):
|
|
item = root.attributes.item(i)
|
|
if fields.get(item.name) is not None:
|
|
attrs[item.name] = fields[item.name](item.value)
|
|
else:
|
|
attrs[item.name] = item.value
|
|
|
|
child = root.firstChild
|
|
while child is not None:
|
|
new_child = {}
|
|
for i in range(0, child.attributes.length):
|
|
item = child.attributes.item(i)
|
|
new_child[item.name] = item.value
|
|
|
|
childname = child.tagName
|
|
|
|
# cast dict into child type
|
|
if response_type.CHILDS.get(childname) is not None:
|
|
new_child = cast_dict_to_childtype(response_type.CHILDS[childname], new_child)
|
|
|
|
if childname in childs:
|
|
childs[childname].append(new_child)
|
|
else:
|
|
childs[childname] = [new_child]
|
|
|
|
child = child.nextSibling
|
|
|
|
|
|
return response_type(name, attrs, childs)
|