almalinux
/
cas_wrapper
14
0
Fork 2
Code Issues Pull Requests Releases Activity

Black/isort syntax

This commit is contained in:
Daniil Anfimov 2023-06-07 16:10:09 +02:00
parent 786c9977ff
commit 46b61be603
1 changed files with 49 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
cas_wrapper.py
View File

@ -1,10 +1,10 @@
import json import json
import logging import logging
import typing import typing
from pathlib import Path
from functools import wraps from functools import wraps
from pathlib import Path
from plumbum import local, ProcessExecutionError from plumbum import ProcessExecutionError, local
def with_env_context(func): def with_env_context(func):
@ -20,6 +20,7 @@ def with_env_context(func):
VCN_LC_PORT=self._vcn_lc_port, VCN_LC_PORT=self._vcn_lc_port,
): ):
return func(self, *args, **kwargs) return func(self, *args, **kwargs)
return wrapper return wrapper
@ -32,17 +33,17 @@ class CasWrapper:
def _is_binary_present(self): def _is_binary_present(self):
if not self._full_binary_path.exists(): if not self._full_binary_path.exists():
raise FileNotFoundError( 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__( def __init__(
self, self,
vcn_lc_api_key: str, vcn_lc_api_key: str,
vcn_lc_host: str = "eval-honeywell.codenotary.com", vcn_lc_host: str = "eval-honeywell.codenotary.com",
vcn_lc_port: int = 443, vcn_lc_port: int = 443,
logger: logging.Logger = None, logger: logging.Logger = None,
binary_name: str = "vcn", binary_name: str = "vcn",
binary_path: str = "/usr/local/bin/", binary_path: str = "/usr/local/bin/",
): ):
self._vcn_lc_api_key = vcn_lc_api_key self._vcn_lc_api_key = vcn_lc_api_key
self._vcn_lc_host = vcn_lc_host self._vcn_lc_host = vcn_lc_host
@ -50,7 +51,7 @@ class CasWrapper:
self._binary_name = binary_name self._binary_name = binary_name
self._binary_path = binary_path self._binary_path = binary_path
self._full_binary_path = Path(self._binary_path, self._binary_name) 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 self._logger = logger
if self._logger is None: if self._logger is None:
self._logger = logging.getLogger() self._logger = logging.getLogger()
@ -58,19 +59,19 @@ class CasWrapper:
def get_version(self): def get_version(self):
self._is_binary_present() self._is_binary_present()
command = self._vcn['--version'] command = self._vcn["--version"]
version = command().split()[-1].split('v')[1] version = command().split()[-1].split("v")[1]
return version return version
@with_env_context @with_env_context
def ensure_login(self): def ensure_login(self):
self._vcn['login']() self._vcn["login"]()
@with_env_context @with_env_context
def notarize( def notarize(
self, self,
local_path: str, local_path: str,
metadata: typing.Dict = None, metadata: typing.Dict = None,
) -> str: ) -> str:
""" """
Wrapper around `vcn notarize` Wrapper around `vcn notarize`
@ -80,25 +81,25 @@ class CasWrapper:
:rtype: str :rtype: str
""" """
command = self._vcn[ command = self._vcn[
'notarize', "notarize",
local_path, local_path,
'-o', "-o",
'json', "json",
] ]
if metadata is not None: if metadata is not None:
for key, value in metadata.items(): for key, value in metadata.items():
command = command[ command = command[
'-a', "-a",
f'{key}={value}', f"{key}={value}",
] ]
result_of_execution = command() result_of_execution = command()
result_of_execution, *_ = json.loads(result_of_execution) result_of_execution, *_ = json.loads(result_of_execution)
return result_of_execution['hash'] return result_of_execution["hash"]
def notarize_no_exc( def notarize_no_exc(
self, self,
local_path: str, local_path: str,
metadata: typing.Dict = None, metadata: typing.Dict = None,
) -> typing.Tuple[bool, str]: ) -> typing.Tuple[bool, str]:
""" """
Wrapper for avoiding raising exceptions during notarization. Wrapper for avoiding raising exceptions during notarization.
@ -114,18 +115,17 @@ class CasWrapper:
vcn_hash = self.notarize(local_path, metadata=metadata) vcn_hash = self.notarize(local_path, metadata=metadata)
success = True success = True
except Exception: except Exception:
self._logger.exception('Cannot notarize artifact: %s', self._logger.exception("Cannot notarize artifact: %s", local_path)
local_path) vcn_hash = ""
vcn_hash = ''
return success, vcn_hash return success, vcn_hash
@with_env_context @with_env_context
def authenticate( def authenticate(
self, self,
local_path: str, local_path: str,
return_json: bool = False, return_json: bool = False,
use_hash: bool = False, use_hash: bool = False,
signer_id: str = "", signer_id: str = "",
) -> typing.Union[bool, dict]: ) -> typing.Union[bool, dict]:
""" """
Wrapper around `vcn authenticate` Wrapper around `vcn authenticate`
@ -138,12 +138,12 @@ class CasWrapper:
or dict with result if return_json param is True or dict with result if return_json param is True
:rtype: bool or dict :rtype: bool or dict
""" """
command_args = ['authenticate', local_path] command_args = ["authenticate", local_path]
if use_hash: if use_hash:
command_args = ['authenticate', '--hash', local_path] command_args = ["authenticate", "--hash", local_path]
if signer_id: if signer_id:
command_args.extend(('--signerID', signer_id)) command_args.extend(("--signerID", signer_id))
command_args.extend(('-o', 'json')) command_args.extend(("-o", "json"))
command = self._vcn[command_args] command = self._vcn[command_args]
try: try:
result_of_execution = command() result_of_execution = command()
@ -153,7 +153,7 @@ class CasWrapper:
json_result = json.loads(result_of_execution) json_result = json.loads(result_of_execution)
if return_json: if return_json:
return json_result return json_result
return not bool(json_result['status']) return not bool(json_result["status"])
def authenticate_source( def authenticate_source(
self, self,
@ -173,12 +173,12 @@ class CasWrapper:
return_json=True, return_json=True,
signer_id=signer_id, signer_id=signer_id,
) )
is_authenticated = result_json['verified'] is_authenticated = result_json["verified"]
commit_vcn_hash = result_json['hash'] commit_vcn_hash = result_json["hash"]
# we can fall with ProcessExecutionError, # we can fall with ProcessExecutionError,
# because source can be not notarized # because source can be not notarized
except ProcessExecutionError: except ProcessExecutionError:
self._logger.exception('Cannot authenticate: %s', local_path) self._logger.exception("Cannot authenticate: %s", local_path)
return is_authenticated, commit_vcn_hash return is_authenticated, commit_vcn_hash
def authenticate_artifact( def authenticate_artifact(
@ -198,12 +198,12 @@ class CasWrapper:
local_path, local_path,
use_hash=use_hash, use_hash=use_hash,
return_json=True, return_json=True,
signer_id=signer_id signer_id=signer_id,
)['verified'] )["verified"]
# we can fall with ProcessExecutionError, # we can fall with ProcessExecutionError,
# because artifact can be not notarized # because artifact can be not notarized
except ProcessExecutionError: except ProcessExecutionError:
self._logger.exception('Cannot authenticate: %s', local_path) self._logger.exception("Cannot authenticate: %s", local_path)
return is_authenticated return is_authenticated
def notarize_artifacts( def notarize_artifacts(
@ -229,8 +229,10 @@ class CasWrapper:
try: try:
vcn_artifact_hash = self.notarize(artifact_path, metadata) vcn_artifact_hash = self.notarize(artifact_path, metadata)
except Exception: except Exception:
self._logger.exception('Cannot notarize artifact: %s', self._logger.exception(
artifact_path) "Cannot notarize artifact: %s",
artifact_path,
)
all_artifacts_is_notarized = False all_artifacts_is_notarized = False
continue continue
notarized_artifacts[artifact_path] = vcn_artifact_hash notarized_artifacts[artifact_path] = vcn_artifact_hash