self.check_data_dir()
self.init_puppet_environments()
self.collect_local_modules()
+ self.get_forge_information()
+
+ if self.verbose > 2:
+ LOG.debug("Found modules:\n{}".format(pp(self.modules)))
print()
dt = datetime.datetime.now().isoformat(' ')
self.read_puppetfile(env)
self.read_metadata_files(env)
- self.get_upstream_versions()
-
- if self.verbose > 2:
- LOG.debug("Found modules:\n{}".format(pp(self.modules)))
-
# -------------------------------------------------------------------------
- def get_upstream_versions(self):
+ def get_forge_information(self):
print()
msg = "Collecting module information from upstream ..."
self.print_out(msg)
for full_name in self.modules.keys():
- self.get_upstream_version(full_name)
+ self.get_forge_module_info(full_name)
if not self.verbose:
- if self.modules[full_name].version_upstream:
+ if self.modules[full_name].forge_avail:
print('.', end='', flush=True)
else:
print('~', end='', flush=True)
print()
# -------------------------------------------------------------------------
- def get_upstream_version(self, full_name):
+ def get_forge_module_info(self, full_name):
module_info = self.modules[full_name]
- url = "{url}/{name}".format(url=self.forge_uri, name=full_name)
-
- LOG.debug(
- "Trying to get current version of module {} from Puppet forge.".format(full_name))
- if self.verbose > 1:
- LOG.debug("URL to request: {}".format(url))
-
- session = requests.Session()
- with warnings.catch_warnings(record=True) as w:
- warnings.simplefilter("always")
- response = session.request('GET', url, timeout=self.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
- if not response.text:
- LOG.warn("No output for URL {!r}".format(url))
- return None
- if self.verbose > 2:
- msg = "Output:\n{}".format(response.text)
- LOG.debug(msg)
+ module_info_forge = ModuleInfo.get_from_forge(
+ full_name, forge_uri=self.forge_uri, http_timeout=self.http_timeout,
+ appname=self.appname, verbose=self.verbose, base_dir=self.base_dir, )
- version = None
-
- js_info = response.json()
- if 'current_release' in js_info and js_info['current_release']:
- if 'version' in js_info['current_release']:
- version = js_info['current_release']['version']
- else:
- msg = "Did not found version of current_release of module {}.".format(full_name)
- LOG.warn(msg)
+ if module_info_forge:
+ module_info.merge_in(module_info_forge)
else:
- msg = "Did not found current_release of module {}.".format(full_name)
- LOG.warn(msg)
-
- LOG.debug("Current version of module {name} is {version}.".format(
- name=full_name, version=version))
-
- if version:
- module_info._version_upstream = version
+ module_info.forge_avail = False
# -------------------------------------------------------------------------
def read_metadata_files(self, env):
import os
import logging
import re
+import copy
+import warnings
# Third party modules
import six
+import requests
# Own modules
-from .common import pp, to_str
+from .common import pp, to_str, to_bool, is_sequence
from .obj import BaseObjectError
from .obj import BaseObject
-__version__ = '0.3.0'
+__version__ = '0.4.0'
LOG = logging.getLogger(__name__)
self._name = None
self._vendor = None
self._full_name_orig = None
- self._version_upstream = None
+ self._forge_version = None
self.local_versions = {}
self.expected_versions = {}
self.repo = None
+ self._forge_avail = None
+ self.forge_releases = None
+ self._forge_homepage_url = None
+ self._forge_source = None
super(ModuleInfo, self).__init__(
appname=appname, verbose=verbose, version=version,
# -------------------------------------------------------------------------
@property
- def version_upstream(self):
+ def forge_version(self):
"""The current version number of the module from upstream."""
- return self._version_upstream
+ return self._forge_version
+
+ @forge_version.setter
+ def forge_version(self, value):
+ if value is None:
+ self._forge_version = None
+ return
+ val = to_str(value).strip()
+ if val == '':
+ val = None
+ self._forge_version = val
+
+ # -------------------------------------------------------------------------
+ @property
+ def forge_homepage_url(self):
+ """The URL of the module home in Puppet forge."""
+ return self._forge_homepage_url
+
+ @forge_homepage_url.setter
+ def forge_homepage_url(self, value):
+ if value is None:
+ self._forge_homepage_url = None
+ return
+ val = to_str(value).strip()
+ if val == '':
+ val = None
+ self._forge_homepage_url = val
+
+ # -------------------------------------------------------------------------
+ @property
+ def forge_source(self):
+ """The source (URL) of the module from Puppet forge."""
+ return self._forge_source
+
+ @forge_source.setter
+ def forge_source(self, value):
+ if value is None:
+ self._forge_source = None
+ return
+ val = to_str(value).strip()
+ if val == '':
+ val = None
+ self._forge_source = val
# -------------------------------------------------------------------------
@property
else:
self._full_name_orig = None
+ # -------------------------------------------------------------------------
+ @property
+ def forge_avail(self):
+ """Is this module available per upstream? None -> not checked."""
+ return self._forge_avail
+
+ @forge_avail.setter
+ def forge_avail(self, value):
+ if value is None:
+ self._forge_avail = None
+ return
+ self._forge_avail = to_bool(value)
+
# -------------------------------------------------------------------------
def __repr__(self):
return str(self)
res['vendor'] = self.vendor
res['full_name'] = self.full_name
res['full_name_orig'] = self.full_name_orig
- res['version_upstream'] = self.version_upstream
+ res['forge_version'] = self.forge_version
+ res['forge_avail'] = self.forge_avail
+ res['forge_homepage_url'] = self.forge_homepage_url
+ res['forge_source'] = self.forge_source
return res
"Parameter {p!r} is not of class ModuleInfoTypeError, but of class {c} "
" instead.").format(p=other, c=other.__class__.__name__))
- if other.version_upstream and not self.version_upstream:
- self._version_upstream = other.version_upstream
+ if other.forge_avail is not None and self.forge_avail is None:
+ self.forge_avail = other.forge_avail
+
+ if other.forge_version and not self.forge_version:
+ self.forge_version = other.forge_version
+
+ if other.forge_homepage_url and not self.forge_homepage_url:
+ self.forge_homepage_url = other.forge_homepage_url
+
+ if other.forge_source and not self.forge_source:
+ self.forge_source = other.forge_source
+
+ if other.forge_releases is not None and self.forge_releases is None:
+ self.forge_releases = copy.copy(other.forge_releases)
if other.repo and not self.repo:
self.repo = other.repo
return defs
+ # -------------------------------------------------------------------------
+ @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.debug("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 ModuleInfoError as e:
+ LOG.warn("{c}: {e}".format(c=e.__class__.__name__, e=e))
+ return None
+
+ version = None
+ source = None
+ module_info.forge_avail = True
+
+ js_info = response.json()
+ if 'current_release' in js_info and js_info['current_release']:
+ if 'version' in js_info['current_release']:
+ version = js_info['current_release']['version']
+ else:
+ msg = "Did not found version of current_release of module {}.".format(full_name)
+ LOG.warn(msg)
+ if 'metadata' in js_info['current_release'] and js_info['current_release']['metadata']:
+ if ('source' in js_info['current_release']['metadata'] and
+ js_info['current_release']['metadata']['source']):
+ source = str(js_info['current_release']['metadata']['source']).strip()
+ if not source:
+ LOG.warn("Did not found source information of module {}.".format(full_name))
+ else:
+ msg = "Did not found current_release of module {}.".format(full_name)
+ LOG.warn(msg)
+
+ LOG.debug("Current version of module {name} is {version}.".format(
+ name=full_name, version=version))
+
+ if version:
+ module_info.forge_version = version
+ if source:
+ module_info.forge_source = source
+
+ if 'homepage_url' in js_info and js_info['homepage_url']:
+ module_info.forge_homepage_url = js_info['homepage_url']
+
+ module_info.forge_releases = []
+ if 'releases' in js_info and is_sequence(js_info['releases']):
+ for release in js_info['releases']:
+ if 'version' in release and release['version']:
+ module_info.forge_releases.append(release['version'].strip())
+
+ return module_info
+
# =============================================================================
if __name__ == "__main__":
projects / pixelpark / puppetmaster-webhooks.git / commitdiff