diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..e682df0 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,53 @@ +exclude: src/ixia_data_model.py +repos: + - repo: https://github.com/timothycrosley/isort + rev: 5.10.1 + hooks: + - id: isort + language_version: python3.7 + args: [--line-length=127] + - repo: https://github.com/python/black + rev: 21.12b0 + hooks: + - id: black + language_version: python3.7 + args: [--line-length=127] + # We can safely ignore flake8 warnings that pylint catches. + - repo: https://gitlab.com/pycqa/flake8 + rev: 4.0.1 + hooks: + - id: flake8 + additional_dependencies: [ + flake8-docstrings, + flake8-builtins, + flake8-comprehensions, + flake8-print, + flake8-eradicate, + ] + language_version: python3.7 + args: [ + --max-line-length=127, + '--ignore=A002,A003,D100,D101,D102,D103,D104,D105,D106,D107,D200,D210,D401,E203,W503' + ] + # See https://stackoverflow.com/questions/61238318/pylint-and-pre-commit-hook-unable-to-import/61238571#61238571 + - repo: local + hooks: + - id: pylint + name: pylint + entry: pylint + language: system + types: [python] + args: [ + --max-line-length=127, + --max-public-methods=40, + --max-statements=80, + --max-args=8, + '--disable=too-few-public-methods,logging-fstring-interpolation,too-many-instance-attributes,no-else-return,too-many-locals,no-self-use,duplicate-code,broad-except,logging-not-lazy,unspecified-encoding', + ] + - repo: https://github.com/pre-commit/mirrors-mypy + rev: v0.930 + hooks: + - id: mypy + verbose: true + entry: bash -c 'mypy "$@" || true' -- + additional_dependencies: [types-PyYAML] diff --git a/Makefile b/Makefile index c1e2dac..bca9b62 100644 --- a/Makefile +++ b/Makefile @@ -4,7 +4,8 @@ user=pypiadmin password=pypiadmin install: - pip install -i http://$(repo):8036 --trusted-host $(repo) -U --pre --use-feature=2020-resolver -r test_requirements.txt + python -m pip install -U pip==20.2.4 + pip install -i http://$(repo):8036 --trusted-host $(repo) -U --pre -r test_requirements.txt .PHONY: build build: diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..6671ae3 --- /dev/null +++ b/setup.cfg @@ -0,0 +1,9 @@ +[isort] +profile = black + +[mypy] +ignore_missing_imports = True +allow_untyped_calls = False +allow_untyped_defs = False +allow_incomplete_defs = False +follow_imports = skip diff --git a/src/ixia_data_model.py b/src/ixia_data_model.py index d42d965..7f8f285 100644 --- a/src/ixia_data_model.py +++ b/src/ixia_data_model.py @@ -1,7 +1,7 @@ -from cloudshell.shell.core.driver_context import ResourceCommandContext, AutoLoadDetails, AutoLoadAttribute, \ - AutoLoadResource from collections import defaultdict +from cloudshell.shell.core.driver_context import AutoLoadAttribute, AutoLoadDetails, AutoLoadResource, ResourceCommandContext + class LegacyUtils(object): def __init__(self): @@ -12,7 +12,7 @@ def migrate_autoload_details(self, autoload_details, context): root_name = context.resource.name root = self.__create_resource_from_datamodel(model_name, root_name) attributes = self.__create_attributes_dict(autoload_details.attributes) - self.__attach_attributes_to_resource(attributes, '', root) + self.__attach_attributes_to_resource(attributes, "", root) self.__build_sub_resoruces_hierarchy(root, autoload_details.resources, attributes) return root @@ -28,58 +28,52 @@ def __create_attributes_dict(self, attributes_lst): def __build_sub_resoruces_hierarchy(self, root, sub_resources, attributes): d = defaultdict(list) for resource in sub_resources: - splitted = resource.relative_address.split('/') - parent = '' if len(splitted) == 1 else resource.relative_address.rsplit('/', 1)[0] + splitted = resource.relative_address.split("/") + parent = "" if len(splitted) == 1 else resource.relative_address.rsplit("/", 1)[0] rank = len(splitted) d[rank].append((parent, resource)) - self.__set_models_hierarchy_recursively(d, 1, root, '', attributes) + self.__set_models_hierarchy_recursively(d, 1, root, "", attributes) def __set_models_hierarchy_recursively(self, dict, rank, manipulated_resource, resource_relative_addr, attributes): - if rank not in dict: # validate if key exists + if rank not in dict: # validate if key exists pass for (parent, resource) in dict[rank]: if parent == resource_relative_addr: - sub_resource = self.__create_resource_from_datamodel( - resource.model.replace(' ', ''), - resource.name) + sub_resource = self.__create_resource_from_datamodel(resource.model.replace(" ", ""), resource.name) self.__attach_attributes_to_resource(attributes, resource.relative_address, sub_resource) manipulated_resource.add_sub_resource( - self.__slice_parent_from_relative_path(parent, resource.relative_address), sub_resource) - self.__set_models_hierarchy_recursively( - dict, - rank + 1, - sub_resource, - resource.relative_address, - attributes) + self.__slice_parent_from_relative_path(parent, resource.relative_address), sub_resource + ) + self.__set_models_hierarchy_recursively(dict, rank + 1, sub_resource, resource.relative_address, attributes) def __attach_attributes_to_resource(self, attributes, curr_relative_addr, resource): for attribute in attributes[curr_relative_addr]: - setattr(resource, attribute.attribute_name.lower().replace(' ', '_'), attribute.attribute_value) + setattr(resource, attribute.attribute_name.lower().replace(" ", "_"), attribute.attribute_value) del attributes[curr_relative_addr] def __slice_parent_from_relative_path(self, parent, relative_addr): - if parent is '': + if parent is "": return relative_addr - return relative_addr[len(parent) + 1:] # + 1 because we want to remove the seperator also + return relative_addr[len(parent) + 1 :] # + 1 because we want to remove the seperator also def __generate_datamodel_classes_dict(self): return dict(self.__collect_generated_classes()) def __collect_generated_classes(self): - import sys, inspect + import inspect + import sys + return inspect.getmembers(sys.modules[__name__], inspect.isclass) class Ixia_Chassis_Shell_2G(object): def __init__(self, name): - """ - - """ + """ """ self.attributes = {} self.resources = {} - self._cloudshell_model_name = 'Ixia Chassis Shell 2G' + self._cloudshell_model_name = "Ixia Chassis Shell 2G" self._name = name def add_sub_resource(self, relative_path, sub_resource): @@ -99,20 +93,24 @@ def create_from_context(cls, context): result.attributes[attr] = context.resource.attributes[attr] return result - def create_autoload_details(self, relative_path=''): + def create_autoload_details(self, relative_path=""): """ :param relative_path: :type relative_path: str :return """ - resources = [AutoLoadResource(model=self.resources[r].cloudshell_model_name, - name=self.resources[r].name, - relative_address=self._get_relative_path(r, relative_path)) - for r in self.resources] + resources = [ + AutoLoadResource( + model=self.resources[r].cloudshell_model_name, + name=self.resources[r].name, + relative_address=self._get_relative_path(r, relative_path), + ) + for r in self.resources + ] attributes = [AutoLoadAttribute(relative_path, a, self.attributes[a]) for a in self.attributes] autoload_details = AutoLoadDetails(resources, attributes) for r in self.resources: - curr_path = relative_path + '/' + r if relative_path else r + curr_path = relative_path + "/" + r if relative_path else r curr_auto_load_details = self.resources[r].create_autoload_details(curr_path) autoload_details = self._merge_autoload_details(autoload_details, curr_auto_load_details) return autoload_details @@ -127,7 +125,7 @@ def _get_relative_path(self, child_path, parent_path): :return: Combined path :rtype str """ - return parent_path + '/' + child_path if parent_path else child_path + return parent_path + "/" + child_path if parent_path else child_path @staticmethod def _merge_autoload_details(autoload_details1, autoload_details2): @@ -152,14 +150,14 @@ def cloudshell_model_name(self): Returns the name of the Cloudshell model :return: """ - return 'Ixia Chassis Shell 2G' + return "Ixia Chassis Shell 2G" @property def user(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.User'] if 'Ixia Chassis Shell 2G.User' in self.attributes else None + return self.attributes["Ixia Chassis Shell 2G.User"] if "Ixia Chassis Shell 2G.User" in self.attributes else None @user.setter def user(self, value): @@ -167,14 +165,16 @@ def user(self, value): User with administrative privileges :type value: str """ - self.attributes['Ixia Chassis Shell 2G.User'] = value + self.attributes["Ixia Chassis Shell 2G.User"] = value @property def password(self): """ :rtype: string """ - return self.attributes['Ixia Chassis Shell 2G.Password'] if 'Ixia Chassis Shell 2G.Password' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.Password"] if "Ixia Chassis Shell 2G.Password" in self.attributes else None + ) @password.setter def password(self, value): @@ -182,14 +182,18 @@ def password(self, value): Password :type value: string """ - self.attributes['Ixia Chassis Shell 2G.Password'] = value + self.attributes["Ixia Chassis Shell 2G.Password"] = value @property def controller_tcp_port(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.Controller TCP Port'] if 'Ixia Chassis Shell 2G.Controller TCP Port' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.Controller TCP Port"] + if "Ixia Chassis Shell 2G.Controller TCP Port" in self.attributes + else None + ) @controller_tcp_port.setter def controller_tcp_port(self, value): @@ -197,14 +201,18 @@ def controller_tcp_port(self, value): The TCP port of the traffic server. Relevant only in case an external server is configured. Default TCP port should be used if kept empty. :type value: str """ - self.attributes['Ixia Chassis Shell 2G.Controller TCP Port'] = value + self.attributes["Ixia Chassis Shell 2G.Controller TCP Port"] = value @property def controller_address(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.Controller Address'] if 'Ixia Chassis Shell 2G.Controller Address' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.Controller Address"] + if "Ixia Chassis Shell 2G.Controller Address" in self.attributes + else None + ) @controller_address.setter def controller_address(self, value): @@ -212,14 +220,18 @@ def controller_address(self, value): The IP address of the traffic server. Relevant only in case an external server is configured. :type value: str """ - self.attributes['Ixia Chassis Shell 2G.Controller Address'] = value + self.attributes["Ixia Chassis Shell 2G.Controller Address"] = value @property def client_install_path(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.Client Install Path'] if 'Ixia Chassis Shell 2G.Client Install Path' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.Client Install Path"] + if "Ixia Chassis Shell 2G.Client Install Path" in self.attributes + else None + ) @client_install_path.setter def client_install_path(self, value): @@ -227,14 +239,18 @@ def client_install_path(self, value): The path in which the traffic client is installed on the Execution Server. For example "C:/Program Files (x86)/Ixia/IxLoad/5.10-GA". :type value: str """ - self.attributes['Ixia Chassis Shell 2G.Client Install Path'] = value + self.attributes["Ixia Chassis Shell 2G.Client Install Path"] = value @property def power_management(self): """ :rtype: bool """ - return self.attributes['Ixia Chassis Shell 2G.Power Management'] if 'Ixia Chassis Shell 2G.Power Management' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.Power Management"] + if "Ixia Chassis Shell 2G.Power Management" in self.attributes + else None + ) @power_management.setter def power_management(self, value=True): @@ -242,14 +258,18 @@ def power_management(self, value=True): Used by the power management orchestration, if enabled, to determine whether to automatically manage the device power status. Enabled by default. :type value: bool """ - self.attributes['Ixia Chassis Shell 2G.Power Management'] = value + self.attributes["Ixia Chassis Shell 2G.Power Management"] = value @property def serial_number(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.Serial Number'] if 'Ixia Chassis Shell 2G.Serial Number' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.Serial Number"] + if "Ixia Chassis Shell 2G.Serial Number" in self.attributes + else None + ) @serial_number.setter def serial_number(self, value): @@ -257,14 +277,18 @@ def serial_number(self, value): The serial number of the resource. :type value: str """ - self.attributes['Ixia Chassis Shell 2G.Serial Number'] = value + self.attributes["Ixia Chassis Shell 2G.Serial Number"] = value @property def server_description(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.Server Description'] if 'Ixia Chassis Shell 2G.Server Description' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.Server Description"] + if "Ixia Chassis Shell 2G.Server Description" in self.attributes + else None + ) @server_description.setter def server_description(self, value): @@ -272,7 +296,7 @@ def server_description(self, value): The full description of the server. Usually includes the OS, exact firmware version and additional characteritics of the device. :type value: str """ - self.attributes['Ixia Chassis Shell 2G.Server Description'] = value + self.attributes["Ixia Chassis Shell 2G.Server Description"] = value @property def name(self): @@ -284,7 +308,7 @@ def name(self): @name.setter def name(self, value): """ - + :type value: str """ self._name = value @@ -299,7 +323,7 @@ def cloudshell_model_name(self): @cloudshell_model_name.setter def cloudshell_model_name(self, value): """ - + :type value: str """ self._cloudshell_model_name = value @@ -309,22 +333,30 @@ def model_name(self): """ :rtype: str """ - return self.attributes['CS_TrafficGeneratorChassis.Model Name'] if 'CS_TrafficGeneratorChassis.Model Name' in self.attributes else None + return ( + self.attributes["CS_TrafficGeneratorChassis.Model Name"] + if "CS_TrafficGeneratorChassis.Model Name" in self.attributes + else None + ) @model_name.setter - def model_name(self, value=''): + def model_name(self, value=""): """ The catalog name of the device model. This attribute will be displayed in CloudShell instead of the CloudShell model. :type value: str """ - self.attributes['CS_TrafficGeneratorChassis.Model Name'] = value + self.attributes["CS_TrafficGeneratorChassis.Model Name"] = value @property def vendor(self): """ :rtype: str """ - return self.attributes['CS_TrafficGeneratorChassis.Vendor'] if 'CS_TrafficGeneratorChassis.Vendor' in self.attributes else None + return ( + self.attributes["CS_TrafficGeneratorChassis.Vendor"] + if "CS_TrafficGeneratorChassis.Vendor" in self.attributes + else None + ) @vendor.setter def vendor(self, value): @@ -332,14 +364,18 @@ def vendor(self, value): The name of the device manufacture. :type value: str """ - self.attributes['CS_TrafficGeneratorChassis.Vendor'] = value + self.attributes["CS_TrafficGeneratorChassis.Vendor"] = value @property def version(self): """ :rtype: str """ - return self.attributes['CS_TrafficGeneratorChassis.Version'] if 'CS_TrafficGeneratorChassis.Version' in self.attributes else None + return ( + self.attributes["CS_TrafficGeneratorChassis.Version"] + if "CS_TrafficGeneratorChassis.Version" in self.attributes + else None + ) @version.setter def version(self, value): @@ -347,17 +383,15 @@ def version(self, value): The firmware version of the resource. :type value: str """ - self.attributes['CS_TrafficGeneratorChassis.Version'] = value + self.attributes["CS_TrafficGeneratorChassis.Version"] = value class GenericTrafficGeneratorModule(object): def __init__(self, name): - """ - - """ + """ """ self.attributes = {} self.resources = {} - self._cloudshell_model_name = 'Ixia Chassis Shell 2G.GenericTrafficGeneratorModule' + self._cloudshell_model_name = "Ixia Chassis Shell 2G.GenericTrafficGeneratorModule" self._name = name def add_sub_resource(self, relative_path, sub_resource): @@ -377,20 +411,24 @@ def create_from_context(cls, context): result.attributes[attr] = context.resource.attributes[attr] return result - def create_autoload_details(self, relative_path=''): + def create_autoload_details(self, relative_path=""): """ :param relative_path: :type relative_path: str :return """ - resources = [AutoLoadResource(model=self.resources[r].cloudshell_model_name, - name=self.resources[r].name, - relative_address=self._get_relative_path(r, relative_path)) - for r in self.resources] + resources = [ + AutoLoadResource( + model=self.resources[r].cloudshell_model_name, + name=self.resources[r].name, + relative_address=self._get_relative_path(r, relative_path), + ) + for r in self.resources + ] attributes = [AutoLoadAttribute(relative_path, a, self.attributes[a]) for a in self.attributes] autoload_details = AutoLoadDetails(resources, attributes) for r in self.resources: - curr_path = relative_path + '/' + r if relative_path else r + curr_path = relative_path + "/" + r if relative_path else r curr_auto_load_details = self.resources[r].create_autoload_details(curr_path) autoload_details = self._merge_autoload_details(autoload_details, curr_auto_load_details) return autoload_details @@ -405,7 +443,7 @@ def _get_relative_path(self, child_path, parent_path): :return: Combined path :rtype str """ - return parent_path + '/' + child_path if parent_path else child_path + return parent_path + "/" + child_path if parent_path else child_path @staticmethod def _merge_autoload_details(autoload_details1, autoload_details2): @@ -430,37 +468,45 @@ def cloudshell_model_name(self): Returns the name of the Cloudshell model :return: """ - return 'GenericTrafficGeneratorModule' + return "GenericTrafficGeneratorModule" @property def version(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Version'] if 'Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Version' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Version"] + if "Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Version" in self.attributes + else None + ) @version.setter - def version(self, value=''): + def version(self, value=""): """ The firmware version of the resource. :type value: str """ - self.attributes['Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Version'] = value + self.attributes["Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Version"] = value @property def serial_number(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Serial Number'] if 'Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Serial Number' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Serial Number"] + if "Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Serial Number" in self.attributes + else None + ) @serial_number.setter - def serial_number(self, value=''): + def serial_number(self, value=""): """ - + :type value: str """ - self.attributes['Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Serial Number'] = value + self.attributes["Ixia Chassis Shell 2G.GenericTrafficGeneratorModule.Serial Number"] = value @property def name(self): @@ -472,7 +518,7 @@ def name(self): @name.setter def name(self, value): """ - + :type value: str """ self._name = value @@ -487,7 +533,7 @@ def cloudshell_model_name(self): @cloudshell_model_name.setter def cloudshell_model_name(self, value): """ - + :type value: str """ self._cloudshell_model_name = value @@ -497,25 +543,27 @@ def model_name(self): """ :rtype: str """ - return self.attributes['CS_TrafficGeneratorModule.Model Name'] if 'CS_TrafficGeneratorModule.Model Name' in self.attributes else None + return ( + self.attributes["CS_TrafficGeneratorModule.Model Name"] + if "CS_TrafficGeneratorModule.Model Name" in self.attributes + else None + ) @model_name.setter - def model_name(self, value=''): + def model_name(self, value=""): """ The catalog name of the device model. This attribute will be displayed in CloudShell instead of the CloudShell model. :type value: str """ - self.attributes['CS_TrafficGeneratorModule.Model Name'] = value + self.attributes["CS_TrafficGeneratorModule.Model Name"] = value class GenericTrafficGeneratorPortGroup(object): def __init__(self, name): - """ - - """ + """ """ self.attributes = {} self.resources = {} - self._cloudshell_model_name = 'Ixia Chassis Shell 2G.GenericTrafficGeneratorPortGroup' + self._cloudshell_model_name = "Ixia Chassis Shell 2G.GenericTrafficGeneratorPortGroup" self._name = name def add_sub_resource(self, relative_path, sub_resource): @@ -535,20 +583,24 @@ def create_from_context(cls, context): result.attributes[attr] = context.resource.attributes[attr] return result - def create_autoload_details(self, relative_path=''): + def create_autoload_details(self, relative_path=""): """ :param relative_path: :type relative_path: str :return """ - resources = [AutoLoadResource(model=self.resources[r].cloudshell_model_name, - name=self.resources[r].name, - relative_address=self._get_relative_path(r, relative_path)) - for r in self.resources] + resources = [ + AutoLoadResource( + model=self.resources[r].cloudshell_model_name, + name=self.resources[r].name, + relative_address=self._get_relative_path(r, relative_path), + ) + for r in self.resources + ] attributes = [AutoLoadAttribute(relative_path, a, self.attributes[a]) for a in self.attributes] autoload_details = AutoLoadDetails(resources, attributes) for r in self.resources: - curr_path = relative_path + '/' + r if relative_path else r + curr_path = relative_path + "/" + r if relative_path else r curr_auto_load_details = self.resources[r].create_autoload_details(curr_path) autoload_details = self._merge_autoload_details(autoload_details, curr_auto_load_details) return autoload_details @@ -563,7 +615,7 @@ def _get_relative_path(self, child_path, parent_path): :return: Combined path :rtype str """ - return parent_path + '/' + child_path if parent_path else child_path + return parent_path + "/" + child_path if parent_path else child_path @staticmethod def _merge_autoload_details(autoload_details1, autoload_details2): @@ -588,7 +640,7 @@ def cloudshell_model_name(self): Returns the name of the Cloudshell model :return: """ - return 'GenericTrafficGeneratorPortGroup' + return "GenericTrafficGeneratorPortGroup" @property def name(self): @@ -600,7 +652,7 @@ def name(self): @name.setter def name(self, value): """ - + :type value: str """ self._name = value @@ -615,7 +667,7 @@ def cloudshell_model_name(self): @cloudshell_model_name.setter def cloudshell_model_name(self, value): """ - + :type value: str """ self._cloudshell_model_name = value @@ -623,12 +675,10 @@ def cloudshell_model_name(self, value): class GenericTrafficGeneratorPort(object): def __init__(self, name): - """ - - """ + """ """ self.attributes = {} self.resources = {} - self._cloudshell_model_name = 'Ixia Chassis Shell 2G.GenericTrafficGeneratorPort' + self._cloudshell_model_name = "Ixia Chassis Shell 2G.GenericTrafficGeneratorPort" self._name = name def add_sub_resource(self, relative_path, sub_resource): @@ -648,20 +698,24 @@ def create_from_context(cls, context): result.attributes[attr] = context.resource.attributes[attr] return result - def create_autoload_details(self, relative_path=''): + def create_autoload_details(self, relative_path=""): """ :param relative_path: :type relative_path: str :return """ - resources = [AutoLoadResource(model=self.resources[r].cloudshell_model_name, - name=self.resources[r].name, - relative_address=self._get_relative_path(r, relative_path)) - for r in self.resources] + resources = [ + AutoLoadResource( + model=self.resources[r].cloudshell_model_name, + name=self.resources[r].name, + relative_address=self._get_relative_path(r, relative_path), + ) + for r in self.resources + ] attributes = [AutoLoadAttribute(relative_path, a, self.attributes[a]) for a in self.attributes] autoload_details = AutoLoadDetails(resources, attributes) for r in self.resources: - curr_path = relative_path + '/' + r if relative_path else r + curr_path = relative_path + "/" + r if relative_path else r curr_auto_load_details = self.resources[r].create_autoload_details(curr_path) autoload_details = self._merge_autoload_details(autoload_details, curr_auto_load_details) return autoload_details @@ -676,7 +730,7 @@ def _get_relative_path(self, child_path, parent_path): :return: Combined path :rtype str """ - return parent_path + '/' + child_path if parent_path else child_path + return parent_path + "/" + child_path if parent_path else child_path @staticmethod def _merge_autoload_details(autoload_details1, autoload_details2): @@ -701,14 +755,18 @@ def cloudshell_model_name(self): Returns the name of the Cloudshell model :return: """ - return 'GenericTrafficGeneratorPort' + return "GenericTrafficGeneratorPort" @property def media_type(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.GenericTrafficGeneratorPort.Media Type'] if 'Ixia Chassis Shell 2G.GenericTrafficGeneratorPort.Media Type' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.GenericTrafficGeneratorPort.Media Type"] + if "Ixia Chassis Shell 2G.GenericTrafficGeneratorPort.Media Type" in self.attributes + else None + ) @media_type.setter def media_type(self, value): @@ -716,7 +774,7 @@ def media_type(self, value): Interface media type. Possible values are Fiber and/or Copper (comma-separated). :type value: str """ - self.attributes['Ixia Chassis Shell 2G.GenericTrafficGeneratorPort.Media Type'] = value + self.attributes["Ixia Chassis Shell 2G.GenericTrafficGeneratorPort.Media Type"] = value @property def name(self): @@ -728,7 +786,7 @@ def name(self): @name.setter def name(self, value): """ - + :type value: str """ self._name = value @@ -743,7 +801,7 @@ def cloudshell_model_name(self): @cloudshell_model_name.setter def cloudshell_model_name(self, value): """ - + :type value: str """ self._cloudshell_model_name = value @@ -753,7 +811,11 @@ def max_speed(self): """ :rtype: str """ - return self.attributes['CS_TrafficGeneratorPort.Max Speed'] if 'CS_TrafficGeneratorPort.Max Speed' in self.attributes else None + return ( + self.attributes["CS_TrafficGeneratorPort.Max Speed"] + if "CS_TrafficGeneratorPort.Max Speed" in self.attributes + else None + ) @max_speed.setter def max_speed(self, value): @@ -761,14 +823,18 @@ def max_speed(self, value): Max speed supported by the interface (default units - MB) :type value: str """ - self.attributes['CS_TrafficGeneratorPort.Max Speed'] = value + self.attributes["CS_TrafficGeneratorPort.Max Speed"] = value @property def logical_name(self): """ :rtype: str """ - return self.attributes['CS_TrafficGeneratorPort.Logical Name'] if 'CS_TrafficGeneratorPort.Logical Name' in self.attributes else None + return ( + self.attributes["CS_TrafficGeneratorPort.Logical Name"] + if "CS_TrafficGeneratorPort.Logical Name" in self.attributes + else None + ) @logical_name.setter def logical_name(self, value): @@ -776,14 +842,18 @@ def logical_name(self, value): The port's logical name in the test configuration. If kept emtpy - allocation will applied in the blue print. :type value: str """ - self.attributes['CS_TrafficGeneratorPort.Logical Name'] = value + self.attributes["CS_TrafficGeneratorPort.Logical Name"] = value @property def configured_controllers(self): """ :rtype: str """ - return self.attributes['CS_TrafficGeneratorPort.Configured Controllers'] if 'CS_TrafficGeneratorPort.Configured Controllers' in self.attributes else None + return ( + self.attributes["CS_TrafficGeneratorPort.Configured Controllers"] + if "CS_TrafficGeneratorPort.Configured Controllers" in self.attributes + else None + ) @configured_controllers.setter def configured_controllers(self, value): @@ -791,17 +861,15 @@ def configured_controllers(self, value): specifies what controller can be used with the ports (IxLoad controller, BP controller etc...) :type value: str """ - self.attributes['CS_TrafficGeneratorPort.Configured Controllers'] = value + self.attributes["CS_TrafficGeneratorPort.Configured Controllers"] = value class GenericPowerPort(object): def __init__(self, name): - """ - - """ + """ """ self.attributes = {} self.resources = {} - self._cloudshell_model_name = 'Ixia Chassis Shell 2G.GenericPowerPort' + self._cloudshell_model_name = "Ixia Chassis Shell 2G.GenericPowerPort" self._name = name def add_sub_resource(self, relative_path, sub_resource): @@ -821,20 +889,24 @@ def create_from_context(cls, context): result.attributes[attr] = context.resource.attributes[attr] return result - def create_autoload_details(self, relative_path=''): + def create_autoload_details(self, relative_path=""): """ :param relative_path: :type relative_path: str :return """ - resources = [AutoLoadResource(model=self.resources[r].cloudshell_model_name, - name=self.resources[r].name, - relative_address=self._get_relative_path(r, relative_path)) - for r in self.resources] + resources = [ + AutoLoadResource( + model=self.resources[r].cloudshell_model_name, + name=self.resources[r].name, + relative_address=self._get_relative_path(r, relative_path), + ) + for r in self.resources + ] attributes = [AutoLoadAttribute(relative_path, a, self.attributes[a]) for a in self.attributes] autoload_details = AutoLoadDetails(resources, attributes) for r in self.resources: - curr_path = relative_path + '/' + r if relative_path else r + curr_path = relative_path + "/" + r if relative_path else r curr_auto_load_details = self.resources[r].create_autoload_details(curr_path) autoload_details = self._merge_autoload_details(autoload_details, curr_auto_load_details) return autoload_details @@ -849,7 +921,7 @@ def _get_relative_path(self, child_path, parent_path): :return: Combined path :rtype str """ - return parent_path + '/' + child_path if parent_path else child_path + return parent_path + "/" + child_path if parent_path else child_path @staticmethod def _merge_autoload_details(autoload_details1, autoload_details2): @@ -874,14 +946,18 @@ def cloudshell_model_name(self): Returns the name of the Cloudshell model :return: """ - return 'GenericPowerPort' + return "GenericPowerPort" @property def model(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.GenericPowerPort.Model'] if 'Ixia Chassis Shell 2G.GenericPowerPort.Model' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.GenericPowerPort.Model"] + if "Ixia Chassis Shell 2G.GenericPowerPort.Model" in self.attributes + else None + ) @model.setter def model(self, value): @@ -889,29 +965,37 @@ def model(self, value): The device model. This information is typically used for abstract resource filtering. :type value: str """ - self.attributes['Ixia Chassis Shell 2G.GenericPowerPort.Model'] = value + self.attributes["Ixia Chassis Shell 2G.GenericPowerPort.Model"] = value @property def serial_number(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.GenericPowerPort.Serial Number'] if 'Ixia Chassis Shell 2G.GenericPowerPort.Serial Number' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.GenericPowerPort.Serial Number"] + if "Ixia Chassis Shell 2G.GenericPowerPort.Serial Number" in self.attributes + else None + ) @serial_number.setter def serial_number(self, value): """ - + :type value: str """ - self.attributes['Ixia Chassis Shell 2G.GenericPowerPort.Serial Number'] = value + self.attributes["Ixia Chassis Shell 2G.GenericPowerPort.Serial Number"] = value @property def version(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.GenericPowerPort.Version'] if 'Ixia Chassis Shell 2G.GenericPowerPort.Version' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.GenericPowerPort.Version"] + if "Ixia Chassis Shell 2G.GenericPowerPort.Version" in self.attributes + else None + ) @version.setter def version(self, value): @@ -919,14 +1003,18 @@ def version(self, value): The firmware version of the resource. :type value: str """ - self.attributes['Ixia Chassis Shell 2G.GenericPowerPort.Version'] = value + self.attributes["Ixia Chassis Shell 2G.GenericPowerPort.Version"] = value @property def port_description(self): """ :rtype: str """ - return self.attributes['Ixia Chassis Shell 2G.GenericPowerPort.Port Description'] if 'Ixia Chassis Shell 2G.GenericPowerPort.Port Description' in self.attributes else None + return ( + self.attributes["Ixia Chassis Shell 2G.GenericPowerPort.Port Description"] + if "Ixia Chassis Shell 2G.GenericPowerPort.Port Description" in self.attributes + else None + ) @port_description.setter def port_description(self, value): @@ -934,7 +1022,7 @@ def port_description(self, value): The description of the port as configured in the device. :type value: str """ - self.attributes['Ixia Chassis Shell 2G.GenericPowerPort.Port Description'] = value + self.attributes["Ixia Chassis Shell 2G.GenericPowerPort.Port Description"] = value @property def name(self): @@ -946,7 +1034,7 @@ def name(self): @name.setter def name(self, value): """ - + :type value: str """ self._name = value @@ -961,7 +1049,7 @@ def cloudshell_model_name(self): @cloudshell_model_name.setter def cloudshell_model_name(self, value): """ - + :type value: str """ self._cloudshell_model_name = value @@ -971,15 +1059,12 @@ def model_name(self): """ :rtype: str """ - return self.attributes['CS_PowerPort.Model Name'] if 'CS_PowerPort.Model Name' in self.attributes else None + return self.attributes["CS_PowerPort.Model Name"] if "CS_PowerPort.Model Name" in self.attributes else None @model_name.setter - def model_name(self, value=''): + def model_name(self, value=""): """ The catalog name of the device model. This attribute will be displayed in CloudShell instead of the CloudShell model. :type value: str """ - self.attributes['CS_PowerPort.Model Name'] = value - - - + self.attributes["CS_PowerPort.Model Name"] = value diff --git a/src/ixia_driver.py b/src/ixia_driver.py index e83e6c5..2934a77 100644 --- a/src/ixia_driver.py +++ b/src/ixia_driver.py @@ -1,19 +1,83 @@ +""" +Ixia chassis shell driver. +""" +import logging +from os import path -from cloudshell.traffic.tg import TgChassisDriver +from cloudshell.logging.qs_logger import get_qs_logger +from cloudshell.shell.core.driver_context import AutoLoadDetails, InitCommandContext, ResourceCommandContext +from cloudshell.shell.core.resource_driver_interface import ResourceDriverInterface +from ixexplorer.ixe_app import init_ixe +from ixexplorer.ixe_hw import IxeCard, IxeChassis, IxePort -from ixia_handler import IxiaHandler +from ixia_data_model import GenericTrafficGeneratorModule, GenericTrafficGeneratorPort, Ixia_Chassis_Shell_2G -class IxiaChassis2GDriver(TgChassisDriver): +class IxiaChassis2GDriver(ResourceDriverInterface): + """Ixia chassis shell driver.""" - def __init__(self): - self.handler = IxiaHandler() + def __init__(self) -> None: + """Initialize object variables, actual initialization is performed in initialize method.""" + self.logger: logging.Logger = None + self.resource: Ixia_Chassis_Shell_2G = None - def initialize(self, context): - super().initialize(context) + def initialize(self, context: InitCommandContext) -> None: + """Initialize Ixia chassis shell (from API).""" + self.logger = get_qs_logger(log_group="traffic_shells", log_file_prefix=context.resource.name) + self.logger.setLevel(logging.DEBUG) - def cleanup(self): + def cleanup(self) -> None: + """Cleanup Ixia chassis shell (from API).""" super().cleanup() - def get_inventory(self, context): - return super().get_inventory(context) + def get_inventory(self, context: ResourceCommandContext) -> AutoLoadDetails: + """Load Ixia chassis inventory to CloudShell (from API).""" + self.resource = Ixia_Chassis_Shell_2G.create_from_context(context) + address = context.resource.address + controller_address = self.resource.controller_address + port = self.resource.controller_tcp_port + + if not controller_address: + controller_address = address + if not port: + port = "4555" + rsa_id = path.join(path.dirname(__file__), "id_rsa") + + ixia = init_ixe(self.logger, host=controller_address, port=int(port), rsa_id=rsa_id) + ixia.connect() + ixia.add(address) + + ixia.discover() + self._load_chassis(list(ixia.chassis_chain.values())[0]) + return self.resource.create_autoload_details() + + def _load_chassis(self, chassis: IxeChassis) -> None: + """Get chassis resource and attributes.""" + self.resource.model_name = chassis.typeName + self.resource.vendor = "Ixia" + self.resource.version = chassis.ixServerVersion + + for card_id, card in chassis.cards.items(): + self._load_module(card_id, card) + + def _load_module(self, card_id: int, card: IxeCard) -> None: + """Get module resource and attributes.""" + gen_module = GenericTrafficGeneratorModule(f"Module{card_id}") + self.resource.add_sub_resource(f"M{card_id}", gen_module) + gen_module.model_name = card.typeName + gen_module.serial_number = card.serialNumber.strip() + gen_module.version = card.hwVersion + + for port_id, port in card.ports.items(): + self._load_port(gen_module, port_id, port) + + def _load_port(self, gen_module: GenericTrafficGeneratorPort, port_id: int, port: IxePort) -> None: + """Get port resource and attributes.""" + gen_port = GenericTrafficGeneratorPort("Port{}".format(port_id)) + gen_module.add_sub_resource("P{}".format(port_id), gen_port) + + supported_speeds = port.supported_speeds() + if not supported_speeds: + supported_speeds = ["0"] + gen_port.max_speed = int(max(supported_speeds, key=int)) + gen_port.configured_controllers = "IxLoad" diff --git a/src/ixia_handler.py b/src/ixia_handler.py deleted file mode 100644 index 4e441d4..0000000 --- a/src/ixia_handler.py +++ /dev/null @@ -1,75 +0,0 @@ - -from os import path - -from cloudshell.traffic.tg import TgChassisHandler - -from ixexplorer.ixe_app import init_ixe - -from ixia_data_model import Ixia_Chassis_Shell_2G, GenericTrafficGeneratorModule, GenericTrafficGeneratorPort - - -class IxiaHandler(TgChassisHandler): - - def initialize(self, context, logger): - """ - :param InitCommandContext context: - """ - resource = Ixia_Chassis_Shell_2G.create_from_context(context) - super().initialize(resource, logger) - - def load_inventory(self, context): - """ - :param InitCommandContext context: - """ - - address = context.resource.address - controller_address = self.resource.controller_address - port = self.resource.controller_tcp_port - - if not controller_address: - controller_address = address - if not port: - port = '4555' - rsa_id = path.join(path.dirname(__file__), 'id_rsa') - - self.ixia = init_ixe(self.logger, host=controller_address, port=int(port), rsa_id=rsa_id) - self.ixia.connect() - self.ixia.add(address) - - self.ixia.discover() - self._load_chassis(list(self.ixia.chassis_chain.values())[0]) - return self.resource.create_autoload_details() - - def _load_chassis(self, chassis): - """ Get chassis resource and attributes. """ - - self.resource.model_name = chassis.typeName - self.resource.vendor = 'Ixia' - self.resource.version = chassis.ixServerVersion - - for card_id, card in chassis.cards.items(): - self._load_module(card_id, card) - - def _load_module(self, card_id, card): - """ Get module resource and attributes. """ - - gen_module = GenericTrafficGeneratorModule(f'Module{card_id}') - self.resource.add_sub_resource(f'M{card_id}', gen_module) - gen_module.model_name = card.typeName - gen_module.serial_number = card.serialNumber.strip() - gen_module.version = card.hwVersion - - for port_id, port in card.ports.items(): - self._load_port(gen_module, port_id, port) - - def _load_port(self, gen_module, port_id, port): - """ Get port resource and attributes. """ - - gen_port = GenericTrafficGeneratorPort('Port{}'.format(port_id)) - gen_module.add_sub_resource('P{}'.format(port_id), gen_port) - - supported_speeds = port.supported_speeds() - if not supported_speeds: - supported_speeds = ['0'] - gen_port.max_speed = int(max(supported_speeds, key=int)) - gen_port.configured_controllers='IxLoad' diff --git a/src/requirements.txt b/src/requirements.txt index d22ab75..c57e34d 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -2,8 +2,6 @@ cloudshell-logging==1.0.3 cloudshell-orch-core>=3.3.0.0,<3.4.0.0 -cloudshell-automation-api>=2020.1.0.0,<2020.2.0.0 cloudshell-shell-core>=5.0.3,<6.0.0 -cloudshell-traffic>3.2,<=3.3 pyixexplorer>=3.0.0,<3.1.0 diff --git a/src/test_ixia_chassis.py b/src/test_ixia_chassis.py index c87bdaf..a7fd45e 100644 --- a/src/test_ixia_chassis.py +++ b/src/test_ixia_chassis.py @@ -1,84 +1,95 @@ """ -Tests for `IxiaChassis2GDriver` +Tests for IxiaChassis2GDriver. """ - -from typing import List +# pylint: disable=redefined-outer-name +from typing import Iterable, List import pytest - +from _pytest.fixtures import SubRequest from cloudshell.api.cloudshell_api import AttributeNameValue, CloudShellAPISession, ResourceInfo from cloudshell.shell.core.driver_context import AutoLoadCommandContext -from cloudshell.traffic.tg import TGN_CHASSIS_FAMILY, IXIA_CHASSIS_MODEL -from shellfoundry_traffic.test_helpers import create_session_from_config, TestHelpers, print_inventory +from cloudshell.traffic.tg import IXIA_CHASSIS_MODEL, TGN_CHASSIS_FAMILY +from shellfoundry_traffic.test_helpers import TestHelpers, create_session_from_config, print_inventory from src.ixia_driver import IxiaChassis2GDriver -@pytest.fixture(params=[('192.168.65.21', '192.168.65.21', '8022'), ('192.168.65.21', 'localhost', '')], - ids=['linux', 'windows']) -def dut(request): - yield request.param +@pytest.fixture(params=[["172.30.150.123", "172.30.150.123", "8022"]], ids=["linux"]) +def dut(request: SubRequest) -> list: + """Yields Ixia device under test parameters.""" + return request.param @pytest.fixture() def session() -> CloudShellAPISession: - """ Yields session. """ + """Yields session.""" yield create_session_from_config() @pytest.fixture() def test_helpers(session: CloudShellAPISession) -> TestHelpers: - """ Yields initialized TestHelpers object. """ + """Yields initialized TestHelpers object.""" yield TestHelpers(session) @pytest.fixture() def autoload_context(test_helpers: TestHelpers, dut: List[str]) -> AutoLoadCommandContext: + """Yields Ixia chassis shell command context for resource commands testing.""" address, controller_address, controller_port = dut # noinspection SpellCheckingInspection - attributes = {f'{IXIA_CHASSIS_MODEL}.Controller Address': controller_address, - f'{IXIA_CHASSIS_MODEL}.Controller TCP Port': controller_port, - f'{IXIA_CHASSIS_MODEL}.User': 'admin', - f'{IXIA_CHASSIS_MODEL}.Password': 'DxTbqlSgAVPmrDLlHvJrsA=='} + attributes = { + f"{IXIA_CHASSIS_MODEL}.Controller Address": controller_address, + f"{IXIA_CHASSIS_MODEL}.Controller TCP Port": controller_port, + f"{IXIA_CHASSIS_MODEL}.User": "admin", + f"{IXIA_CHASSIS_MODEL}.Password": "DxTbqlSgAVPmrDLlHvJrsA==", + } yield test_helpers.autoload_command_context(TGN_CHASSIS_FAMILY, IXIA_CHASSIS_MODEL, address, attributes) @pytest.fixture() -def driver(test_helpers: TestHelpers, dut: List[str]) -> IxiaChassis2GDriver: - """ Yields initialized IxiaChassis2GDriver. """ +def driver(test_helpers: TestHelpers, dut: List[str]) -> Iterable[IxiaChassis2GDriver]: + """Yields initialized IxiaChassis2GDriver.""" address, controller_address, controller_port = dut # noinspection SpellCheckingInspection - attributes = {f'{IXIA_CHASSIS_MODEL}.Controller Address': controller_address, - f'{IXIA_CHASSIS_MODEL}.Controller TCP Port': controller_port, - f'{IXIA_CHASSIS_MODEL}.User': 'admin', - f'{IXIA_CHASSIS_MODEL}.Password': 'DxTbqlSgAVPmrDLlHvJrsA=='} - init_context = test_helpers.resource_init_command_context(TGN_CHASSIS_FAMILY, IXIA_CHASSIS_MODEL, address, - attributes, 'test-ixia') + attributes = { + f"{IXIA_CHASSIS_MODEL}.Controller Address": controller_address, + f"{IXIA_CHASSIS_MODEL}.Controller TCP Port": controller_port, + f"{IXIA_CHASSIS_MODEL}.User": "admin", + f"{IXIA_CHASSIS_MODEL}.Password": "DxTbqlSgAVPmrDLlHvJrsA==", + } + init_context = test_helpers.resource_init_command_context( + TGN_CHASSIS_FAMILY, IXIA_CHASSIS_MODEL, address, attributes, "test-ixia" + ) driver = IxiaChassis2GDriver() driver.initialize(init_context) - print(driver.logger.handlers[0].baseFilename) yield driver + driver.cleanup() @pytest.fixture() def autoload_resource(session: CloudShellAPISession, test_helpers: TestHelpers, dut: List[str]) -> ResourceInfo: + """Yields Ixia chassis resource for shell autoload testing.""" address, controller_address, controller_port = dut - attributes = [AttributeNameValue(f'{IXIA_CHASSIS_MODEL}.Controller Address', controller_address), - AttributeNameValue(f'{IXIA_CHASSIS_MODEL}.Controller TCP Port', controller_port), - AttributeNameValue(f'{IXIA_CHASSIS_MODEL}.User', 'admin'), - AttributeNameValue(f'{IXIA_CHASSIS_MODEL}.Password', 'admin')] - resource = test_helpers.create_autoload_resource(IXIA_CHASSIS_MODEL, 'test-ixia', address, attributes) + attributes = [ + AttributeNameValue(f"{IXIA_CHASSIS_MODEL}.Controller Address", controller_address), + AttributeNameValue(f"{IXIA_CHASSIS_MODEL}.Controller TCP Port", controller_port), + AttributeNameValue(f"{IXIA_CHASSIS_MODEL}.User", "admin"), + AttributeNameValue(f"{IXIA_CHASSIS_MODEL}.Password", "admin"), + ] + resource = test_helpers.create_autoload_resource(IXIA_CHASSIS_MODEL, "test-folder/test-ixia", address, attributes) yield resource session.DeleteResource(resource.Name) def test_autoload(driver: IxiaChassis2GDriver, autoload_context: AutoLoadCommandContext) -> None: + """Test direct (driver) auto load command.""" inventory = driver.get_inventory(autoload_context) print_inventory(inventory) def test_autoload_session(session: CloudShellAPISession, autoload_resource: ResourceInfo, dut: List[str]) -> None: + """Test indirect (shell) auto load command.""" session.AutoLoad(autoload_resource.Name) resource_details = session.GetResourceDetails(autoload_resource.Name) assert len(resource_details.ChildResources) == 1 - assert resource_details.ChildResources[0].FullAddress == f'{dut[0]}/M1' + assert resource_details.ChildResources[0].FullAddress == f"{dut[0]}/M1" diff --git a/test_requirements.txt b/test_requirements.txt index 5d24de5..8c34c44 100644 --- a/test_requirements.txt +++ b/test_requirements.txt @@ -2,4 +2,9 @@ pytest pyyaml +cloudshell-traffic>=3.3,<3.4 shellfoundry_traffic + +pylint +pre-commit +black[d]