sai_test: add opt-in SIMULATE_SONIC to emulate SONiC teamd/IntfMgr netdevs for standalone benches (#2316)

Context / motivation
Part of the SAIVPP unit-test framework (see the sonic-sairedis docker-sai-test-vpp harness). When the OCP sai_test T0 suite runs against a SAI backend on a standalone bench with no SONiC control plane, there is no teamd to create PortChannel netdevs and no IntfMgr to assign LAG/SVI interface IPs — which some backends (e.g. the VPP virtual switch) rely on for LAG bring-up and routed-to-LAG/SVI forwarding. This adds an opt-in way for the test setup to emulate that.

What this change does

test/sai_test/config/simulate_sonic.py (new). Opt-in helper, active only when SIMULATE_SONIC=1; every entry point is a no-op otherwise.
test/sai_test/config/lag_configer.py — create_lag(). When enabled, create the PortChannel<N> netdev teamd would provide, before the SAI LAG create.
test/sai_test/config/route_configer.py — create_router_interface(). When enabled, assign the connected IPs IntfMgr would set on a LAG or VLAN(SVI) router interface.
Vendor-neutral by construction. PortChannel netdevs use generic Linux ip … type bond; LAG RIF IPs use ip addr add. A VLAN SVI has no host-interface netdev to mirror from, so its address is programmed via command templates supplied by the caller (SVI_RIF_SET_IP_CMD / SVI_RIF_PROBE_CMD, with {ifname}/{addr} placeholders) — no backend-specific tooling is hardcoded in sai_test. All interface-name prefixes and address patterns are env-overridable.
Scope / risk

3 files (one new module + two setUp-path hooks) — test-config helpers only; no change to SAI/backend code or packet semantics.
Fully opt-in with default-off: with SIMULATE_SONIC unset, all entry points no-op, so real-HW and other OCP consumers are unchanged unless they opt in.
Dependencies
None. Independent of #2299 / #2300 / #2301 (edits different functions); targets master.

Author: nicholasching

test/sai_test/config/lag_configer.py

@@ -78,6 +78,13 @@ def create_lag(self, lag_port_idxs):
             Lag: lag object
         """
         lag: Lag = Lag(None, [], [])
+        lag_index = len(self.test_obj.dut.lag_list)
+        try:
+            from config import simulate_sonic
+        except ImportError:
+            simulate_sonic = None
+        if simulate_sonic is not None:
+            simulate_sonic.ensure_portchannel(lag_index)
         lag_oid = sai_thrift_create_lag(self.client)
         lag.oid = lag_oid
         self.create_lag_member(lag, lag_port_idxs)

test/sai_test/config/route_configer.py

@@ -549,8 +549,25 @@ def create_router_interface(self, net_interface: route_item, virtual_router=None
             if not self.test_obj.dut.rif_list:
                 self.test_obj.dut.rif_list = []
             self.test_obj.dut.rif_list.append(rif)
+            self._simulate_sonic_assign_rif_ips(net_interface)
         return net_interface.rif_list[-1]
 
+    def _simulate_sonic_assign_rif_ips(self, net_interface):
+        """When SIMULATE_SONIC=1, assign the connected IPs IntfMgr would set on a
+        LAG/SVI router interface (no-op otherwise)."""
+        try:
+            from config import simulate_sonic
+        except ImportError:
+            return
+        if isinstance(net_interface, Lag):
+            try:
+                lag_index = self.test_obj.dut.lag_list.index(net_interface)
+            except ValueError:
+                return
+            simulate_sonic.assign_lag_rif_ips(lag_index)
+        elif isinstance(net_interface, Vlan):
+            simulate_sonic.assign_svi_rif_ips(net_interface.vlan_id)
+
     def create_nexthop_by_rif(self, rif, nexthop_device: Device):
         """
         Create nexthop by bridge port index for a port.

test/sai_test/config/simulate_sonic.py

@@ -0,0 +1,173 @@
+# Copyright (c) 2026 Microsoft Open Technologies, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+#    THIS CODE IS PROVIDED ON AN *AS IS* BASIS, WITHOUT WARRANTIES OR
+#    CONDITIONS OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT
+#    LIMITATION ANY IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS
+#    FOR A PARTICULAR PURPOSE, MERCHANTABILITY OR NON-INFRINGEMENT.
+#
+#    See the Apache Version 2.0 License for specific language governing
+#    permissions and limitations under the License.
+#
+#    Microsoft would like to thank the following companies for their review and
+#    assistance with these files: Intel Corporation, Mellanox Technologies Ltd,
+#    Dell Products, L.P., Facebook, Inc., Marvell International Ltd.
+#
+#
+
+# Optional SONiC control-plane simulation for standalone sai_test benches.
+#
+# In a real SONiC device teamd creates PortChannel netdevs and IntfMgr assigns the
+# interface IPs that the SAI backend mirrors onto LAG/SVI router interfaces. A
+# standalone sai_test bench (no teamd / IntfMgr) has neither, so LAG bring-up and
+# routed-to-LAG/SVI forwarding can fail. When SIMULATE_SONIC=1 the config helpers
+# call into this module to emulate that setup from the test's setUp path.
+#
+# This is opt-in and default-preserving: with SIMULATE_SONIC unset (the default) all
+# entry points are no-ops, so real ASICs and other harnesses are unaffected. All
+# interface names and address patterns are environment-overridable so the defaults
+# can be retargeted without code changes. This module is backend-neutral: it uses
+# standard Linux "ip" for PortChannel netdevs and LAG router-interface IPs, and for
+# a VLAN (SVI) router interface -- which has no host-interface netdev to mirror from
+# -- it runs the command templates the harness provides via SVI_RIF_SET_IP_CMD (and
+# optional SVI_RIF_PROBE_CMD), so no backend-specific tooling is hardcoded here.
+
+import logging
+import os
+import shlex
+import subprocess
+import time
+
+_logger = logging.getLogger(__name__)
+
+
+def enabled():
+    return os.environ.get("SIMULATE_SONIC", "0") == "1"
+
+
+def _run(cmd, check=False, quiet=False):
+    result = subprocess.run(cmd, check=False, capture_output=True, text=True)
+    if result.returncode != 0 and not quiet:
+        msg = "command failed (rc={}): {}".format(result.returncode, cmd)
+        stderr = result.stderr.strip() if result.stderr else ""
+        if stderr:
+            msg = "{} stderr: {}".format(msg, stderr)
+        if check:
+            _logger.error(msg)
+            result.check_returncode()
+        else:
+            _logger.warning(msg)
+    elif check:
+        result.check_returncode()
+    return result
+
+
+def _mtu():
+    return int(os.environ.get("MTU", "9100"))
+
+
+def _templated(cmd_tmpl, **kw):
+    """Split a command template and substitute {ifname}/{addr} placeholders."""
+    return shlex.split(cmd_tmpl.format(**kw))
+
+
+def ensure_portchannel(lag_index):
+    """Create the PortChannel<lag_index> netdev teamd would provide for a LAG."""
+    if not enabled():
+        return
+
+    pc_if = "PortChannel{}".format(lag_index)
+    mtu = _mtu()
+
+    if _run(["ip", "link", "show", pc_if], quiet=True).returncode != 0:
+        if _run(["ip", "link", "add", pc_if, "type", "bond"]).returncode != 0:
+            _run(["ip", "link", "add", pc_if, "type", "dummy"], check=False)
+
+    _run(["ip", "link", "set", "dev", pc_if, "mtu", str(mtu)], check=False)
+    _run(["ip", "link", "set", "dev", pc_if, "up"], check=False)
+
+
+def assign_lag_rif_ips(lag_index, retries=20, interval=0.3):
+    """Assign the connected IPs IntfMgr would put on a LAG router interface.
+
+    The IP is placed on the backend's LAG host-interface netdev (default "be<N>"),
+    which the backend mirrors onto the LAG router interface.
+    """
+    if not enabled() or os.environ.get("LAG_RIF_IPS", "1") != "1":
+        return
+
+    be_prefix = os.environ.get("LAG_BE_TAP_PREFIX", "be")
+    v4_pattern = os.environ.get("LAG_RIF_IPV4_PATTERN", "10.1.%d.1/24")
+    v6_pattern = os.environ.get("LAG_RIF_IPV6_PATTERN", "fc00:1::%d:1/112")
+    subnet_id = lag_index + 1
+    be_if = "{}{}".format(be_prefix, lag_index)
+    v4_addr = v4_pattern % subnet_id
+    v6_addr = v6_pattern % subnet_id
+
+    for _ in range(retries):
+        if _run(["ip", "link", "show", be_if], quiet=True).returncode == 0:
+            break
+        time.sleep(interval)
+    else:
+        return
+
+    disable_v6 = "/proc/sys/net/ipv6/conf/{}/disable_ipv6".format(be_if)
+    accept_dad = "/proc/sys/net/ipv6/conf/{}/accept_dad".format(be_if)
+    try:
+        with open(disable_v6, "w", encoding="ascii") as fh:
+            fh.write("0")
+        with open(accept_dad, "w", encoding="ascii") as fh:
+            fh.write("0")
+    except OSError:
+        pass
+
+    _run(["ip", "addr", "add", v4_addr, "dev", be_if], check=False)
+    _run(["ip", "-6", "addr", "add", v6_addr, "dev", be_if, "nodad"], check=False)
+
+
+def _svi_group_id(vlan_id):
+    for entry in os.environ.get("SVI_RIF_VLANS", "10:1 20:2").split():
+        vid, gid = entry.split(":", 1)
+        if int(vid) == int(vlan_id):
+            return int(gid)
+    return int(vlan_id)
+
+
+def assign_svi_rif_ips(vlan_id, retries=20, interval=0.3):
+    """Assign the connected IPs IntfMgr would put on a VLAN (SVI) router interface.
+
+    A VLAN SVI has no host-interface netdev to mirror from, so the IP must be set on
+    the backend interface directly. The backend-specific commands are injected by the
+    harness as templates (with {ifname}/{addr} placeholders): SVI_RIF_SET_IP_CMD sets
+    the address, and the optional SVI_RIF_PROBE_CMD waits until the interface exists.
+    With SVI_RIF_SET_IP_CMD unset this is a no-op, keeping this module backend-neutral.
+    """
+    if not enabled() or os.environ.get("SVI_RIF_IPS", "1") != "1":
+        return
+
+    set_ip_cmd = os.environ.get("SVI_RIF_SET_IP_CMD")
+    if not set_ip_cmd:
+        return
+
+    probe_cmd = os.environ.get("SVI_RIF_PROBE_CMD")
+    bvi_prefix = os.environ.get("SVI_BVI_PREFIX", "bvi")
+    v4_pattern = os.environ.get("SVI_RIF_IPV4_PATTERN", "192.168.%d.1/24")
+    v6_pattern = os.environ.get("SVI_RIF_IPV6_PATTERN", "fc02::%d:1/112")
+    group_id = _svi_group_id(vlan_id)
+    bvi_if = "{}{}".format(bvi_prefix, vlan_id)
+    v4_addr = v4_pattern % group_id
+    v6_addr = v6_pattern % group_id
+
+    if probe_cmd:
+        for _ in range(retries):
+            if _run(_templated(probe_cmd, ifname=bvi_if, addr=""), quiet=True).returncode == 0:
+                break
+            time.sleep(interval)
+        else:
+            return
+
+    for addr in (v4_addr, v6_addr):
+        _run(_templated(set_ip_cmd, ifname=bvi_if, addr=addr), check=False)