Format Code and fix typo

Author: yma955

charon/config.py

@@ -13,6 +13,7 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+
 import logging
 import os
 from typing import Dict, List, Optional
@@ -45,6 +46,7 @@ def __init__(self, data: Dict):
         self.__quay_radas_registry: str = data.get("quay_radas_registry", None)
         self.__quay_radas_username: str = data.get("quay_radas_username", None)
         self.__quay_radas_password: str = data.get("quay_radas_password", None)
+        self.__radas_sign_enabled: bool = data.get("radas_sign_enabled", False)
         self.__radas_sign_timeout_count: int = data.get("radas_sign_timeout_count", 10)
         self.__radas_sign_wait_interval_sec: int = data.get("radas_sign_wait_interval_sec", 60)
 
@@ -93,24 +95,26 @@ def get_quay_radas_username(self) -> str:
     def get_quay_radas_password(self) -> str:
         return self.__quay_radas_password
 
+    def is_radas_sign_enabled(self) -> bool:
+        return self.__radas_sign_enabled
+
     def get_radas_sign_timeout_count(self) -> int:
         return self.__radas_sign_timeout_count
 
     def get_radas_sign_wait_interval_sec(self) -> int:
         return self.__radas_sign_wait_interval_sec
 
+
 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)
-    data = read_yaml_from_file_path(config_file_path, 'schemas/charon.json')
+    data = read_yaml_from_file_path(config_file_path, "schemas/charon.json")
     return CharonConfig(data)
 
 
 def get_template(template_file: str) -> str:
-    template = os.path.join(
-        os.getenv("HOME", ''), ".charon/template", template_file
-    )
+    template = os.path.join(os.getenv("HOME", ""), ".charon/template", template_file)
     if os.path.isfile(template):
         with open(template, encoding="utf-8") as file_:
             return file_.read()

charon/constants.py

@@ -177,4 +177,4 @@
 DEFAULT_REGISTRY = "localhost"
 DEFAULT_SIGN_RESULT_LOC = "/tmp/sign"
 DEFAULT_RADAS_SIGN_TIMEOUT_COUNT = 10
-DEFAULT_RADAS_SIGN_WAIT_INTERVAL_SEC = 60
+DEFAULT_RADAS_SIGN_WAIT_INTERVAL_SEC = 60

charon/pkgs/maven.py

