From 330367c5e7ad52ef8eff0db4affc44e552aa67cd Mon Sep 17 00:00:00 2001 From: cogu Date: Wed, 2 Oct 2024 22:56:42 +0200 Subject: [PATCH] Refactor event creation API - Add new BehaviorSettings class to Workspace - Event names are created dynamically from settings --- src/autosar/xml/element.py | 162 +++++++++++++++++-- src/autosar/xml/workspace.py | 5 +- tests/xml/test_software_component_builder.py | 64 +++++--- tests/xml/test_workspace.py | 54 +++++++ 4 files changed, 244 insertions(+), 41 deletions(-) diff --git a/src/autosar/xml/element.py b/src/autosar/xml/element.py index 469febc..efc5ba6 100644 --- a/src/autosar/xml/element.py +++ b/src/autosar/xml/element.py @@ -2930,13 +2930,51 @@ def ref(self) -> PackageRef: return None if ref_str is None else PackageRef(ref_str) +class BehaviorSettings: + """ + Enables users to customize settings in SwcInternalBehavior such as naming conventions. + """ + + def __init__(self) -> None: + + # Default prefix used for generating event names + self.background_event_prefix: str | None = None # BackgroundEvent prefix + self.data_receive_error_event_prefix: str | None = None # DataReceiveErrorEvent prefix + self.data_receive_event_prefix: str | None = None # DataReceivedEvent prefix + self.init_event_prefix: str | None = None # InitEvent prefix + self.operation_invoked_event_prefix: str | None = None # OperationInvokedEvent prefix + self.swc_mode_manager_error_event_prefix: str | None = None # SwcModeManagerErrorEvent prefix + self.swc_mode_switch_event_prefix: str | None = None # SwcModeSwitchEvent prefix + self.timing_event_prefix: str | None = None # TimingEvent prefix + + def set_value(self, name: str, value: str): + """ + Updates a single value with error check + """ + if hasattr(self, name): + if not isinstance(value, str): + raise TypeError(f"value: Expected string type. Got {str(type(value))}") + setattr(self, name, value) + else: + raise KeyError(f"name: Invalid name '{name}'") + + def update(self, value_map: dict[str, str]): + """ + Updates multiple values using keys in value_map, with error-check + """ + for name, value in value_map.items(): + self.set_value(name, value) + + class PackageCollection: """ Base class that maintains a collection of AUTOSAR packages """ - def __init__(self, packages: list[Package] | None = None) -> None: + def __init__(self, packages: list[Package] | None = None, + behavior_settings: BehaviorSettings | None = None) -> None: self.parent = None + self.behavior_settings = behavior_settings self.packages: list[Package] = [] # .PACKAGES self._package_dict = {} # internal package map if packages is not None: @@ -6304,6 +6342,16 @@ def _get_valid_parent(self) -> SwComponentType: raise RuntimeError("Behavior object doesn't have a valid parent") return self.parent + def _get_valid_behavior_settings(self) -> BehaviorSettings: + """ + Verifies that the root collection has a valid behavior_settings object and returns it + """ + swc = self._get_valid_parent() + workspace = swc.root_collection() + if not isinstance(workspace.behavior_settings, BehaviorSettings): + raise TypeError("Root collection doesn't seem to be a valid workspace") + return workspace.behavior_settings + def ref(self) -> SwcInternalBehaviorRef | None: """ Returns a reference to this element or @@ -6377,8 +6425,8 @@ def _make_unique_event_name(self, event_name: str) -> str: return event_name def create_background_event(self, - event_name: str, runnable_name: str, + event_name: str | None = None, **kwargs ) -> BackgroundEvent: """ @@ -6387,15 +6435,24 @@ def create_background_event(self, runnable = self.find_runnable(runnable_name) if runnable is None: raise KeyError(f"Found no runnable with name '{runnable_name}'") + if event_name is None: + behavior_settings = self._get_valid_behavior_settings() + if behavior_settings.background_event_prefix: + event_name = behavior_settings.background_event_prefix + runnable_name + else: + msg = "event_name: Unable to dynamically create event name,"\ + " background_event_prefix is not set in behavior settings" + raise RuntimeError(msg) + assert isinstance(event_name, str) unique_event_name = self._make_unique_event_name(event_name) event = BackgroundEvent(unique_event_name, runnable.ref(), **kwargs) self.append_event(event) return event def create_data_receive_error_event(self, - event_name: str, runnable_name: str, port_data_element: str, + event_name: str | None = None, **kwargs ) -> DataReceiveErrorEvent: """ @@ -6408,7 +6465,6 @@ def create_data_receive_error_event(self, runnable = self.find_runnable(runnable_name) if runnable is None: raise KeyError(f"Found no runnable with name '{runnable_name}'") - unique_event_name = self._make_unique_event_name(event_name) name_parts = split_ref_strict(port_data_element) if len(name_parts) == 1: port_name, data_element_name = name_parts[0], None @@ -6421,6 +6477,18 @@ def create_data_receive_error_event(self, if target_data_element is None: msg = f"port_data_element: '{port_data_element}' does not name an existing data element in port interface" raise ValueError(msg) + if event_name is None: + behavior_settings = self._get_valid_behavior_settings() + if behavior_settings.data_receive_error_event_prefix: + event_name = behavior_settings.data_receive_error_event_prefix + "_".join([runnable_name, + context_port.name, + target_data_element.name]) + else: + msg = "event_name: Unable to dynamically create event name,"\ + " data_receive_error_event_prefix is not set in behavior settings" + raise RuntimeError(msg) + assert isinstance(event_name, str) + unique_event_name = self._make_unique_event_name(event_name) event = DataReceiveErrorEvent.make(unique_event_name, runnable.ref(), context_port.ref(), @@ -6430,9 +6498,9 @@ def create_data_receive_error_event(self, return event def create_data_received_event(self, - event_name: str, runnable_name: str, data_element_ref: str, + event_name: str | None = None, **kwargs ) -> DataReceivedEvent: """ @@ -6445,7 +6513,6 @@ def create_data_received_event(self, runnable = self.find_runnable(runnable_name) if runnable is None: raise KeyError(f"Found no runnable with name '{runnable_name}'") - unique_event_name = self._make_unique_event_name(event_name) name_parts = split_ref_strict(data_element_ref) if len(name_parts) == 1: port_name, data_element_name = name_parts[0], None @@ -6458,6 +6525,18 @@ def create_data_received_event(self, if target_data_element is None: msg = f"data_element_ref: '{data_element_ref}' does not name an existing data element in port interface" raise ValueError(msg) + if event_name is None: + behavior_settings = self._get_valid_behavior_settings() + if behavior_settings.data_receive_event_prefix: + event_name = behavior_settings.data_receive_event_prefix + "_".join([runnable_name, + context_port.name, + target_data_element.name]) + else: + msg = "event_name: Unable to dynamically create event name,"\ + " data_receive_event_prefix is not set in behavior settings" + raise RuntimeError(msg) + assert isinstance(event_name, str) + unique_event_name = self._make_unique_event_name(event_name) event = DataReceivedEvent.make(unique_event_name, runnable.ref(), context_port.ref(), @@ -6467,9 +6546,9 @@ def create_data_received_event(self, return event def create_data_send_completed_event(self, - event_name: str, runnable_name: str, data_element_ref: str, + event_name: str | None = None, **kwargs ) -> DataSendCompletedEvent: """ @@ -6483,9 +6562,9 @@ def create_data_send_completed_event(self, raise NotImplementedError("References to data send points are not yet supported") def create_data_write_completed_event(self, - event_name: str, runnable_name: str, data_element_ref: str, + event_name: str | None = None, **kwargs ) -> DataWriteCompletedEvent: """ @@ -6499,8 +6578,8 @@ def create_data_write_completed_event(self, raise NotImplementedError("References to data write access are not yet supported") def create_init_event(self, - event_name: str, runnable_name: str, + event_name: str | None = None, **kwargs ) -> InitEvent: """ @@ -6509,15 +6588,24 @@ def create_init_event(self, runnable = self.find_runnable(runnable_name) if runnable is None: raise KeyError(f"Found no runnable with name '{runnable_name}'") + if event_name is None: + behavior_settings = self._get_valid_behavior_settings() + if behavior_settings.init_event_prefix: + event_name = behavior_settings.init_event_prefix + runnable_name + else: + msg = "event_name: Unable to dynamically create event name,"\ + " init_event_prefix is not set in behavior settings" + raise RuntimeError(msg) + assert isinstance(event_name, str) unique_event_name = self._make_unique_event_name(event_name) event = InitEvent(unique_event_name, runnable.ref(), **kwargs) self.append_event(event) return event def create_operation_invoked_event(self, - event_name: str, runnable_name: str, operation_ref: str, + event_name: str | None = None, **kwargs ) -> OperationInvokedEvent: """ @@ -6528,7 +6616,6 @@ def create_operation_invoked_event(self, runnable = self.find_runnable(runnable_name) if runnable is None: raise KeyError(f"Found no runnable with name '{runnable_name}'") - unique_event_name = self._make_unique_event_name(event_name) name_parts = split_ref_strict(operation_ref) if len(name_parts) == 1: port_name, operation_name = name_parts[0], None @@ -6540,6 +6627,19 @@ def create_operation_invoked_event(self, target_provided_operation = swc.get_operation_in_port(context_port, operation_name) if target_provided_operation is None: raise ValueError(f"operation_ref: '{operation_ref}' does not name a valid operation in port interface") + if event_name is None: + behavior_settings = self._get_valid_behavior_settings() + if behavior_settings.operation_invoked_event_prefix: + prefix = behavior_settings.operation_invoked_event_prefix + event_name = prefix + "_".join([runnable_name, + context_port.name, + target_provided_operation.name]) + else: + msg = "event_name: Unable to dynamically create event name,"\ + " operation_invoked_event_prefix is not set in behavior settings" + raise RuntimeError(msg) + assert isinstance(event_name, str) + unique_event_name = self._make_unique_event_name(event_name) event = OperationInvokedEvent.make(unique_event_name, runnable.ref(), context_port.ref(), @@ -6549,9 +6649,9 @@ def create_operation_invoked_event(self, return event def create_swc_mode_manager_error_event(self, - event_name: str, runnable_name: str, port_name: str, + event_name: str | None = None, **kwargs ) -> SwcModeManagerErrorEvent: """ @@ -6561,7 +6661,6 @@ def create_swc_mode_manager_error_event(self, runnable = self.find_runnable(runnable_name) if runnable is None: raise KeyError(f"Found no runnable with name '{runnable_name}'") - unique_event_name = self._make_unique_event_name(event_name) context_port = swc.find_p_port(port_name) if context_port is None: raise ValueError(f"port_name: '{port_name}' does not name an existing P-PORT or PR-PORT") @@ -6569,6 +6668,19 @@ def create_swc_mode_manager_error_event(self, if context_mode_declaration_group is None: msg = f"port_name: '{port_name}' does not name a valid ModeDeclarationGroupPrototype in port interface" raise ValueError(msg) + if event_name is None: + behavior_settings = self._get_valid_behavior_settings() + if behavior_settings.swc_mode_manager_error_event_prefix: + prefix = behavior_settings.swc_mode_manager_error_event_prefix + event_name = prefix + "_".join([runnable_name, + context_port.name, + context_mode_declaration_group.name]) + else: + msg = "event_name: Unable to dynamically create event name,"\ + " swc_mode_manager_error_event_prefix is not set in behavior settings" + raise RuntimeError(msg) + assert isinstance(event_name, str) + unique_event_name = self._make_unique_event_name(event_name) event = SwcModeManagerErrorEvent.make(unique_event_name, runnable.ref(), context_port.ref(), @@ -6578,10 +6690,10 @@ def create_swc_mode_manager_error_event(self, return event def create_swc_mode_mode_switch_event(self, - event_name: str, runnable_name: str, mode_ref: str | list[str] | tuple[str, str], activation: ar_enum.ModeActivationKind | None = None, + event_name: str | None = None, **kwargs ) -> SwcModeSwitchEvent: """ @@ -6596,6 +6708,16 @@ def create_swc_mode_mode_switch_event(self, runnable = self.find_runnable(runnable_name) if runnable is None: raise KeyError(f"Found no runnable with name '{runnable_name}'") + if event_name is None: + behavior_settings = self._get_valid_behavior_settings() + if behavior_settings.swc_mode_switch_event_prefix: + prefix = behavior_settings.swc_mode_switch_event_prefix + event_name = prefix + runnable_name + else: + msg = "event_name: Unable to dynamically create event name,"\ + " swc_mode_switch_event_prefix is not set in behavior settings" + raise RuntimeError(msg) + assert isinstance(event_name, str) unique_event_name = self._make_unique_event_name(event_name) if isinstance(mode_ref, str): context_port, context_mode_declaration_group, target_mode_declaration = ( @@ -6628,10 +6750,10 @@ def create_swc_mode_mode_switch_event(self, return event def create_timing_event(self, - event_name: str, runnable_name: str, period: int | float | None = None, offset: int | float | None = None, + event_name: str | None = None, **kwargs) -> TimingEvent: """ Adds a new TimingEvent to this object @@ -6639,6 +6761,16 @@ def create_timing_event(self, runnable = self.find_runnable(runnable_name) if runnable is None: raise KeyError(f"Found no runnable with name '{runnable_name}'") + if event_name is None: + behavior_settings = self._get_valid_behavior_settings() + if behavior_settings.timing_event_prefix: + prefix = behavior_settings.timing_event_prefix + event_name = prefix + runnable_name + else: + msg = "event_name: Unable to dynamically create event name,"\ + " timing_event_prefix is not set in behavior settings" + raise RuntimeError(msg) + assert isinstance(event_name, str) unique_event_name = self._make_unique_event_name(event_name) event = TimingEvent(unique_event_name, runnable.ref(), period, offset, **kwargs) self.append_event(event) diff --git a/src/autosar/xml/workspace.py b/src/autosar/xml/workspace.py index b6ca8e9..a343455 100644 --- a/src/autosar/xml/workspace.py +++ b/src/autosar/xml/workspace.py @@ -79,12 +79,12 @@ class Workspace(ar_element.PackageCollection): """ def __init__(self, config_file_path: str | None = None, document_root: str | None = None) -> None: + super().__init__(behavior_settings=ar_element.BehaviorSettings()) self.namespaces: dict[str, Namespace] = {} self.documents: list[DocumentConfig] = [] self.document_mappings: list[PackageToDocumentMapping] = [] self.document_root = document_root self.package_map: dict[str, ar_element.Package] = {} # Each key is user-defined - super().__init__() if config_file_path is not None: self.load_config(config_file_path) @@ -219,6 +219,9 @@ def load_config(self, file_path: str) -> None: if document_mapping is not None: for mapping in document_mapping.values(): self._create_document_mapping_from_config(mapping) + behavior_settings = config.get("behavior", None) + if behavior_settings is not None: + self.behavior_settings.update(behavior_settings) def _write_document_from_config(self, writer: Writer, diff --git a/tests/xml/test_software_component_builder.py b/tests/xml/test_software_component_builder.py index 44750d8..5e1ecb2 100644 --- a/tests/xml/test_software_component_builder.py +++ b/tests/xml/test_software_component_builder.py @@ -389,131 +389,143 @@ def create_swc(self, workspace: autosar.xml.Workspace) -> ar_element.Application def test_create_background_event(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("background_event_prefix", "BT_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("MyApplication_Background") - event = behavior.create_background_event("BGR_MyApplication_Background", "MyApplication_Background") + event = behavior.create_background_event("MyApplication_Background") self.assertIsInstance(event, ar_element.BackgroundEvent) + self.assertEqual(event.name, "BT_MyApplication_Background") self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/MyApplication_Background") def test_create_data_receive_error_event_using_port_name_only(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("data_receive_error_event_prefix", "DRET_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("EngineSpeed_Updated") - event = behavior.create_data_receive_error_event("RCVERR_EngineSpeed", "EngineSpeed_Updated", "EngineSpeed") + event = behavior.create_data_receive_error_event("EngineSpeed_Updated", "EngineSpeed") self.assertIsInstance(event, ar_element.DataReceiveErrorEvent) + self.assertEqual(event.name, "DRET_EngineSpeed_Updated_EngineSpeed_EngineSpeed") self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/EngineSpeed_Updated") self.assertEqual(str(event.data.context_port), "/ComponentTypes/MyApplication/EngineSpeed") self.assertEqual(str(event.data.target_data_element), "/PortInterfaces/EngineSpeed_I/EngineSpeed") def test_create_data_receive_error_event_using_port_and_data_element(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("data_receive_error_event_prefix", "DRET_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("EngineSpeed_Updated") - event = behavior.create_data_receive_error_event("RCVERR_EngineSpeed", - "EngineSpeed_Updated", - "EngineSpeed/EngineSpeed") + event = behavior.create_data_receive_error_event("EngineSpeed_Updated", "EngineSpeed/EngineSpeed") self.assertIsInstance(event, ar_element.DataReceiveErrorEvent) + self.assertEqual(event.name, "DRET_EngineSpeed_Updated_EngineSpeed_EngineSpeed") self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/EngineSpeed_Updated") self.assertEqual(str(event.data.context_port), "/ComponentTypes/MyApplication/EngineSpeed") self.assertEqual(str(event.data.target_data_element), "/PortInterfaces/EngineSpeed_I/EngineSpeed") def test_create_data_received_event_using_port_name_only(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("data_receive_event_prefix", "DRT_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("EngineSpeed_Updated") - event = behavior.create_data_received_event("RCV_EngineSpeed", "EngineSpeed_Updated", "EngineSpeed") + event = behavior.create_data_received_event("EngineSpeed_Updated", "EngineSpeed") self.assertIsInstance(event, ar_element.DataReceivedEvent) + self.assertEqual(event.name, "DRT_EngineSpeed_Updated_EngineSpeed_EngineSpeed") self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/EngineSpeed_Updated") self.assertEqual(str(event.data.context_port), "/ComponentTypes/MyApplication/EngineSpeed") self.assertEqual(str(event.data.target_data_element), "/PortInterfaces/EngineSpeed_I/EngineSpeed") def test_create_data_received_event_using_port_and_data_element(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("data_receive_event_prefix", "DRT_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("EngineSpeed_Updated") - event = behavior.create_data_received_event("RCV_EngineSpeed", "EngineSpeed_Updated", "EngineSpeed/EngineSpeed") + event = behavior.create_data_received_event("EngineSpeed_Updated", "EngineSpeed/EngineSpeed") self.assertIsInstance(event, ar_element.DataReceivedEvent) + self.assertEqual(event.name, "DRT_EngineSpeed_Updated_EngineSpeed_EngineSpeed") self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/EngineSpeed_Updated") self.assertEqual(str(event.data.context_port), "/ComponentTypes/MyApplication/EngineSpeed") self.assertEqual(str(event.data.target_data_element), "/PortInterfaces/EngineSpeed_I/EngineSpeed") def test_create_init_event(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("init_event_prefix", "IT_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("MyApplication_Init") - event = behavior.create_background_event("INIT_MyApplication_Init", "MyApplication_Init") - self.assertIsInstance(event, ar_element.BackgroundEvent) + event = behavior.create_init_event("MyApplication_Init") + self.assertIsInstance(event, ar_element.InitEvent) + self.assertEqual(event.name, "IT_MyApplication_Init") self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/MyApplication_Init") def test_create_operation_invoked_event_port_name_only(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("operation_invoked_event_prefix", "OIT_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("MyApplication_GetTime") - event = behavior.create_operation_invoked_event("CALL_MyApplication_GetTime", - "MyApplication_GetTime", - "Timer") + event = behavior.create_operation_invoked_event("MyApplication_GetTime", "Timer") self.assertIsInstance(event, ar_element.OperationInvokedEvent) + self.assertEqual(event.name, "OIT_MyApplication_GetTime_Timer_GetTime") self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/MyApplication_GetTime") self.assertEqual(str(event.operation.context_port), "/ComponentTypes/MyApplication/Timer") self.assertEqual(str(event.operation.target_provided_operation), "/PortInterfaces/Timer_I/GetTime") def test_create_operation_invoked_event_with_operation_name(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("operation_invoked_event_prefix", "OIT_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("MyApplication_GetTime") - event = behavior.create_operation_invoked_event("CALL_MyApplication_GetTime", - "MyApplication_GetTime", - "Timer/GetTime") + event = behavior.create_operation_invoked_event("MyApplication_GetTime", "Timer/GetTime") self.assertIsInstance(event, ar_element.OperationInvokedEvent) + self.assertEqual(event.name, "OIT_MyApplication_GetTime_Timer_GetTime") self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/MyApplication_GetTime") self.assertEqual(str(event.operation.context_port), "/ComponentTypes/MyApplication/Timer") self.assertEqual(str(event.operation.target_provided_operation), "/PortInterfaces/Timer_I/GetTime") def test_create_swc_mode_manager_error_event(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("swc_mode_manager_error_event_prefix", "MMET_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("MyApplication_ApplicationModeError") - event = behavior.create_swc_mode_manager_error_event("MDERR_MyApplication_ApplicationModeError", - "MyApplication_ApplicationModeError", - "ApplicationMode") + event = behavior.create_swc_mode_manager_error_event("MyApplication_ApplicationModeError", "ApplicationMode") self.assertIsInstance(event, ar_element.SwcModeManagerErrorEvent) + self.assertEqual(event.name, "MMET_MyApplication_ApplicationModeError_ApplicationMode_mode",) self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/MyApplication_ApplicationModeError") self.assertEqual(str(event.mode_group.context_port), "/ComponentTypes/MyApplication/ApplicationMode") self.assertEqual(str(event.mode_group.context_mode_declaration_group_prototype), "/PortInterfaces/ApplicationMode_I/mode") - def test_create_swc_mode_mode_switch_event(self): + def test_create_swc_mode_switch_event(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("swc_mode_switch_event_prefix", "MST_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("MyApplication_Init") - event = behavior.create_swc_mode_mode_switch_event("MSW_MyApplication_Init", - "MyApplication_Init", + event = behavior.create_swc_mode_mode_switch_event("MyApplication_Init", "VehicleMode/ACCESSORY") self.assertIsInstance(event, ar_element.SwcModeSwitchEvent) + self.assertEqual(event.name, "MST_MyApplication_Init",) self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/MyApplication_Init") self.assertEqual(str(event.mode.context_port), "/ComponentTypes/MyApplication/VehicleMode") self.assertEqual(str(event.mode.context_mode_declaration_group_prototype), "/PortInterfaces/VehicleMode_I/mode") self.assertEqual(str(event.mode.target_mode_declaration), "/ModeDeclarations/VehicleMode/ACCESSORY") - def test_create_swc_mode_mode_switch_event_with_transition(self): + def test_create_swc_mode_switch_event_with_transition(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("swc_mode_switch_event_prefix", "MST_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("MyApplication_Init") - event = behavior.create_swc_mode_mode_switch_event("MSW_MyApplication_Init", - "MyApplication_Init", + event = behavior.create_swc_mode_mode_switch_event("MyApplication_Init", ["VehicleMode/PARKING", "VehicleMode/ACCESSORY"]) self.assertIsInstance(event, ar_element.SwcModeSwitchEvent) + self.assertEqual(event.name, "MST_MyApplication_Init",) self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/MyApplication_Init") self.assertEqual(str(event.mode[0].context_port), "/ComponentTypes/MyApplication/VehicleMode") self.assertEqual(str(event.mode[0].context_mode_declaration_group_prototype), @@ -526,11 +538,13 @@ def test_create_swc_mode_mode_switch_event_with_transition(self): def test_create_timing_event_without_offset(self): workspace = autosar.xml.Workspace() + workspace.behavior_settings.set_value("timing_event_prefix", "TMT_") swc = self.create_swc(workspace) behavior = swc.internal_behavior behavior.create_runnable("MyApplication_Run") - event = behavior.create_timing_event("TM_MyApplication_Run", "MyApplication_Run", 0.02) + event = behavior.create_timing_event("MyApplication_Run", 0.02) self.assertIsInstance(event, ar_element.TimingEvent) + self.assertEqual(event.name, "TMT_MyApplication_Run") self.assertEqual(str(event.start_on_event), self.expected_behavior_ref + "/MyApplication_Run") self.assertAlmostEqual(event.period, 0.02) self.assertIsNone(event.offset) diff --git a/tests/xml/test_workspace.py b/tests/xml/test_workspace.py index be1df52..57897c1 100644 --- a/tests/xml/test_workspace.py +++ b/tests/xml/test_workspace.py @@ -146,5 +146,59 @@ def test_get_package_ref_by_role_with_custom_role(self): "/DataTypes") +class BehaviorSettingsTests(unittest.TestCase): + + def test_background_event_prefix(self): + workspace = ar_workspace.Workspace() + workspace.behavior_settings.set_value("background_event_prefix", "BT_") + self.assertEqual(workspace.behavior_settings.background_event_prefix, "BT_") + + def test_data_receive_error_event_prefix(self): + workspace = ar_workspace.Workspace() + workspace.behavior_settings.set_value("data_receive_error_event_prefix", "DRET_") + self.assertEqual(workspace.behavior_settings.data_receive_error_event_prefix, "DRET_") + + def test_data_receive_event_prefix(self): + workspace = ar_workspace.Workspace() + workspace.behavior_settings.set_value("data_receive_event_prefix", "DRT_") + self.assertEqual(workspace.behavior_settings.data_receive_event_prefix, "DRT_") + + def test_init_event_prefix(self): + workspace = ar_workspace.Workspace() + workspace.behavior_settings.set_value("init_event_prefix", "IT_") + self.assertEqual(workspace.behavior_settings.init_event_prefix, "IT_") + + def test_operation_invoked_event_prefix(self): + workspace = ar_workspace.Workspace() + workspace.behavior_settings.set_value("operation_invoked_event_prefix", "OIT_") + self.assertEqual(workspace.behavior_settings.operation_invoked_event_prefix, "OIT_") + + def test_swc_mode_manager_error_event_prefix(self): + workspace = ar_workspace.Workspace() + workspace.behavior_settings.set_value("swc_mode_manager_error_event_prefix", "MMET_") + self.assertEqual(workspace.behavior_settings.swc_mode_manager_error_event_prefix, "MMET_") + + def test_swc_mode_switch_event_prefix(self): + workspace = ar_workspace.Workspace() + workspace.behavior_settings.set_value("swc_mode_switch_event_prefix", "MST_") + self.assertEqual(workspace.behavior_settings.swc_mode_switch_event_prefix, "MST_") + + def test_timing_event_prefix(self): + workspace = ar_workspace.Workspace() + workspace.behavior_settings.set_value("timing_event_prefix", "TMT_") + self.assertEqual(workspace.behavior_settings.timing_event_prefix, "TMT_") + + def test_set_multiple(self): + workspace = ar_workspace.Workspace() + workspace.behavior_settings.update({"init_event_prefix": "IT_", + "operation_invoked_event_prefix": "OIT_", + "swc_mode_switch_event_prefix": "MST_", + "timing_event_prefix": "TMT_"}) + self.assertEqual(workspace.behavior_settings.init_event_prefix, "IT_") + self.assertEqual(workspace.behavior_settings.operation_invoked_event_prefix, "OIT_") + self.assertEqual(workspace.behavior_settings.swc_mode_switch_event_prefix, "MST_") + self.assertEqual(workspace.behavior_settings.timing_event_prefix, "TMT_") + + if __name__ == '__main__': unittest.main()