Add python typechecking for mypy on src/ci_workflow and tests/tests_ci_workflow

.github/workflows/python-tests.yml

@@ -32,7 +32,7 @@ jobs:
           pipenv run flake8 .
       - name: Run Type Checker
         run: |
-          pipenv run mypy src/build_workflow tests/tests_build_workflow src/checkout_workflow tests/tests_checkout_workflow src/run_assemble.py tests/test_run_assemble.py src/assemble_workflow tests/tests_assemble_workflow src/manifests tests/tests_manifests src/paths tests/tests_paths src/system tests/tests_system
+          pipenv run mypy src/build_workflow tests/tests_build_workflow src/checkout_workflow tests/tests_checkout_workflow src/run_assemble.py tests/test_run_assemble.py src/assemble_workflow tests/tests_assemble_workflow src/manifests tests/tests_manifests src/paths tests/tests_paths src/system tests/tests_system src/ci_workflow tests/tests_ci_workflow
       - name: Run Tests with Coverage
         run: |
           pipenv run coverage run -m pytest --cov=./src --cov-report=xml

.pre-commit-config.yaml

@@ -19,7 +19,7 @@ repos:
       - id: mypy
         stages: [commit]
         name: mypy
-        entry: bash -c 'pipenv run mypy src/build_workflow tests/tests_build_workflow src/checkout_workflow tests/tests_checkout_workflow src/run_assemble.py tests/test_run_assemble.py src/assemble_workflow tests/tests_assemble_workflow src/manifests tests/tests_manifests src/paths tests/tests_paths src/system tests/tests_system'
+        entry: bash -c 'pipenv run mypy src/build_workflow tests/tests_build_workflow src/checkout_workflow tests/tests_checkout_workflow src/run_assemble.py tests/test_run_assemble.py src/assemble_workflow tests/tests_assemble_workflow src/manifests tests/tests_manifests src/paths tests/tests_paths src/system tests/tests_system src/ci_workflow tests/tests_ci_workflow'
         language: system
       - id: pytest
         stages: [commit]

src/ci_workflow/ci_args.py

@@ -5,15 +5,16 @@
 # compatible open source license.
 
 import argparse
+import io
 import logging
 import sys
 
 
 class CiArgs:
-    manifest: str
+    manifest: io.TextIOWrapper
     snapshot: bool
 
