Refactor utils library for state machine checks

Signed-off-by: romanodanilo <[email protected]>

qc_otx/checks/models.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from lxml import etree
-from typing import Union
+from typing import Union, List
 
 from qc_baselib import Configuration, Result
 
@@ -24,3 +24,38 @@ class CheckerData:
     config: Configuration
     result: Result
     schema_version: str
+
+
+@dataclass
+class SMTrigger:
+    id: str
+    name: Union[str, None]
+    xml_element: etree._Element
+
+
+@dataclass
+class SMTransition:
+    id: str
+    name: Union[str, None]
+    target: Union[str, None]
+    xml_element: etree._Element
+
+
+@dataclass
+class SMState:
+    id: str
+    name: Union[str, None]
+    is_initial: bool
+    is_completed: bool
+    transitions: List[SMTransition]
+    target_state_ids: List[str]
+    triggers: List[SMTrigger]
+    xml_element: etree._Element
+
+
+@dataclass
+class StateMachine:
+    id: str
+    name: str
+    states: List[SMState]
+    xml_element: etree._Element

qc_otx/checks/state_machine_checker/__init__.py

@@ -2,5 +2,3 @@
 from . import state_machine_checker as state_machine_checker
 from . import no_procedure_realization as no_procedure_realization
 from . import mandatory_target_state as mandatory_target_state
-from . import models as models
-from . import utils as utils

qc_otx/checks/state_machine_checker/mandatory_target_state.py

@@ -3,14 +3,10 @@
 from qc_baselib import IssueSeverity
 
 from qc_otx import constants
-from qc_otx.checks import models
+from qc_otx.checks import models, utils
 
 from qc_otx.checks.state_machine_checker import state_machine_constants
 
-from qc_otx.checks.state_machine_checker import utils as state_machine_utils
-
-logging.basicConfig(level=logging.DEBUG)
-
 
 def check_rule(checker_data: models.CheckerData) -> None:
     """
@@ -40,17 +36,15 @@ def check_rule(checker_data: models.CheckerData) -> None:
 
     tree = checker_data.input_file_xml_root
     root = tree.getroot()
-    nsmap = {k: v for k, v in root.nsmap.items() if k is not None}
+    nsmap = utils.get_namespace_map(root)
 
     if "smp" not in nsmap:
         logging.error(
             'No state machine procedure prefix "smp" found in document namespaces. Abort state machine procedure checks...'
         )
         return
 
-    state_machine_procedures = tree.xpath(
-        "//*[@xsi:type='smp:StateMachineProcedure']", namespaces=nsmap
-    )
+    state_machine_procedures = utils.get_state_machine_procedures(tree, nsmap)
 
     if state_machine_procedures is None:
         return
@@ -60,9 +54,7 @@ def check_rule(checker_data: models.CheckerData) -> None:
     # smp = "state machine procedure"
     for state_machine_procedure in state_machine_procedures:
 
-        state_machine = state_machine_utils.get_state_machine(
-            state_machine_procedure, nsmap
-        )
+        state_machine = utils.get_state_machine(state_machine_procedure, nsmap)
 
         if state_machine is None:
             return

qc_otx/checks/state_machine_checker/no_procedure_realization.py

@@ -3,12 +3,10 @@
 from qc_baselib import IssueSeverity
 
 from qc_otx import constants
-from qc_otx.checks import models
+from qc_otx.checks import models, utils
 
 from qc_otx.checks.state_machine_checker import state_machine_constants
 
-logging.basicConfig(level=logging.DEBUG)
-
 
 def check_rule(checker_data: models.CheckerData) -> None:
     """
@@ -38,17 +36,15 @@ def check_rule(checker_data: models.CheckerData) -> None:
 
     tree = checker_data.input_file_xml_root
     root = tree.getroot()
-    nsmap = {k: v for k, v in root.nsmap.items() if k is not None}
+    nsmap = utils.get_namespace_map(root)
 
     if "smp" not in nsmap:
         logging.error(
             'No state machine procedure prefix "smp" found in document namespaces. Abort state machine procedure checks...'
         )
         return
 
