Merge pull request #5030 from c-po/op-mode-cli-verify

opmode: T8343: implement common verify_cli_exists() helper

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__])