app: main: support custom tools

scripts/custom_tools/custom_tool.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python3
+
+"""Example out-of-tree tool"""
+
+__author__ = "Jordan Yates"
+__copyright__ = "Copyright 2025, Embeint Inc"
+
+from infuse_iot.commands import InfuseCommand
+
+
+class SubCommand(InfuseCommand):
+    NAME = "custom_tool"
+    HELP = "Test out-of-tree tool"
+    DESCRIPTION = "Test out-of-tree tool"
+
+    @classmethod
+    def add_parser(cls, parser):
+        parser.add_argument("--echo", "-e", required=True, type=str)
+
+    def __init__(self, args):
+        self.echo_string = args.echo
+
+    def run(self):
+        print("Echoing provided string:")
+        print(self.echo_string)

src/infuse_iot/app/main.py

@@ -8,13 +8,16 @@
 
 import argparse
 import importlib.util
+import pathlib
 import pkgutil
 import sys
+import types
 
 import argcomplete
 
 import infuse_iot.tools
 from infuse_iot.commands import InfuseCommand
+from infuse_iot.credentials import get_custom_tool_path
 from infuse_iot.version import __version__
 
 
@@ -38,24 +41,42 @@ def run(self, argv):
         tool = self.args.tool_class(self.args)
         tool.run()
 
+    def _load_from_module(self, parent_parser: argparse._SubParsersAction, module: types.ModuleType):
+        tool_cls: InfuseCommand = module.SubCommand
+        parser = parent_parser.add_parser(
+            tool_cls.NAME,
+            help=tool_cls.HELP,
+            description=tool_cls.DESCRIPTION,
+            formatter_class=argparse.RawDescriptionHelpFormatter,
+        )
+        parser.set_defaults(tool_class=tool_cls)
+        tool_cls.add_parser(parser)
+
     def _load_tools(self, parser: argparse.ArgumentParser):
         tools_parser = parser.add_subparsers(title="commands", metavar="<command>", required=True)
 
-        # Iterate over tools
+        # Iterate over local tools
         for _, name, _ in pkgutil.walk_packages(infuse_iot.tools.__path__):
             full_name = f"{infuse_iot.tools.__name__}.{name}"
             module = importlib.import_module(full_name)
-
-            # Add tool to parser
-            tool_cls: InfuseCommand = module.SubCommand
-            parser = tools_parser.add_parser(
-                tool_cls.NAME,
-                help=tool_cls.HELP,
-                description=tool_cls.DESCRIPTION,
-                formatter_class=argparse.RawDescriptionHelpFormatter,
-            )
-            parser.set_defaults(tool_class=tool_cls)
-            tool_cls.add_parser(parser)
+            self._load_from_module(tools_parser, module)
+
+        # Load custom tools, if configured
+        if extension_tools := get_custom_tool_path():
+            extension_path = pathlib.Path(extension_tools)
+            for _, name, _ in pkgutil.walk_packages([extension_tools]):
+                full_name = f"{infuse_iot.tools.__name__}.{name}"
+                full_path = str(extension_path / f"{name}.py")
+                spec = importlib.util.spec_from_file_location(full_name, full_path)
+                if spec is None or spec.loader is None:
+                    continue
+                module = importlib.util.module_from_spec(spec)
+                try:
+                    spec.loader.exec_module(module)
+                except Exception as _:
+                    continue
+                if hasattr(module, "SubCommand"):
+                    self._load_from_module(tools_parser, module)
 
 
 def main(argv=None):

src/infuse_iot/credentials.py

@@ -4,14 +4,14 @@
 import yaml
 
 
-def set_api_key(api_key: str):
+def set_api_key(api_key: str) -> None:
     """
     Save the Infuse-IoT API key to the keyring module
     """
     keyring.set_password("infuse-iot", "api-key", api_key)
 
 
-def get_api_key():
+def get_api_key() -> str:
     """
     Retrieve the Infuse-IoT API key from the keyring module
     """
@@ -20,8 +20,14 @@ def get_api_key():
         raise FileNotFoundError("API key does not exist in keyring")
     return key
 
+def delete_api_key() -> None:
+    """
+    Delete the Infuse-IoT API key from the keyring module
+    """
+    keyring.delete_password("infuse-iot", "api-key")
 
-def save_network(network_id: int, network_info: str):
+
+def save_network(network_id: int, network_info: str) -> None:
     """
     Save an Infuse-IoT network key to the keyring module
     """
@@ -38,3 +44,24 @@ def load_network(network_id: int):
     if key is None:
         raise FileNotFoundError(f"Network key {network_id:06x} does not exist in keyring")
     return yaml.safe_load(key)
