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

Fix review comments

This commit is contained in:
Daniil Anfimov 2022-06-30 15:05:12 +02:00
parent e5b832a30a
commit 07becc78df
1 changed files with 19 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
cas_wrapper.py
View File

@ -113,19 +113,22 @@ class CasWrapper:
self,
local_path: str,
) -> typing.Tuple[bool, typing.Optional[str]]:
"""
Authenticates source by git path.
Returns authenticate result and source commit hash.
"""
is_authenticated = False
commit_cas_hash = None
self.ensure_login()
try:
result_json = self.authenticate(local_path, return_json=True)
# it should return 0 for authenticated and trusted commits
is_authenticated = not bool(
result_json.get('status', 1))
commit_cas_hash = result_json.get('hash')
is_authenticated = not bool(result_json['status'])
commit_cas_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_cas_hash
def authenticate_artifact(
@ -134,6 +137,10 @@ class CasWrapper:
use_hash: bool = False,
return_json: bool = False,
) -> bool:
"""
Authenticates artifact by artifact path or hash if `use_hash` is True.
Returns authenticate result.
"""
is_authenticated = False
self.ensure_login()
try:
@ -145,7 +152,7 @@ class CasWrapper:
# 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
def notarize_artifacts(
@ -153,6 +160,11 @@ class CasWrapper:
artifact_paths: typing.List[str],
metadata: typing.Dict[str, typing.Any],
) -> typing.Tuple[bool, typing.Dict[str, str]]:
"""
Notarize artifacts by their paths.
Returns `True` if all artifacts was succesful notarizated
and dict with CAS hashes.
"""
all_artifacts_is_notarized = True
notarized_artifacts = {}
self.ensure_login()
@ -166,7 +178,8 @@ class CasWrapper:
try:
cas_artifact_hash = future.result()
except Exception:
self._logger.exception('Cannot notarize artifact:')
self._logger.exception('Cannot notarize artifact: %s',
artifact_path)
all_artifacts_is_notarized = False
continue
notarized_artifacts[artifact_path] = cas_artifact_hash