vpp: T8230: Add support for PPPoE on bonding interfaces

1 files changed, 61 insertions, 19 deletions

diff --git a/python/vyos/vpp/control_vpp.py b/python/vyos/vpp/control_vpp.py
index b37bb865f..2daaed423 100644
--- a/python/vyos/vpp/control_vpp.py
+++ b/python/vyos/vpp/control_vpp.py
@@ -15,9 +15,10 @@
 # with this program; if not, write to the Free Software Foundation, Inc.,
 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 
+import re
+
 from collections.abc import Callable
 from functools import wraps
-from re import search as re_search, MULTILINE as re_M
 from systemd import journal
 from time import sleep
 from typing import TypeVar, ParamSpec, Literal
@@ -330,7 +331,7 @@ class VPPControl:
         hw_info = self.cli_cmd(f'show hardware-interfaces {ifname}').reply
 
         regex_filter = r'^\s+pci: device (?P<device>\w+:\w+) subsystem (?P<subsystem>\w+:\w+) address (?P<address>\w+:\w+:\w+\.\w+) numa (?P<numa>\w+)$'
-        re_obj = re_search(regex_filter, hw_info, re_M)
+        re_obj = re.search(regex_filter, hw_info, re.MULTILINE)
 
         # return empty string if no interface or no PCI info was found
         if not hw_info or not re_obj:
@@ -439,7 +440,6 @@ class VPPControl:
                 return iface.interface_dev_type
         return None
 
-
     @property
     def connected(self) -> bool:
         """Check if VPP API is connected
@@ -460,47 +460,89 @@ class VPPControl:
         return self.__vpp_api_client.api
 
     @_Decorators.api_call
-    def map_pppoe_interface(self, ifname: str, is_add: bool) -> None:
-        """Create or delete PPPoE mapping between data-plain and control-plane interfaces
+    def map_pppoe_interface(self, ifname: str) -> None:
+        """
+        Create PPPoE mapping between data-plane and control-plane interfaces.
 
         Args:
-            ifname (str): name of an interface in kernel
-            is_add (bool): create or delete mapping
+            ifname (str): Name of an interface in kernel.
         """
         vpp_pair = self.lcp_pair_find(kernel_name=ifname)
         if vpp_pair:
+            vpp_iface_name = vpp_pair['vpp_name_hw']
             vpp_pair_name = vpp_pair['vpp_name_kernel']
             self.__vpp_api_client.api.pppoe_add_del_cp(
-                dp_sw_if_index=self.get_sw_if_index(ifname),
+                dp_sw_if_index=self.get_sw_if_index(vpp_iface_name),
                 cp_sw_if_index=self.get_sw_if_index(vpp_pair_name),
-                is_add=is_add,
+                is_add=True,
             )
 
     @_Decorators.api_call
     def get_pppoe_interface_mapping(self) -> dict:
         """
-        Get PPPoE mapping between data-plain and control-plane interfaces
+        Parse the output of 'show pppoe control-plane binding' to build a mapping
+        between data-plane and control-plane interfaces.
 
         Returns:
-            dict: Dict where key is the dataplane interface and value is the control interface
+            dict: Mapping of data-plane interface names/indices to control-plane interface names/indices.
         """
-
         reply = self.cli_cmd('show pppoe control-plane binding').reply
+        result = {}
 
-        if 'No PPPoE control plane interface configured' in reply:
-            return {}
+        deleted_re = re.compile(r'DELETED\s*\((\d+)\)')
 
-        lines = reply.splitlines()
-        lines = lines[2:]  # Ignore two lines with headers
+        lines = reply.strip().splitlines()
+
+        # If there are less than 2 lines, assume no PPPoE configured
+        if len(lines) < 2:
+            return result
+
+        # Skip header lines (first 2 lines)
+        for line in lines[2:]:
+            line = line.strip()
+            if not line:
+                continue
+
+            # Split into two columns by 2+ spaces
+            cols = re.split(r'\s{2,}', line, maxsplit=1)
+            if len(cols) != 2:
+                continue
+
+            # Determine key/value depending on which column has "DELETED (index)"
+            col_a, col_b = cols
+
+            match_a = deleted_re.search(col_a)
+            match_b = deleted_re.search(col_b)
+
+            key = int(match_a.group(1)) if match_a else col_a
+            value = int(match_b.group(1)) if match_b else col_b
 
-        result = {}
-        for line in lines:
-            key, value = line.split()
             result[key] = value
 
         return result
 
     @_Decorators.api_call
+    def delete_pppoe_mapping(self, dp_iface: str | int, cp_iface: str | int) -> None:
+        """
+        Delete PPPoE mapping between data-plane and control-plane interfaces.
+
+        Args:
+            dp_iface (str|int): Name or index of an interface in data-plane.
+            cp_iface (str|int): Name or index of an interface in control plane.
+        """
+        dp_index = dp_iface
+        if isinstance(dp_iface, str):
+            dp_index = self.get_sw_if_index(dp_iface)
+        cp_index = cp_iface
+        if isinstance(cp_iface, str):
+            cp_index = self.get_sw_if_index(cp_iface)
+        self.__vpp_api_client.api.pppoe_add_del_cp(
+            dp_sw_if_index=dp_index,
+            cp_sw_if_index=cp_index,
+            is_add=False,
+        )
+
+    @_Decorators.api_call
     def enable_dhcp_client(self, ifname: str) -> None:
         """Enable DHCP client detection on a given interface