+
+
+def set_custom_tool_path(path: str):
+    """
+    Save the location of custom Infuse-IoT tools on the filesystem
+    """
+    keyring.set_password("infuse-iot", "custom-tools", path)
+
+
+def get_custom_tool_path() -> str | None:
+    """
+    Retrieve the location of custom Infuse-IoT tools on the filesystem
+    """
+    return keyring.get_password("infuse-iot", "custom-tools")
+
+
+def delete_custom_tool_path() -> None:
+    """
+    Delete the location of custom Infuse-IoT tools on the filesystem
+    """
+    return keyring.delete_password("infuse-iot", "custom-tools")

src/infuse_iot/tools/credentials.py

@@ -9,7 +9,7 @@
 
 from infuse_iot import credentials
 from infuse_iot.commands import InfuseCommand
-from infuse_iot.util.argparse import ValidFile
+from infuse_iot.util.argparse import ValidDir, ValidFile
 
 
 class SubCommand(InfuseCommand):
@@ -22,6 +22,7 @@ def add_parser(cls, parser):
         parser.add_argument("--api-key", type=str, help="Set Infuse-IoT API key")
         parser.add_argument("--api-key-print", action="store_true", help="Print Infuse-IoT API key")
         parser.add_argument("--network", type=ValidFile, help="Load network credentials from file")
+        parser.add_argument("--custom-tools", type=ValidDir, help="Location of custom tools")
 
     def __init__(self, args):
         self.args = args
@@ -30,11 +31,16 @@ def run(self):
         if self.args.api_key is not None:
             credentials.set_api_key(self.args.api_key)
         if self.args.api_key_print:
-            print(f"API Key: {credentials.get_api_key()}")
+            try:
+                print(f"API Key: {credentials.get_api_key()}")
+            except FileNotFoundError:
+                print("API Key: N/A")
         if self.args.network is not None:
             # Read the file
             with self.args.network.open("r") as f:
                 content = f.read()
             # Validate it is valid yaml
             network_info = yaml.safe_load(content)
             credentials.save_network(network_info["id"], content)
+        if self.args.custom_tools:
+            credentials.set_custom_tool_path(str(self.args.custom_tools.absolute()))

tests/test_credentials.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+
+import os
+import subprocess
+
+import pytest
+
+import infuse_iot.credentials as cred
+
+assert "TOXTEMPDIR" in os.environ, "you must run these tests using tox"
+
+
+def test_credentials():
+    # Validate the credentials API
+
+    try:
+        cred.delete_api_key()
+    except Exception as _:
+        pass
+
+    with pytest.raises(FileNotFoundError):
+        cred.get_api_key()
+
+    test_api_key = "ABCDEFGHIJKLMNOP"
+    test_api_key_2 = "ABCDEFGHIJKLMNOP123456"
+
+    output = subprocess.check_output(["infuse", "credentials", "--api-key-print"]).decode()
+    assert "API Key: N/A" in output
+
+    subprocess.check_output(["infuse", "credentials", "--api-key", test_api_key]).decode()
+    assert test_api_key == cred.get_api_key()
+
+    output = subprocess.check_output(["infuse", "credentials", "--api-key-print"]).decode()
+    assert test_api_key in output
+
+    cred.set_api_key(test_api_key_2)
+
+    output = subprocess.check_output(["infuse", "credentials", "--api-key-print"]).decode()
+    assert test_api_key_2 in output
+
+    cred.delete_api_key()
+
+    output = subprocess.check_output(["infuse", "credentials", "--api-key-print"]).decode()
+    assert "API Key: N/A" in output

tests/test_custom_tools.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+
+import os
+import pathlib
+import subprocess
+
+import pytest
+
+import infuse_iot.credentials as cred
+
+assert "TOXTEMPDIR" in os.environ, "you must run these tests using tox"
+
+
+def test_custom_tool_integration():
+    # Validate custom tool integration
+    echo_string = "test_string"
+
+    try:
+        cred.delete_custom_tool_path()
+    except Exception as _:
+        pass
+
+    with pytest.raises(subprocess.CalledProcessError):
+        subprocess.check_output(["infuse", "custom_tool", "--echo", echo_string])
+
+    custom_tools_path = pathlib.Path(__file__).parent.parent / 'scripts' / 'custom_tools'
+
+    subprocess.check_output(["infuse", "credentials", "--custom-tools", str(custom_tools_path)])
+
+    output = subprocess.check_output(["infuse", "custom_tool", "--echo", echo_string]).decode()
+    assert echo_string in output
+
+    cred.delete_custom_tool_path()
+
+    with pytest.raises(subprocess.CalledProcessError):
+        subprocess.check_output(["infuse", "custom_tool", "--echo", echo_string])

tox.ini

@@ -16,6 +16,7 @@ deps =
     types-PyYAML
     pandas-stubs
     types-tabulate
+    keyrings-alt
     mypy
     ruff
 setenv =