@@ -409,12 +409,13 @@ def handle_maven_uploading(
                 if cf_enable:
                     cf_invalidate_paths.extend(archetype_files)
 
-        # 10. Generate signature file if radas sign is enabled, or do detached sign if contain_signature is set to True
+        # 10. Generate signature file if radas sign is enabled,
+        # or do detached sign if contain_signature is set to True
         conf = get_config(config)
         if not conf:
             sys.exit(1)
 
-        if conf.get_radas_sign_enabled():
+        if conf.is_radas_sign_enabled():
             logger.info("Start generating radas signature files for s3 bucket %s\n", bucket_name)
             (_failed_metas, _generated_signs) = radas_signature.generate_radas_sign(top_level)
             if not _generated_signs:

charon/pkgs/oras_client.py

@@ -13,13 +13,15 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+
 import oras.client
 import logging
 from charon.config import get_config
 from typing import List
 
 logger = logging.getLogger(__name__)
 
+
 class OrasClient:
     """
     Wrapper for oras‑py’s OrasClient, deciding whether to login based on config.
@@ -49,16 +51,19 @@ def pull(self, result_reference_url: str, sign_result_loc: str) -> List[str]:
         """
         Call oras‑py’s pull method to pull the remote file to local.
         Args:
-            result_reference_url (str): Reference of the remote file (e.g. “quay.io/repository/signing/radas@hash”).
-            sign_result_loc (str): Local save path (e.g. “/tmp/sign”).
+            result_reference_url (str):
+                Reference of the remote file (e.g. “quay.io/repository/signing/radas@hash”).
+            sign_result_loc (str):
+                Local save path (e.g. “/tmp/sign”).
         """
         files = []
         try:
             self.login_if_needed()
-            # the filename should be possibly named by the digest hash value based on the oras source code
             files = self.client.pull(target=result_reference_url, outdir=sign_result_loc)
             logger.info("Pull file from %s to %s", result_reference_url, sign_result_loc)
         except Exception as e:
-            logger.error("Failed to pull file from %s to %s: %s", result_reference_url, sign_result_loc, e)
+            logger.error(
+                "Failed to pull file from %s to %s: %s", result_reference_url, sign_result_loc, e
+            )
         finally:
-            return files
+            return files

charon/pkgs/radas_signature_handler.py

@@ -13,13 +13,16 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+
 import proton
 import proton.handlers
 import threading
 import logging
 import json
 import os
 import asyncio
+import sys
+import time
 from typing import List, Any, Tuple, Callable, Dict
 from charon.config import get_config
 from charon.constants import DEFAULT_SIGN_RESULT_LOC
@@ -29,10 +32,12 @@
 
 logger = logging.getLogger(__name__)
 
+
 class SignHandler:
     """
     Handle the sign result status management
     """
+
     _is_processing: bool = True
     _downloaded_files: List[str] = []
 
@@ -52,6 +57,7 @@ def set_processing(cls, value: bool) -> None:
     def set_downloaded_files(cls, files: List[str]) -> None:
         cls._downloaded_files = files
 
+
 class UmbListener(proton.handlers.MessagingHandler):
     """
     UmbListener class (AMQP version), register this when setup UmbClient
@@ -74,10 +80,7 @@ def on_message(self, event: proton.Event) -> None:
         On message callback
         """
         # handle response from radas in a thread
-        thread = threading.Thread(
-            target=self._process_message,
-            args=[event.message.body]
-        )
+        thread = threading.Thread(target=self._process_message, args=[event.message.body])
         thread.start()
 
     def on_error(self, event: proton.Event) -> None:
@@ -103,8 +106,8 @@ def _process_message(msg: Any) -> None:
             result_reference_url = msg_dict.get("result_reference")
 
             if not result_reference_url:
-                 logger.warning("Not found result_reference in message，ignore.")
-                 return
+                logger.warning("Not found result_reference in message，ignore.")
+                return
 
             conf = get_config()
             if not conf:
@@ -117,20 +120,24 @@ def _process_message(msg: Any) -> None:
 
             oras_client = OrasClient()
             files = oras_client.pull(
-                result_reference_url=result_reference_url,
-                sign_result_loc=sign_result_loc
+                result_reference_url=result_reference_url, sign_result_loc=sign_result_loc
             )
             SignHandler.set_downloaded_files(files)
         finally:
             SignHandler.set_processing(False)
 
+
 def generate_radas_sign(top_level: str) -> Tuple[List[str], List[str]]:
     """
     Generate .asc files based on RADAS sign result json file
     """
     conf = get_config()
-    timeout_count = conf.get_radas_sign_timeout_count() if conf else DEFAULT_RADAS_SIGN_TIMEOUT_COUNT
-    wait_interval_sec = conf.get_radas_sign_wait_interval_sec() if conf else DEFAULT_RADAS_SIGN_WAIT_INTERVAL_SEC
+    timeout_count = (
+        conf.get_radas_sign_timeout_count() if conf else DEFAULT_RADAS_SIGN_TIMEOUT_COUNT
+    )
+    wait_interval_sec = (
+        conf.get_radas_sign_wait_interval_sec() if conf else DEFAULT_RADAS_SIGN_WAIT_INTERVAL_SEC
+    )
     wait_count = 0
     while SignHandler.is_processing():
         wait_count += 1
@@ -146,23 +153,25 @@ def generate_radas_sign(top_level: str) -> Tuple[List[str], List[str]]:
     # should only have the single sign result json file from the radas registry
     json_file_path = files[0]
     try:
-        with open(json_file_path, 'r') as f:
+        with open(json_file_path, "r") as f:
             data = json.load(f)
     except Exception as e:
         logger.error(f"Failed to read or parse the JSON file: {e}")
         raise
 
     async def generate_single_sign_file(
-        file_path: str, signature: str, failed_paths: List[str], generated_signs: List[str],
-        sem: asyncio.BoundedSemaphore
+        file_path: str,
+        signature: str,
+        failed_paths: List[str],
+        generated_signs: List[str],
+        sem: asyncio.BoundedSemaphore,
     ):
         async with sem:
             if not file_path or not signature:
-                logger.error(f"Invalid JSON entry")
+                logger.error("Invalid JSON entry")
                 return
             # remove the root path maven-repository
             filename = file_path.split("/", 1)[1]
-            signature = item.get("signature")
 
             artifact_path = os.path.join(top_level, filename)
             asc_filename = f"{filename}.asc"
@@ -173,7 +182,7 @@ async def generate_single_sign_file(
                 return
 
             try:
-                with open(signature_path, 'w') as asc_file:
+                with open(signature_path, "w") as asc_file:
                     asc_file.write(signature)
                 generated_signs.append(signature_path)
                 logger.info(f"Generated .asc file: {signature_path}")
@@ -182,14 +191,11 @@ async def generate_single_sign_file(
                 logger.error(f"Failed to write .asc file for {artifact_path}: {e}")
 
     result = data.get("result", [])
-    return __do_path_cut_and(
-            path_handler=generate_single_sign_file,
-            data=result
-        )
+    return __do_path_cut_and(path_handler=generate_single_sign_file, data=result)
+
 
 def __do_path_cut_and(
-    path_handler: Callable,
-    data: List[Dict[str, str]]
+    path_handler: Callable, data: List[Dict[str, str]]
 ) -> Tuple[List[str], List[str]]:
 
     failed_paths: List[str] = []
@@ -207,4 +213,4 @@ def __do_path_cut_and(
 
     loop = asyncio.get_event_loop()
     loop.run_until_complete(asyncio.gather(*tasks))
-    return (failed_paths, generated_signs)
+    return (failed_paths, generated_signs)