diff --git a/broadworks_ocip/api.py b/broadworks_ocip/api.py index 22496a2..841be5d 100644 --- a/broadworks_ocip/api.py +++ b/broadworks_ocip/api.py @@ -10,9 +10,10 @@ import socket import sys import uuid +from typing import Callable +from typing import Dict -from classforge import Class -from classforge import Field +import attr from lxml import etree import broadworks_ocip.base @@ -25,7 +26,8 @@ VERBOSE_DEBUG = 9 -class BroadworksAPI(Class): +@attr.s(slots=True, kw_only=True) +class BroadworksAPI: """ BroadworksAPI - A class encapsulating the Broadworks OCI-P API @@ -47,19 +49,19 @@ class BroadworksAPI(Class): """ - host: str = Field(type=str, required=True, mutable=False) - port: int = Field(type=int, default=2208, mutable=False) - username: str = Field(type=str, required=True, mutable=False) - password: str = Field(type=str, required=True, mutable=False) - logger = Field(type=logging.Logger) - authenticated: bool = Field(type=bool, default=False) - connect_timeout: int = Field(type=int, default=8) - command_timeout: int = Field(type=int, default=30) - socket = Field(type=socket.socket, default=None) # type: socket.socket - session_id: str = Field(type=str) - _despatch_table = Field(type=dict) - - def on_init(self): + host: str = attr.ib() + port: int = attr.ib(default=2208) + username: str = attr.ib() + password: str = attr.ib() + logger: logging.Logger = attr.ib(default=None) + authenticated: bool = attr.ib(default=False) + connect_timeout: int = attr.ib(default=8) + command_timeout: int = attr.ib(default=30) + socket = attr.ib(default=None) + session_id: str = attr.ib(default=None) + _despatch_table: Dict[str, Callable] = attr.ib(default=None) + + def __attrs_post_init__(self): """ Initialise the API object. diff --git a/broadworks_ocip/base.py b/broadworks_ocip/base.py index a34c2e3..8ebf578 100644 --- a/broadworks_ocip/base.py +++ b/broadworks_ocip/base.py @@ -6,12 +6,15 @@ """ import re from collections import namedtuple +from typing import Any +from typing import Dict +from typing import Optional +from typing import Tuple import attr -from classforge import Class -from classforge import Field from lxml import etree +from broadworks_ocip.exceptions import OCIErrorAPISetup from broadworks_ocip.exceptions import OCIErrorResponse @@ -41,19 +44,32 @@ class ElementInfo: is_table: bool = attr.ib(default=False) -class OCIType(Class): +@attr.s(slots=True, frozen=True, kw_only=True) +class OCIType: """ OCIType - Base type for all the OCI-P component classes """ # Namespace maps used for various XML build tasks - DEFAULT_NSMAP = {None: "", "xsi": "http://www.w3.org/2001/XMLSchema-instance"} - DOCUMENT_NSMAP = {None: "C", "xsi": "http://www.w3.org/2001/XMLSchema-instance"} - ERROR_NSMAP = { - "c": "C", - None: "", - "xsi": "http://www.w3.org/2001/XMLSchema-instance", - } + @classmethod + def _default_nsmap(cls): + return {None: "", "xsi": "http://www.w3.org/2001/XMLSchema-instance"} + + @classmethod + def _document_nsmap(cls): + return {None: "C", "xsi": "http://www.w3.org/2001/XMLSchema-instance"} + + @classmethod + def _error_nsmap(cls): + return { + "c": "C", + None: "", + "xsi": "http://www.w3.org/2001/XMLSchema-instance", + } + + @classmethod + def _elements(cls) -> Tuple[ElementInfo, ...]: + raise OCIErrorAPISetup(message="_elements should be defined in the subclass.") @property def type_(self): @@ -81,7 +97,7 @@ def etree_components_(self, name=None): """ if name is None: name = self.type_ - element = etree.Element(name, nsmap=self.DEFAULT_NSMAP) + element = etree.Element(name, nsmap=self._default_nsmap()) return self.etree_sub_components_(element) def etree_sub_components_(self, element: "etree._Element"): @@ -94,7 +110,7 @@ def etree_sub_components_(self, element: "etree._Element"): Returns: etree: etree.Element() for this class """ - for sub_element in self._ELEMENTS: + for sub_element in self._elements(): value = getattr(self, sub_element.name) if sub_element.is_array: if value is not None: @@ -127,7 +143,7 @@ def etree_sub_element_( element, sub_element.xmlname, {"{http://www.w3.org/2001/XMLSchema-instance}nil": "true"}, - nsmap=self.DEFAULT_NSMAP, + nsmap=self._default_nsmap(), ) elif sub_element.is_table: # any table should be a list of namedtuple elements @@ -135,7 +151,7 @@ def etree_sub_element_( elem = etree.SubElement( element, sub_element.xmlname, - nsmap=self.DEFAULT_NSMAP, + nsmap=self._default_nsmap(), ) first = value[0] for col in first._fields: @@ -150,14 +166,14 @@ def etree_sub_element_( elem = etree.SubElement( element, sub_element.xmlname, - nsmap=self.DEFAULT_NSMAP, + nsmap=self._default_nsmap(), ) value.etree_sub_components_(elem) else: elem = etree.SubElement( element, sub_element.xmlname, - nsmap=self.DEFAULT_NSMAP, + nsmap=self._default_nsmap(), ) if sub_element.type == bool: elem.text = "true" if value else "false" @@ -270,7 +286,7 @@ def build_from_etree_(cls, element: "etree._Element"): results: Object instance for this class """ initialiser = {} - for elem in cls._ELEMENTS: + for elem in cls._elements(): if elem.is_array: result = [] nodes = element.findall(elem.xmlname) @@ -289,7 +305,31 @@ def build_from_etree_(cls, element: "etree._Element"): # use that to build a new object return cls(**initialiser) + def to_dict(self) -> Dict[str, Any]: + """ + Convert object to dict representation of itself + This was provided as part of the Classforge system, which we have moved away + from, so this is a local re-implementation. This is only used within the test + suite at present. + """ + elements = {} + for elem in self._elements(): + value = getattr(self, elem.name) + if elem.is_table: + pass + elif elem.is_complex: + if elem.is_array: + value = [x.to_dict() for x in value] + else: + value = value.to_dict() + elif elem.is_array: + value = [x for x in value] + elements[elem.name] = value + return elements + + +@attr.s(slots=True, frozen=True, kw_only=True) class OCICommand(OCIType): """ OCICommand - base class for all OCI Command (Request/Response) types @@ -302,7 +342,7 @@ class OCICommand(OCIType): there to give a known value for testing. """ - session_id: str = Field(type=str, default="00000000-1111-2222-3333-444444444444") + session_id: str = attr.ib(default="00000000-1111-2222-3333-444444444444") def build_xml_(self): """ @@ -317,11 +357,11 @@ def build_xml_(self): root = etree.Element( "{C}BroadsoftDocument", {"protocol": "OCI"}, - nsmap=self.DOCUMENT_NSMAP, + nsmap=self._document_nsmap(), ) # # add the session - session = etree.SubElement(root, "sessionId", nsmap=self.DEFAULT_NSMAP) + session = etree.SubElement(root, "sessionId", nsmap=self._default_nsmap()) session.text = self.session_id # # and the command @@ -350,7 +390,7 @@ def build_xml_command_element_(self, root: "etree._Element"): root, "command", {"{http://www.w3.org/2001/XMLSchema-instance}type": self.type_}, - nsmap=self.DEFAULT_NSMAP, + nsmap=self._default_nsmap(), ) @classmethod @@ -368,7 +408,20 @@ def build_from_etree_non_parameters_( if node is not None: initialiser["session_id"] = node.text + def to_dict(self) -> Dict[str, Any]: + """ + Convert object to dict representation of itself + + This was provided as part of the Classforge system, which we have moved away + from, so this is a local re-implementation. This is only used within the test + suite at present. + """ + elements = super().to_dict() # pick up the base object data + elements["session_id"] = self.session_id + return elements + +@attr.s(slots=True, frozen=True, kw_only=True) class OCIRequest(OCICommand): """ OCIRequest - base class for all OCI Command Request types @@ -377,6 +430,7 @@ class OCIRequest(OCICommand): pass +@attr.s(slots=True, frozen=True, kw_only=True) class OCIResponse(OCICommand): """ OCIResponse - base class for all OCI Command Response types @@ -393,19 +447,23 @@ def build_xml_command_element_(self, root: "etree._Element"): root, "command", {"echo": "", "{http://www.w3.org/2001/XMLSchema-instance}type": self.type_}, - nsmap=self.DEFAULT_NSMAP, + nsmap=self._default_nsmap(), ) +@attr.s(slots=True, frozen=True, kw_only=True) class SuccessResponse(OCIResponse): """ The SuccessResponse is concrete response sent whenever a transaction is successful and does not return any data. """ - _ELEMENTS = () # type: ignore # type: Tuple[Tuple] + @classmethod + def _elements(cls) -> Tuple[ElementInfo, ...]: + return () +@attr.s(slots=True, frozen=True, kw_only=True) class ErrorResponse(OCIResponse): """ The ErrorResponse is concrete response sent whenever a transaction fails @@ -415,18 +473,21 @@ class ErrorResponse(OCIResponse): `OCIErrorResponse` exception is raised in `post_xml_decode_`. """ - _ELEMENTS = ( - ElementInfo("error_code", "errorCode", int), - ElementInfo("summary", "summary", str, is_required=True), - ElementInfo("summary_english", "summaryEnglish", str, is_required=True), - ElementInfo("detail", "detail", str), - ElementInfo("type", "type", str), - ) - error_code = Field(type=int, required=False) - summary = Field(type=str, required=True) - summary_english = Field(type=str, required=True) - detail = Field(type=str, required=False) - type = Field(type=str, required=False) + error_code: Optional[int] = attr.ib(default=None) + summary: str = attr.ib() + summary_english: str = attr.ib() + detail: Optional[str] = attr.ib(default=None) + type: Optional[str] = attr.ib(default=None) + + @classmethod + def _elements(cls) -> Tuple[ElementInfo, ...]: + return ( + ElementInfo("error_code", "errorCode", int), + ElementInfo("summary", "summary", str, is_required=True), + ElementInfo("summary_english", "summaryEnglish", str, is_required=True), + ElementInfo("detail", "detail", str), + ElementInfo("type", "type", str), + ) def post_xml_decode_(self): """Raise an exception as this is an error""" @@ -444,7 +505,7 @@ def build_xml_command_element_(self, root): "echo": "", "{http://www.w3.org/2001/XMLSchema-instance}type": "c:" + self.type_, }, - nsmap=self.ERROR_NSMAP, + nsmap=self._error_nsmap(), ) diff --git a/broadworks_ocip/exceptions.py b/broadworks_ocip/exceptions.py index 73901f7..e27f2d6 100644 --- a/broadworks_ocip/exceptions.py +++ b/broadworks_ocip/exceptions.py @@ -22,6 +22,7 @@ def __str__(self): return f"{self.__class__.__name__}({self.message})" +@attr.s(slots=True, frozen=True) class OCIErrorResponse(OCIError): """ Exception raised when an ErrorResponse is received and decoded. @@ -32,6 +33,7 @@ class OCIErrorResponse(OCIError): pass +@attr.s(slots=True, frozen=True) class OCIErrorTimeOut(OCIError): """ Exception raised when nothing is head back from the server. @@ -42,6 +44,7 @@ class OCIErrorTimeOut(OCIError): pass +@attr.s(slots=True, frozen=True) class OCIErrorUnknown(OCIError): """ Exception raised when life becomes too much for the software. @@ -52,4 +55,15 @@ class OCIErrorUnknown(OCIError): pass +@attr.s(slots=True, frozen=True) +class OCIErrorAPISetup(OCIError): + """ + Exception raised when life becomes too much for the software. + + Subclass of OCIError() + """ + + pass + + # end diff --git a/poetry.lock b/poetry.lock index 7586db0..fa41458 100644 --- a/poetry.lock +++ b/poetry.lock @@ -82,17 +82,6 @@ python-versions = ">=3.5.0" [package.extras] unicode_backport = ["unicodedata2"] -[[package]] -name = "classforge" -version = "0.92" -description = "A new python object system" -category = "main" -optional = false -python-versions = ">=3.6" - -[package.dependencies] -PyYAML = ">=5.3.1" - [[package]] name = "click" version = "8.0.3" @@ -611,7 +600,7 @@ python-versions = "*" name = "pyyaml" version = "6.0" description = "YAML parser and emitter for Python" -category = "main" +category = "dev" optional = false python-versions = ">=3.6" @@ -883,7 +872,7 @@ testing = ["pytest (>=4.6)", "pytest-checkdocs (>=2.4)", "pytest-flake8", "pytes [metadata] lock-version = "1.1" python-versions = ">=3.6.2,<4.0" -content-hash = "2da26844a52ac2332d7a6f1840dca8ff7f705fd665a5caa45b369156d2c5b265" +content-hash = "5d61d1f23cddcd8a3dca1b3d06955b1d8f0080f493be3c73cffeb7063604c044" [metadata.files] astunparse = [ @@ -966,9 +955,6 @@ charset-normalizer = [ {file = "charset-normalizer-2.0.10.tar.gz", hash = "sha256:876d180e9d7432c5d1dfd4c5d26b72f099d503e8fcc0feb7532c9289be60fcbd"}, {file = "charset_normalizer-2.0.10-py3-none-any.whl", hash = "sha256:cb957888737fc0bbcd78e3df769addb41fd1ff8cf950dc9e7ad7793f1bf44455"}, ] -classforge = [ - {file = "classforge-0.92.tar.gz", hash = "sha256:eb9a5d14b2d5d9fa78e761d1d51ae3b7e1c564efd7fe59884a30158c0972dab7"}, -] click = [ {file = "click-8.0.3-py3-none-any.whl", hash = "sha256:353f466495adaeb40b6b5f592f9f91cb22372351c84caeb068132442a4518ef3"}, {file = "click-8.0.3.tar.gz", hash = "sha256:410e932b050f5eed773c4cda94de75971c89cdb3155a72a0831139a79e5ecb5b"}, diff --git a/pyproject.toml b/pyproject.toml index d5385f3..dedac7f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,6 @@ exclude = [ [tool.poetry.dependencies] python = ">=3.6.2,<4.0" -classforge = "^0.92" lxml = "^4.7.1" attrs = "^21.4.0"