-    def __init__(self):
+    def __init__(self) -> None:
         parser = argparse.ArgumentParser(description="Sanity test the OpenSearch Bundle")
         parser.add_argument("manifest", type=argparse.FileType("r"), help="Manifest file.")
         parser.add_argument(
@@ -47,7 +48,7 @@ def __init__(self):
         self.logging_level = args.logging_level
         self.script_path = sys.argv[0].replace("/src/run_ci.py", "/ci.sh")
 
-    def component_command(self, name):
+    def component_command(self, name: str) -> str:
         return " ".join(
             filter(
                 None,

src/ci_workflow/ci_check.py

@@ -5,16 +5,21 @@
 # compatible open source license.
 
 from abc import ABC, abstractmethod
+from typing import Any
+
+from ci_workflow.ci_target import CiTarget
+from git.git_repository import GitRepository
+from manifests.input_manifest import Component
 
 
 class CiCheck(ABC):
-    def __init__(self, component, target, args=None):
+    def __init__(self, component: Any, target: CiTarget, args: Any = None) -> None:
         self.component = component
         self.target = target
         self.args = args
 
     @abstractmethod
-    def check(self):
+    def check(self) -> None:
         pass
 
 
@@ -23,6 +28,6 @@ class CiCheckDist(CiCheck):
 
 
 class CiCheckSource(CiCheck):
-    def __init__(self, component, git_repo, target, args=None):
+    def __init__(self, component: Component, git_repo: GitRepository, target: CiTarget, args: Any = None) -> None:
         super().__init__(component, target, args)
         self.git_repo = git_repo

src/ci_workflow/ci_check_gradle_dependencies.py

@@ -6,18 +6,22 @@
 
 import logging
 import re
+from typing import Any
 
 from ci_workflow.ci_check import CiCheckSource
+from ci_workflow.ci_target import CiTarget
+from git.git_repository import GitRepository
+from manifests.input_manifest import InputComponent
 from system.properties_file import PropertiesFile
 
 
 class CiCheckGradleDependencies(CiCheckSource):
-    def __init__(self, component, git_repo, target, args):
+    def __init__(self, component: InputComponent, git_repo: GitRepository, target: CiTarget, args: Any) -> None:
         super().__init__(component, git_repo, target, args)
         self.gradle_project = args if args else None
         self.dependencies = self.__get_dependencies()
 
-    def __get_dependencies(self):
+    def __get_dependencies(self) -> PropertiesFile:
         cmd = " ".join(
             [
                 f"./gradlew {self.gradle_project or ''}:dependencies",

src/ci_workflow/ci_check_gradle_dependencies_opensearch.py

@@ -10,6 +10,6 @@
 
 
 class CiCheckGradleDependenciesOpenSearchVersion(CiCheckGradleDependencies):
-    def check(self):
+    def check(self) -> None:
         self.dependencies.check_value("org.opensearch:opensearch", self.target.opensearch_version)
         logging.info(f"Checked {self.component.name} OpenSearch dependency ({self.target.opensearch_version}).")

src/ci_workflow/ci_check_gradle_properties.py

@@ -4,16 +4,21 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
+from typing import Any
+
 from ci_workflow.ci_check import CiCheckSource
+from ci_workflow.ci_target import CiTarget
+from git.git_repository import GitRepository
+from manifests.input_manifest import Component
 from system.properties_file import PropertiesFile
 
 
 class CiCheckGradleProperties(CiCheckSource):
-    def __init__(self, component, git_repo, target, args=None):
+    def __init__(self, component: Component, git_repo: GitRepository, target: CiTarget, args: Any = None) -> None:
         super().__init__(component, git_repo, target, args)
         self.properties = self.__get_properties()
 
-    def __get_properties(self):
+    def __get_properties(self) -> PropertiesFile:
         cmd = " ".join(
             [
                 "./gradlew properties",

src/ci_workflow/ci_check_gradle_properties_version.py

@@ -11,12 +11,12 @@
 
 class CiCheckGradlePropertiesVersion(CiCheckGradleProperties):
     @property
-    def checked_version(self):
+    def checked_version(self) -> str:
         if self.component.name == "OpenSearch":
             return self.target.opensearch_version
         else:
             return self.target.component_version
 
-    def check(self):
+    def check(self) -> None:
         self.properties.check_value("version", self.checked_version)
         logging.info(f"Checked {self.component.name} ({self.checked_version}).")

src/ci_workflow/ci_check_gradle_publish_to_maven_local.py

@@ -8,7 +8,7 @@
 
 
 class CiCheckGradlePublishToMavenLocal(CiCheckSource):
-    def check(self):
+    def check(self) -> None:
         cmd = " ".join(
             [
                 "./gradlew publishToMavenLocal",

src/ci_workflow/ci_check_list.py

@@ -5,17 +5,20 @@
 # compatible open source license.
 
 from abc import ABC, abstractmethod
+from typing import Any
+
+from ci_workflow.ci_target import CiTarget
 
 
 class CiCheckList(ABC):
-    def __init__(self, component, target):
+    def __init__(self, component: Any, target: CiTarget) -> None:
         self.component = component
         self.target = target
 
     @abstractmethod
-    def checkout(self, work_dir):
+    def checkout(self, work_dir: str) -> None:
         pass
 
     @abstractmethod
-    def check(self):
+    def check(self) -> None:
         pass

src/ci_workflow/ci_check_list_dist.py

@@ -4,22 +4,24 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
+from typing import Any
+
 from ci_workflow.ci_check_list import CiCheckList
 from ci_workflow.ci_check_manifest_component import CiCheckManifestComponent
 
 
 class CiCheckListDist(CiCheckList):
-    def checkout(self, work_dir):
+    def checkout(self, work_dir: str) -> None:
         pass
 
     CHECKS = {"manifest:component": CiCheckManifestComponent}
 
     class InvalidCheckError(Exception):
-        def __init__(self, check):
+        def __init__(self, check: Any) -> None:
             self.check = check
             super().__init__(f"Invalid check: {check.name}, must be one of {CiCheckListDist.CHECKS.keys()}.")
 
-    def check(self):
+    def check(self) -> None:
         for check in self.component.checks:
             klass = CiCheckListDist.CHECKS.get(check.name, None)
             if klass is None:

src/ci_workflow/ci_check_list_source.py

@@ -5,6 +5,7 @@
 # compatible open source license.
 
 import os
+from typing import Any
 
 from ci_workflow.ci_check_gradle_dependencies_opensearch import CiCheckGradleDependenciesOpenSearchVersion
 from ci_workflow.ci_check_gradle_properties_version import CiCheckGradlePropertiesVersion
@@ -15,7 +16,7 @@
 
 
 class CiCheckListSource(CiCheckList):
-    def checkout(self, work_dir):
+    def checkout(self, work_dir: str) -> None:
         self.git_repo = GitRepository(
             self.component.repository, self.component.ref, os.path.join(work_dir, self.component.name), self.component.working_directory
         )
@@ -30,11 +31,11 @@ def checkout(self, work_dir):
     }
 
     class InvalidCheckError(Exception):
-        def __init__(self, check):
+        def __init__(self, check: Any):
             self.check = check
             super().__init__(f"Invalid check: {check.name}, must be one of {CiCheckListSource.CHECKS.keys()}.")
 
-    def check(self):
+    def check(self) -> None:
         for check in self.component.checks:
             klass = CiCheckListSource.CHECKS.get(check.name, None)
             if klass is None:

src/ci_workflow/ci_check_list_source_ref.py

@@ -15,10 +15,10 @@ class MissingRefError(Exception):
         def __init__(self, url: str, ref: str) -> None:
             super().__init__(f"Missing {url}@{ref}.")
 
-    def checkout(self, work_dir):
+    def checkout(self, work_dir: str) -> None:
         return super().checkout(work_dir)
 
-    def check(self):
+    def check(self) -> None:
         logging.info(f"Checking {self.component.repository} at {self.component.ref}.")
         results = subprocess.check_output(f"git ls-remote {self.component.repository} {self.component.ref}", shell=True).decode().strip().split("\t")
         if len(results) != 2:

src/ci_workflow/ci_check_lists.py

@@ -5,16 +5,18 @@
 # compatible open source license.
 
 from abc import ABC
+from typing import Any
 
 from ci_workflow.ci_check_list_dist import CiCheckListDist
 from ci_workflow.ci_check_list_source import CiCheckListSource
 from ci_workflow.ci_check_list_source_ref import CiCheckListSourceRef
+from ci_workflow.ci_target import CiTarget
 from manifests.input_manifest import InputComponentFromDist, InputComponentFromSource
 
 
 class CiCheckLists(ABC):
     @classmethod
-    def from_component(self, component, target):
+    def from_component(self, component: Any, target: CiTarget) -> Any:
         if type(component) is InputComponentFromDist:
             return CiCheckListDist(component, target)
         elif type(component) is InputComponentFromSource:

src/ci_workflow/ci_check_manifest_component.py

@@ -14,10 +14,10 @@
 
 class CiCheckManifestComponent(CiCheckDist):
     class MissingComponentError(Exception):
-        def __init__(self, component, url):
+        def __init__(self, component: str, url: str) -> None:
             super().__init__(f"Missing {component} in {url}.")
 
-    def check(self):
+    def check(self) -> None:
         for architecture in BuildArgs.SUPPORTED_ARCHITECTURES:
             # Since we only have 'linux' builds now we hard code it to 'linux'
             # Once we have all platform builds on S3 we can then add a second loop for 'BuildArgs.SUPPORTED_PLATFORMS'

src/ci_workflow/ci_check_npm_package_version.py

@@ -11,12 +11,12 @@
 
 class CiCheckNpmPackageVersion(CiCheckPackage):
     @property
-    def checked_version(self):
+    def checked_version(self) -> str:
         if self.component.name == "OpenSearch-Dashboards":
-            return self.target.opensearch_version.replace('-SNAPSHOT', '')
+            return self.target.opensearch_version.replace("-SNAPSHOT", "")
         else:
-            return self.target.component_version.replace('-SNAPSHOT', '')
+            return self.target.component_version.replace("-SNAPSHOT", "")
 
-    def check(self):
+    def check(self) -> None:
         self.properties.check_value("version", self.checked_version)
         logging.info(f"Checked {self.component.name} ({self.checked_version}).")

src/ci_workflow/ci_check_package.py

@@ -6,27 +6,31 @@
 
 import json
 import os
+from typing import Any
 
 from ci_workflow.ci_check import CiCheckSource
+from ci_workflow.ci_target import CiTarget
+from git.git_repository import GitRepository
+from manifests.input_manifest import Component
 from system.properties_file import PropertiesFile
 
 
 class CiCheckPackage(CiCheckSource):
-    def __init__(self, component, git_repo, target, args=None):
+    def __init__(self, component: Component, git_repo: GitRepository, target: CiTarget, args: Any = None) -> None:
         super().__init__(component, git_repo, target, args)
         self.properties = self.__get_properties()
 
     @property
-    def package_json_path(self):
+    def package_json_path(self) -> str:
         return os.path.join(self.git_repo.working_directory, "package.json")
 
-    def __get_properties(self):
+    def __get_properties(self) -> PropertiesFile:
         with open(self.package_json_path, "r") as f:
             return PropertiesFile(CiCheckPackage.__flattenDict(json.load(f)))
 
     # https://gist.github.com/higarmi/6708779
     @classmethod
-    def __flattenDict(cls, d, result=None, index=None, parent_key=None):
+    def __flattenDict(cls, d: dict, result: dict = None, index: int = None, parent_key: str = None) -> dict:
         if result is None:
             result = {}
         if isinstance(d, (list, tuple)):

src/ci_workflow/ci_input_manifest.py

@@ -5,7 +5,9 @@
 # compatible open source license.
 
 import logging
+from io import TextIOWrapper
 
+from ci_workflow.ci_args import CiArgs
 from ci_workflow.ci_check_lists import CiCheckLists
 from ci_workflow.ci_manifest import CiManifest
 from ci_workflow.ci_target import CiTarget
@@ -14,10 +16,10 @@
 
 
 class CiInputManifest(CiManifest):
-    def __init__(self, file, args):
+    def __init__(self, file: TextIOWrapper, args: CiArgs) -> None:
         super().__init__(InputManifest.from_file(file), args)
 
-    def __check__(self):
+    def __check__(self) -> None:
 
         target = CiTarget(version=self.manifest.build.version, name=self.manifest.build.filename, snapshot=self.args.snapshot)
 

src/ci_workflow/ci_manifest.py

@@ -4,18 +4,25 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
-import abc
 import logging
+from abc import ABC, abstractmethod
+from typing import Any
 
+from ci_workflow.ci_args import CiArgs
 
-class CiManifest(abc.ABC):
-    def __init__(self, manifest, args):
+
+class CiManifest(ABC):
+    def __init__(self, manifest: Any, args: CiArgs) -> None:
         self.manifest = manifest
         self.args = args
 
-    def check(self):
+    def check(self) -> None:
         try:
             self.__check__()
         except:
             logging.error("CI Manifest check failed")
             raise
+
+    @abstractmethod
+    def __check__(self) -> None:
+        pass