From 3775528bfaef33aa2b14a8059017943872dffb31 Mon Sep 17 00:00:00 2001 From: fpfeng Date: Tue, 14 Feb 2023 11:35:26 +0800 Subject: [PATCH] fix: #17 use enum name --- xoa_converter/converters/fabric.py | 4 +- xoa_converter/converters/rfc2889/adapter.py | 46 ++++++++++----------- xoa_converter/converters/rfc2889/const.py | 21 ++++------ xoa_converter/converters/rfc2889/model.py | 3 +- xoa_converter/entry.py | 2 +- 5 files changed, 37 insertions(+), 39 deletions(-) diff --git a/xoa_converter/converters/fabric.py b/xoa_converter/converters/fabric.py index 08b274b..5bb6998 100644 --- a/xoa_converter/converters/fabric.py +++ b/xoa_converter/converters/fabric.py @@ -1,4 +1,4 @@ -from typing import Type, Protocol, Dict +from typing import Any, Type, Protocol, Dict from xoa_converter import exceptions from xoa_converter import types from .rfc2544.adapter import Converter2544 @@ -7,7 +7,7 @@ class XoaConverter(Protocol): def __init__(self, source_config: str) -> None: ... - def gen(self) -> Dict: ... + def gen(self) -> Dict[str, Any]: ... def get_converter(test_suite_type: types.TestSuiteType) -> Type[XoaConverter]: diff --git a/xoa_converter/converters/rfc2889/adapter.py b/xoa_converter/converters/rfc2889/adapter.py index ba99034..afc1709 100644 --- a/xoa_converter/converters/rfc2889/adapter.py +++ b/xoa_converter/converters/rfc2889/adapter.py @@ -1,6 +1,6 @@ import base64 import hashlib -from typing import Dict, TYPE_CHECKING +from typing import Any, Dict, TYPE_CHECKING from .model import ( LegacyPortRoleHandler, ValkyrieConfiguration2889 as old_model, @@ -49,7 +49,7 @@ def __gen_port_identity(self) -> Dict[str, "PortIdentity"]: count += 1 return port_identity - def __gen_ipv4_addr(self, entity: "LegacyPortEntity"): + def __gen_ipv4_addr(self, entity: "LegacyPortEntity") -> Dict[str, Any]: return dict( address=entity.ip_v4_address, routing_prefix=entity.ip_v4_routing_prefix, @@ -63,7 +63,7 @@ def __gen_ipv4_addr(self, entity: "LegacyPortEntity"): else "0.0.0.0", ) - def __gen_ipv6_addr(self, entity: "LegacyPortEntity"): + def __gen_ipv6_addr(self, entity: "LegacyPortEntity") -> Dict[str, Any]: return dict( address=entity.ip_v6_address, routing_prefix=entity.ip_v6_routing_prefix, @@ -77,7 +77,7 @@ def __gen_ipv6_addr(self, entity: "LegacyPortEntity"): else "::", ) - def __gen_port_conf(self, entity: "LegacyPortEntity"): + def __gen_port_conf(self, entity: "LegacyPortEntity") -> Dict[str, Any]: profile_id = self.data.stream_profile_handler.profile_assignment_map.get(f"guid_{entity.item_id}") return dict( port_slot=self.id_map[entity.item_id][1], @@ -109,13 +109,13 @@ def __gen_port_conf(self, entity: "LegacyPortEntity"): item_id=entity.item_id, ) - def __gen_port_config(self) -> Dict: + def __gen_port_config(self) -> Dict[str, Any]: port_conf: Dict = {} for entity in self.data.port_handler.entity_list: port_conf[self.id_map[entity.item_id][0]] = self.__gen_port_conf(entity) return port_conf - def __gen_frame_size(self): + def __gen_frame_size(self) -> Dict[str, Any]: packet_size = self.data.test_options.packet_sizes packet_size_type = packet_size.packet_size_type fz = packet_size.mixed_length_config.frame_sizes @@ -131,7 +131,7 @@ def __gen_frame_size(self): mixed_length_config=LegacyFrameSizesOptions(**fz).dict(), ) - def __gen_rate_definition(self): + def __gen_rate_definition(self) -> Dict[str, Any]: return dict( rate_type=self.data.test_options.rate_definition.rate_type, rate_fraction=self.data.test_options.rate_definition.rate_fraction, @@ -142,7 +142,7 @@ def __gen_rate_definition(self): rate_bps_l2_unit=self.data.test_options.rate_definition.rate_bps_l2_unit, ) - def __gen_general_test_config(self): + def __gen_general_test_config(self) -> Dict[str, Any]: return dict( frame_sizes=self.__gen_frame_size(), rate_definition=self.__gen_rate_definition(), @@ -158,7 +158,7 @@ def __gen_general_test_config(self): tid_allocation_scope=self.data.tid_allocation_scope, ) - def __gen_rate_iteration_options(self, rate_iteration_options): + def __gen_rate_iteration_options(self, rate_iteration_options) -> Dict[str, Any]: return dict( initial_value=rate_iteration_options.initial_value, minimum_value=rate_iteration_options.minimum_value, @@ -167,14 +167,14 @@ def __gen_rate_iteration_options(self, rate_iteration_options): use_pass_threshold=rate_iteration_options.use_pass_threshold, pass_threshold=rate_iteration_options.pass_threshold, ) - def __gen_rate_sweep_option(self, rate_sweep_options): + def __gen_rate_sweep_option(self, rate_sweep_options) -> Dict[str, Any]: return dict( start_value=rate_sweep_options.start_value, end_value=rate_sweep_options.end_value, step_value=rate_sweep_options.step_value, ) - def __gather_test_case_common_config(self, test_case_config): + def __gather_test_case_common_config(self, test_case_config) -> Dict[str, Any]: return dict( enabled=test_case_config.enabled, duration=test_case_config.duration, @@ -184,14 +184,14 @@ def __gather_test_case_common_config(self, test_case_config): label=test_case_config.label, ) - def __gen_test_port_role(self, port_role_handler: LegacyPortRoleHandler): + def __gen_test_port_role(self, port_role_handler: LegacyPortRoleHandler) -> Dict[str, Any]: role_map = {} for k, v in port_role_handler.role_map.items(): v.role = v.role role_map[k] = v.dict() return {'role_map': role_map} - def __gen_rate_test(self): + def __gen_rate_test(self) -> Dict[str, Any]: rate_test_config = self.data.test_options.test_type_option_map.rate_test all_rate_sub_tests = [] @@ -215,13 +215,13 @@ def __gen_rate_test(self): **self.__gather_test_case_common_config(rate_test_config), ) - def __gen_congestion_control(self): + def __gen_congestion_control(self) -> Dict[str, Any]: return dict( port_role_handler=self.__gen_test_port_role(self.data.test_options.test_type_option_map.congestion_control.port_role_handler), **self.__gather_test_case_common_config(self.data.test_options.test_type_option_map.congestion_control) ) - def __gen_forward_pressure(self): + def __gen_forward_pressure(self) -> Dict[str, Any]: forward_pressure = self.data.test_options.test_type_option_map.forward_pressure return dict( port_role_handler=self.__gen_test_port_role(forward_pressure.port_role_handler), @@ -230,7 +230,7 @@ def __gen_forward_pressure(self): **self.__gather_test_case_common_config(forward_pressure) ) - def __gen_max_forwarding_rate(self): + def __gen_max_forwarding_rate(self) -> Dict[str, Any]: max_forwarding_rate = self.data.test_options.test_type_option_map.max_forwarding_rate return dict( port_role_handler=self.__gen_test_port_role(max_forwarding_rate.port_role_handler), @@ -239,7 +239,7 @@ def __gen_max_forwarding_rate(self): **self.__gather_test_case_common_config(max_forwarding_rate), ) - def __gen_address_caching_capacity(self): + def __gen_address_caching_capacity(self) -> Dict[str, Any]: address_caching_capacity = self.data.test_options.test_type_option_map.address_caching_capacity return dict( port_role_handler=self.__gen_test_port_role(address_caching_capacity.port_role_handler), @@ -259,7 +259,7 @@ def __gen_address_caching_capacity(self): **self.__gather_test_case_common_config(address_caching_capacity), ) - def __gen_address_learning_rate(self): + def __gen_address_learning_rate(self) -> Dict[str, Any]: address_learning_rate = self.data.test_options.test_type_option_map.address_learning_rate return dict( port_role_handler=self.__gen_test_port_role(address_learning_rate.port_role_handler), @@ -280,7 +280,7 @@ def __gen_address_learning_rate(self): **self.__gather_test_case_common_config(address_learning_rate), ) - def __gen_errored_frames_filtering(self): + def __gen_errored_frames_filtering(self) -> Dict[str, Any]: errored_frames_filtering = self.data.test_options.test_type_option_map.errored_frames_filtering return dict( port_role_handler=self.__gen_test_port_role(errored_frames_filtering.port_role_handler), @@ -293,7 +293,7 @@ def __gen_errored_frames_filtering(self): **self.__gather_test_case_common_config(errored_frames_filtering), ) - def __gen_broadcast_forwarding(self): + def __gen_broadcast_forwarding(self) -> Dict[str, Any]: broadcast_forwarding = self.data.test_options.test_type_option_map.broadcast_forwarding return dict( port_role_handler=self.__gen_test_port_role(broadcast_forwarding.port_role_handler), @@ -301,7 +301,7 @@ def __gen_broadcast_forwarding(self): **self.__gather_test_case_common_config(broadcast_forwarding), ) - def __gen_test_types_config(self): + def __gen_test_types_config(self) -> Dict[str, Any]: return dict( rate_test=self.__gen_rate_test(), congestion_control=self.__gen_congestion_control(), @@ -313,7 +313,7 @@ def __gen_test_types_config(self): broadcast_forwarding=self.__gen_broadcast_forwarding(), ) - def __gen_protocol_segments(self) -> Dict: + def __gen_protocol_segments(self) -> Dict[str, Any]: protocol_segments_profile = {} for profile in self.data.stream_profile_handler.entity_list: @@ -344,7 +344,7 @@ def __gen_protocol_segments(self) -> Dict: protocol_segments_profile[profile.item_id] = dict(header_segments=header_segments) return protocol_segments_profile - def gen(self) -> "Dict": + def gen(self) -> Dict[str, Any]: port_identities = self.__gen_port_identity() config = dict( ports_configuration=self.__gen_port_config(), diff --git a/xoa_converter/converters/rfc2889/const.py b/xoa_converter/converters/rfc2889/const.py index 5d56f76..e7d99c5 100644 --- a/xoa_converter/converters/rfc2889/const.py +++ b/xoa_converter/converters/rfc2889/const.py @@ -83,14 +83,6 @@ class LegacyRateResultScopeType(Enum): PER_SOURCE_PORT = "persrcportresult" -# special_type_map = { -# "ip": "ipv4", -# "mldv2_ar": "mldv2ar", -# "igmpv3_mr": "igmpv3mr", -# "igmpv3_gr": "igmpv3gr", -# } - - class LegacyTestType(Enum): RATE_TEST = "RateTest" CONGESTION_CONTROL = "CongestionControl" @@ -152,10 +144,6 @@ class LegacySegmentType(Enum): for i in range(1, 65): OSegmentType[f"RAW_{i}"] = f"raw_{i}" # type: ignore - @property - def core(self): - return SegmentType[self.name] - @property def is_raw(self) -> bool: return self.value.lower().startswith("raw") @@ -182,6 +170,7 @@ class TestPortMacMode(Enum): def is_use_learning_base_address(self): return self == TestPortMacMode.USE_LEARNING_MAC_BASE_ADDRESS + class LearningPortDMacMode(Enum): USE_TEST_PORT_MAC = "UseTestPortMac" USE_BROADCAST = "UseBroadcast" @@ -190,6 +179,7 @@ class LearningPortDMacMode(Enum): def is_use_broadcast(self): return self == LearningPortDMacMode.USE_BROADCAST + class LearningSequencePortDMacMode(Enum): USE_INCREMENTING_MAC_ADDRESSES = "UseIncrementingMacAddresses" USE_RANDOM_MAC_ADDRESSES = "UseRandomMacAddresses" @@ -333,3 +323,10 @@ class LegacyFecMode(Enum): ON = "on" OFF = "off" FC_FEC = "FIRECODE" + + +class LatencyMode(Enum): + FIRST2LAST = "First_To_Last" + LAST2LAST = "Last_To_Last" + FIRST2FIRST = "First_To_First" + LAST2FIRST = "Last_To_First" \ No newline at end of file diff --git a/xoa_converter/converters/rfc2889/model.py b/xoa_converter/converters/rfc2889/model.py index f18dbe6..5e4bc2e 100644 --- a/xoa_converter/converters/rfc2889/model.py +++ b/xoa_converter/converters/rfc2889/model.py @@ -4,6 +4,7 @@ from pydantic import BaseModel, Field, validator from .const import ( BRRModeStr, + LatencyMode, LearningPortDMacMode, LearningSequencePortDMacMode, LegacyDurationType, @@ -231,7 +232,7 @@ class TestOptions(BaseModel): packet_sizes: PacketSizes = Field(alias="PacketSizes") rate_definition: LegacyRateDefinition = Field(alias="RateDefinition") flow_creation_options: FlowCreationOptions = Field(alias="FlowCreationOptions") - latency_mode: str = Field(alias="LatencyMode") + latency_mode: LatencyMode = Field(alias="LatencyMode") toggle_sync_state: bool = Field(alias="ToggleSyncState") sync_off_duration: int = Field(alias="SyncOffDuration") sync_on_duration: int = Field(alias="SyncOnDuration") diff --git a/xoa_converter/entry.py b/xoa_converter/entry.py index b1d3701..0e38062 100644 --- a/xoa_converter/entry.py +++ b/xoa_converter/entry.py @@ -9,7 +9,7 @@ def default(self, obj): if isinstance(obj, Enum): return obj.name.upper() else: - return json.JSONEncoder.default(self, obj) + return super().default(obj) def converter(