From 53b670a142f3a30d5bcc34999064426e69c59747 Mon Sep 17 00:00:00 2001 From: Frank Brehm Date: Wed, 27 Feb 2019 16:10:23 +0100 Subject: [PATCH] Removing lib/webhooks/get_forge_modules.py --- lib/webhooks/forge_module_info.py | 1930 ----------------------------- lib/webhooks/get_forge_modules.py | 2 - 2 files changed, 1932 deletions(-) delete mode 100644 lib/webhooks/forge_module_info.py diff --git a/lib/webhooks/forge_module_info.py b/lib/webhooks/forge_module_info.py deleted file mode 100644 index 8adda96..0000000 --- a/lib/webhooks/forge_module_info.py +++ /dev/null @@ -1,1930 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -@author: Frank Brehm -@contact: frank.brehm@pixelpark.com -@copyright: © 2018 by Frank Brehm, Publicies Pixelpark GmbH, Berlin -@summary: A module for encapsulating all information about a Puppet module -""" -from __future__ import absolute_import - -# Standard modules -import logging -import re -import copy -import warnings -import time -import datetime -import collections - -# Third party modules -import requests -import pytz - -# Own modules -from fb_tools.common import pp, to_str, to_bool, is_sequence -from fb_tools.obj import FbBaseObjectError, FbBaseObject - -from .xlate import XLATOR - -from .base_module_info import BaseModuleInfoError, BaseModuleInfo - -from .module_meta_info import ModuleMetadata - -__version__ = '0.7.1' - -LOG = logging.getLogger(__name__) - -_ = XLATOR.gettext -ngettext = XLATOR.ngettext - - -# ============================================================================= -def parse_forge_date(dt): - - return datetime.datetime.strptime(dt, '%Y-%m-%d %H:%M:%S %z') - - -# ============================================================================= -class ReleaseInfoError(BaseModuleInfoError): - - pass - - -# ============================================================================= -class ForgeModuleInfoError(BaseModuleInfoError): - - pass - - -# ============================================================================= -class ForgeModuleInfoTypeError(ForgeModuleInfoError, TypeError): - - pass - - -# ============================================================================= -class ForgeOwnerInfo(FbBaseObject): - """Class for encapsulating information about an module owner in Puppet forge.""" - - # ------------------------------------------------------------------------- - def __init__( - self, gravatar_id=None, slug=None, uri=None, username=None, - appname=None, verbose=0, version=__version__, base_dir=None, initialized=None): - - self._gravatar_id = None - self._slug = None - self._uri = None - self._username = None - - super(ForgeOwnerInfo, self).__init__( - appname=appname, verbose=verbose, version=version, base_dir=base_dir, initialized=False) - - self.gravatar_id = gravatar_id - self.slug = slug - self.uri = uri - self.username = username - - if initialized is not None: - self.initialized = initialized - - # ------------------------------------------------------------------------- - def as_dict(self, short=True): - """ - Transforms the elements of the object into a dict - - @return: structure as dict - @rtype: dict - """ - - res = super(ForgeOwnerInfo, self).as_dict(short=short) - - res['gravatar_id'] = self.gravatar_id - res['slug'] = self.slug - res['uri'] = self.uri - res['username'] = self.username - - return res - - # ------------------------------------------------------------------------- - @property - def gravatar_id(self): - """The Gravatar-Id of this owner.""" - return self._gravatar_id - - @gravatar_id.setter - def gravatar_id(self, value): - if value is None: - self._gravatar_id = None - return - v = str(value).strip() - if value == '': - self._gravatar_id = None - return - self._gravatar_id = v - - # ------------------------------------------------------------------------- - @property - def slug(self): - """The slug of this owner.""" - return self._slug - - @slug.setter - def slug(self, value): - if value is None: - self._slug = None - return - v = str(value).strip() - if value == '': - self._slug = None - return - self._slug = v - - # ------------------------------------------------------------------------- - @property - def uri(self): - """The URI of this owner.""" - return self._uri - - @uri.setter - def uri(self, value): - if value is None: - self._uri = None - return - v = str(value).strip() - if value == '': - self._uri = None - return - self._uri = v - - # ------------------------------------------------------------------------- - @property - def username(self): - """The username of this owner.""" - return self._username - - @username.setter - def username(self, value): - if value is None: - self._username = None - return - v = str(value).strip() - if value == '': - self._username = None - return - self._username = v - - # ------------------------------------------------------------------------- - def to_data(self): - """Returning a dict, which can be used to re-instantiate this owner info.""" - - res = { - 'gravatar_id': self.gravatar_id, - 'slug': self.slug, - 'uri': self.uri, - 'username': self.username, - } - - return res - - # ------------------------------------------------------------------------- - def __copy__(self): - - owner = ForgeOwnerInfo( - gravatar_id=self.gravatar_id, slug=self.slug, uri=self.uri, username=self.username, - appname=self.appname, verbose=self.verbose, base_dir=self.base_dir, - initialized=self.initialized) - return owner - - # ------------------------------------------------------------------------- - def __eq__(self, other): - - if self.verbose > 4: - LOG.debug(_("Comparing {} objects ...").format(self.__class__.__name__)) - - if not isinstance(other, ForgeOwnerInfo): - return False - - if self.gravatar_id != other.gravatar_id: - return False - if self.slug != other.slug: - return False - if self.uri != other.uri: - return False - if self.username != other.username: - return False - - return True - - # ------------------------------------------------------------------------- - @classmethod - def from_data(cls, data, appname=None, verbose=0, base_dir=None): - - owner = cls(appname=appname, verbose=verbose, base_dir=base_dir) - - if 'gravatar_id' in data: - owner.gravatar_id = data['gravatar_id'] - if 'slug' in data: - owner.slug = data['slug'] - if 'uri' in data: - owner.uri = data['uri'] - if 'username' in data: - owner.username = data['username'] - owner.initialized = True - - if verbose > 3: - LOG.debug(_("Got {}:").format(cls.__name__) + '\n' + pp(owner.as_dict())) - - return owner - - -# ============================================================================= -class BaseForgeModuleInfo(FbBaseObject): - """Class for encapsulating base information about a Puppet module from Puppet Forge.""" - - # ------------------------------------------------------------------------- - def __init__( - self, appname=None, verbose=0, version=__version__, base_dir=None, - initialized=None): - - self._deprecated_at = None - self._name = None - self.owner = None - self._slug = None - self._uri = None - - super(BaseForgeModuleInfo, self).__init__( - appname=appname, verbose=verbose, version=version, base_dir=base_dir, initialized=False) - - if initialized is not None: - self.initialized = initialized - - # ------------------------------------------------------------------------- - def as_dict(self, short=True): - """ - Transforms the elements of the object into a dict - - @return: structure as dict - @rtype: dict - """ - - res = super(BaseForgeModuleInfo, self).as_dict(short=short) - - res['deprecated_at'] = self.deprecated_at - res['name'] = self.name - res['slug'] = self.slug - res['uri'] = self.uri - - return res - - # ------------------------------------------------------------------------- - @property - def deprecated_at(self): - """Date of deprecation of this module.""" - return self._deprecated_at - - @deprecated_at.setter - def deprecated_at(self, value): - if value is None: - self._deprecated_at = None - return - if isinstance(value, datetime.datetime): - self._deprecated_at = value - return - v = str(value).strip() - if value == '': - self._deprecated_at = None - return - self._deprecated_at = parse_forge_date(v) - - # ------------------------------------------------------------------------- - @property - def name(self): - """The name of this module.""" - return self._name - - @name.setter - def name(self, value): - if value is None: - self._name = None - return - v = str(value).strip() - if value == '': - self._name = None - return - self._name = v - - # ------------------------------------------------------------------------- - @property - def slug(self): - """The slug of this module.""" - return self._slug - - @slug.setter - def slug(self, value): - if value is None: - self._slug = None - return - v = str(value).strip() - if value == '': - self._slug = None - return - self._slug = v - - # ------------------------------------------------------------------------- - @property - def uri(self): - """The URI of this module.""" - return self._uri - - @uri.setter - def uri(self, value): - if value is None: - self._uri = None - return - v = str(value).strip() - if value == '': - self._uri = None - return - self._uri = v - - # ------------------------------------------------------------------------- - def to_data(self): - """Returning a dict, which can be used to re-instantiate this module info.""" - - res = {} - - res['deprecated_at'] = None - if self.deprecated_at: - res['deprecated_at'] = self.deprecated_at.strftime('%Y-%m-%d %H:%M:%S %z') - - res['owner'] = None - if self.owner: - res['owner'] = self.owner.to_data() - - res['name'] = self.name - res['slug'] = self.slug - res['uri'] = self.uri - - return res - - # ------------------------------------------------------------------------- - def __copy__(self): - - mod = BaseForgeModuleInfo( - appname=self.appname, verbose=self.verbose, base_dir=self.base_dir) - mod.deprecated_at = self.deprecated_at - mod.name = self.name - if self.owner: - mod.owner = copy.copy(self.owner) - mod.slug = self.slug - mod.uri = self.uri - - mod.initialized = self.initialized - return mod - - # ------------------------------------------------------------------------- - def __eq__(self, other): - - if self.verbose > 4: - LOG.debug(_("Comparing {} objects ...").format(self.__class__.__name__)) - - if not isinstance(other, BaseForgeModuleInfo): - return False - - if self.deprecated_at != other.deprecated_at: - return False - if self.name != other.name: - return False - if self.owner != other.owner: - return False - if self.slug != other.slug: - return False - if self.uri != other.uri: - return False - - return True - - # ------------------------------------------------------------------------- - @classmethod - def from_data(cls, data, appname=None, verbose=0, base_dir=None): - - mod = cls(appname=appname, verbose=verbose, base_dir=base_dir) - - if 'deprecated_at' in data: - mod.deprecated_at = data['deprecated_at'] - if 'slug' in data: - mod.slug = data['slug'] - if 'uri' in data: - mod.uri = data['uri'] - if 'name' in data: - mod.name = data['name'] - - if 'owner' in data and data['owner']: - mod.owner = ForgeOwnerInfo.from_data( - data['owner'], appname=appname, verbose=verbose, base_dir=base_dir) - - mod.initialized = True - - if verbose > 3: - LOG.debug(_("Got {}:").format(cls.__name__) + '\n' + pp(mod.as_dict())) - - return mod - - -# ============================================================================= -class ModuleReleaseInfo(FbBaseObject): - """Class for encapsulating information about a Puppet module release from Puppet Forge.""" - - # ------------------------------------------------------------------------- - def __init__( - self, appname=None, verbose=0, version=__version__, base_dir=None, - initialized=None): - - self._created_at = None - self._deleted_at = None - self._file_size = None - self._file_uri = None - self._slug = None - self._supported = None - self._uri = None - self._release_version = None - - super(ModuleReleaseInfo, self).__init__( - appname=appname, verbose=verbose, version=version, base_dir=base_dir, initialized=False) - - if initialized is not None: - self.initialized = initialized - - # ------------------------------------------------------------------------- - def as_dict(self, short=True): - """ - Transforms the elements of the object into a dict - - @return: structure as dict - @rtype: dict - """ - - res = super(ModuleReleaseInfo, self).as_dict(short=short) - - res['created_at'] = self.created_at - res['deleted_at'] = self.deleted_at - res['file_size'] = self.file_size - res['file_uri'] = self.file_uri - res['slug'] = self.slug - res['supported'] = self.supported - res['uri'] = self.uri - res['release_version'] = self.release_version - - return res - - # ------------------------------------------------------------------------- - @property - def created_at(self): - """Creation date of this release.""" - return self._created_at - - @created_at.setter - def created_at(self, value): - if value is None: - self._created_at = None - return - if isinstance(value, datetime.datetime): - self._created_at = value - return - v = str(value).strip() - if value == '': - self._created_at = None - return - self._created_at = parse_forge_date(v) - - # ------------------------------------------------------------------------- - @property - def deleted_at(self): - """Deletion date of this release.""" - return self._deleted_at - - @deleted_at.setter - def deleted_at(self, value): - if value is None: - self._deleted_at = None - return - if isinstance(value, datetime.datetime): - self._deleted_at = value - return - v = str(value).strip() - if value == '': - self._deleted_at = None - return - self._deleted_at = parse_forge_date(v) - - # ------------------------------------------------------------------------- - @property - def file_size(self): - """The file size in bytes of this release.""" - return self._file_size - - @file_size.setter - def file_size(self, value): - if value is None: - self._file_size = None - return - v = int(value) - if v < 0: - msg = _( - "The file size of a release must be greater or equal to zero " - "(Given: {}).").format(value) - raise ValueError(msg) - self._file_size = v - - # ------------------------------------------------------------------------- - @property - def file_uri(self): - """The file URI of this release.""" - return self._file_uri - - @file_uri.setter - def file_uri(self, value): - if value is None: - self._file_uri = None - return - v = str(value).strip() - if value == '': - self._file_uri = None - return - self._file_uri = v - - # ------------------------------------------------------------------------- - @property - def slug(self): - """The slug of this release.""" - return self._slug - - @slug.setter - def slug(self, value): - if value is None: - self._slug = None - return - v = str(value).strip() - if value == '': - self._slug = None - return - self._slug = v - - # ------------------------------------------------------------------------- - @property - def supported(self): - """The URI of this release.""" - return self._supported - - @supported.setter - def supported(self, value): - if value is None: - self._supported = None - return - self._supported = to_bool(value) - - # ------------------------------------------------------------------------- - @property - def uri(self): - """The URI of this release.""" - return self._uri - - @uri.setter - def uri(self, value): - if value is None: - self._uri = None - return - v = str(value).strip() - if value == '': - self._uri = None - return - self._uri = v - - # ------------------------------------------------------------------------- - @property - def release_version(self): - """The URI of this release.""" - return self._release_version - - @release_version.setter - def release_version(self, value): - if value is None: - self._release_version = None - return - v = str(value).strip() - if value == '': - self._release_version = None - return - self._release_version = v - - # ------------------------------------------------------------------------- - def to_data(self): - """Returning a dict, which can be used to re-instantiate this module info.""" - - res = {} - - res['created_at'] = None - if self.created_at: - res['created_at'] = self.created_at.strftime('%Y-%m-%d %H:%M:%S %z') - - res['deleted_at'] = None - if self.deleted_at: - res['deleted_at'] = self.deleted_at.strftime('%Y-%m-%d %H:%M:%S %z') - - res['file_size'] = self.file_size - res['file_uri'] = self.file_uri - res['slug'] = self.slug - res['supported'] = self.supported - res['uri'] = self.uri - res['version'] = self.release_version - - return res - - # ------------------------------------------------------------------------- - def __copy__(self): - - release = ModuleReleaseInfo( - appname=self.appname, verbose=self.verbose, base_dir=self.base_dir) - release.created_at = self.created_at - release.deleted_at = self.deleted_at - release.file_size = self.file_size - release.file_uri = self.file_uri - release.slug = self.slug - release.supported = self.supported - release.uri = self.uri - release.release_version = self.release_version - - release.initialized = self.initialized - return release - - # ------------------------------------------------------------------------- - def __eq__(self, other): - - if self.verbose > 4: - LOG.debug(_("Comparing {} objects ...").format(self.__class__.__name__)) - - if not isinstance(other, ModuleReleaseInfo): - return False - - if self.created_at != other.created_at: - return False - if self.deleted_at != other.deleted_at: - return False - if self.file_size != other.file_size: - return False - if self.file_uri != other.file_uri: - return False - if self.slug != other.slug: - return False - if self.supported != other.supported: - return False - if self.uri != other.uri: - return False - if self.release_version != other.release_version: - return False - - return True - - # ------------------------------------------------------------------------- - @classmethod - def get_from_forge(cls, data, appname=None, verbose=0, base_dir=None): - - release = cls(appname=appname, verbose=verbose, base_dir=base_dir) - release.apply_forge_data(data) - - if verbose > 2: - LOG.debug(_("Got {}:").format(cls.__name__) + '\n' + pp(release.as_dict())) - - return release - - # ------------------------------------------------------------------------- - def apply_forge_data(self, data): - - if 'created_at' in data and data['created_at']: - self.created_at = data['created_at'] - if 'deleted_at' in data and data['deleted_at']: - self.deleted_at = data['deleted_at'] - if 'file_size' in data and data['file_size']: - self.file_size = data['file_size'] - if 'file_uri' in data and data['file_uri']: - self.file_uri = data['file_uri'] - if 'slug' in data and data['slug']: - self.slug = data['slug'] - if 'supported' in data and data['supported']: - self.supported = data['supported'] - if 'uri' in data and data['uri']: - self.uri = data['uri'] - if 'version' in data and data['version']: - self.release_version = data['version'] - - -# ============================================================================= -class CurrentModuleReleaseInfo(ModuleReleaseInfo): - """Class for encapsulating information about the current release of a Puppet module - from Puppet Forge.""" - - # ------------------------------------------------------------------------- - def __init__( - self, appname=None, verbose=0, version=__version__, base_dir=None, - initialized=None): - - self._changelog = None - self._deleted_for = None - self._downloads = None - self._file_md5 = None - self._license = None - self.metadata = None - self.module = None - self._pdk = False - self._readme = None - self._reference = None - self.tags = [] - self.tasks = [] - self._updated_at = None - self._validation_score = None - - super(CurrentModuleReleaseInfo, self).__init__( - appname=appname, verbose=verbose, version=version, base_dir=base_dir, initialized=False) - - if initialized is not None: - self.initialized = initialized - - # ------------------------------------------------------------------------- - @property - def changelog(self): - """The change log of this module release.""" - return self._changelog - - @changelog.setter - def changelog(self, value): - if value is None: - self._changelog = None - return - v = str(value).strip() - if v == '': - self._changelog = None - return - self._changelog = v - - # ------------------------------------------------------------------------- - @property - def deleted_for(self): - """The reason, why this module release was deleted.""" - return self._deleted_for - - @deleted_for.setter - def deleted_for(self, value): - if value is None: - self._deleted_for = None - return - v = str(value).strip() - if v == '': - self._deleted_for = None - return - self._deleted_for = v - - # ------------------------------------------------------------------------- - @property - def downloads(self): - "The number of downloads of this module release.""" - return self._downloads - - @downloads.setter - def downloads(self, value): - if value is None: - self._downloads = None - return - try: - v = int(value) - self._downloads = v - return - except Exception as e: - LOG.error(_("Got a {c} setting {w}: {e}").format( - c=e.__class__.__name__, w='downloads', e=e)) - - # ------------------------------------------------------------------------- - @property - def file_md5(self): - """The MD5-sum of the current release package.""" - return self._file_md5 - - @file_md5.setter - def file_md5(self, value): - if value is None: - self._file_md5 = None - return - v = str(value).strip() - if v == '': - self._file_md5 = None - return - self._file_md5 = v - - # ------------------------------------------------------------------------- - @property - def license(self): - """The license of module source.""" - return self._license - - @license.setter - def license(self, value): - if value is None: - self._license = None - return - v = str(value).strip() - if v == '': - self._license = None - return - self._license = v - - # ------------------------------------------------------------------------- - @property - def pdk(self): - """The pdk of this release.""" - return self._pdk - - @pdk.setter - def pdk(self, value): - if value is None: - self._pdk = None - return - self._pdk = to_bool(value) - - # ------------------------------------------------------------------------- - @property - def readme(self): - """The readme of module release.""" - return self._readme - - @readme.setter - def readme(self, value): - if value is None: - self._readme = None - return - v = str(value).strip() - if v == '': - self._readme = None - return - self._readme = v - - # ------------------------------------------------------------------------- - @property - def reference(self): - """The reference of module release.""" - return self._reference - - @reference.setter - def reference(self, value): - if value is None: - self._reference = None - return - v = str(value).strip() - if v == '': - self._reference = None - return - self._reference = v - - # ------------------------------------------------------------------------- - @property - def updated_at(self): - """Update date of this release.""" - return self._updated_at - - @updated_at.setter - def updated_at(self, value): - if value is None: - self._updated_at = None - return - if isinstance(value, datetime.datetime): - self._updated_at = value - return - v = str(value).strip() - if value == '': - self._updated_at = None - return - self._updated_at = parse_forge_date(v) - - # ------------------------------------------------------------------------- - @property - def validation_score(self): - "The validation score of this module release.""" - return self._validation_score - - @validation_score.setter - def validation_score(self, value): - if value is None: - self._validation_score = None - return - try: - v = int(value) - self._validation_score = v - return - except Exception as e: - LOG.error(_("Got a {c} setting {w}: {e}").format( - c=e.__class__.__name__, w='validation_score', e=e)) - - # ------------------------------------------------------------------------- - def as_dict(self, short=True): - """ - Transforms the elements of the object into a dict - - @return: structure as dict - @rtype: dict - """ - - res = super(CurrentModuleReleaseInfo, self).as_dict(short=short) - - res['changelog'] = self.changelog - res['deleted_for'] = self.deleted_for - res['downloads'] = self.downloads - res['file_md5'] = self.file_md5 - res['license'] = self.license - res['pdk'] = self.pdk - res['readme'] = self.readme - res['reference'] = self.reference - res['updated_at'] = self.updated_at - res['validation_score'] = self.validation_score - - return res - - # ------------------------------------------------------------------------- - def apply_forge_data(self, data): - - super(CurrentModuleReleaseInfo, self).apply_forge_data(data) - - if 'changelog' in data and data['changelog']: - self.changelog = data['changelog'] - if 'deleted_for' in data and data['deleted_for']: - self.deleted_for = data['deleted_for'] - if 'downloads' in data and data['downloads']: - self.downloads = data['downloads'] - if 'file_md5' in data and data['file_md5']: - self.file_md5 = data['file_md5'] - if 'license' in data and data['license']: - self.license = data['license'] - - if 'metadata' in data and data['metadata']: - self.metadata = ModuleMetadata.from_json_data( - data['metadata'], appname=self.appname, - verbose=self.verbose, base_dir=self.base_dir) - - if 'module' in data and data['module']: - self.module = BaseForgeModuleInfo.from_data( - data['module'], appname=self.appname, - verbose=self.verbose, base_dir=self.base_dir) - - if 'pdk' in data: - self.pdk = data['pdk'] - if 'readme' in data: - self.readme = data['readme'] - if 'reference' in data: - self.reference = data['reference'] - if 'updated_at' in data: - self.reference = data['updated_at'] - if 'validation_score' in data: - self.validation_score = data['validation_score'] - - self.tags = [] - if 'tags' in data and data['tags']: - if self.verbose > 2: - LOG.debug( - "Tags in current release of {!r}:".format(self.slug) + '\n' + pp(data['tags'])) - for tag in data['tags']: - self.tags.append(tag) - - self.tasks = [] - if 'tasks' in data and data['tasks']: - if self.verbose > 3: - LOG.debug( - "Tasks in current release of {!r}:".format(self.slug) + - '\n' + pp(data['tasks'])) - for task in data['tasks']: - self.tasks.append(copy.copy(task)) - - # ------------------------------------------------------------------------- - def to_data(self): - """Returning a dict, which can be used to re-instantiate this module info.""" - - res = super(CurrentModuleReleaseInfo, self).to_data() - - res['changelog'] = self.changelog - res['deleted_for'] = self.deleted_for - res['downloads'] = self.downloads - res['file_md5'] = self.file_md5 - res['license'] = self.license - res['pdk'] = self.pdk - res['readme'] = self.readme - res['reference'] = self.reference - res['validation_score'] = self.validation_score - - res['metadata'] = None - if self.metadata: - res['metadata'] = self.metadata.to_data() - - res['module'] = None - if self.module: - res['module'] = self.module.to_data() - - res['updated_at'] = None - if self.updated_at: - res['updated_at'] = self.updated_at.strftime('%Y-%m-%d %H:%M:%S %z') - - res['tags'] = [] - for tag in self.tags: - res['tags'].append(tag) - - res['tasks'] = [] - for task in self.tasks: - res['tasks'].append(copy.copy(task)) - - return res - - # ------------------------------------------------------------------------- - def __copy__(self): - - release = super(CurrentModuleReleaseInfo, self).__copy__() - release.initialized = False - - release.changelog = self.changelog - release.deleted_for = self.deleted_for - release.downloads = self.downloads - release.file_md5 = self.file_md5 - release.license = self.license - release.pdk = self.pdk - release.readme = self.readme - release.reference = self.reference - release.updated_at = self.updated_at - release.validation_score = self.validation_score - - release.metadata = None - if self.metadata: - release.metadata = copy.copy(self.metadata) - - release.module = None - if self.module: - release.module = copy.copy(self.module) - - release.tags = [] - if self.tags: - for tag in self.tags: - release.tags.append(tag) - - release.tasks = [] - if self.tasks: - for task in self.tasks: - release.tasks.append(copy.copy(task)) - - release.initialized = self.initialized - return release - -# ============================================================================= -class ModuleReleaseList(collections.MutableSequence, FbBaseObject): - - msg_no_release = _("Invalid type {t!r} as an item of a {c}, only {o} objects are allowed.") - - # ------------------------------------------------------------------------- - def __init__( - self, appname=None, verbose=0, version=__version__, base_dir=None, - initialized=None, *releases): - - self._list = [] - - super(ModuleReleaseList, self).__init__( - appname=appname, verbose=verbose, version=version, base_dir=base_dir, initialized=False) - - for release in releases: - self.append(release) - - # ------------------------------------------------------------------------- - def as_dict(self, short=True): - - res = super(ModuleReleaseList, self).as_dict(short=short) - - res['list'] = [] - for release in self: - res['list'].append(release.as_dict(short=short)) - - return res - - # ------------------------------------------------------------------------- - def to_data(self): - """Returning a list, which can be used to re-instantiate this module info.""" - - res = [] - for release in self: - res.append(release.to_data()) - - return res - - # ------------------------------------------------------------------------- - def index(self, release, *args): - - i = None - j = None - - if len(args) > 0: - if len(args) > 2: - raise TypeError(_("{m} takes at most {max} arguments ({n} given).").format( - m='index()', max=3, n=len(args) + 1)) - i = int(args[0]) - if len(args) > 1: - j = int(args[1]) - - index = 0 - if i is not None: - start = i - if i < 0: - start = len(self._list) + i - wrap = False - end = len(self._list) - if j is not None: - if j < 0: - end = len(self._list) + j - if end < index: - wrap = True - else: - end = j - for index in list(range(len(self._list))): - item = self._list[index] - if index < start: - continue - if index >= end and not wrap: - break - if item == release: - return index - - if wrap: - for index in list(range(len(self._list))): - item = self._list[index] - if index >= end: - break - if item == release: - return index - - msg = _("Release {!r} is not in release list.").format(release.release_version) - raise ValueError(msg) - - # ------------------------------------------------------------------------- - def __contains__(self, release): - - if not isinstance(release, ModuleReleaseInfo): - raise TypeError(self.msg_no_release.format( - t=release.__class__.__name__, c=self.__class__.__name__, o='ModuleReleaseInfo')) - - if not self._list: - return False - - for item in self._list: - if item == release: - return True - - return False - - # ------------------------------------------------------------------------- - def count(self, release): - - if not isinstance(release, ModuleReleaseInfo): - raise TypeError(self.msg_no_release.format( - t=release.__class__.__name__, c=self.__class__.__name__, o='ModuleReleaseInfo')) - - if not self._list: - return 0 - - num = 0 - for item in self._list: - if item == release: - num += 1 - return num - - # ------------------------------------------------------------------------- - def __len__(self): - return len(self._list) - - # ------------------------------------------------------------------------- - def __getitem__(self, key): - return self._list.__getitem__(key) - - # ------------------------------------------------------------------------- - def __reversed__(self): - - return reversed(self._list) - - # ------------------------------------------------------------------------- - def __setitem__(self, key, release): - - if not isinstance(release, ModuleReleaseInfo): - raise TypeError(self.msg_no_release.format( - t=release.__class__.__name__, c=self.__class__.__name__, o='ModuleReleaseInfo')) - - self._list.__setitem__(key, release) - - # ------------------------------------------------------------------------- - def __delitem__(self, key): - - del self._list[key] - - # ------------------------------------------------------------------------- - def append(self, release): - - if not isinstance(release, ModuleReleaseInfo): - raise TypeError(self.msg_no_release.format( - t=release.__class__.__name__, c=self.__class__.__name__, o='ModuleReleaseInfo')) - - self._list.append(release) - - # ------------------------------------------------------------------------- - def insert(self, index, release): - - if not isinstance(release, ModuleReleaseInfo): - raise TypeError(self.msg_no_release.format( - t=release.__class__.__name__, c=self.__class__.__name__, o='ModuleReleaseInfo')) - - self._list.insert(index, release) - - # ------------------------------------------------------------------------- - def __copy__(self): - - new_list = self.__class__() - for release in self._list: - new_list.append(copy.copy(release)) - return new_list - - -# ============================================================================= -class ForgeModuleInfo(BaseModuleInfo): - """Class for encapsulating all information about a Puppet module from Puppet Forge.""" - - # ------------------------------------------------------------------------- - def __init__( - self, appname=None, verbose=0, version=__version__, base_dir=None, - initialized=None, name=None, vendor=None, full_name=None): - - self.current_release = None - self.releases = None - self.owner = None - - self._created_at = None - self._deprecated_at = None - self._deprecated_for = None - self._downloads = None - self._endorsement = None - self._feedback_score = None - self._homepage_url = None - self._issues_url = None - self._module_group = None - self._slug = None - self._superseded_by = None - self._supported = None - self._updated_at = None - self._uri = None - - super(ForgeModuleInfo, self).__init__( - appname=appname, verbose=verbose, version=version, base_dir=base_dir, - initialized=False, name=name, vendor=vendor, full_name=full_name - ) - - if initialized is not None: - self.initialized = initialized - - # ------------------------------------------------------------------------- - @property - def created_at(self): - """Creation date of this forge module.""" - return self._created_at - - @created_at.setter - def created_at(self, value): - if value is None: - self._created_at = None - return - if isinstance(value, datetime.datetime): - self._created_at = value - return - v = str(value).strip() - if value == '': - self._created_at = None - return - self._created_at = parse_forge_date(v) - - # ------------------------------------------------------------------------- - @property - def deprecated_at(self): - """Date of deprecation of this forge module.""" - return self._deprecated_at - - @deprecated_at.setter - def deprecated_at(self, value): - if value is None: - self._deprecated_at = None - return - if isinstance(value, datetime.datetime): - self._deprecated_at = value - return - v = str(value).strip() - if value == '': - self._deprecated_at = None - return - self._deprecated_at = parse_forge_date(v) - - # ------------------------------------------------------------------------- - @property - def deprecated_for(self): - """The reason of deprecation of this forge module.""" - return self._deprecated_for - - @deprecated_for.setter - def deprecated_for(self, value): - if value is None: - self._deprecated_for = None - return - v = str(value).strip() - if value == '': - self._deprecated_for = None - return - self._deprecated_for = v - - # ------------------------------------------------------------------------- - @property - def downloads(self): - """The number of downloads of this module.""" - return self._downloads - - @downloads.setter - def downloads(self, value): - if value is None: - self._downloads = None - return - v = int(value) - if v < 0: - msg = _( - "The number of {w!r} must be greater or equal to zero " - "(Given: {v}).").format(w='downloads', v=value) - raise ValueError(msg) - self._downloads = v - - # ------------------------------------------------------------------------- - @property - def endorsement(self): - """The endorsement of this forge module..""" - return self._endorsement - - @endorsement.setter - def endorsement(self, value): - if value is None: - self._endorsement = None - return - v = str(value).strip() - if value == '': - self._endorsement = None - return - self._endorsement = v - - # ------------------------------------------------------------------------- - @property - def feedback_score(self): - """The number of feedback_scores of this module.""" - return self._feedback_score - - @feedback_score.setter - def feedback_score(self, value): - if value is None: - self._feedback_score = None - return - v = int(value) - if v < 0: - msg = _( - "The number of {w!r} must be greater or equal to zero " - "(Given: {v}).").format(w='feedback_score', v=value) - raise ValueError(msg) - self._feedback_score = v - - # ------------------------------------------------------------------------- - @property - def homepage_url(self): - """The homepage URL of this forge module.""" - return self._homepage_url - - @homepage_url.setter - def homepage_url(self, value): - if value is None: - self._homepage_url = None - return - v = str(value).strip() - if value == '': - self._homepage_url = None - return - self._homepage_url = v - - # ------------------------------------------------------------------------- - @property - def issues_url(self): - """The issues URL of this forge module.""" - return self._issues_url - - @issues_url.setter - def issues_url(self, value): - if value is None: - self._issues_url = None - return - v = str(value).strip() - if value == '': - self._issues_url = None - return - self._issues_url = v - - # ------------------------------------------------------------------------- - @property - def module_group(self): - """The module group of this forge module.""" - return self._module_group - - @module_group.setter - def module_group(self, value): - if value is None: - self._module_group = None - return - v = str(value).strip() - if value == '': - self._module_group = None - return - self._module_group = v - - # ------------------------------------------------------------------------- - @property - def slug(self): - """The slug of this forge module.""" - return self._slug - - @slug.setter - def slug(self, value): - if value is None: - self._slug = None - return - v = str(value).strip() - if value == '': - self._slug = None - return - self._slug = v - - # ------------------------------------------------------------------------- - @property - def superseded_by(self): - """The name of the superseding module this forge module.""" - return self._superseded_by - - @superseded_by.setter - def superseded_by(self, value): - if value is None: - self._superseded_by = None - return - if isinstance(value, collections.Mapping): - self._superseded_by = copy.copy(value) - return - - v = str(value).strip() - if value == '': - self._superseded_by = None - return - self._superseded_by = v - - # ------------------------------------------------------------------------- - @property - def supported(self): - """Is this forge module supported by Puppetlabs?.""" - return self._supported - - @supported.setter - def supported(self, value): - if value is None: - self._supported = None - return - self._supported = to_bool(value) - - # ------------------------------------------------------------------------- - @property - def updated_at(self): - """Last update date of this forge module.""" - return self._updated_at - - @updated_at.setter - def updated_at(self, value): - if value is None: - self._updated_at = None - return - if isinstance(value, datetime.datetime): - self._updated_at = value - return - v = str(value).strip() - if value == '': - self._updated_at = None - return - self._updated_at = parse_forge_date(v) - - # ------------------------------------------------------------------------- - @property - def uri(self): - """The URI of this forge module.""" - return self._uri - - @uri.setter - def uri(self, value): - if value is None: - self._uri = None - return - v = str(value).strip() - if value == '': - self._uri = None - return - self._uri = v - - # ------------------------------------------------------------------------- - def as_dict(self, short=True): - - res = super(ForgeModuleInfo, self).as_dict(short=short) - - res['created_at'] = self.created_at - res['deprecated_at'] = self.deprecated_at - res['deprecated_for'] = self.deprecated_for - res['downloads'] = self.downloads - res['endorsement'] = self.endorsement - res['feedback_score'] = self.feedback_score - res['homepage_url'] = self.homepage_url - res['issues_url'] = self.issues_url - res['module_group'] = self.module_group - res['slug'] = self.slug - res['superseded_by'] = self.superseded_by - res['supported'] = self.supported - res['updated_at'] = self.updated_at - res['uri'] = self.uri - - return res - - # ------------------------------------------------------------------------- - def to_data(self): - """Returning a dict, which can be used to re-instantiate this module info.""" - - res = {} - - res['deprecated_for'] = self.deprecated_for - res['downloads'] = self.downloads - res['endorsement'] = self.endorsement - res['feedback_score'] = self.feedback_score - res['homepage_url'] = self.homepage_url - res['issues_url'] = self.issues_url - res['module_group'] = self.module_group - res['name'] = self.name - res['slug'] = self.slug - res['superseded_by'] = self.superseded_by - res['supported'] = self.supported - res['uri'] = self.uri - - res['created_at'] = None - if self.created_at: - res['created_at'] = self.created_at.strftime('%Y-%m-%d %H:%M:%S %z') - - res['deprecated_at'] = None - if self.deprecated_at: - res['deprecated_at'] = self.deprecated_at.strftime('%Y-%m-%d %H:%M:%S %z') - - res['updated_at'] = None - if self.updated_at: - res['updated_at'] = self.updated_at.strftime('%Y-%m-%d %H:%M:%S %z') - - res['releases'] = [] - for release in self.releases: - res['releases'].append(release.to_data()) - - res['current_release'] = None - if self.current_release: - res['current_release'] = self.current_release.to_data() - - res['owner'] = None - if self.owner: - res['owner'] = self.owner.to_data() - - return res - - # ------------------------------------------------------------------------- - @classmethod - def get_from_forge( - cls, full_name, forge_uri, http_timeout=30, - appname=None, verbose=0, base_dir=None): - - url = "{url}/{name}".format(url=forge_uri, name=full_name) - module_info = None - - LOG.info(_("Trying to get module {m!r} from Puppet forge {u!r} ...").format( - m=full_name, u=url)) - - session = requests.Session() - with warnings.catch_warnings(record=True) as w: - warnings.simplefilter("always") - response = session.request('GET', url, timeout=http_timeout) - if w: - warn_class = w[-1].category.__name__ - warn_msg = '{}: {}'.format( - warn_class, w[-1].message) - if warn_class == 'SubjectAltNameWarning': - LOG.debug(warn_msg) - else: - LOG.warn(warn_msg) - - LOG.debug("Got status code: {}.".format(response.status_code)) - if not response.ok: - LOG.debug("Did not found module {} on Puppet forge.".format(full_name)) - return None - - if not response.text: - LOG.warn(_("No output for URL {!r}.").format(url)) - return None - if verbose > 3: - msg = "Output:\n{}".format(response.text) - LOG.debug(msg) - - try: - module_info = cls( - appname=appname, verbose=verbose, base_dir=base_dir, - full_name=full_name, - ) - except ForgeModuleInfoError as e: - LOG.warn("{c}: {e}".format(c=e.__class__.__name__, e=e)) - return None - - js_info = response.json() - if verbose > 4: - LOG.debug("Performing forge data:\n" + pp(js_info)) - - module_info.releases = ModuleReleaseList( - appname=appname, verbose=verbose, base_dir=base_dir) - - for prop_name in ( - 'created_at', 'deprecated_at', 'deprecated_for', 'downloads', 'endorsement', - 'feedback_score', 'homepage_url', 'issues_url', 'module_group', 'slug', - 'superseded_by', 'supported', 'updated_at', 'uri'): - if prop_name in js_info and js_info[prop_name]: - setattr(module_info, prop_name, js_info[prop_name]) - - if 'current_release' in js_info and js_info['current_release']: - module_info.current_release = CurrentModuleReleaseInfo.get_from_forge( - js_info['current_release'], appname=appname, verbose=verbose, base_dir=base_dir) - - if 'releases' in js_info: - for rel in js_info['releases']: - release = ModuleReleaseInfo.get_from_forge( - rel, appname=appname, verbose=verbose, base_dir=base_dir) - if release: - module_info.releases.append(release) - - module_info.releases.initialized = True - - if 'owner' in js_info and js_info['owner']: - module_info.owner = ForgeOwnerInfo.from_data( - js_info['owner'], appname=appname, verbose=verbose, base_dir=base_dir) - - if module_info.superseded_by: - subst = module_info.superseded_by - if verbose > 2: - LOG.debug("Superseded info:\n" + pp(subst)) - if 'slug' in subst: - subst = subst['slug'] - LOG.warning(_( - "Module {c!r} is deprecated at Puppet forge and should be substituted " - "by module {n!r}.").format(c=module_info.slug, n=subst)) - - return module_info - - # ------------------------------------------------------------------------- - @classmethod - def from_data(cls, data, appname=None, verbose=0, base_dir=None): - - if verbose > 3: - LOG.debug(_( - "Trying to instantiate a {}-object from:").format( - cls.__name__) + '\n' + pp(data)) - - if 'slug' not in data: - msg = _("Did not found {!r}-definition in data for forge module:").format( - 'slug') + '\n' + pp(data) - LOG.error(msg) - return None - - full_name = data['slug'].strip() - if full_name == '': - msg = _("Found empty {!r}-definition in data for forge module:").format( - 'slug') + '\n' + pp(data) - LOG.error(msg) - return None - - module_info = cls(appname=appname, verbose=verbose, base_dir=base_dir, full_name=full_name) - - module_info.releases = ModuleReleaseList( - appname=appname, verbose=verbose, base_dir=base_dir) - - for prop_name in ( - 'created_at', 'deprecated_at', 'deprecated_for', 'downloads', 'endorsement', - 'feedback_score', 'homepage_url', 'issues_url', 'module_group', 'slug', - 'superseded_by', 'supported', 'updated_at', 'uri'): - if prop_name in data and data[prop_name]: - setattr(module_info, prop_name, data[prop_name]) - - if 'current_release' in data and data['current_release']: - module_info.current_release = CurrentModuleReleaseInfo.get_from_forge( - data['current_release'], appname=appname, verbose=verbose, base_dir=base_dir) - - if 'releases' in data: - for rel in data['releases']: - release = ModuleReleaseInfo.get_from_forge( - rel, appname=appname, verbose=verbose, base_dir=base_dir) - if release: - module_info.releases.append(release) - - module_info.releases.initialized = True - - if 'owner' in data and data['owner']: - module_info.owner = ForgeOwnerInfo.from_data( - data['owner'], appname=appname, verbose=verbose, base_dir=base_dir) - - return module_info - - -# ============================================================================= -class ForgeModulesList(collections.MutableSequence, FbBaseObject): - - msg_no_release = _("Invalid type {t!r} as an item of a {c}, only {o} objects are allowed.") - - # ------------------------------------------------------------------------- - def __init__( - self, appname=None, verbose=0, version=__version__, base_dir=None, - initialized=None, *modules): - - self._list = [] - - super(ForgeModulesList, self).__init__( - appname=appname, verbose=verbose, version=version, base_dir=base_dir, initialized=False) - - for mod in modules: - self.append(mod) - - # ------------------------------------------------------------------------- - def as_dict(self, short=True): - - res = super(ForgeModulesList, self).as_dict(short=short) - - res['list'] = [] - for mod in self: - res['list'].append(mod.as_dict(short=short)) - - return res - - # ------------------------------------------------------------------------- - def to_data(self): - """Returning a list, which can be used to re-instantiate this module info.""" - - res = [] - for mod in self: - res.append(mod.to_data()) - - return res - - # ------------------------------------------------------------------------- - def index(self, mod, *args): - - i = None - j = None - - if len(args) > 0: - if len(args) > 2: - raise TypeError(_("{m} takes at most {max} arguments ({n} given).").format( - m='index()', max=3, n=len(args) + 1)) - i = int(args[0]) - if len(args) > 1: - j = int(args[1]) - - index = 0 - if i is not None: - start = i - if i < 0: - start = len(self._list) + i - wrap = False - end = len(self._list) - if j is not None: - if j < 0: - end = len(self._list) + j - if end < index: - wrap = True - else: - end = j - for index in list(range(len(self._list))): - item = self._list[index] - if index < start: - continue - if index >= end and not wrap: - break - if item == mod: - return index - - if wrap: - for index in list(range(len(self._list))): - item = self._list[index] - if index >= end: - break - if item == mod: - return index - - msg = _("Module {!r} is not in modules list.").format(mod.full_name) - raise ValueError(msg) - - # ------------------------------------------------------------------------- - def __contains__(self, mod): - - if not isinstance(mod, ForgeModuleInfo): - raise TypeError(self.msg_no_release.format( - t=mod.__class__.__name__, c=self.__class__.__name__, o='ForgeModuleInfo')) - - if not self._list: - return False - - for item in self._list: - if item == mod: - return True - - return False - - # ------------------------------------------------------------------------- - def count(self, mod): - - if not isinstance(mod, ForgeModuleInfo): - raise TypeError(self.msg_no_release.format( - t=mod.__class__.__name__, c=self.__class__.__name__, o='ForgeModuleInfo')) - - if not self._list: - return 0 - - num = 0 - for item in self._list: - if item == mod: - num += 1 - return num - - # ------------------------------------------------------------------------- - def __len__(self): - return len(self._list) - - # ------------------------------------------------------------------------- - def __getitem__(self, key): - return self._list.__getitem__(key) - - # ------------------------------------------------------------------------- - def __reversed__(self): - - return reversed(self._list) - - # ------------------------------------------------------------------------- - def __setitem__(self, key, mod): - - if not isinstance(mod, ForgeModuleInfo): - raise TypeError(self.msg_no_release.format( - t=mod.__class__.__name__, c=self.__class__.__name__, o='ForgeModuleInfo')) - - self._list.__setitem__(key, mod) - - # ------------------------------------------------------------------------- - def __delitem__(self, key): - - del self._list[key] - - # ------------------------------------------------------------------------- - def append(self, mod): - - if not isinstance(mod, ForgeModuleInfo): - raise TypeError(self.msg_no_release.format( - t=mod.__class__.__name__, c=self.__class__.__name__, o='ForgeModuleInfo')) - - self._list.append(mod) - - # ------------------------------------------------------------------------- - def insert(self, index, mod): - - if not isinstance(mod, ForgeModuleInfo): - raise TypeError(self.msg_no_release.format( - t=mod.__class__.__name__, c=self.__class__.__name__, o='ForgeModuleInfo')) - - self._list.insert(index, mod) - - # ------------------------------------------------------------------------- - def __copy__(self): - - new_list = self.__class__() - for mod in self._list: - new_list.append(copy.copy(mod)) - return new_list - - -# ============================================================================= -if __name__ == "__main__": - - pass - -# ============================================================================= - -# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 list diff --git a/lib/webhooks/get_forge_modules.py b/lib/webhooks/get_forge_modules.py index 89d639e..e8181f6 100644 --- a/lib/webhooks/get_forge_modules.py +++ b/lib/webhooks/get_forge_modules.py @@ -38,8 +38,6 @@ from . import __version__ from .base_app import BaseHookError, BaseHookApp -#from .forge_module_info import ForgeModuleInfo, ForgeModulesList - from .module_info import ModuleInfo from .module_list import ModuleInfoDict -- 2.39.5