From b63c2ec345bc272f72a0d88cc280afc736b164c6 Mon Sep 17 00:00:00 2001 From: Deichmann Date: Thu, 21 Nov 2024 10:54:53 +0100 Subject: [PATCH] fixed ruff findings in mockstuff.py --- tests/mockstuff.py | 98 ++++++++++++++++++++++------------------ tests/test_operations.py | 36 +++++++-------- 2 files changed, 73 insertions(+), 61 deletions(-) diff --git a/tests/mockstuff.py b/tests/mockstuff.py index 8783c5b0..fd9fcd76 100644 --- a/tests/mockstuff.py +++ b/tests/mockstuff.py @@ -1,3 +1,8 @@ +"""The module implements classes needed for testing. + +It contains simplified replacements for WsDiscovery and Subscription on devices side. +It also contains SdcProvider implementations that simplify the instantiation of a device. +""" from __future__ import annotations import logging @@ -21,13 +26,15 @@ from sdc11073.xml_types.eventing_types import Subscribe if TYPE_CHECKING: + import ipaddress import uuid import sdc11073.certloader from sdc11073.provider.components import SdcProviderComponents from sdc11073.provider.providerimpl import WsDiscoveryProtocol + from sdc11073.pysoap.msgfactory import MessageFactory from sdc11073.pysoap.soapclientpool import SoapClientPool - + from sdc11073.xml_utils import LxmlElement ports_lock = threading.Lock() _ports = 10000 @@ -36,32 +43,31 @@ _logger = logging.getLogger('sdc.mock') -def dec_list(*args): +def dec_list(*args: list[int | float]) -> list[Decimal]: + """Convert a list of numbers to decimal.""" return [Decimal(x) for x in args] -def _findServer(netloc): - dev_addr = netloc.split(':') - dev_addr = tuple([dev_addr[0], int(dev_addr[1])]) # make port number an integer - for key, srv in _mockhttpservers.items(): - if tuple(key) == dev_addr: - return srv - raise KeyError(f'{dev_addr} is not in {_mockhttpservers.keys()}') +class MockWsDiscovery: + """Implementation of a minimal WsDiscovery interface. + The class does nothing except logging. + """ -class MockWsDiscovery: - def __init__(self, ipaddress): + def __init__(self, ipaddress: str | ipaddress.IPv4Address): self._ipaddress = ipaddress - def get_active_addresses(self): + def get_active_addresses(self) -> str: + """Return the ip address.""" return [self._ipaddress] - def clear_service(self, epr): - _logger.info(f'clear_service "{epr}"') + def clear_service(self, epr: str): + """Clear services.""" + _logger.info('clear_service "%r"', epr) class TestDevSubscription(BicepsSubscription): - """Can be used instead of real Subscription objects""" + """Can be used instead of real Subscription objects.""" mode = 'SomeMode' notify_to = 'http://self.com:123' @@ -69,12 +75,13 @@ class TestDevSubscription(BicepsSubscription): expires = 60 notify_ref = 'a ref string' - def __init__(self, filter_, + def __init__(self, + filter_: list[str], soap_client_pool: SoapClientPool, - msg_factory): + msg_factory: MessageFactory): notify_ref_node = etree.Element(ns_hlp.WSE.tag('References')) - identNode = etree.SubElement(notify_ref_node, ns_hlp.WSE.tag('Identifier')) - identNode.text = self.notify_ref + ident_node = etree.SubElement(notify_ref_node, ns_hlp.WSE.tag('Identifier')) + ident_node.text = self.notify_ref base_urls = [SplitResult('https', 'www.example.com:222', 'no_uuid', query=None, fragment=None)] accepted_encodings = ['foo'] # not needed here subscribe_request = Subscribe() @@ -88,29 +95,36 @@ def __init__(self, filter_, soap_client_pool, msg_factory=msg_factory, log_prefix='test') self.reports = [] - def send_notification_report(self, body_node, action: str): + def send_notification_report(self, body_node: LxmlElement, action: str): + """Send notification to subscriber.""" info_block = HeaderInformationBlock(action=action, addr_to=self.notify_to_address, reference_parameters=self.notify_ref_params) message = self._mk_notification_message(info_block, body_node) self.reports.append(message) - async def async_send_notification_report(self, body_node, action): + async def async_send_notification_report(self, body_node: LxmlElement, action: str): + """Send notification to subscriber.""" info_block = HeaderInformationBlock(action=action, addr_to=self.notify_to_address, reference_parameters=self.notify_ref_params) message = self._mk_notification_message(info_block, body_node) self.reports.append(message) - async def async_send_notification_end_message(self, code='SourceShuttingDown', - reason='Event source going off line.'): - pass + async def async_send_notification_end_message(self, + code: str = 'SourceShuttingDown', + reason: str = 'Event source going off line.'): + """Do nothing. + + Implementation not needed for tests. + """ class SomeDevice(SdcProvider): """A device used for unit tests. Some values are predefined.""" - def __init__(self, wsdiscovery: WsDiscoveryProtocol, + def __init__(self, # noqa: PLR0913 + wsdiscovery: WsDiscoveryProtocol, mdib_xml_data: bytes, epr: str | uuid.UUID | None = None, validate: bool = True, @@ -134,13 +148,13 @@ def __init__(self, wsdiscovery: WsDiscoveryProtocol, device_mdib_container = ProviderMdib.from_string(mdib_xml_data, log_prefix=log_prefix) device_mdib_container.instance_id = 1 # set the optional value # set Metadata - mdsDescriptors = device_mdib_container.descriptions.NODETYPE.get(pm.MdsDescriptor) - for mdsDescriptor in mdsDescriptors: - if mdsDescriptor.MetaData is not None: - mdsDescriptor.MetaData.Manufacturer.append(pm_types.LocalizedText('Example Manufacturer')) - mdsDescriptor.MetaData.ModelName.append(pm_types.LocalizedText(model.ModelName[0].text)) - mdsDescriptor.MetaData.SerialNumber.append('ABCD-1234') - mdsDescriptor.MetaData.ModelNumber = '0.99' + mds_descriptors = device_mdib_container.descriptions.NODETYPE.get(pm.MdsDescriptor) + for mds_descriptor in mds_descriptors: + if mds_descriptor.MetaData is not None: + mds_descriptor.MetaData.Manufacturer.append(pm_types.LocalizedText('Example Manufacturer')) + mds_descriptor.MetaData.ModelName.append(pm_types.LocalizedText(model.ModelName[0].text)) + mds_descriptor.MetaData.SerialNumber.append('ABCD-1234') + mds_descriptor.MetaData.ModelNumber = '0.99' super().__init__(wsdiscovery, model, device, device_mdib_container, epr, validate, ssl_context_container=ssl_context_container, max_subscription_duration=max_subscription_duration, @@ -151,7 +165,7 @@ def __init__(self, wsdiscovery: WsDiscoveryProtocol, alternative_hostname=alternative_hostname) @classmethod - def from_mdib_file(cls, + def from_mdib_file(cls, # noqa: PLR0913 wsdiscovery: WsDiscoveryProtocol, epr: str | uuid.UUID | None, mdib_xml_path: str | pathlib.Path, @@ -162,7 +176,7 @@ def from_mdib_file(cls, default_components: SdcProviderComponents | None = None, specific_components: SdcProviderComponents | None = None, chunk_size: int = 0, - alternative_hostname: str | None = None): + alternative_hostname: str | None = None) -> SomeDevice: """Construct class with path to a mdib file.""" mdib_xml_path = pathlib.Path(mdib_xml_path) if not mdib_xml_path.is_absolute(): @@ -175,12 +189,11 @@ def from_mdib_file(cls, alternative_hostname=alternative_hostname) - - class SomeDeviceEntityMdib(SdcProvider): """A device used for unit tests. Some values are predefined.""" - def __init__(self, wsdiscovery: WsDiscoveryProtocol, + def __init__(self, # noqa: PLR0913 + wsdiscovery: WsDiscoveryProtocol, mdib_xml_data: bytes, epr: str | uuid.UUID | None = None, validate: bool = True, @@ -204,7 +217,6 @@ def __init__(self, wsdiscovery: WsDiscoveryProtocol, device_mdib_container = EntityProviderMdib.from_string(mdib_xml_data, log_prefix=log_prefix) device_mdib_container.instance_id = 1 # set the optional value # set Metadata - # mds_entities = device_mdib_container.parent_handle.get(None) mds_entities = device_mdib_container.entities.parent_handle(None) # Todo: write that meta data back to dom tree for mds_entity in mds_entities: @@ -216,7 +228,7 @@ def __init__(self, wsdiscovery: WsDiscoveryProtocol, mds_descriptor.MetaData.ModelNumber = '0.99' super().__init__(wsdiscovery, model, device, device_mdib_container, epr, validate, ssl_context_container=ssl_context_container, - max_subscription_duration = max_subscription_duration, + max_subscription_duration=max_subscription_duration, log_prefix=log_prefix, default_components=default_components, specific_components=specific_components, @@ -224,24 +236,24 @@ def __init__(self, wsdiscovery: WsDiscoveryProtocol, alternative_hostname=alternative_hostname) @classmethod - def from_mdib_file(cls, + def from_mdib_file(cls, # noqa: PLR0913 wsdiscovery: WsDiscoveryProtocol, epr: str | uuid.UUID | None, mdib_xml_path: str | pathlib.Path, - validate: bool =True, + validate: bool = True, ssl_context_container: sdc11073.certloader.SSLContextContainer | None = None, max_subscription_duration: int = 15, log_prefix: str = '', default_components: SdcProviderComponents | None = None, specific_components: SdcProviderComponents | None = None, chunk_size: int = 0, - alternative_hostname: str | None = None): + alternative_hostname: str | None = None) -> SomeDeviceEntityMdib: """Construct class with path to a mdib file.""" mdib_xml_path = pathlib.Path(mdib_xml_path) if not mdib_xml_path.is_absolute(): mdib_xml_path = pathlib.Path(__file__).parent.joinpath(mdib_xml_path) return cls(wsdiscovery, mdib_xml_path.read_bytes(), epr, validate, ssl_context_container, - max_subscription_duration = max_subscription_duration, + max_subscription_duration=max_subscription_duration, log_prefix=log_prefix, default_components=default_components, specific_components=specific_components, chunk_size=chunk_size, diff --git a/tests/test_operations.py b/tests/test_operations.py index a5870730..a416a3f9 100644 --- a/tests/test_operations.py +++ b/tests/test_operations.py @@ -4,17 +4,16 @@ import unittest from decimal import Decimal -from sdc11073 import commlog -from sdc11073 import loghelper -from sdc11073 import observableproperties -from sdc11073.xml_types import pm_types, msg_types, pm_qnames as pm +from sdc11073 import commlog, loghelper, observableproperties +from sdc11073.consumer import SdcConsumer +from sdc11073.consumer.components import SdcConsumerComponents +from sdc11073.dispatch import RequestDispatcher from sdc11073.loghelper import basic_logging_setup from sdc11073.mdib import ConsumerMdib from sdc11073.roles.nomenclature import NomenclatureCodes -from sdc11073.consumer import SdcConsumer from sdc11073.wsdiscovery import WSDiscovery -from sdc11073.consumer.components import SdcConsumerComponents -from sdc11073.dispatch import RequestDispatcher +from sdc11073.xml_types import msg_types, pm_types +from sdc11073.xml_types import pm_qnames as pm from tests import utils from tests.mockstuff import SomeDevice @@ -51,7 +50,7 @@ def setUp(self): x_addr = self.sdc_device.get_xaddrs() # no deferred action handling for easier debugging specific_components = SdcConsumerComponents( - action_dispatcher_class=RequestDispatcher + action_dispatcher_class=RequestDispatcher, ) self.sdc_client = SdcConsumer(x_addr[0], sdc_definitions=self.sdc_device.mdib.sdc_definitions, @@ -80,9 +79,10 @@ def tearDown(self): self._logger.info('############### tearDown %s done ##############\n', self._testMethodName) def test_set_patient_context_operation(self): - """client calls corresponding operation of GenericContextProvider. + """Client calls corresponding operation of GenericContextProvider. - verify that operation is successful. - verify that a notification device->client also updates the client mdib.""" + verify that a notification device->client also updates the client mdib. + """ client_mdib = ConsumerMdib(self.sdc_client) client_mdib.init_mdib() patient_descriptor_container = client_mdib.descriptions.NODETYPE.get_one(pm.PatientContextDescriptor) @@ -360,7 +360,7 @@ def test_audio_pause_two_clients(self): x_addr = self.sdc_device.get_xaddrs() # no deferred action handling for easier debugging specific_components = SdcConsumerComponents( - action_dispatcher_class=RequestDispatcher + action_dispatcher_class=RequestDispatcher, ) sdc_client2 = SdcConsumer(x_addr[0], sdc_definitions=self.sdc_device.mdib.sdc_definitions, @@ -427,7 +427,7 @@ def test_set_ntp_server(self): # verify that the corresponding state has been updated state = client_mdib.states.descriptor_handle.get_one(my_operation_descriptor.OperationTarget) - if state.NODETYPE == pm.MdsState: + if pm.MdsState == state.NODETYPE: # look for the ClockState child clock_descriptors = client_mdib.descriptions.NODETYPE.get(pm.ClockDescriptor, []) clock_descriptors = [c for c in clock_descriptors if c.descriptor_handle == state.descriptor_handle] @@ -455,7 +455,7 @@ def test_set_time_zone(self): # verify that the corresponding state has been updated state = client_mdib.states.descriptor_handle.get_one(my_operation_descriptor.OperationTarget) - if state.NODETYPE == pm.MdsState: + if pm.MdsState == state.NODETYPE: # look for the ClockState child clock_descriptors = client_mdib.descriptions.NODETYPE.get(pm.ClockDescriptor, []) clock_descriptors = [c for c in clock_descriptors if c.parent_handle == state.DescriptorHandle] @@ -510,7 +510,7 @@ def test_set_metric_state(self): self.assertAlmostEqual(updated_metric_state.LifeTimePeriod, newLifeTimePeriod) def test_set_component_state(self): - """ tests GenericSetComponentStateOperationProvider""" + """Tests GenericSetComponentStateOperationProvider""" operation_target_handle = '2.1.2.1' # a channel # first we need to add a set_component_state Operation sco_descriptors = self.sdc_device.mdib.descriptions.NODETYPE.get(pm.ScoDescriptor) @@ -635,8 +635,8 @@ def test_set_operating_mode(self): def test_set_string_value(self): """Verify that metricprovider instantiated an operation for SetString call. - OperationTarget of operation 0815 is an EnumStringMetricState. - """ + OperationTarget of operation 0815 is an EnumStringMetricState. + """ set_service = self.sdc_client.client('Set') client_mdib = ConsumerMdib(self.sdc_client) client_mdib.init_mdib() @@ -660,8 +660,8 @@ def test_set_string_value(self): def test_set_metric_value(self): """Verify that metricprovider instantiated an operation for SetNumericValue call. - OperationTarget of operation 0815-1 is a NumericMetricState. - """ + OperationTarget of operation 0815-1 is a NumericMetricState. + """ set_service = self.sdc_client.client('Set') client_mdib = ConsumerMdib(self.sdc_client) client_mdib.init_mdib()