opmode: T8343: implement common verify_cli_exists() helper

Instead of re-defining a per module helper which verifies that a given CLI path
exists during execution, rather create a generic representation which takes a
CLI path (list) as argument with an optional error message to display on the
CLI.

This is the initial implementation of the generic helper with some migrations
for VPP, WireGuard and Wireless interfaces. More opmode scripts should follow.

4 files changed, 18 insertions, 93 deletions

diff --git a/src/op_mode/interfaces_wireguard.py b/src/op_mode/interfaces_wireguard.py
index 3c63d83ff..b600bd3a4 100644
--- a/src/op_mode/interfaces_wireguard.py
+++ b/src/op_mode/interfaces_wireguard.py
@@ -18,31 +18,13 @@ import sys
 import vyos.opmode
 
 from vyos.ifconfig import WireGuardIf
-from vyos.configquery import ConfigTreeQuery
 
-
-def _verify(func):
-    """Decorator checks if WireGuard interface config exists"""
-    from functools import wraps
-
-    @wraps(func)
-    def _wrapper(*args, **kwargs):
-        config = ConfigTreeQuery()
-        interface = kwargs.get('intf_name')
-        if not config.exists(['interfaces', 'wireguard', interface]):
-            unconf_message = f'WireGuard interface {interface} is not configured'
-            raise vyos.opmode.UnconfiguredSubsystem(unconf_message)
-        return func(*args, **kwargs)
-
-    return _wrapper
-
-
-@_verify
+@vyos.opmode.verify_cli_exists(['interfaces', 'wireguard'],
+    'WireGuard interface {interface} is not configured!')
 def show_summary(raw: bool, intf_name: str):
     intf = WireGuardIf(intf_name, create=False, debug=False)
     return intf.operational.show_interface()
 
-
 if __name__ == '__main__':
     try:
         res = vyos.opmode.run(sys.modules[__name__])
diff --git a/src/op_mode/interfaces_wireless.py b/src/op_mode/interfaces_wireless.py
index 5a708ad27..58cc9e5db 100755
--- a/src/op_mode/interfaces_wireless.py
+++ b/src/op_mode/interfaces_wireless.py
@@ -23,18 +23,9 @@ from tabulate import tabulate
 from vyos.utils.process import popen
 from vyos.configquery import ConfigTreeQuery
 
-def _verify(func):
-    """Decorator checks if Wireless LAN config exists"""
-    from functools import wraps
-
-    @wraps(func)
-    def _wrapper(*args, **kwargs):
-        config = ConfigTreeQuery()
-        if not config.exists(['interfaces', 'wireless']):
-            unconf_message = 'No Wireless interfaces configured'
-            raise vyos.opmode.UnconfiguredSubsystem(unconf_message)
-        return func(*args, **kwargs)
-    return _wrapper
+verify_path = ['interfaces', 'wireless']
+verify_error = 'Wireless/WiFi subsystem unconfigured!'
+verify_interface_error = 'Wireless interface {interface} is not configured!'
 
 def _get_raw_info_data():
     output_data = []
@@ -156,7 +147,7 @@ def _format_station_data(raw_data):
     headers = ["Station", "Signal", "RX bytes", "RX packets", "TX bytes", "TX packets"]
     return tabulate(output, headers, numalign="left")
 
-@_verify
+@vyos.opmode.verify_cli_exists(verify_path, verify_error)
 def show_info(raw: bool):
     info_data = _get_raw_info_data()
     if raw:
@@ -169,7 +160,7 @@ def show_scan(raw: bool, intf_name: str):
         return data
     return _format_scan_data(data)
 
-@_verify
+@vyos.opmode.verify_cli_exists(verify_path, verify_interface_error)
 def show_stations(raw: bool, intf_name: str):
     data = _get_raw_station_data(intf_name)
     if raw:
diff --git a/src/op_mode/reset_wireguard.py b/src/op_mode/reset_wireguard.py
index 60aeef52d..ad8ea0346 100755
--- a/src/op_mode/reset_wireguard.py
+++ b/src/op_mode/reset_wireguard.py
@@ -16,35 +16,16 @@
 
 import sys
 import typing
-
 import vyos.opmode
 
 from vyos.ifconfig import WireGuardIf
