Run Command and Constants refactor (WLAN-Pi#47)

Co-authored-by: Michael Ketchel <[email protected]>

testing.in

@@ -5,6 +5,7 @@ autoflake
 mypy
 flake8
 pytest
+pytest-asyncio
 pytest-cov
 coverage-badge
 pytest-mock

testing.txt

@@ -4,6 +4,8 @@
 #
 #    pip-compile testing.in
 #
+--extra-index-url https://www.piwheels.org/simple
+
 autoflake==2.3.1
     # via -r testing.in
 black==24.8.0
@@ -72,8 +74,11 @@ pyproject-api==1.8.0
 pytest==8.3.3
     # via
     #   -r testing.in
+    #   pytest-asyncio
     #   pytest-cov
     #   pytest-mock
+pytest-asyncio==0.24.0
+    # via -r testing.in
 pytest-cov==5.0.0
     # via -r testing.in
 pytest-mock==3.14.0
@@ -95,3 +100,6 @@ typing-extensions==4.7.1
     #   mypy
 virtualenv==20.26.6
     # via tox
+
+# The following packages are considered to be unsafe in a requirements file:
+# setuptools

wlanpi_core/api/api_v1/endpoints/network_api.py

@@ -3,6 +3,7 @@
 
 from fastapi import APIRouter, Response
 
+from wlanpi_core.constants import API_DEFAULT_TIMEOUT
 from wlanpi_core.models.network.vlan.vlan_errors import VLANError
 from wlanpi_core.models.validation_error import ValidationError
 from wlanpi_core.schemas import network
@@ -12,8 +13,6 @@
 
 router = APIRouter()
 
-API_DEFAULT_TIMEOUT = 20
-
 log = logging.getLogger("uvicorn")
 
 

wlanpi_core/api/api_v1/endpoints/system_api.py

@@ -6,6 +6,7 @@
 from wlanpi_core.models.validation_error import ValidationError
 from wlanpi_core.schemas import system
 from wlanpi_core.services import system_service
+from wlanpi_core.utils.general import run_command
 
 router = APIRouter()
 
@@ -77,7 +78,7 @@ async def show_device_model():
     # get output of wlanpi-model
     model_cmd = "wlanpi-model -b"
     try:
-        platform = subprocess.check_output(model_cmd, shell=True).decode().strip()
+        platform = run_command(model_cmd).stdout.strip()
 
         if platform.endswith("?"):
             platform = "Unknown"

wlanpi_core/api/api_v1/endpoints/utils_api.py

@@ -23,7 +23,7 @@ async def reachability():
     """
 
     try:
-        reachability = utils_service.show_reachability()
+        reachability = await utils_service.show_reachability()
 
         if reachability.get("error"):
             return Response(
@@ -78,7 +78,7 @@ async def usb_interfaces():
     """
 
     try:
-        result = utils_service.show_usb()
+        result = await utils_service.show_usb()
 
         if result.get("error"):
             return Response(
@@ -103,7 +103,7 @@ async def usb_interfaces():
     """
 
     try:
-        result = utils_service.show_ufw()
+        result = await utils_service.show_ufw()
 
         if result.get("error"):
             return Response(

wlanpi_core/constants.py

@@ -0,0 +1,58 @@
+# Core config
+API_V1_STR: str = "/api/v1"
+PROJECT_NAME: str = "wlanpi-core"
+PROJECT_DESCRIPTION: str = (
+    "The wlanpi-core API provides endpoints for applications on the WLAN Pi to share data. 🚀"
+)
+
+# Linux programs
+IFCONFIG_FILE: str = "/sbin/ifconfig"
+IW_FILE: str = "/sbin/iw"
+IP_FILE: str = "/usr/sbin/ip"
+UFW_FILE: str = "/usr/sbin/ufw"
+ETHTOOL_FILE: str = "/sbin/ethtool"
+
+# Mode changer scripts
+MODE_FILE: str = "/etc/wlanpi-state"
+
+# Version file for WLAN Pi image
+WLANPI_IMAGE_FILE: str = "/etc/wlanpi-release"
+
+WCONSOLE_SWITCHER_FILE: str = "/opt/wlanpi-wconsole/wconsole_switcher"
+HOTSPOT_SWITCHER_FILE: str = "/opt/wlanpi-hotspot/hotspot_switcher"
+WIPERF_SWITCHER_FILE: str = "/opt/wlanpi-wiperf/wiperf_switcher"
+SERVER_SWITCHER_FILE: str = "/opt/wlanpi-server/server_switcher"
+BRIDGE_SWITCHER_FILE: str = "/opt/wlanpi-bridge/bridge_switcher"
+
+REG_DOMAIN_FILE: str = "/usr/bin/wlanpi-reg-domain"
+TIME_ZONE_FILE: str = "/usr/bin/wlanpi-timezone"
+
+# WPA Supplicant dbus service and interface
+WPAS_DBUS_SERVICE: str = "fi.w1.wpa_supplicant1"
+WPAS_DBUS_INTERFACE: str = "fi.w1.wpa_supplicant1"
+WPAS_DBUS_OPATH: str = "/fi/w1/wpa_supplicant1"
+WPAS_DBUS_INTERFACES_INTERFACE: str = "fi.w1.wpa_supplicant1.Interface"
+WPAS_DBUS_INTERFACES_OPATH: str = "/fi/w1/wpa_supplicant1/Interfaces"
+WPAS_DBUS_BSS_INTERFACE: str = "fi.w1.wpa_supplicant1.BSS"
+WPAS_DBUS_NETWORK_INTERFACE: str = "fi.w1.wpa_supplicant1.Network"
+
+# Core API configuration
+API_DEFAULT_TIMEOUT: int = 20
+
+# VLAN model constants
+DEFAULT_VLAN_INTERFACE_FILE = "/etc/network/interfaces.d/vlans"
+DEFAULT_INTERFACE_FILE = "/etc/network/interfaces"
+
+# Service Constants
+BT_ADAPTER = "hci0"
+
+#### Paths below here are relative to script dir or /tmp fixed paths ###
+
+# Networkinfo data file names
+LLDPNEIGH_FILE: str = "/tmp/lldpneigh.txt"
+CDPNEIGH_FILE: str = "/tmp/cdpneigh.txt"
+IPCONFIG_FILE: str = "/opt/wlanpi-common/networkinfo/ipconfig.sh 2>/dev/null"
+REACHABILITY_FILE: str = "/opt/wlanpi-common/networkinfo/reachability.sh"
+PUBLICIP_CMD: str = "/opt/wlanpi-common/networkinfo/publicip.sh"
+PUBLICIP6_CMD: str = "/opt/wlanpi-common/networkinfo/publicip6.sh"
+BLINKER_FILE: str = "/opt/wlanpi-common/networkinfo/portblinker.sh"

wlanpi_core/core/config.py

@@ -2,15 +2,15 @@
 
 from pydantic_settings import BaseSettings
 
+from wlanpi_core import constants
+
 
 class Settings(BaseSettings):
-    API_V1_STR: str = "/api/v1"
+    API_V1_STR: str = constants.API_V1_STR
 
-    PROJECT_NAME: str = "wlanpi-core"
+    PROJECT_NAME: str = constants.PROJECT_NAME
 
-    PROJECT_DESCRIPTION: str = """
-    The wlanpi-core API provides endpoints for applications on the WLAN Pi to share data. 🚀
-    """
+    PROJECT_DESCRIPTION: str = constants.PROJECT_DESCRIPTION
 
     TAGS_METADATA: list = [
         {

wlanpi_core/models/command_result.py

@@ -1,19 +1,52 @@
 import json
+import re
 from json import JSONDecodeError
+from re import RegexFlag
 from typing import Union
 
 
 class CommandResult:
     """Returned by run_command"""
 
-    def __init__(self, output: str, error: str, status_code: int):
-        self.output = output
-        self.error = error
-        self.status_code = status_code
-        self.success = self.status_code == 0
+    def __init__(self, stdout: str, stderr: str, return_code: int):
+        self.stdout = stdout
+        self.stderr = stderr
+        self.return_code = return_code
+        self.success = self.return_code == 0
 
     def output_from_json(self) -> Union[dict, list, int, float, str, None]:
         try:
-            return json.loads(self.output)
+            return json.loads(self.stdout)
         except JSONDecodeError:
             return None
+
+    def grep_stdout_for_string(
+        self, string: str, negate: bool = False, split: bool = False
+    ) -> Union[str, list[str]]:
+        if negate:
+            filtered = list(filter(lambda x: string not in x, self.stdout.split("\n")))
+        else:
+            filtered = list(filter(lambda x: string in x, self.stdout.split("\n")))
+        return filtered if split else "\n".join(filtered)
+
+    def grep_stdout_for_pattern(
+        self,
+        pattern: Union[re.Pattern[str], str],
+        flags: Union[int, RegexFlag] = 0,
+        negate: bool = False,
+        split: bool = False,
+    ) -> Union[str, list[str]]:
+        if negate:
+            filtered = list(
+                filter(
+                    lambda x: not re.match(pattern, x, flags=flags),
+                    self.stdout.split("\n"),
+                )
+            )
+        else:
+            filtered = list(
+                filter(
+                    lambda x: re.match(pattern, x, flags=flags), self.stdout.split("\n")
+                )
+            )
+        return filtered if split else "\n".join(filtered)

wlanpi_core/models/network/common.py

@@ -21,7 +21,7 @@ def get_interfaces(
             interface["link_speed"] = int(
                 run_command(
                     ["cat", f"/sys/class/net/{interface['ifname']}/speed"]
-                ).output
+                ).stdout
             )
 
     if custom_filter:

wlanpi_core/models/network/namespace/namespace.py

@@ -78,7 +78,7 @@ def list_namespaces() -> list:
         """
         result = run_command("ip -j netns list".split(), raise_on_fail=False)
         if not result.success:
-            raise NetworkNamespaceError(f"Error listing namespaces: {result.error}")
+            raise NetworkNamespaceError(f"Error listing namespaces: {result.stderr}")
         return result.output_from_json() or []
 
     @staticmethod
@@ -153,7 +153,7 @@ def destroy_namespace(namespace_name: str):
                 res = run_command(
                     f"ip netns exec {namespace_name} iw dev {interface['name']} info".split()
                 )
-                phynum = re.findall(r"wiphy ([0-9]+)", res.output)[0]
+                phynum = re.findall(r"wiphy ([0-9]+)", res.stdout)[0]
                 phy = f"phy{phynum}"
 
                 res = run_command(
@@ -162,7 +162,7 @@ def destroy_namespace(namespace_name: str):
                 )
                 if not res.success:
                     raise NetworkNamespaceError(
-                        f"Failed to move wireless interface {interface['name']} to default namespace: {res.error}"
+                        f"Failed to move wireless interface {interface['name']} to default namespace: {res.stderr}"
                     )
 
             elif interface["name"].startswith("eth"):
@@ -172,7 +172,7 @@ def destroy_namespace(namespace_name: str):
                 )
                 if not res.success:
                     raise NetworkNamespaceError(
-                        f"Failed to move wired interface {interface['name']} to default namespace: {res.error}"
+                        f"Failed to move wired interface {interface['name']} to default namespace: {res.stderr}"
                     )
 
             elif interface["name"].startswith("lo"):
@@ -188,7 +188,7 @@ def destroy_namespace(namespace_name: str):
         res = run_command(f"ip netns del {namespace_name}".split(), raise_on_fail=False)
         if not res.success:
             raise NetworkNamespaceError(
-                f"Unable to destroy namespace {namespace_name} {res.error}"
+                f"Unable to destroy namespace {namespace_name} {res.stderr}"
             )
 
     @staticmethod
@@ -201,6 +201,6 @@ def processes_using_namespace(namespace_name: str):
         )
         if not result.success:
             raise NetworkNamespaceError(
-                f"Error getting namespace processes: {result.error}"
+                f"Error getting namespace processes: {result.stderr}"
             )
