+++ /dev/null
-#!/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
projects / pixelpark / puppetmaster-webhooks.git / commitdiff