Pull request #5: vhallivu/xr_pytest_with_http_mock

pytz==2021.3

redis==4.1.2

requests==2.27.1

requests-mock==1.9.3

xmltodict==0.12.0

# pip's dependency resolver does not take into account installed packages.

        self.__started = threading.Event()

        self.__terminate = threading.Event()

        self.__timeout = int(settings.get('timeout', 120))

        self.__cm_address = address

        # Mandatory key, an exception will get thrown if missing

        self.__hub_module_name = settings["hub_module_name"]

        LOGGER.info(f"XrDriver instantiated, cm {address}:{port}, {settings=}")

    def __str__(self):

        return f"{self.__hub_module_name}@{self.__cm_address}"

    def Connect(self) -> bool:

        LOGGER.info(f"Connect[{self}]")

        with self.__lock:

            if self.__started.is_set():

                return True

                return True

    def Disconnect(self) -> bool:

        LOGGER.info(f"Disconnect[{self}]")

        with self.__lock:

            self.__terminate.set()

            return True

    def GetInitialConfig(self) -> List[Tuple[str, Any]]:

        LOGGER.info(f"GetInitialConfig[{self}]")

        with self.__lock:

            return []

    #pylint: disable=dangerous-default-value

    def GetConfig(self, resource_keys : List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:

        LOGGER.info(f"GetConfig[{self}]: {resource_keys=}")

        chk_type('resources', resource_keys, list)

        # Empty resource_keys means all resources. As we only have endpoints, we ignore parameter and always

        # return everything.

        with self.__lock:

            constellation = self.__cm_connection.get_constellation_by_hub_name(self.__hub_module_name)

            if constellation:

                self.__constellation = constellation

            else:

                return []

    def SetConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:

        LOGGER.info(f"SetConfig {resources=}")

        LOGGER.info(f"SetConfig[{self}]: {resources=}")

        # Logged config seems like:

        #[('/service[52ff5f0f-fda4-40bd-a0b1-066f4ff04079:optical]', '{"capacity_unit": "GHz", "capacity_value": 1, "direction": "UNIDIRECTIONAL", "input_sip": "XR HUB 1|XR-T4", "layer_protocol_name": "PHOTONIC_MEDIA", "layer_protocol_qualifier": "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_NMC", "output_sip": "XR LEAF 1|XR-T1", "uuid": "52ff5f0f-fda4-40bd-a0b1-066f4ff04079:optical"}')]

        with self.__lock:

            if self.__constellation is None:

                self.__constellation = self.__cm_connection.get_constellation_by_hub_name(self.__hub_module_name)

            return results

    def DeleteConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:

        LOGGER.info(f"DeleteConfig {resources=}")

        LOGGER.info(f"DeleteConfig[{self}]: {resources=}")

        # Input looks like:

        #  resources=[('/service[c8a35e81-88d8-4468-9afc-a8abd92a64d0:optical]', '{"uuid": "c8a35e81-88d8-4468-9afc-a8abd92a64d0:optical"}')]

        with self.__lock:

            results = []

            if len(resources) == 0:

                return results