Configure linters and fix errors. Clean up pyproject.toml

Author: razvan

.gitignore

@@ -2,4 +2,3 @@
 build
 dist
 src/beku_stackabletech.egg-info
-

.pre-commit-config.yaml

@@ -0,0 +1,34 @@
+# See https://pre-commit.com for more information
+# See https://pre-commit.com/hooks.html for more hooks
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v3.2.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-added-large-files
+  - repo: https://github.com/Yelp/detect-secrets
+    rev: v1.4.0
+    hooks:
+      - id: detect-secrets
+  - repo: https://github.com/pre-commit/mirrors-autopep8
+    rev: v2.0.1
+    hooks:
+      - id: autopep8
+        entry: autopep8
+  - repo: local
+    # These hooks are run from the local virtual env (and it's assumed they are there)
+    # to allow these tools to find project dependencies.
+    hooks:
+      - id: pylint
+        name: pylint
+        entry: pylint
+        language: system
+        types: [ python ]
+      - id: mypy
+        name: mypy
+        entry: mypy
+        language: system
+        types: [ python ]
+        args: ['--install-types', '--non-interactive']

README.md

@@ -1,6 +1,6 @@
 # beku
 
-Version: 0.0.5
+Version: 0.0.6-dev
 
 ## Installation
 
@@ -23,13 +23,12 @@ Fast Kuttl tests expander for Stackable integration tests.
 Update the version in:
 
 * `pyptoject.toml`
-* `setup.cfg`
 * `__init__.py`
 * `README.md`
 
 Update the CHANGELOG.
 Commit and tag.
 Build and publish:
     rm -rf dist/
