Fix ATVM watcher test flow plugin filtering

atvm/watcher-service/atvm_run_watcher.py

@@ -1242,6 +1242,69 @@ def extract_test_flow_from_generated_spec(
     if not spec_list:
         return []
 
+    runtime_settings: Dict[str, object] = {}
+    config_file = metadata.get("config_file")
+    if isinstance(config_file, str) and config_file:
+        config_path = project_root / config_file
+        if config_path.exists():
+            config_text = config_path.read_text(encoding="utf-8", errors="replace")
+            for key in ("pure_plugin_type", "debug-type"):
+                match = re.search(rf'"{re.escape(key)}"\s*:\s*"([^"]*)"', config_text)
+                if match:
+                    runtime_settings[key] = match.group(1)
+            for key in ("test-unaligned-fio", "isRegularCutover", "test-install-only"):
+                match = re.search(rf'"{re.escape(key)}"\s*:\s*(true|false)', config_text)
+                if match:
+                    runtime_settings[key] = match.group(1) == "true"
+
+    def selected_plugin_gates() -> Optional[set[str]]:
+        pure_plugin_type = runtime_settings.get("pure_plugin_type")
+        if isinstance(pure_plugin_type, str):
+            lowered = pure_plugin_type.lower()
+            if lowered == "both":
+                return {"fc", "iscsi"}
+            if lowered in {"fc", "iscsi"}:
+                return {lowered}
+        integration_plugin = metadata.get("integration_plugin")
+        if not isinstance(integration_plugin, str):
+            return None
+        lowered = integration_plugin.lower()
+        if not lowered or lowered == "unknown":
+            return None
+        if "both" in lowered:
+            return {"fc", "iscsi"}
+        selected: set[str] = set()
+        if re.search(r"\bfc\b", lowered):
+            selected.add("fc")
+        if "iscsi" in lowered:
+            selected.add("iscsi")
+        return selected or None
+
+    def evaluate_gate_line(line: str, allowed_plugins: Optional[set[str]]) -> Tuple[bool, Optional[bool]]:
+        normalized = re.sub(r"\s+", "", line)
+        if normalized.startswith("if(useFCPlugin)"):
+            return True, allowed_plugins is None or "fc" in allowed_plugins
+        if normalized.startswith("if(useIscsiPlugin)"):
+            return True, allowed_plugins is None or "iscsi" in allowed_plugins
+        if normalized.startswith('if(Cypress.env("test-unaligned-fio")==true)'):
+            value = runtime_settings.get("test-unaligned-fio")
+            return True, value if isinstance(value, bool) else None
+        if normalized.startswith('if(Cypress.env("isRegularCutover")==false)'):
+            value = runtime_settings.get("isRegularCutover")
+            return True, (value is False) if isinstance(value, bool) else None
+        if normalized.startswith('if(Cypress.env("isRegularCutover")===true)'):
+            value = runtime_settings.get("isRegularCutover")
+            return True, (value is True) if isinstance(value, bool) else None
+        if normalized.startswith("if(!enabled_percpu_debug)"):
+            debug_type = runtime_settings.get("debug-type")
+            return True, debug_type != "percpu" if isinstance(debug_type, str) else None
+        if normalized.startswith("if(enabled_percpu_debug)"):
+            debug_type = runtime_settings.get("debug-type")
+            return True, debug_type == "percpu" if isinstance(debug_type, str) else None
+        return False, None
+
+    allowed_plugins = selected_plugin_gates()
+
     for entry in spec_list:
         if not isinstance(entry, str) or "check-xml-files.ts" in entry:
             continue
@@ -1249,12 +1312,32 @@ def extract_test_flow_from_generated_spec(
         if not spec_path.exists():
             continue
         steps: List[str] = []
+        active_gate_blocks: List[Tuple[Optional[bool], int]] = []
+        pending_gate_block: Optional[Optional[bool]] = None
+        current_depth = 0
         for line in spec_path.read_text(encoding="utf-8", errors="replace").splitlines():
+            matched_gate, gate_result = evaluate_gate_line(line, allowed_plugins)
+            if matched_gate:
+                if "{" in line:
+                    active_gate_blocks.append((gate_result, current_depth + line.count("{")))
+                else:
+                    pending_gate_block = gate_result
+            elif pending_gate_block is not None and "{" in line:
+                active_gate_blocks.append((pending_gate_block, current_depth + line.count("{")))
+                pending_gate_block = None
+
             match = re.search(r'it\(\s*`?\$\{numStep\+\+\}\.\s*(.*?)`\s*,', line)
             if match:
                 step_text = match.group(1).strip()
-                if step_text:
+                include_step = (
+                    step_text
+                    and not any(gate_result is False for gate_result, _ in active_gate_blocks)
+                )
+                if include_step:
                     steps.append(f"{len(steps) + 1}. {step_text}")
+            current_depth += line.count("{") - line.count("}")
+            while active_gate_blocks and current_depth < active_gate_blocks[-1][1]:
+                active_gate_blocks.pop()
         if steps:
             return steps
     return []