moved helper functions to ha_utils from conftest file

Author: nnelluri-cisco

tests/ha/conftest.py

@@ -2,24 +2,31 @@
 import pytest
 from pathlib import Path
 from collections import defaultdict
-import re
 import os
 import json
-import ast
 import time
 
 from tests.common.config_reload import config_reload
 from tests.common.helpers.constants import DEFAULT_NAMESPACE
-from tests.common.utilities import wait_until
-from common.ha.smartswitch_ha_helper import PtfTcpTestAdapter
-from common.ha.smartswitch_ha_io import SmartSwitchHaTrafficTest
-from common.ha.smartswitch_ha_helper import (
+from tests.common.ha.smartswitch_ha_helper import PtfTcpTestAdapter
+from tests.common.ha.smartswitch_ha_io import SmartSwitchHaTrafficTest
+from tests.common.ha.smartswitch_ha_helper import (
     add_port_to_namespace,
     remove_namespace,
     add_static_route_to_ptf,
     add_static_route_to_dut
 )
 
+from ha_utils import (
+
+    build_dash_ha_scope_args,
+    wait_for_pending_operation_id,
+    build_dash_ha_scope_activate_args,
+    wait_for_ha_state,
+    build_dash_ha_set_args,
+    proto_utils_hset
+)
+
 logger = logging.getLogger(__name__)
 
 
@@ -417,194 +424,7 @@ def setup_ha_config(duthosts):
     return final_cfg
 
 
-def build_dash_ha_set_args(fields):
-    """
-    Build args for DASH_HA_SET_CONFIG_TABLE
-    EXACTLY following the working CLI
-    """
-
-    version = str(fields["version"])
-    if version.endswith(".0"):
-        version = version[:-2]
-
-    return (
-        f'version \\"{version}\\" '
-        f'vip_v4 "{fields["vip_v4"]}" '
-        f'vip_v6 "{fields["vip_v6"]}" '
-        f'scope "{fields["scope"]}" '
-        f'preferred_vdpu_id "{fields["preferred_vdpu_id"]}" '
-        f'preferred_standalone_vdpu_index 0 '
-        f'vdpu_ids \'["vdpu0_0","vdpu1_0"]\''
-    )
-
-
-def build_dash_ha_scope_args(fields):
-    """
-    Build args for DASH_HA_SCOPE_CONFIG_TABLE
-    EXACTLY following the working CLI
-    """
-
-    version = str(fields["version"])
-    if version.endswith(".0"):
-        version = version[:-2]
-
-    return (
-        f'version \\"{version}\\" '
-        f'disabled "{fields["disabled"]}" '
-        f'desired_ha_state "{fields["desired_ha_state"]}" '
-        f'ha_set_id "{fields["ha_set_id"]}" '
-        f'owner "{fields["owner"]}"'
-    )
-
-
-def extract_pending_operations(text):
-    """
-    Extract pending_operation_ids and pending_operation_types
-    and return list of (type, id) tuples.
-    """
-    ids_match = re.search(
-        r'pending_operation_ids\s*\|\s*([^\|\r\n]+)',
-        text,
-        re.DOTALL,
-    )
-    types_match = re.search(
-        r'pending_operation_types\s*\|\s*([^\|\r\n]+)',
-        text,
-        re.DOTALL,
-    )
-    if not ids_match or not types_match:
-        return []
-
-    try:
-        ids = ast.literal_eval(f"'{ids_match.group(1)}'")
-        id_list = ids.split()
-        ids = id_list[0].split(',')
-        types = ast.literal_eval(f"'{types_match.group(1)}'")
-        type_list = types.split()
-        types = type_list[0].split(',')
-    except Exception:
-        return []
-
-    return list(zip(types, ids))
-
-
-def get_pending_operation_id(duthost, scope_key, expected_op_type):
-    """
-    scope_key example: vdpu0_0:haset0_0
-    expected_op_type example: ACTIVATE_ROLE
-    """
-    cmd = (
-        "docker exec dash-hadpu0 swbus-cli show hamgrd actor "
-        f"/hamgrd/0/ha-scope/{scope_key}"
-    )
-    res = duthost.shell(cmd)
-
-    pending_ops = extract_pending_operations(res["stdout"])
-
-    for op_type, op_id in pending_ops:
-        if op_type == expected_op_type:
-            return op_id
-
-    return None
-
-
-def build_dash_ha_scope_activate_args(fields, pending_id):
-    return (
-        f'version \\"{fields["version"]}\\" '
-        f'disabled {fields["disabled"]} '
-        f'desired_ha_state "{fields["desired_ha_state"]}" '
-        f'ha_set_id "{fields["ha_set_id"]}" '
-        f'owner "{fields["owner"]}" '
-        f'approved_pending_operation_ids '
-        f'[\\\"{pending_id}\\\"]'
-    )
-
-
-def proto_utils_hset(duthost, table, key, args):
-    """
-    Wrapper around proto_utils.py hset
-
-    Args:
-        duthost: pytest duthost fixture
-        table (str): Redis table name
-        key (str): Redis key
-        args (str): Already-built proto args string
-    """
-    cmd = (
-        "docker exec swss python /etc/sonic/proto_utils.py hset "
-        f'"{table}:{key}" {args}'
-    )
-    duthost.shell(cmd)
-
-
-def wait_for_pending_operation_id(
-    duthost,
-    scope_key,
-    expected_op_type,
-    timeout=60,
-    interval=2,
-):
-    """
-    Wait until the expected pending_operation_id appears.
-    """
-    pending_id = None
-
-    def _condition():
-        nonlocal pending_id
-        pending_id = get_pending_operation_id(
-            duthost,
-            scope_key,
-            expected_op_type,
-        )
-        return pending_id is not None
-
-    success = wait_until(
-        timeout,
-        interval,
-        0,           # REQUIRED delay argument
-        _condition,  # condition callable
-    )
-
-    return pending_id if success else None
-
-
-def extract_ha_state(text):
-    """
-    Extract ha_state from swbus-cli output
-    """
-    match = re.search(r'ha_state\s+\|\s+(\w+)', text)
-    return match.group(1) if match else None
-
-
-def wait_for_ha_state(
-    duthost,
-    scope_key,
-    expected_state,
-    timeout=120,
-    interval=5,
-):
-    """
-    Wait until HA reaches the expected state
-    """
-    def _check_ha_state():
-        cmd = (
-            "docker exec dash-hadpu0 swbus-cli show hamgrd actor "
-            f"/hamgrd/0/ha-scope/{scope_key}"
-        )
-        res = duthost.shell(cmd)
-        return extract_ha_state(res["stdout"]) == expected_state
-
-    success = wait_until(
-        timeout,
-        interval,
-        0,
-        _check_ha_state
-    )
-
-    return success
-
-
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="function")
 def setup_dash_ha_from_json(duthosts):
     base_dir = "/data/tests/common/ha"
     ha_set_file = os.path.join(base_dir, "dash_ha_set_dpu_config_table.json")
@@ -659,7 +479,7 @@ def setup_dash_ha_from_json(duthosts):
         )
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="function")
 def activate_dash_ha_from_json(duthosts):
     # -------------------------------------------------
     # Step 4: Activate Role (using pending_operation_ids)