Use of pylint and mypy

Author: Alan Christie

.github/workflows/build.yaml

@@ -41,6 +41,8 @@ jobs:
         pip install -r package-requirements.txt
     - name: Test
       run: |
+        pylint jote
+        mypy jote --install-types --non-interactive
         pyroma .
     - name: Build
       run: |

.pylintrc

@@ -0,0 +1,11 @@
+[messages control]
+
+# Disable some distracting checks.
+# See http://pylint.pycqa.org/en/latest/user_guide/message-control.html
+#
+# R : (Refactoring) warnings
+disable = import-error,
+          too-many-arguments,
+          too-many-locals,
+          too-many-branches,
+          too-many-statements

build-requirements.txt

@@ -1,3 +1,5 @@
 pyroma == 3.2
 pytest == 6.2.5
 pytest-cov == 3.0.0
+mypy == 0.931
+pylint == 2.12.2

jote/__main__.py

@@ -9,7 +9,7 @@
 from .jote import main as jote_main
 
 
-def main():
+def main() -> int:
     """The entry-point of the component."""
     return jote_main()
 

jote/compose.py

@@ -35,14 +35,13 @@
 
 def _get_docker_compose_version() -> str:
 
-    result: subprocess.CompletedProcess =\
-        subprocess.run(['docker-compose', 'version'],
-                       capture_output=True, timeout=4)
+    result = subprocess.run(['docker-compose', 'version'],
+                            capture_output=True, check=False, timeout=4)
 
     # stdout will contain the version on the first line: -
     # "docker-compose version 1.29.2, build unknown"
     # Ignore the first 23 characters of the first line...
-    return result.stdout.decode("utf-8").split('\n')[0][23:]
+    return str(result.stdout.decode("utf-8").split('\n')[0][23:])
 
 
 def get_test_root() -> str:
@@ -53,6 +52,9 @@ def get_test_root() -> str:
 
 
 class Compose:
+    """A class handling the execution of 'docker-compose'
+    for an individual test.
+    """
 
     # The docker-compose version (for the first test)
     _COMPOSE_VERSION: Optional[str] = None
@@ -104,7 +106,7 @@ def create(self) -> str:
             print(f'# docker-compose ({Compose._COMPOSE_VERSION})')
 
         # Make the test directory...
-        test_path: str = self.get_test_path()
+        test_path = self.get_test_path()
         project_path: str = self.get_test_project_path()
         inst_path: str = f'{project_path}/{_INSTANCE_DIRECTORY}'
         if not os.path.exists(inst_path):
@@ -142,15 +144,16 @@ def run(self) -> Tuple[int, str, str]:
         try:
             # Run the container
             # and then cleanup
-            test: subprocess.CompletedProcess =\
-                subprocess.run(['docker-compose', 'up',
-                                '--exit-code-from', 'job',
-                                '--abort-on-container-exit'],
-                               capture_output=True,
-                               timeout=timeout)
+            test = subprocess.run(['docker-compose', 'up',
+                                   '--exit-code-from', 'job',
+                                   '--abort-on-container-exit'],
+                                  capture_output=True,
+                                  timeout=timeout,
+                                  check=False)
             _ = subprocess.run(['docker-compose', 'down'],
                                capture_output=True,
-                               timeout=120)
+                               timeout=120,
+                               check=False)
         finally:
             os.chdir(cwd)
 

jote/jote.py

@@ -20,11 +20,11 @@
 
 # Where can we expect to find Job definitions?
 _DEFINITION_DIRECTORY: str = 'data-manager'
-# Data directory
+# Where can we expect to find test data?
 _DATA_DIRECTORY: str = 'data'
 
 # The yamllint configuration file of the repository under test.
-# Expected to be in the repo we're running from.
+# It must exist in the root of the repo we're running in.
 _YAMLLINT_FILE: str = '.yamllint'
 
 
@@ -105,9 +105,9 @@ def _load(skip_lint: bool = False) -> Tuple[List[DefaultMunch], int]:
                 return [], -1
 
         with open(jd_filename, 'r') as jd_file:
-            jd: Dict[str, Any] = yaml.load(jd_file, Loader=yaml.FullLoader)
-        if jd:
-            jd_munch: DefaultMunch = DefaultMunch.fromDict(jd)
+            job_def: Dict[str, Any] = yaml.load(jd_file, Loader=yaml.FullLoader)
+        if job_def:
+            jd_munch: DefaultMunch = DefaultMunch.fromDict(job_def)
             for jd_name in jd_munch.jobs:
                 if jd_munch.jobs[jd_name].tests:
                     num_tests += len(jd_munch.jobs[jd_name].tests)
@@ -158,7 +158,7 @@ def _check_exists(name: str, path: str, expected: bool) -> bool:
         print('! FAILURE')
         print(f'! Check exists "{name}" (does not exist)')
         return False
-    elif not expected and exists:
+    if not expected and exists:
         print(f'#   exists ({expected}) [FAILED]')
         print('! FAILURE')
         print(f'! Check does not exist "{name}" (exists)')
@@ -329,7 +329,8 @@ def _test(args: argparse.Namespace,
         # Run the container
         if test_status and not args.dry_run:
             # Run the container
-            exit_code, out, err = t_compose.run()
+            assert t_compose
+            exit_code, out, _ = t_compose.run()
 
             # Delete the test directory?
             # Not if there's an error
@@ -353,12 +354,15 @@ def _test(args: argparse.Namespace,
         if test_status \
                 and not args.dry_run \
                 and job_definition.tests[job_test_name].checks.outputs:
+
+            assert t_compose
             test_status = \
                 _check(t_compose,
                        job_definition.tests[job_test_name].checks.outputs)
 
         # Clean-up
         if test_status and not args.keep_results:
+            assert t_compose
             t_compose.delete()
 
         # Count?
@@ -383,7 +387,7 @@ def _wipe() -> None:
 # -----------------------------------------------------------------------------
 # main
 # -----------------------------------------------------------------------------
-def main() -> None:
+def main() -> int:
     """The console script entry-point. Called when jote is executed
     or from __main__.py, which is used by the installed console script.
     """
@@ -468,7 +472,7 @@ def main() -> None:
     if args.wipe:
         _wipe()
         print(f'Done [Wiped]')
-        return
+        return 0
 
     # Load all the files we can and then run the tests.
     job_definitions, num_tests = _load(args.skip_lint)
@@ -503,7 +507,7 @@ def main() -> None:
                     continue
 
                 if job_definition.jobs[job_name].tests:
-                    num_passed, num_ignored, num_failed =\
+                    _, num_ignored, num_failed =\
                         _test(args,
                               collection,
                               job_name,
@@ -547,6 +551,8 @@ def main() -> None:
     if total_fail_count == 0 and not args.keep_results:
         _wipe()
 
+    return 0
+
 
 # -----------------------------------------------------------------------------
 # MAIN

mypy.ini

@@ -0,0 +1,5 @@
+[mypy]
+# Strict
+strict = True
+# Now enable more rules
+ignore_missing_imports = True

setup.py

@@ -29,7 +29,7 @@ def get_long_description():
     author_email='[email protected]',
     url='https://github.com/informaticsmatters/data-manager-job-tester',
     license=copyright,
-    description='test',
+    description='The IM Data Manager Job Tester (jote)',
     long_description=get_long_description(),
     keywords='configuration',
     platforms=['any'],