Code refactoring, added functions for build and sign node integration

This commit is contained in:
Daniil Anfimov 2022-06-28 21:05:51 +02:00
parent 4d05f388a2
commit 0a94dc6f8f
2 changed files with 72 additions and 9 deletions

View File

@ -1,7 +1,15 @@
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
from typing import Dict
import logging
import typing
from plumbum import local, ProcessExecutionError
from pydantic import BaseModel
class CasArtifact(BaseModel):
path: str
cas_hash: typing.Optional[str]
class CasWrapper:
@ -16,6 +24,7 @@ class CasWrapper:
self,
cas_api_key: str,
cas_signer_id: str,
logger: logging.Logger = None,
):
if self.binary_name not in local:
raise FileNotFoundError(
@ -23,19 +32,16 @@ class CasWrapper:
)
self._cas_api_key = cas_api_key
self._cas_signer_id = cas_signer_id
with local.env(
CAS_API_KEY=self._cas_api_key,
SIGNER_ID=self._cas_signer_id
):
self._cas = local['cas']
self._cas['login']()
self._logger = logger
if self._logger is None:
self._logger = logging.getLogger()
def __enter__(self):
with local.env(
CAS_API_KEY=self._cas_api_key,
SIGNER_ID=self._cas_signer_id,
):
self._cas = local['cas']
self._cas['login']()
return self
@ -45,7 +51,7 @@ class CasWrapper:
def notarize(
self,
local_path: str,
metadata: Dict = None,
metadata: typing.Dict = None,
) -> str:
"""
Wrapper around `cas notarize`
@ -110,3 +116,59 @@ class CasWrapper:
if return_json:
return json_result
return not bool(json_result['status'])
def authenticate_source(
self,
local_path: str,
) -> typing.Tuple[bool, typing.Optional[str]]:
is_authenticated = False
commit_cas_hash = None
with self as cas:
try:
result_json = cas.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')
# we can fall with ProcessExecutionError,
# because source can be not notarized
except ProcessExecutionError:
self._logger.exception('Cannot authenticate %s:', local_path)
return is_authenticated, commit_cas_hash
def authenticate_artifact(
self,
local_path: str,
) -> bool:
is_authenticated = False
with self as cas:
try:
is_authenticated = cas.authenticate(local_path)
# we can fall with ProcessExecutionError,
# because source can be not notarized
except ProcessExecutionError:
self._logger.exception('Cannot authenticate %s:', local_path)
return is_authenticated
def notarize_artifacts(
self,
artifacts: typing.List[CasArtifact],
metadata: typing.Dict[str, typing.Any],
) -> bool:
all_artifacts_is_notarized = True
with self as cas, ThreadPoolExecutor(max_workers=4) as executor:
futures = {
executor.submit(cas.notarize, artifact.path, metadata): artifact
for artifact in artifacts
if not artifact.cas_hash
}
for future in as_completed(futures):
artifact = futures[future]
try:
cas_artifact_hash = future.result()
except Exception:
self._logger.exception('Cannot notarize artifact:')
all_artifacts_is_notarized = False
continue
artifact.cas_hash = cas_artifact_hash
return all_artifacts_is_notarized

View File

@ -21,6 +21,7 @@ setup(
scripts=['cas_wrapper.py'],
install_requires=[
'plumbum>=1.7.2',
'pydantic>=1.8.1',
],
python_requires=">=3.6",
)