Enable mypy types check and fix reported issues

Signed-off-by: Gang Li <[email protected]>

.github/workflows/linters.yaml

@@ -29,17 +29,6 @@ jobs:
       - name: Run flake8 on python${{ matrix.python-version }}
         run: python -m tox -e flake8
 
-  # markdownlint:
-  #  name: Markdownlint
-  #  runs-on: ubuntu-latest
-
-  #  steps:
-  #    - name: Check out repo
-  #      uses: actions/checkout@v2
-
-  #    - name: Run markdownlint
-  #      uses: containerbuildsystem/actions/markdownlint@master
-
   pylint:
     name: Pylint analyzer for Python ${{ matrix.python-version }}
     runs-on: ubuntu-latest
@@ -71,25 +60,25 @@ jobs:
       - name: Run ShellCheck
         uses: containerbuildsystem/actions/shellcheck@master
 
-#  mypy:
-#    name: mypy type checker for Python ${{ matrix.python-version }}
-#    runs-on: ubuntu-latest
-#
-#    strategy:
-#      matrix:
-#        python-version: [ "3.8" ]
-#
-#    steps:
-#      - uses: actions/checkout@v3
-#      - uses: actions/setup-python@v4
-#        with:
-#          python-version: ${{ matrix.python-version }}
-#      - name: Install dependencies
-#        run: |
-#          python -m pip install --upgrade pip setuptools tox
-#
-#      - name: Run mypy on python${{ matrix.python-version }}
-#        run: python -m tox -e mypy
+  mypy:
+    name: mypy type checker for Python ${{ matrix.python-version }}
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        python-version: [ "3.9" ]
+
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip setuptools tox
+
+      - name: Run mypy on python${{ matrix.python-version }}
+        run: python -m tox -e mypy
 
   # bandit:
   #   name: Bandit analyzer for Python ${{ matrix.python-version }}

charon/cache.py

@@ -1,6 +1,6 @@
 from boto3 import session
 from botocore.exceptions import ClientError
-from typing import Dict, List
+from typing import Dict, List, Optional
 import os
 import logging
 import uuid
@@ -57,7 +57,7 @@ def __init_aws_client(
             endpoint_url=endpoint_url
         )
 
-    def __get_endpoint(self, extra_conf) -> str:
+    def __get_endpoint(self, extra_conf) -> Optional[str]:
         endpoint_url = os.getenv(ENDPOINT_ENV)
         if not endpoint_url or not endpoint_url.strip():
             if isinstance(extra_conf, Dict):
@@ -97,14 +97,14 @@ def invalidate_paths(
                     " will take more than %d seconds",
                     len(real_paths), total_time_approx)
         results = []
-        current_invalidation = {}
+        current_invalidation: Dict[str, str] = {}
         processed_count = 0
         for batch_paths in real_paths:
             while (current_invalidation and
                     INVALIDATION_STATUS_INPROGRESS == current_invalidation.get('Status', '')):
                 time.sleep(INPRO_W_SECS)
                 try:
-                    result = self.check_invalidation(distr_id, current_invalidation.get('Id'))
+                    result = self.check_invalidation(distr_id, current_invalidation.get('Id', ''))
                     if result:
                         current_invalidation = {
                             'Id': result.get('Id', None),
@@ -159,7 +159,7 @@ def invalidate_paths(
             results.append(current_invalidation)
         return results
 
-    def check_invalidation(self, distr_id: str, invalidation_id: str) -> dict:
+    def check_invalidation(self, distr_id: str, invalidation_id: str) -> Optional[dict]:
         try:
             response = self.__client.get_invalidation(
                 DistributionId=distr_id,
@@ -177,8 +177,9 @@ def check_invalidation(self, distr_id: str, invalidation_id: str) -> dict:
                 "[CloudFront] Error occurred while check invalidation of id %s, "
                 "error: %s", invalidation_id, err
             )
+        return None
 
-    def get_dist_id_by_domain(self, domain: str) -> str:
+    def get_dist_id_by_domain(self, domain: str) -> Optional[str]:
         """Get distribution id by a domain name. The id can be used to send invalidating
         request through #invalidate_paths function
            * Domain are Ronda domains, like "maven.repository.redhat.com"
@@ -200,5 +201,5 @@ def get_dist_id_by_domain(self, domain: str) -> str:
             )
         return None
 
-    def get_domain_by_bucket(self, bucket: str) -> str:
+    def get_domain_by_bucket(self, bucket: str) -> Optional[str]:
         return DEFAULT_BUCKET_TO_DOMAIN.get(bucket, None)

charon/cmd/cmd_cache.py

@@ -15,11 +15,12 @@
 """
 
 from charon.config import get_config
-from charon.cmd.internal import _decide_mode, _get_buckets
+from charon.cmd.internal import _decide_mode, _get_targets
 from charon.cache import CFClient
 from charon.pkgs.pkg_utils import invalidate_cf_paths
+from charon.types import TARGET_TYPE
 from click import command, option, argument, group
-from typing import List, Tuple
+from typing import List, Tuple, Optional
 
 import traceback
 import logging
@@ -87,7 +88,7 @@ def invalidate(
     target: str,
     paths: List[str],
     path_file: str,
-    config: str = None,
+    config: str = "",
     quiet: bool = False,
     debug: bool = False
 ):
@@ -181,7 +182,7 @@ def invalidate(
 def check(
     invalidation_id: str,
     target: str,
-    config: str = None,
+    config: str = "",
     quiet: bool = False,
     debug: bool = False
 ):
@@ -200,7 +201,7 @@ def check(
         for b in buckets:
             cf_client = CFClient(aws_profile=aws_profile)
             bucket_name = b[1]
-            domain = b[4]
+            domain: Optional[str] = b[4]
             if not domain:
                 domain = cf_client.get_domain_by_bucket(bucket_name)
             if domain:
@@ -221,7 +222,7 @@ def check(
         sys.exit(2)
 
 
-def _init_cmd(target: str, config: str) -> Tuple[List[Tuple[str, str, str, str, str]], str]:
+def _init_cmd(target: str, config: str) -> Tuple[List[TARGET_TYPE], str]:
     conf = get_config(config)
     if not conf:
         sys.exit(1)
@@ -231,7 +232,7 @@ def _init_cmd(target: str, config: str) -> Tuple[List[Tuple[str, str, str, str,
         logger.error("No AWS profile specified!")
         sys.exit(1)
 
-    return (_get_buckets([target], conf), aws_profile)
+    return (_get_targets([target], conf), aws_profile)
 
 
 @group()

charon/cmd/cmd_checksum.py

@@ -104,7 +104,7 @@
 def validate(
     path: str,
     target: str,
-    includes: List[str],
+    includes: str,
     report_file_path: str,
     skips: List[str],
     recursive: bool = False,
@@ -259,12 +259,12 @@ def _init_cmd(target: str) -> Tuple[str, str]:
     conf = get_config()
     if not conf:
         sys.exit(1)
-    aws_bucket = ""
     t = conf.get_target(target)
     if not t:
         sys.exit(1)
+    aws_bucket = ''
     for b in t:
-        aws_bucket = b.get('bucket')
+        aws_bucket = b.get('bucket', '')
         prefix = b.get('prefix', '')
     return (aws_bucket, prefix)
 

charon/cmd/cmd_delete.py

@@ -21,7 +21,7 @@
 from charon.pkgs.npm import handle_npm_del
 from charon.cmd.internal import (
     _decide_mode, _validate_prod_key,
-    _get_local_repo, _get_buckets,
+    _get_local_repo, _get_targets,
     _get_ignore_patterns, _safe_delete
 )
 from click import command, option, argument
@@ -158,19 +158,19 @@ def delete(
         npm_archive_type = detect_npm_archive(archive_path)
         product_key = f"{product}-{version}"
         manifest_bucket_name = conf.get_manifest_bucket()
-        buckets = _get_buckets(targets, conf)
-        if not buckets:
+        targets_ = _get_targets(targets, conf)
+        if not targets_:
             logger.error(
                 "The targets %s can not be found! Please check"
                 " your charon configuration to confirm the targets"
-                " are set correctly.", targets
+                " are set correctly.", targets_
             )
         if npm_archive_type != NpmArchiveType.NOT_NPM:
             logger.info("This is a npm archive")
             tmp_dir, succeeded = handle_npm_del(
                 archive_path,
                 product_key,
-                buckets=buckets,
+                targets=targets_,
                 aws_profile=aws_profile,
                 dir_=work_dir,
                 cf_enable=conf.is_aws_cf_enable(),
@@ -191,7 +191,7 @@ def delete(
                 product_key,
                 ignore_patterns_list,
                 root=root_path,
-                buckets=buckets,
+                targets=targets_,
                 aws_profile=aws_profile,
                 dir_=work_dir,
                 cf_enable=conf.is_aws_cf_enable(),
@@ -204,5 +204,5 @@ def delete(
         print(traceback.format_exc())
         sys.exit(2)  # distinguish between exception and bad config or bad state
     finally:
-        if not debug:
+        if not debug and tmp_dir:
             _safe_delete(tmp_dir)

charon/cmd/cmd_index.py

@@ -97,7 +97,7 @@ def index(
             sys.exit(1)
 
         for b in tgt:
-            aws_bucket = b.get('bucket')
+            aws_bucket = b.get('bucket', '')
 
             package_type = None
             if "maven" in aws_bucket:
@@ -115,6 +115,7 @@ def index(
                     "The target %s is not supported. Only maven or npm target is supported.",
                     target
                 )
+                continue
 
             if not aws_bucket:
                 logger.error("No bucket specified for target %s!", target)

charon/cmd/cmd_upload.py

@@ -21,7 +21,7 @@
 from charon.pkgs.npm import handle_npm_uploading
 from charon.cmd.internal import (
     _decide_mode, _validate_prod_key,
-    _get_local_repo, _get_buckets,
+    _get_local_repo, _get_targets,
     _get_ignore_patterns, _safe_delete
 )
 from click import command, option, argument
@@ -177,20 +177,20 @@ def upload(
         npm_archive_type = detect_npm_archive(archive_path)
         product_key = f"{product}-{version}"
         manifest_bucket_name = conf.get_manifest_bucket()
-        buckets = _get_buckets(targets, conf)
-        if not buckets:
+        targets_ = _get_targets(targets, conf)
+        if not targets_:
             logger.error(
                 "The targets %s can not be found! Please check"
                 " your charon configuration to confirm the targets"
-                " are set correctly.", targets
+                " are set correctly.", targets_
             )
             sys.exit(1)
         if npm_archive_type != NpmArchiveType.NOT_NPM:
             logger.info("This is a npm archive")
             tmp_dir, succeeded = handle_npm_uploading(
                 archive_path,
                 product_key,
-                buckets=buckets,
+                targets=targets_,
                 aws_profile=aws_profile,
                 dir_=work_dir,
                 gen_sign=contain_signature,
@@ -213,7 +213,7 @@ def upload(
                 product_key,
                 ignore_patterns_list,
                 root=root_path,
-                buckets=buckets,
+                targets=targets_,
                 aws_profile=aws_profile,
                 dir_=work_dir,
                 gen_sign=contain_signature,
@@ -229,5 +229,5 @@ def upload(
         print(traceback.format_exc())
         sys.exit(2)  # distinguish between exception and bad config or bad state
     finally:
-        if not debug:
+        if not debug and tmp_dir:
             _safe_delete(tmp_dir)

charon/cmd/internal.py

@@ -13,10 +13,11 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
-from typing import List, Tuple
+from typing import List, Optional
 
 from charon.config import CharonConfig
 from charon.constants import DEFAULT_REGISTRY
+from charon.types import TARGET_TYPE
 from charon.utils.logs import set_logging
 from charon.utils.archive import download_archive
 from json import loads
@@ -28,13 +29,13 @@
 logger = logging.getLogger(__name__)
 
 
-def _get_buckets(
-    targets: List[str], conf: CharonConfig
-) -> List[Tuple[str, str, str, str, str]]:
-    buckets = []
-    for target in targets:
+def _get_targets(
+    target_names: List[str], conf: CharonConfig
+) -> List[TARGET_TYPE]:
+    buckets: List[TARGET_TYPE] = []
+    for target in target_names:
         for bucket in conf.get_target(target):
-            aws_bucket = bucket.get('bucket')
+            aws_bucket = bucket.get('bucket', '')
             prefix = bucket.get('prefix', '')
             registry = bucket.get('registry', DEFAULT_REGISTRY)
             cf_domain = bucket.get('domain', None)
@@ -51,7 +52,7 @@ def _safe_delete(tmp_dir: str):
             logger.error("Failed to clear work directory. %s", e)
 
 
-def _get_ignore_patterns(conf: CharonConfig) -> List[str]:
+def _get_ignore_patterns(conf: CharonConfig) -> Optional[List[str]]:
     ignore_patterns = os.getenv("CHARON_IGNORE_PATTERNS")
     if ignore_patterns:
         try:

charon/config.py

@@ -55,8 +55,8 @@ def get_aws_profile(self) -> str:
     def get_manifest_bucket(self) -> str:
         return self.__manifest_bucket
 
-    def get_ignore_signature_suffix(self, package_type: str) -> List[str]:
-        xartifact_list: List = self.__ignore_signature_suffix.get(package_type)
+    def get_ignore_signature_suffix(self, package_type: str) -> Optional[List[str]]:
+        xartifact_list = self.__ignore_signature_suffix.get(package_type)
         if not xartifact_list:
             logger.error("package type %s does not have ignore artifact config.", package_type)
         return xartifact_list
@@ -68,10 +68,10 @@ def is_aws_cf_enable(self) -> bool:
         return self.__aws_cf_enable
 
 
-def get_config(cfgPath=None) -> Optional[CharonConfig]:
+def get_config(cfgPath=None) -> CharonConfig:
     config_file_path = cfgPath
     if not config_file_path or not os.path.isfile(config_file_path):
-        config_file_path = os.path.join(os.getenv("HOME"), ".charon", CONFIG_FILE)
+        config_file_path = os.path.join(os.getenv("HOME", ""), ".charon", CONFIG_FILE)
     data = read_yaml_from_file_path(config_file_path, 'schemas/charon.json')
     return CharonConfig(data)