- Fixed bugs in asserting and validating responses within the features.

- Fixed bug in writing name/attribute values with config by casting all values as strings (as before pytpp 2.0.0)
- Changed behavior of serializing returned content that is not a mapping to not raise an error, but instead to ignore the content and return what we can.
- Pushed 2.0.2.

docs/conf.py

@@ -19,7 +19,6 @@
 # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
 # ones.
 extensions = [
-    'sphinx.ext.autodoc',
     'sphinx.ext.napoleon',
     'sphinx.ext.inheritance_diagram',
     'sphinx_rtd_dark_mode',
@@ -72,11 +71,11 @@
 napoleon_use_ivar = True
 napoleon_attr_annotations = True
 autodoc_unqualified_typehints = True
-
 autodoc_pydantic_model_show_json = False
 autodoc_pydantic_settings_show_json = False
 autodoc_pydantic_validator_list_fields = False
 autodoc_pydantic_field_list_validators = False
+
 autodoc_pydantic_model_show_field_summary = False
 autodoc_pydantic_model_show_validator_members = False
 autodoc_pydantic_model_show_config_summary = False

pytpp/_about.py

@@ -2,4 +2,4 @@
 __author_email__ = 'spi@venafi.com'
 __project_name__ = 'pytpp'
 __project_url__ = 'https://github.com/Venafi/pytpp'
-__version__ = '2.0.1'
+__version__ = '2.0.2'

pytpp/api/api_base.py

@@ -303,11 +303,7 @@ def generate_output(response: Response, output_cls: Type[T_], root_field: str =
     except:
         result = {}
     if not isinstance(result, dict):
-        if not root_field:
-            raise AttributeError('Unable to assign ')
-        result = {
-            str(root_field): result
-        }
+        result = {str(root_field): result} if root_field else {}
     elif root_field:
         result = {
             str(root_field): result,
@@ -386,5 +382,5 @@ class WebSdkOutputModel(RootOutputModel):
 class InvalidResponseError(Exception):
     def __init__(self, msg: str, response: Response):
         self.response = response
-        super().__init__(msg)
+        super().__init__(msg, response)
 # endregion Response Definitions

pytpp/api/websdk/models/config.py

@@ -31,6 +31,9 @@ class Object(ObjectModel):
     revision: Optional[int] = ApiField(alias='Revision')
     type_name: str = ApiField(alias='TypeName')
 
+    def __str__(self):
+        return self.dn
+
 
 class Policy(ObjectModel):
     attribute_name: str = ApiField(alias='AttributeName')

pytpp/features/application.py

@@ -308,14 +308,14 @@ def create(self, name: 'str', device: 'Union[config.Object, str]', policy_folder
                 attribute_name=AdaptableAppAttributes.script_hash_mismatch_error,
                 values=[retry_after_script_hash_mismatch],
                 locked=locked
-            ).assert_valid_response()
+            )
         self._api.websdk.Config.WritePolicy.post(
             object_dn=policy_folder_dn,
             class_name=Classes.adaptable_app,
             attribute_name=AdaptableAppAttributes.powershell_script,
             values=[powershell_script_name],
             locked=locked
-        ).assert_valid_response()
+        )
         vault_id = self._api.websdk.SecretStore.Add.post(
             base_64_data=self._calculate_hash(powershell_script_content),
             keyname=KeyNames.software_default,
@@ -330,7 +330,7 @@ def create(self, name: 'str', device: 'Union[config.Object, str]', policy_folder
             attribute_name=AdaptableAppAttributes.powershell_script_hash_vault_id,
             values=[vault_id],
             locked=locked
-        ).assert_valid_response()
+        )
         # endregion Create The Policy Attributes
 
         app_attrs = {
@@ -653,19 +653,17 @@ def convert(self, basic_application: 'Union[config.Object, str]', new_class_name
             :ref:`config_object` of the application.
         """
         basic_application_dn = self._get_dn(basic_application)
-        result = self._api.websdk.Config.MutateObject.post(
+        self._api.websdk.Config.MutateObject.post(
             object_dn=basic_application_dn,
             class_name=str(new_class_name)
         )
-        result.assert_valid_response()
 
         if attributes:
             attributes = {k: ([str(v)] if not isinstance(v, list) else v) for k, v in attributes.items()}
-            result = self._api.websdk.Config.Write.post(
+            self._api.websdk.Config.Write.post(
                 object_dn=basic_application_dn,
                 attribute_data=self._name_value_list(attributes)
             )
-            result.assert_valid_response()
 
         new_object = self._api.websdk.Config.IsValid.post(object_dn=basic_application_dn).object
         if new_object.type_name != new_class_name:

pytpp/features/bases/feature_base.py

@@ -91,17 +91,20 @@ def _get_identity_object(self, prefixed_name: str = None, prefixed_universal: st
         if isinstance(prefixed_universal, ident.Identity):
             return prefixed_universal
 
-        result = self._api.websdk.Identity.Validate.post(
-            identity=self._identity_dict(prefixed_name=prefixed_name, prefixed_universal=prefixed_universal)
-        )
-        if result.is_valid_response() and result.api_response.content:
-            identity = result.identity
-        elif raise_error_if_not_exists:
-            target = prefixed_name or prefixed_universal
-            raise ObjectDoesNotExist(f'Could not find identity "{target}".')
+        try:
+            response = self._api.websdk.Identity.Validate.post(
+                identity=self._identity_dict(prefixed_name=prefixed_name, prefixed_universal=prefixed_universal)
+            )
+            identity = response.identity if response.api_response.content else None
+        except:
+            identity = None
+
+        if identity is not None:
+            return identity
+        elif not raise_error_if_not_exists:
+            return ident.Identity()
         else:
-            identity = ident.Identity()
-        return identity
+            raise ObjectDoesNotExist(f'Could not find identity "{prefixed_name or prefixed_universal}".')
 
     @staticmethod
     def _identity_dict(prefixed_name: str = None, prefixed_universal: str = None):
@@ -204,11 +207,10 @@ def _name_value_list(attributes: Dict[str, List[str]]):
         for n, v in attributes.items():
             if v is None:
                 continue
-            if not isinstance(v, list):
-                if isinstance(v, (tuple, set)):
-                    v = list(v)
-                else:
-                    v = [v]
+            if not isinstance(v, (list, tuple, set)):
+                v = [str(v)]
+            else:
+                v = list(map(str, v))
             nvl.append(config.NameAttribute(name=n, value=v))
         return nvl
 

pytpp/features/certificate.py

@@ -1,6 +1,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
 from datetime import datetime
+from pytpp.api.api_base import InvalidResponseError
 from pytpp.attributes.x509_certificate import X509CertificateAttributes
 from pytpp.features.bases.feature_base import FeatureBase, feature
 from pytpp.features.definitions.exceptions import FeatureException, UnexpectedValue
@@ -32,17 +33,17 @@ def _get(self, certificate: 'Union[config.Object, str]'):
         certificate_guid = self._get_guid(certificate)
 
         retries = 3
-        result = self._api.websdk.Certificates.Guid(certificate_guid).get()
-        for retry in range(retries):
-            if 'Rerun the transaction' not in result.api_response.reason:
-                result.assert_valid_response()
-                return result
-            features_logger.debug('Rerunning Certificates/Get transaction due to deadlock...')
-            result = self._api.websdk.Certificates.Guid(certificate_guid).get()
-        # It's likely at this point that we failed to get the certificate. Let the result.assert_valid_response()
-        # handle the errors. In some unknown case where the response is valid, return it just in case.
-        result.assert_valid_response()
-        return result
+        while retries > 0:
+            try:
+                result = self._api.websdk.Certificates.Guid(certificate_guid).get()
+                if 'Rerun the transaction' not in result.api_response.reason:
+                    return result
+            except InvalidResponseError as err:
+                if 'Rerun the transaction' in err.response.reason:
+                    # A deadlock issue may have caused this. Try getting the certificate again.
+                    features_logger.debug('Rerunning Certificates/Get transaction due to deadlock...')
+            retries -= 1
+        return self._api.websdk.Certificates.Guid(certificate_guid).get()
 
     def associate_application(self, certificate: 'Union[config.Object, str]', applications: 'List[Union[config.Object, str]]',
                               push_to_new: bool = False):
@@ -157,10 +158,10 @@ def delete(self, certificate: 'Union[config.Object, str]'):
             certificate: :ref:`config_object` or :ref:`dn` of the certificate object.
         """
         certificate_guid = self._get_guid(certificate)
-        result = self._api.websdk.Certificates.Guid(certificate_guid).delete()
-        if not result.is_valid_response():
-            certificate_dn = self._get_dn(certificate)
-            raise FeatureException(f'Could not delete certificate {certificate_dn}.')
+        try:
+            self._api.websdk.Certificates.Guid(certificate_guid).delete()
+        except InvalidResponseError:
+            raise FeatureException(f'Could not delete certificate {str(certificate)}.')
 
     def details(self, certificate: 'Union[config.Object, str]'):
         """
@@ -185,12 +186,11 @@ def dissociate_application(self, certificate: 'Union[config.Object, str]', appli
                             device objects.
         """
         certificate_dn = self._get_dn(certificate)
-        result = self._api.websdk.Certificates.Dissociate.post(
+        self._api.websdk.Certificates.Dissociate.post(
             certificate_dn=certificate_dn,
             application_dn=[self._get_dn(a) for a in applications],
             delete_orphans=delete_orphans
         )
-        result.assert_valid_response()
 
     # noinspection ALL
     def download(self, format: str, certificate: 'Union[config.Object, str]' = None, friendly_name: str = None,
@@ -593,8 +593,7 @@ def renew(self, certificate: 'Union[config.Object, str]', csr: str = None, re_en
         """
         result = self._get(certificate)
         current_thumbprint = result.certificate_details.thumbprint
-        result = self._api.websdk.Certificates.Renew.post(certificate_dn=result.dn, pkcs10=csr, reenable=re_enable)
-        result.assert_valid_response()
+        self._api.websdk.Certificates.Renew.post(certificate_dn=result.dn, pkcs10=csr, reenable=re_enable)
         return current_thumbprint
 
     def reset(self, certificate: 'Union[config.Object, str]'):
@@ -606,7 +605,6 @@ def reset(self, certificate: 'Union[config.Object, str]'):
         """
         certificate_dn = self._get_dn(certificate)
         result = self._api.websdk.Certificates.Reset.post(certificate_dn=certificate_dn, restart=False)
-        result.assert_valid_response()
         if not result.processing_reset_completed:
             raise UnexpectedValue(f'Processing reset was not completed for {certificate_dn}.')
 
@@ -618,8 +616,7 @@ def retry_from_current_stage(self, certificate: 'Union[config.Object, str]'):
             certificate: :ref:`config_object` or :ref:`dn` of the certificate object.
         """
         certificate_dn = self._get_dn(certificate)
-        result = self._api.websdk.Certificates.Retry.post(certificate_dn=certificate_dn)
-        result.assert_valid_response()
+        self._api.websdk.Certificates.Retry.post(certificate_dn=certificate_dn)
 
     def retry_from_stage_0(self, certificate: 'Union[config.Object, str]'):
         """
@@ -631,7 +628,6 @@ def retry_from_stage_0(self, certificate: 'Union[config.Object, str]'):
         """
         certificate_dn = self._get_dn(certificate)
         result = self._api.websdk.Certificates.Reset.post(certificate_dn=certificate_dn, restart=True)
-        result.assert_valid_response()
         if not result.restart_completed:
             raise UnexpectedValue(f'Restart renewal from stage 0 was not triggered on {certificate_dn}.')
 
@@ -656,7 +652,6 @@ def revoke(self, certificate: 'Union[config.Object, str]', comments: str = None,
             reason=reason,
             thumbprint=thumbprint
         )
-        result.assert_valid_response()
         if result.success is not True:
             raise UnexpectedValue(
                 f'Cannot revoke {certificate_dn} due to this error:\n{result.error}.'
@@ -697,7 +692,6 @@ def upload(self, certificate_data: str, parent_folder: 'Union[config.Object, str
             private_key_data=private_key_data,
             reconcile=reconcile
         )
-        result.assert_valid_response()
         return self._api.websdk.Config.IsValid.post(object_dn=result.certificate_dn).object
 
     def validate(self, certificates: 'List[Union[config.Object, str]]'):
@@ -739,7 +733,6 @@ def wait_for_enrollment_to_complete(self, certificate: 'Union[config.Object, str
         cert = self._get(certificate=certificate)
         with self._Timeout(timeout=timeout) as to:
             while not to.is_expired(poll=poll_interval):
-                cert.assert_valid_response()
                 thumbprint = cert.certificate_details.thumbprint
                 if thumbprint and thumbprint != current_thumbprint and cert.processing_details.stage in {None, 800}:
                     return cert.certificate_details