.gitignore

@@ -1,4 +0,0 @@
-# Byte-compiled / optimized / DLL files
-__pycache__/
-*.py[cod]
-*$py.class

cas_wrapper.py

@@ -1,6 +1,5 @@
 import json
-import logging
+from typing import Dict
-import typing
 
 from plumbum import local, ProcessExecutionError
 
@@ -13,45 +12,28 @@ class CasWrapper:
 
     binary_name = 'cas'
 
-    @classmethod
-    def _is_binary_present(cls):
-        if cls.binary_name not in local:
-            raise FileNotFoundError(
-                'Binary CAS is not found in PATH on the machine',
-            )
-
     def __init__(
             self,
             cas_api_key: str,
             cas_signer_id: str,
-            logger: logging.Logger = None,
     ):
-        self._is_binary_present()
+        if self.binary_name not in local:
+            raise FileNotFoundError(
+                'Binary CAS is not found in PATH on the machine',
+            )
         self._cas_api_key = cas_api_key
         self._cas_signer_id = cas_signer_id
-        self._cas = local['cas']
-        self._logger = logger
-        if self._logger is None:
-            self._logger = logging.getLogger()
-
-    @classmethod
-    def get_version(cls):
-        cls._is_binary_present()
-        command = local['cas']['--version']
-        version = command().split()[-1].split('v')[1]
-        return version
-
-    def ensure_login(self):
         with local.env(
-            CAS_API_KEY=self._cas_api_key,
+                CAS_API_KEY=self._cas_api_key,
-            SIGNER_ID=self._cas_signer_id,
+                SIGNER_ID=self._cas_signer_id
         ):
+            self._cas = local['cas']
             self._cas['login']()
 
     def notarize(
             self,
             local_path: str,
-            metadata: typing.Dict = None,
+            metadata: Dict = None,
     ) -> str:
         """
         Wrapper around `cas notarize`
@@ -79,55 +61,24 @@ class CasWrapper:
             result_of_execution = command()
         return json.loads(result_of_execution)['hash']
 
-    def notarize_no_exc(
-            self,
-            local_path: str,
-            metadata: typing.Dict = None,
-    ) -> typing.Tuple[bool, str]:
-        """
-        Wrapper for avoiding raising exceptions during notarization.
-        Return `success` flag instead for library user to react respectively.
-        :param local_path: path to a local Git repo
-        :param metadata: additional metadata
-        :return: boolean flag for operation success and the hash
-            of the notarized artifact.
-        :rtype: tuple
-        """
-        success = False
-        try:
-            cas_hash = self.notarize(local_path, metadata=metadata)
-            success = True
-        except Exception:
-            self._logger.exception('Cannot notarize artifact: %s',
-                                   local_path)
-            cas_hash = ''
-        return success, cas_hash
-
     def authenticate(
             self,
             local_path: str,
-            return_json: bool = False,
-            use_hash: bool = False,
-            signer_id: str = None,
     ):
         """
         Wrapper around `cas authenticate`
         :param local_path: path to a local Git repo
                (should be started from `git://`)
-               or to a single local file or hash
+               or to a single local file
-        :param return_json: flag for return json response
-        :param use_hash: flag for authenticate by hash
         :return: true if a commit is trusted, vice versa - false
-                 or dict with result if return_json param is True
+        :rtype: bool
-        :rtype: bool or dict
         """
-        command_args = ['authenticate', local_path]
+        command = self._cas[
-        if use_hash:
+            'authenticate',
-            command_args = ['authenticate', '--hash', local_path]
+            local_path,
-        if signer_id:
+            '-o',
-            command_args.extend(('--signerID', signer_id))
+            'json',
-        command_args.extend(('-o', 'json'))
+        ]
-        command = self._cas[command_args]
         try:
             with local.env(
                     CAS_API_KEY=self._cas_api_key,
@@ -141,88 +92,4 @@ class CasWrapper:
             ):
                 # in case if commit is untrusted
                 result_of_execution = command(retcode=1)
-        json_result = json.loads(result_of_execution)
+        return not bool(json.loads(result_of_execution)['status'])
-        if return_json:
-            return json_result
-        return not bool(json_result['status'])
-
-    def authenticate_source(
-        self,
-        local_path: str,
-        signer_id: str = None,
-    ) -> 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,
-                signer_id=signer_id
-            )
-            is_authenticated = result_json['verified']
-            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)
-        return is_authenticated, commit_cas_hash
-
-    def authenticate_artifact(
-        self,
-        local_path: str,
-        use_hash: bool = False,
-        signer_id: str = None,
-    ) -> bool:
-        """
-        Authenticates artifact by artifact path or hash if `use_hash` is True.
-        Returns authenticate result.
-        """
-        is_authenticated = False
-        self.ensure_login()
-        try:
-            is_authenticated = self.authenticate(
-                local_path,
-                use_hash=use_hash,
-                return_json=True,
-                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)
-        return is_authenticated
-
-    def notarize_artifacts(
-        self,
-        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()
-
-        # ALBS-576: We stopped doing this process in parallel due to the
-        # problems experienced and described in this CAS issue:
-        # https://github.com/codenotary/cas/issues/275
-        # Hence, we decided to go sequential here until the problem is
-        # resolved in CAS itself.
-        for artifact_path in artifact_paths:
-            try:
-                cas_artifact_hash = self.notarize(artifact_path, metadata)
-            except Exception:
-                self._logger.exception('Cannot notarize artifact: %s',
-                                       artifact_path)
-                all_artifacts_is_notarized = False
-                continue
-            notarized_artifacts[artifact_path] = cas_artifact_hash
-        return all_artifacts_is_notarized, notarized_artifacts

setup.py

@@ -2,7 +2,7 @@ from setuptools import setup
 
 setup(
     name="cas_wrapper",
-    version="0.0.6",
+    version="0.0.1",
     author="Stepan Oksanichenko",
     author_email="soksanichenko@almalinux.org",
     description="The python wrapper around binary cas from "