-        return [int(x) for x in filter(None, result.output.split("\n") or [])]
+        return [int(x) for x in filter(None, result.stdout.split("\n") or [])]

wlanpi_core/models/network/vlan/vlan_file.py

@@ -1,9 +1,10 @@
 from collections import defaultdict
 from typing import Optional, Union
 
+from wlanpi_core.constants import DEFAULT_INTERFACE_FILE, DEFAULT_VLAN_INTERFACE_FILE
 from wlanpi_core.models.validation_error import ValidationError
 from wlanpi_core.schemas.network.config import Vlan
-from wlanpi_core.services.helpers import run_cli_async
+from wlanpi_core.utils.general import run_command_async
 
 
 class VLANFile:
@@ -28,8 +29,6 @@ class VLANFile:
         "wvdial",
         "ipv4ll",
     )
-    DEFAULT_VLAN_INTERFACE_FILE = "/etc/network/interfaces.d/vlans"
-    DEFAULT_INTERFACE_FILE = "/etc/network/interfaces"
 
     def __init__(
         self,
@@ -173,9 +172,13 @@ def generate_if_config_from_object(configuration: Vlan) -> str:
 
     @staticmethod
     async def check_interface_exists(interface: str) -> bool:
-        ethernet_interfaces = (
-            await run_cli_async("ls /sys/class/net/ | grep eth")
-        ).split("\n")
+        ethernet_interfaces = [
+            x
+            for x in (
+                await run_command_async("ls /sys/class/net/", raise_on_fail=True)
+            ).stdout.split("\n")
+            if "eth" in x
+        ]
         ethernet_interfaces = set([i.split(".")[0] for i in ethernet_interfaces if i])
         return interface in ethernet_interfaces
 

wlanpi_core/models/runcommand_error.py

@@ -1,8 +1,8 @@
 class RunCommandError(Exception):
     """Raised when runcommand returns stderr"""
 
-    def __init__(self, error_msg: str, status_code: int):
+    def __init__(self, error_msg: str, return_code: int):
         super().__init__(error_msg)
 
-        self.status_code = status_code
+        self.return_code = return_code
         self.error_msg = error_msg