diff --git a/src/bin/shipyard_airflow/shipyard_airflow/control/helpers/configdocs_helper.py b/src/bin/shipyard_airflow/shipyard_airflow/control/helpers/configdocs_helper.py index 8a5a3099..7a07dd85 100644 --- a/src/bin/shipyard_airflow/shipyard_airflow/control/helpers/configdocs_helper.py +++ b/src/bin/shipyard_airflow/shipyard_airflow/control/helpers/configdocs_helper.py @@ -628,6 +628,97 @@ class ConfigdocsHelper(object): return False + def _get_ordered_versions(self, versions=None): + """returns a list of ordered versions""" + + # Default ordering + def_order = [SUCCESSFUL_SITE_ACTION, + LAST_SITE_ACTION, + COMMITTED, + BUFFER] + + # Defaults to COMMITTED and BUFFER + if versions is None: + versions = [COMMITTED, BUFFER] + + elif not len(versions) == 2: + raise AppError( + title='Incorrect number of versions for comparison', + description=( + 'User must pass in 2 valid versions for comparison'), + status=falcon.HTTP_400, + retry=False) + + elif versions[0] == versions[1]: + raise AppError( + title='Versions must be different for comparison', + description=( + 'Versions must be unique in order to perform comparison'), + status=falcon.HTTP_400, + retry=False) + + for version in versions: + if version not in def_order: + raise AppError( + title='Invalid version detected', + description=( + '{} is not a valid version, which include: ' + '{}'.format(version, ', '.join(def_order))), + status=falcon.HTTP_400, + retry=False) + + # Higher index in the def_order list will mean that it is a newer + # version. We will swap the order and sort the version if need be. + if def_order.index(versions[0]) > def_order.index(versions[1]): + ordered_versions = list(reversed(versions)) + else: + ordered_versions = versions + + return ordered_versions + + def _get_versions_name_id(self, ordered_versions): + + # Get version id + old_version_id = self.get_revision_id(ordered_versions[0]) + new_version_id = self.get_revision_id(ordered_versions[1]) + + # Get revision name + old_version_name = ordered_versions[0] + new_version_name = ordered_versions[1] + + # Check that revision id of LAST_SITE_ACTION and SUCCESSFUL_SITE_ACTION + # is not None + for name, rev_id in [(old_version_name, old_version_id), + (new_version_name, new_version_id)]: + if (name in [LAST_SITE_ACTION, SUCCESSFUL_SITE_ACTION] and + rev_id is None): + raise AppError( + title='Version does not exist', + description='{} version does not exist'.format(name), + status=falcon.HTTP_404, + retry=False) + + # Set to 0 if there is no committed version + if old_version_name == COMMITTED and old_version_id is None: + old_version_id = 0 + new_version_id = self.get_revision_id(BUFFER) or 0 + + # Set new_version_id if None + if new_version_id is None: + new_version_id = ( + self.get_revision_id(BUFFER) or old_version_id or 0) + + return ( + old_version_name, new_version_name, old_version_id, new_version_id) + +# +# End of ConfigdocsHelper class +# + + +# +# functions for module. +# def _get_validation_endpoints(): # returns the list of validation endpoint supported @@ -807,86 +898,3 @@ def _format_validations_to_status(val_msgs, error_count): }, "code": code } - - def _get_ordered_versions(self, versions=None): - """returns a list of ordered versions""" - - # Default ordering - def_order = [SUCCESSFUL_SITE_ACTION, - LAST_SITE_ACTION, - COMMITTED, - BUFFER] - - # Defaults to COMMITTED and BUFFER - if versions is None: - versions = [COMMITTED, BUFFER] - - elif not len(versions) == 2: - raise AppError( - title='Incorrect number of versions for comparison', - description=( - 'User must pass in 2 valid versions for comparison'), - status=falcon.HTTP_400, - retry=False) - - elif versions[0] == versions[1]: - raise AppError( - title='Versions must be different for comparison', - description=( - 'Versions must be unique in order to perform comparison'), - status=falcon.HTTP_400, - retry=False) - - for version in versions: - if version not in def_order: - raise AppError( - title='Invalid version detected', - description=( - '{} is not a valid version, which include: ' - '{}'.format(version, ', '.join(def_order))), - status=falcon.HTTP_400, - retry=False) - - # Higher index in the def_order list will mean that it is a newer - # version. We will swap the order and sort the version if need be. - if def_order.index(versions[0]) > def_order.index(versions[1]): - ordered_versions = list(reversed(versions)) - else: - ordered_versions = versions - - return ordered_versions - - def _get_versions_name_id(self, ordered_versions): - - # Get version id - old_version_id = self.get_revision_id(ordered_versions[0]) - new_version_id = self.get_revision_id(ordered_versions[1]) - - # Get revision name - old_version_name = ordered_versions[0] - new_version_name = ordered_versions[1] - - # Check that revision id of LAST_SITE_ACTION and SUCCESSFUL_SITE_ACTION - # is not None - for name, rev_id in [(old_version_name, old_version_id), - (new_version_name, new_version_id)]: - if (name in [LAST_SITE_ACTION, SUCCESSFUL_SITE_ACTION] and - rev_id is None): - raise AppError( - title='Version does not exist', - description='{} version does not exist'.format(name), - status=falcon.HTTP_404, - retry=False) - - # Set to 0 if there is no committed version - if old_version_name == COMMITTED and old_version_id is None: - old_version_id = 0 - new_version_id = self.get_revision_id(BUFFER) or 0 - - # Set new_version_id if None - if new_version_id is None: - new_version_id = ( - self.get_revision_id(BUFFER) or old_version_id or 0) - - return ( - old_version_name, new_version_name, old_version_id, new_version_id)