pr checks

Author: rorsten

.github/workflows/checks.yaml

@@ -0,0 +1,29 @@
+---
+name: static-analysis
+on: push
+jobs:
+  static-analysis:
+    name: Static Analysis
+    runs-on: ubuntu-latest
+    container: python:3.12
+    steps:
+      - uses: actions/checkout@v4
+      - name: install prerequisites
+        run: |
+          pip install .[check] .[test] black pylint
+          curl -L https://codeclimate.com/downloads/test-reporter/test-reporter-latest-linux-amd64 --output cc-test-reporter
+          chmod +x cc-test-reporter
+      - name: run black
+        run: |
+          black --check --diff .
+      - name: run pylint
+        run: |
+          pylint --fail-under 10 src
+          pylint --fail-under 10 --disable=protected-access tests
+      - name: run tests
+        env:
+          CC_TEST_REPORTER_ID: ${{ secrets.CC_TEST_REPORTER_ID }}
+        run:
+          ./cc-test-reporter before-build
+          pytest . --cov-report lcov
+          ./cc-test-reporter after-build --exit-code $?

pyproject.toml

@@ -40,4 +40,13 @@ log_cli = "true"
 testpaths = ["tests"]
 pythonpath = "src"
 addopts = "-n 4 -Walways --cov=cli_wrapper --cov-branch --cov-report html:coverage"
-asyncio_default_fixture_loop_scope = "function"
+asyncio_default_fixture_loop_scope = "function"
+
+[tool.pylint."messages control"]
+disable = [
+    "missing-module-docstring",
+    "missing-class-docstring",
+    "missing-function-docstring",
+    "logging-fstring-interpolation",
+    "fixme"
+]

src/cli_wrapper/cli_wrapper.py

@@ -46,7 +46,7 @@ def validate_pod_name(name):
         to `command --arg=val`. If you want to use spaces instead, set this to ' '
 """
 
-import asyncio
+import asyncio.subprocess
 import logging
 import os
 import subprocess
@@ -87,7 +87,7 @@ def from_dict(cls, arg_dict):
             transformer=arg_dict.get("transformer", None),
         )
 
-    def _to_dict(self):
+    def to_dict(self):
         """
         Convert the Argument to a dictionary
         :return: the dictionary representation of the Argument
@@ -96,7 +96,7 @@ def _to_dict(self):
         return {
             "literal_name": self.literal_name,
             "default": self.default,
-            "validator": self.validator._to_dict() if self.validator is not None else None,
+            "validator": self.validator.to_dict() if self.validator is not None else None,
         }
 
     def is_valid(self, value):
@@ -139,7 +139,7 @@ def arg_converter(value: dict):
 
 
 @define
-class Command(object):
+class Command:  # pylint: disable=too-many-instance-attributes
     """
     Command represents a command to be run with the cli_wrapper
     """
@@ -154,7 +154,7 @@ class Command(object):
     arg_separator: str = field(repr=False, default="=")
 
     @classmethod
-    def _from_dict(cls, command_dict, **kwargs):
+    def from_dict(cls, command_dict, **kwargs):
         """
         Create a Command from a dictionary
         :param command_dict: the dictionary to be converted
@@ -172,7 +172,7 @@ def _from_dict(cls, command_dict, **kwargs):
             **kwargs,
         )
 
-    def _to_dict(self):
+    def to_dict(self):
         """
         Convert the Command to a dictionary.
         Excludes prefixes/separators, because they are set in the CLIWrapper
@@ -182,8 +182,8 @@ def _to_dict(self):
         return {
             "cli_command": self.cli_command,
             "default_flags": self.default_flags,
-            "args": {k: v._to_dict() for k, v in self.args.items()},
-            "parse": self.parse._to_dict() if self.parse is not None else None,
+            "args": {k: v.to_dict() for k, v in self.args.items()},
+            "parse": self.parse.to_dict() if self.parse is not None else None,
         }
 
     def validate_args(self, *args, **kwargs):
@@ -231,12 +231,13 @@ def build_args(self, *args, **kwargs):
 
 
 @define