-    python -m build --sdist --wheel .
+    python -m build --wheel .
     twine upload dist/*

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "beku-stackabletech"
-version = "0.0.5"
+version = "0.0.6-dev"
 authors = [
   { name="Razvan Mihai", email="razvan.mihai@stackable.tech" }
 ]
@@ -12,16 +12,16 @@ classifiers = [
     "License :: OSI Approved :: MIT License",
     "Operating System :: OS Independent",
 ]
+dependencies = [
+    "Jinja2>=3.1.2",
+    "PyYAML>=6.0"
+]
+[project.optional-dependencies]
+lint = ['pre-commit', 'pylint', 'mypy']
 
 [project.scripts]
 beku = "beku:main"
 
-[build-system]
-requires = [
-    "setuptools >= 40.6.2",
-    "wheel >= 0.30.0"
-]
-build-backend = "setuptools.build_meta"
-
 [project.urls]
-Homepage = "https://github.com/stackabletech/beku.py"
+"Homepage" = "https://github.com/stackabletech/beku.py"
+"Bug Tracker" = "https://github.com/stackabletech/beku.py/issues"

setup.cfg

@@ -1,18 +1,19 @@
-from typing import Dict, List, TypeVar, Type, Tuple
+"""Expand kuttl tests from templates."""
+import logging
+import os
+import re
+import sys
 from argparse import ArgumentParser, Namespace
 from dataclasses import dataclass, field
 from itertools import product, chain
-from jinja2 import Environment, FileSystemLoader
 from os import walk, path, makedirs
 from shutil import copy2
-from sys import exit
-from yaml import safe_load
-import logging
-import re
-import os
+from typing import Dict, List, TypeVar, Type, Tuple
 
+from jinja2 import Environment, FileSystemLoader
+from yaml import safe_load
 
-__version_info__ = (0, 0, 5)
+__version_info__ = (0, 0, '6-dev')
 __version__ = ".".join(map(str, __version_info__))
 
 
@@ -22,7 +23,7 @@ def ansible_lookup(loc: str, what: str) -> str:
     Simulates the Ansible `lookup()` function which is made available by Ansible Jinja templates.
     Raises an exception if `loc` is not `env`.
     """
-    if not loc is 'env':
+    if loc != 'env':
         raise ValueError("Can only lookup() in 'env'")
     result = ""
     try:
@@ -34,6 +35,7 @@ def ansible_lookup(loc: str, what: str) -> str:
 
 @dataclass
 class TestCase:
+    """Test case definition."""
     name: str
     values: Dict[str, str]
     tid: str = field(init=False)
@@ -48,7 +50,8 @@ def __post_init__(self):
         )
 
     def expand(self, template_dir: str, target_dir: str) -> None:
-        logging.info(f"Expanding test case id [{self.tid}]")
+        """Expand test case."""
+        logging.info("Expanding test case id [%s]", self.tid)
         td_root = path.join(template_dir, self.name)
         tc_root = path.join(target_dir, self.name, self.tid)
         _mkdir_ignore_exists(tc_root)
@@ -63,115 +66,121 @@ def expand(self, template_dir: str, target_dir: str) -> None:
             if sub_level == 5:
                 # Sanity check
                 raise ValueError("Maximum recursive level (5) reached.")
-            for d in dirs:
+            for dir_name in dirs:
                 _mkdir_ignore_exists(
-                    path.join(tc_root, root[len(td_root) + 1:], d))
-            for f in files:
-                source = path.join(root, f)
+                    path.join(tc_root, root[len(td_root) + 1:], dir_name))
+            for file_name in files:
+                source = path.join(root, file_name)
                 dest = ""
                 f_mode = os.stat(source).st_mode
-                if f.endswith(".j2"):
-                    logging.debug(f"Render template {f} to {dest}")
-                    dest = path.join(tc_root, root[len(td_root) + 1:], f[:-3:])
-                    self._expand_template(f, dest, test_env)
+                if file_name.endswith(".j2"):
+                    logging.debug("Render template %s to %s", file_name, dest)
+                    dest = path.join(
+                        tc_root, root[len(td_root) + 1:], file_name[:-3:])
+                    self._expand_template(file_name, dest, test_env)
                 else:
-                    dest = path.join(tc_root, root[len(td_root) + 1:], f)
-                    logging.debug(f"Copy file {f} to {dest}")
+                    dest = path.join(
+                        tc_root, root[len(td_root) + 1:], file_name)
+                    logging.debug("Copy file %s to %s", file_name, dest)
                     copy2(source, dest)
                 # restore file permissions (especially the executable bit is important here)
-                logging.debug(f"Update file mode for {dest}")
+                logging.debug("Update file mode for %s", dest)
                 os.chmod(dest, f_mode)
 
-    def _expand_template(self, template: str, dest: str, env: Environment) -> None:
-        logging.debug(f"Expanding template {template}")
-        t = env.get_template(template)
+    def _expand_template(self, template_file: str, dest: str, env: Environment) -> None:
+        logging.debug("Expanding template %s", template_file)
+        template = env.get_template(template_file)
         with open(dest, encoding="utf8", mode="w") as stream:
             print(
-                t.render({"test_scenario": {"values": self.values}}), file=stream)
+                template.render({"test_scenario": {"values": self.values}}), file=stream)
 
 
 @dataclass
 class TestDimension:
+    """Test dimension."""
     name: str
     values: List[str]
 
     def expand(self) -> List[Tuple[str, str]]:
+        """Return a list of tuples in the form of (<dimension>, <value>)"""
         return [(self.name, v) for v in self.values]
 
 
 @dataclass
 class TestDefinition:
+    """Test case definition."""
     name: str
     dimensions: List[str]
 
 
-TTestSuite = TypeVar("TTestSuite", bound="TestSuite")
+TTestSuite = TypeVar(    # pylint: disable=invalid-name
+    "TTestSuite", bound="TestSuite")
 
 
-@dataclass
+@dataclass(frozen=True)
 class TestSuite:
-    source: str
-    test_cases: List[TestCase]
+    """Test suite template."""
+    source: str = field()
+    test_cases: List[TestCase] = field(default_factory=list)
 
-    def __init__(self, source: str) -> None:
-        self.source = source
-        with open(source, encoding="utf8") as stream:
+    def __post_init__(self) -> None:
+        with open(self.source, encoding="utf8") as stream:
             tin = safe_load(stream)
             dimensions = [
                 TestDimension(d["name"], d["values"]) for d in tin["dimensions"]
             ]
             test_def = [
                 TestDefinition(t["name"], t["dimensions"]) for t in tin["tests"]
             ]
-            self.test_cases = self._build(dimensions, test_def)
+            self.test_cases.extend(self._build(dimensions, test_def))
 
     @classmethod
     def _build(
-        cls: Type[TTestSuite], dims: List[TestDimension], tests: List[TestDefinition]
+            cls: Type[TTestSuite], dims: List[TestDimension], tests: List[TestDefinition]
     ) -> List[TestCase]:
         """
         >>> TestSuite._build([TestDimension(name='trino', values=['234', '235'])],
         ... [TestDefinition(name='smoke', dimensions=['trino'])])
-        [TestCase(name='smoke', values={'trino': '234'}, name='smoke_trino-234'), TestCase(name='smoke', values={'trino': '235'}, name='smoke_trino-235')]
+        [TestCase(name='smoke', values={'trino': '234'}, name='smoke_trino-234'), \
+        TestCase(name='smoke', values={'trino': '235'}, name='smoke_trino-235')]
         """
         result = []
-        for t in tests:
-            used_dims = [d for d in dims if d.name in t.dimensions]
+        for test in tests:
+            used_dims = [d for d in dims if d.name in test.dimensions]
             expanded_test_dims: List[List[Tuple[str, str]]] = [
                 d.expand() for d in used_dims
             ]
             for tc_dim in product(*expanded_test_dims):
-                result.append(TestCase(name=t.name, values=dict(tc_dim)))
+                result.append(TestCase(name=test.name, values=dict(tc_dim)))
         return result
 
     def __repr__(self) -> str:
         return f"TestSuite(source={self.source})"
 
     def expand(self, template_dir: str, output_dir: str, kuttl_tests: str) -> int:
-        logging.info(f"Expanding test suite from {self.source}")
-        self._sanity_checks(template_dir, output_dir, kuttl_tests)
+        """Expand test suite."""
+        logging.info("Expanding test suite from %s", self.source)
+        self._sanity_checks(template_dir, kuttl_tests)
         _mkdir_ignore_exists(output_dir)
         self._expand_kuttl_tests(output_dir, kuttl_tests)
         for test_case in self.test_cases:
             test_case.expand(template_dir, output_dir)
         return 0
 
-    def _sanity_checks(
-        self, template_dir: str, output_dir: str, kuttl_tests: str
-    ) -> None:
-        for tc in self.test_cases:
-            td_root = path.join(template_dir, tc.name)
+    def _sanity_checks(self, template_dir: str, kuttl_tests: str) -> None:
+        for test_case in self.test_cases:
+            td_root = path.join(template_dir, test_case.name)
             if not path.isdir(td_root):
                 raise ValueError(
                     f"Test definition directory not found [{td_root}]")
-        if not (path.isfile(kuttl_tests)):
+        if not path.isfile(kuttl_tests):
             raise ValueError(
                 f"Kuttl test config template not found [{kuttl_tests}]")
 
     def _expand_kuttl_tests(self, output_dir: str, kuttl_tests: str) -> None:
         env = Environment(loader=FileSystemLoader(path.dirname(kuttl_tests)))
         kt_base_name = path.basename(kuttl_tests)
-        t = env.get_template(kt_base_name)
+        template = env.get_template(kt_base_name)
         kt_dest_name = re.sub(r"\.j(inja)?2$", "", kt_base_name)
         # Compatibility warning: Assume output_dir ends with 'tests' and remove
         # it from the destination file
@@ -181,20 +190,21 @@ def _expand_kuttl_tests(self, output_dir: str, kuttl_tests: str) -> None:
                 "tests": [{"name": tn} for tn in {tc.name for tc in self.test_cases}]
             }
         }
-        logging.debug(f"kuttl vars {kuttl_vars}")
+        logging.debug("kuttl vars %s", kuttl_vars)
         with open(dest, encoding="utf8", mode="w") as stream:
-            print(t.render(kuttl_vars), file=stream)
+            print(template.render(kuttl_vars), file=stream)
 
 
 def parse_cli_args() -> Namespace:
+    """Parse command line args."""
     parser = ArgumentParser(
         description="Kuttl test expander for the Stackable Data Platform")
     parser.add_argument(
         "-v",
         "--version",
         help="Display application version",
         action='version',
-        version='%(prog)s {version}'.format(version=__version__)
+        version=f'%(prog)s {__version__}'
     )
 
     parser.add_argument(
@@ -247,27 +257,27 @@ def parse_cli_args() -> Namespace:
 def _cli_log_level(cli_arg: str) -> int:
     if cli_arg == "debug":
         return logging.DEBUG
-    else:
-        return logging.INFO
+    return logging.INFO
 
 
-def _mkdir_ignore_exists(path: str) -> None:
+def _mkdir_ignore_exists(dir_name: str) -> None:
     try:
-        logging.debug(f"Creating directory {path}")
-        makedirs(path)
+        logging.debug("Creating directory %s", dir_name)
+        makedirs(dir_name)
     except FileExistsError:
         pass
 
 
 def main() -> int:
+    """Main"""
     cli_args = parse_cli_args()
     logging.basicConfig(
         encoding="utf-8", level=_cli_log_level(cli_args.log_level))
-    ts = TestSuite(cli_args.test_definition)
+    test_suite = TestSuite(cli_args.test_definition)
     # Compatibility warning: add 'tests' to output_dir
     output_dir = path.join(cli_args.output_dir, "tests")
-    return ts.expand(cli_args.template_dir, output_dir, cli_args.kuttl_test)
+    return test_suite.expand(cli_args.template_dir, output_dir, cli_args.kuttl_test)
 
 
 if __name__ == "__main__":
-    exit(main())
+    sys.exit(main())