-    state_machine_procedures = tree.xpath(
-        "//*[@xsi:type='smp:StateMachineProcedure']", namespaces=nsmap
-    )
+    state_machine_procedures = utils.get_state_machine_procedures(tree, nsmap)
 
     if state_machine_procedures is None:
         return

qc_otx/checks/utils.py

@@ -1,7 +1,15 @@
 from lxml import etree
-from typing import Union, List
-from qc_otx.checks.models import QueueNode, AttributeInfo
+from typing import Union, List, Dict
+from qc_otx.checks.models import (
+    QueueNode,
+    AttributeInfo,
+    SMState,
+    SMTransition,
+    SMTrigger,
+    StateMachine,
+)
 import re
+import logging
 
 
 def get_data_model_version(root: etree._Element):
@@ -94,3 +102,91 @@ def compare_versions(version1: str, version2: str) -> int:
         return 1
     else:
         return 0
+
+
+def get_state_machine(input_node: etree._Element, nsmap: Dict) -> StateMachine:
+
+    smp_id = input_node.get("id")
+    smp_name = input_node.get("name")
+
+    smp_realisation = input_node.xpath("./smp:realisation", namespaces=nsmap)
+    logging.debug(f"smp_realisation: {smp_realisation}")
+
+    if smp_realisation is None or len(smp_realisation) != 1:
+        logging.error(
+            f"Invalid realisation found in current state machine procedure named {smp_name} with id {smp_id}"
+        )
+        return
+
+    smp_realisation = smp_realisation[0]
+    initial_state_name = smp_realisation.get("initialState")
+    completed_state_name = smp_realisation.get("completedState")
+    logging.debug(f"initial_state_name: {initial_state_name}")
+    logging.debug(f"completed_state_name: {completed_state_name}")
+
+    smp_states = smp_realisation.xpath("./smp:states/smp:state", namespaces=nsmap)
+
+    logging.debug(f"smp_states: {smp_states}")
+
+    sm_state_list = []
+
+    for smp_state in smp_states:
+        state_name = smp_state.get("name")
+        state_id = smp_state.get("id")
+        is_initial = state_name == initial_state_name
+        is_completed = state_name == completed_state_name
+
+        state_transitions = smp_state.xpath(
+            "./smp:transitions/smp:transition", namespaces=nsmap
+        )
+        transitions = []
+        for state_transition in state_transitions:
+            current_id = state_transition.get("id")
+            current_name = state_transition.get("name")
+            current_target = state_transition.get("target")
+            transitions.append(
+                SMTransition(current_id, current_name, current_target, state_transition)
+            )
+
+        target_state_ids = []
+
+        for transition in transitions:
+            if transition.target is not None:
+                target_state_ids.append(transition.id)
+
+        logging.debug(f"transitions: {transitions}")
+
+        state_triggers = smp_state.xpath("./smp:triggers/smp:trigger", namespaces=nsmap)
+        logging.debug(f"state_triggers: {state_triggers}")
+
+        triggers = []
+        for state_trigger in state_triggers:
+            logging.debug(f"state_trigger: {state_trigger}")
+            current_id = state_trigger.get("id")
+            current_name = state_trigger.get("name")
+            triggers.append(SMTrigger(current_id, current_name, state_trigger))
+
+        current_sm_state = SMState(
+            state_id,
+            state_name,
+            is_initial,
+            is_completed,
+            transitions,
+            target_state_ids,
+            triggers,
+            smp_state,
+        )
+
+        sm_state_list.append(current_sm_state)
+
+    state_machine_object = StateMachine(smp_id, smp_name, sm_state_list, input_node)
+
+    return state_machine_object
+
+
+def get_state_machine_procedures(tree: etree._ElementTree, nsmap: Dict) -> List:
+    return tree.xpath("//*[@xsi:type='smp:StateMachineProcedure']", namespaces=nsmap)
+
+
+def get_namespace_map(root: etree._Element) -> Dict:
+    return {k: v for k, v in root.nsmap.items() if k is not None}