-from vyos.configquery import ConfigTreeQuery
-
-
-def _verify(func):
-    """Decorator checks if WireGuard interface config exists"""
-    from functools import wraps
-
-    @wraps(func)
-    def _wrapper(*args, **kwargs):
-        config = ConfigTreeQuery()
-        interface = kwargs.get('interface')
-        if not config.exists(['interfaces', 'wireguard', interface]):
-            unconf_message = f'WireGuard interface {interface} is not configured'
-            raise vyos.opmode.UnconfiguredSubsystem(unconf_message)
-        return func(*args, **kwargs)
 
-    return _wrapper
-
-
-@_verify
+@vyos.opmode.verify_cli_exists(['interfaces', 'wireguard'],
+    'WireGuard interface {interface} is not configured!')
 def reset_peer(interface: str, peer: typing.Optional[str] = None):
     intf = WireGuardIf(interface, create=False, debug=False)
     return intf.operational.reset_peer(peer)
 
-
 if __name__ == '__main__':
     try:
         res = vyos.opmode.run(sys.modules[__name__])
diff --git a/src/op_mode/vpp.py b/src/op_mode/vpp.py
index 54b46cf68..f108392fb 100755
--- a/src/op_mode/vpp.py
+++ b/src/op_mode/vpp.py
@@ -23,31 +23,8 @@ from vyos.vpp import VPPControl
 from vyos.configquery import ConfigTreeQuery
 import vyos.opmode
 
-
 NO_INDEX = 0xFFFFFFFF
 
-
-def _verify(target: typing.Optional[str]):
-    """Decorator checks if config for VPP feature exists"""
-    from functools import wraps
-
-    path = target.split() if target else []
-
-    def _verify_target(func):
-        @wraps(func)
-        def _wrapper(*args, **kwargs):
-            config = ConfigTreeQuery()
-            if not config.exists(path):
-                raise vyos.opmode.UnconfiguredSubsystem(
-                    f'"{" ".join(path)}" is not configured'
-                )
-            return func(*args, **kwargs)
-
-        return _wrapper
-
-    return _verify_target
-
-
 class VPPShow:
     RX_STATES = {
         0: 'INITIALIZE',
@@ -361,6 +338,7 @@ class VPPShow:
         cmd_command = f'show bridge-domain {bd_id} detail'
         data = self.vpp.cli_cmd(cmd_command)
 
+
         if raw:
             return [data.reply]
 
@@ -370,47 +348,40 @@ class VPPShow:
 # -----------------------------
 # VyOS IPFIX op-mode entries
 # -----------------------------
-@_verify('vpp ipfix interface')
+@vyos.opmode.verify_cli_exists(['vpp', 'ipfix', 'interface'])
 def show_ipfix_interfaces(raw: bool):
     return VPPShow().ipfix_interfaces(raw)
 
-
-@_verify('vpp ipfix collector')
+@vyos.opmode.verify_cli_exists(['vpp', 'ipfix', 'collector'])
 def show_ipfix_collectors(raw: bool):
     return VPPShow().ipfix_collectors(raw)
 
-
-@_verify('vpp ipfix')
+@vyos.opmode.verify_cli_exists(['vpp', 'ipfix'])
 def show_ipfix_table(raw: bool):
     return VPPShow().ipfix_table(raw)
 
-
 # -----------------------------
 # VPP LACP information
 # -----------------------------
-@_verify('interfaces vpp bonding')
+@vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bonding'])
 def show_lacp(raw: bool, ifname: typing.Optional[str]):
     return VPPShow().lacp_info(raw, ifname)
 
-
-@_verify('interfaces vpp bonding')
+@vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bonding'])
 def show_lacp_details(raw: bool, ifname: typing.Optional[str]):
     return VPPShow().lacp_details(raw, ifname)
 
-
 # -----------------------------
-# Bridge op-mode entries
+# Bridge op-mode entrie
 # -----------------------------
-@_verify('interfaces vpp bridge')
+@vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bridge'])
 def show_bridge(raw: bool, ifname: typing.Optional[str] = None):
     return VPPShow().bridge_domain(raw, ifname)
 
-
-@_verify('interfaces vpp bridge')
+@vyos.opmode.verify_cli_exists(['interfaces', 'vpp', 'bridge'])
 def show_bridge_details(raw: bool, ifname: typing.Optional[str] = None):
     return VPPShow().bridge_domain_details(raw, ifname)
 
-
 if __name__ == '__main__':
     try:
         res = vyos.opmode.run(sys.modules[__name__])