-class CLIWrapper:
+class CLIWrapper:  # pylint: disable=too-many-instance-attributes
     path: str
     env: dict[str, str] = None
     commands: dict[str, Command] = {}
 
     trusting: bool = True
+    raise_exc: bool = False
     async_: bool = False
     default_transformer: str = "snake2kebab"
     short_prefix: str = "-"
@@ -263,17 +264,19 @@ def _get_command(self, command: str):
             return c
         return self.commands[command]
 
-    def _update_command(
+    def _update_command(  # pylint: disable=too-many-arguments
         self,
         command: str,
-        cli_command: str = None,
+        *,
+        cli_command: str | list[str] = None,
         args: dict[str | int, any] = None,
         default_flags: dict = None,
         parse=None,
     ):
         """
         update the command to be run with the cli_wrapper
-        :param command: the subcommand for the cli tool
+        :param command: the command name for the wrapper
+        :param cli_command: the command to be run, if different from the command name
         :param default_flags: default flags to be used with the command
         :param parse: function to parse the output of the command
         :return:
@@ -303,7 +306,7 @@ def _run(self, command: str, *args, **kwargs):
         env = os.environ.copy().update(self.env if self.env is not None else {})
         logger.debug(f"Running command: {' '.join(command_args)}")
         # run the command
-        result = subprocess.run(command_args, capture_output=True, text=True, env=env)
+        result = subprocess.run(command_args, capture_output=True, text=True, env=env, check=self.raise_exc)
         if result.returncode != 0:
             raise RuntimeError(f"Command {command} failed with error: {result.stderr}")
         return command_obj.parse(result.stdout)
@@ -314,7 +317,7 @@ async def _run_async(self, command: str, *args, **kwargs):
         command_args = [self.path] + list(command_obj.build_args(*args, **kwargs))
         env = os.environ.copy().update(self.env if self.env is not None else {})
         logger.error(f"Running command: {', '.join(command_args)}")
-        proc = await asyncio.subprocess.create_subprocess_exec(
+        proc = await asyncio.subprocess.create_subprocess_exec(  # pylint: disable=no-member
             *command_args,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
@@ -351,22 +354,22 @@ def from_dict(cls, cliwrapper_dict):
             else:
                 if "cli_command" not in config:
                     config["cli_command"] = command
-            commands[command] = Command._from_dict(config)
+            commands[command] = Command.from_dict(config)
 
         return CLIWrapper(
             commands=commands,
             **cliwrapper_dict,
         )
 
-    def _to_dict(self):
+    def to_dict(self):
         """
         Convert the CLIWrapper to a dictionary
         :return:
         """
         return {
             "path": self.path,
             "env": self.env,
-            "commands": {k: v._to_dict() for k, v in self.commands.items()},
+            "commands": {k: v.to_dict() for k, v in self.commands.items()},
             "trusting": self.trusting,
             "async_": self.async_,
         }

src/cli_wrapper/parsers.py

@@ -1,6 +1,6 @@
 import logging
 
-from .util.callable_chain import params_from_kwargs, CallableChain
+from .util.callable_chain import CallableChain
 from .util.callable_registry import CallableRegistry
 
 logger = logging.getLogger(__name__)
@@ -71,13 +71,15 @@ def dotted_dictify(src, *args, **kwargs):
 
 
 class Parser(CallableChain):
-
+    """
+    Parser class that allows for the chaining of multiple parsers.
+    """
     def __init__(self, config):
         super().__init__(config, parsers)
 
-    def __call__(self, input):
+    def __call__(self, src):
         # For now, parser expects to be called with one input.
-        result = input
+        result = src
         for parser in self.chain:
             logger.debug(result)
             result = parser(result)

src/cli_wrapper/util/callable_chain.py

@@ -24,7 +24,7 @@ def __init__(self, config, source):
             name, args, kwargs = params_from_kwargs(config)
             self.chain = [source.get(name, args, kwargs)]
 
-    def _to_dict(self):
+    def to_dict(self):
         return self.config
 
     @abstractmethod

src/cli_wrapper/util/callable_registry.py

@@ -40,22 +40,22 @@ def get(self, name: str | Callable, args=None, kwargs=None) -> Callable:
             raise KeyError(f"{self.callable_name} '{name}' not found.")
         return lambda *fargs: callable_(*fargs, *args, **kwargs)
 
-    def register(self, name: str, callable: callable, group="core"):
+    def register(self, name: str, callable_: callable, group="core"):
         """
         Registers a new parser function with the specified name.
 
         :param name: The name to associate with the parser.
-        :param callable: The callable function to register.
+        :param callable_: The callable function to register.
         """
         ngroup, name = self._parse_name(name)
         if ngroup is not None:
             if group != "core":
                 # approximately, raise an exception if a group is specified in the name and the group arg
-                raise KeyError(f"'{callable}' already specifies a group.")
+                raise KeyError(f"'{callable_}' already specifies a group.")
             group = ngroup
         if name in self._all[group]:
             raise KeyError(f"{self.callable_name} '{name}' already registered.")
-        self._all[group][name] = callable
+        self._all[group][name] = callable_
 
     def register_group(self, name: str, parsers: dict = None):
         """
@@ -87,6 +87,6 @@ def _parse_name(self, name: str) -> tuple[str, str]:
             return None, name
         try:
             group, name = name.split(".")
-        except ValueError:
-            raise KeyError(f"{self.callable_name} name '{name}' is not valid.")
+        except ValueError as err:
+            raise KeyError(f"{self.callable_name} name '{name}' is not valid.") from err
         return group, name

src/cli_wrapper/validators.py

@@ -31,9 +31,9 @@ class Validator(CallableChain):
 
     def __init__(self, config):
         if callable(config):
-            id = str(uuid4())
-            validators.register(id, config)
-            config = id
+            id_ = str(uuid4())
+            validators.register(id_, config)
+            config = id_
         self.config = config
         super().__init__(config, validators)
 
@@ -46,7 +46,7 @@ def __call__(self, value):
             result = result and validator_result
         return result
 
-    def _to_dict(self):
+    def to_dict(self):
         """
         Converts the validator configuration to a dictionary.
         """

tests/test_cli_wrapper.py

@@ -1,11 +1,9 @@
-import json
 import logging
 from json import loads
 
 import pytest
 
 from cli_wrapper.cli_wrapper import CLIWrapper, Argument, Command
-from cli_wrapper.transformers import transformers
 from cli_wrapper.validators import validators
 
 logger = logging.getLogger(__name__)
@@ -98,7 +96,7 @@ def validator(name):
         validators._all.pop("test_command_group")
 
     def test_command_from_dict(self):
-        command = Command._from_dict(
+        command = Command.from_dict(
             {
                 "cli_command": "get",
                 "default_flags": {"namespace": "default"},
@@ -120,7 +118,7 @@ def test_command_from_dict(self):
             "--namespace=kube-system",
         ]
 
-        command = Command._from_dict({"cli_command": "get", "args": {}})
+        command = Command.from_dict({"cli_command": "get", "args": {}})
 
 
 class TestCLIWrapper:

tests/test_parsers.py

@@ -22,7 +22,7 @@ def custom_extract(src):
         parser = Parser(["json", custom_extract])
         assert parser(testdata) == "baz"
 
-        testdata = "foo:\n" "  bar: baz\n"
+        testdata = "foo:\n  bar: baz\n"
         parser = Parser(["yaml", "dotted_dict"])
         assert parser(testdata).foo.bar == "baz"
 

tests/test_serialization.py

@@ -14,11 +14,11 @@ def test_wrapper_to_dict(self):
             },
         )
 
-        config = kubectl._to_dict()
+        config = kubectl.to_dict()
 
         kubectl2 = CLIWrapper.from_dict(config)
 
-        assert kubectl2._to_dict() == config
+        assert kubectl2.to_dict() == config
 
     def test_argument_to_dict(self):
         arg = Argument(
@@ -27,8 +27,8 @@ def test_argument_to_dict(self):
             validator=["is_alnum", "starts_alpha"],
         )
 
-        config = arg._to_dict()
+        config = arg.to_dict()
 
         arg2 = Argument.from_dict(config)
 
-        assert arg2._to_dict() == config
+        assert arg2.to_dict() == config