From 46b61be603409dbb02fd952650fb83a1870ada7b Mon Sep 17 00:00:00 2001 From: Daniil Anfimov Date: Wed, 7 Jun 2023 16:10:09 +0200 Subject: [PATCH] Black/isort syntax --- cas_wrapper.py | 96 ++++++++++++++++++++++++++------------------------ 1 file changed, 49 insertions(+), 47 deletions(-) diff --git a/cas_wrapper.py b/cas_wrapper.py index 1f66e6f..ed2855e 100644 --- a/cas_wrapper.py +++ b/cas_wrapper.py @@ -1,10 +1,10 @@ import json import logging import typing -from pathlib import Path from functools import wraps +from pathlib import Path -from plumbum import local, ProcessExecutionError +from plumbum import ProcessExecutionError, local def with_env_context(func): @@ -20,6 +20,7 @@ def with_env_context(func): VCN_LC_PORT=self._vcn_lc_port, ): return func(self, *args, **kwargs) + return wrapper @@ -32,17 +33,17 @@ class CasWrapper: def _is_binary_present(self): if not self._full_binary_path.exists(): raise FileNotFoundError( - f'Binary VCN is not found in {self._binary_path} on the machine', + f"Binary VCN is not found in {self._binary_path} on the machine", ) def __init__( - self, - vcn_lc_api_key: str, - vcn_lc_host: str = "eval-honeywell.codenotary.com", - vcn_lc_port: int = 443, - logger: logging.Logger = None, - binary_name: str = "vcn", - binary_path: str = "/usr/local/bin/", + self, + vcn_lc_api_key: str, + vcn_lc_host: str = "eval-honeywell.codenotary.com", + vcn_lc_port: int = 443, + logger: logging.Logger = None, + binary_name: str = "vcn", + binary_path: str = "/usr/local/bin/", ): self._vcn_lc_api_key = vcn_lc_api_key self._vcn_lc_host = vcn_lc_host @@ -50,7 +51,7 @@ class CasWrapper: self._binary_name = binary_name self._binary_path = binary_path self._full_binary_path = Path(self._binary_path, self._binary_name) - self._vcn = local[self._full_binary_path] + self._vcn = local[str(self._full_binary_path)] self._logger = logger if self._logger is None: self._logger = logging.getLogger() @@ -58,19 +59,19 @@ class CasWrapper: def get_version(self): self._is_binary_present() - command = self._vcn['--version'] - version = command().split()[-1].split('v')[1] + command = self._vcn["--version"] + version = command().split()[-1].split("v")[1] return version @with_env_context def ensure_login(self): - self._vcn['login']() + self._vcn["login"]() @with_env_context def notarize( - self, - local_path: str, - metadata: typing.Dict = None, + self, + local_path: str, + metadata: typing.Dict = None, ) -> str: """ Wrapper around `vcn notarize` @@ -80,25 +81,25 @@ class CasWrapper: :rtype: str """ command = self._vcn[ - 'notarize', + "notarize", local_path, - '-o', - 'json', + "-o", + "json", ] if metadata is not None: for key, value in metadata.items(): command = command[ - '-a', - f'{key}={value}', + "-a", + f"{key}={value}", ] result_of_execution = command() result_of_execution, *_ = json.loads(result_of_execution) - return result_of_execution['hash'] + return result_of_execution["hash"] def notarize_no_exc( - self, - local_path: str, - metadata: typing.Dict = None, + self, + local_path: str, + metadata: typing.Dict = None, ) -> typing.Tuple[bool, str]: """ Wrapper for avoiding raising exceptions during notarization. @@ -114,18 +115,17 @@ class CasWrapper: vcn_hash = self.notarize(local_path, metadata=metadata) success = True except Exception: - self._logger.exception('Cannot notarize artifact: %s', - local_path) - vcn_hash = '' + self._logger.exception("Cannot notarize artifact: %s", local_path) + vcn_hash = "" return success, vcn_hash @with_env_context def authenticate( - self, - local_path: str, - return_json: bool = False, - use_hash: bool = False, - signer_id: str = "", + self, + local_path: str, + return_json: bool = False, + use_hash: bool = False, + signer_id: str = "", ) -> typing.Union[bool, dict]: """ Wrapper around `vcn authenticate` @@ -138,12 +138,12 @@ class CasWrapper: or dict with result if return_json param is True :rtype: bool or dict """ - command_args = ['authenticate', local_path] + command_args = ["authenticate", local_path] if use_hash: - command_args = ['authenticate', '--hash', local_path] + command_args = ["authenticate", "--hash", local_path] if signer_id: - command_args.extend(('--signerID', signer_id)) - command_args.extend(('-o', 'json')) + command_args.extend(("--signerID", signer_id)) + command_args.extend(("-o", "json")) command = self._vcn[command_args] try: result_of_execution = command() @@ -153,7 +153,7 @@ class CasWrapper: json_result = json.loads(result_of_execution) if return_json: return json_result - return not bool(json_result['status']) + return not bool(json_result["status"]) def authenticate_source( self, @@ -173,12 +173,12 @@ class CasWrapper: return_json=True, signer_id=signer_id, ) - is_authenticated = result_json['verified'] - commit_vcn_hash = result_json['hash'] + is_authenticated = result_json["verified"] + commit_vcn_hash = result_json["hash"] # we can fall with ProcessExecutionError, # because source can be not notarized except ProcessExecutionError: - self._logger.exception('Cannot authenticate: %s', local_path) + self._logger.exception("Cannot authenticate: %s", local_path) return is_authenticated, commit_vcn_hash def authenticate_artifact( @@ -198,12 +198,12 @@ class CasWrapper: local_path, use_hash=use_hash, return_json=True, - signer_id=signer_id - )['verified'] + signer_id=signer_id, + )["verified"] # we can fall with ProcessExecutionError, # because artifact can be not notarized except ProcessExecutionError: - self._logger.exception('Cannot authenticate: %s', local_path) + self._logger.exception("Cannot authenticate: %s", local_path) return is_authenticated def notarize_artifacts( @@ -229,8 +229,10 @@ class CasWrapper: try: vcn_artifact_hash = self.notarize(artifact_path, metadata) except Exception: - self._logger.exception('Cannot notarize artifact: %s', - artifact_path) + self._logger.exception( + "Cannot notarize artifact: %s", + artifact_path, + ) all_artifacts_is_notarized = False continue notarized_artifacts[artifact_path] = vcn_artifact_hash