--- /dev/null
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@author: Frank Brehm
+@contact: frank.brehm@pixelpark.com
+@copyright: © 2023 by Frank Brehm, Digitas Pixelpark GmbH, Berlin
+@summary: A module for encapsulating all information from a metadata.json file
+ of a Puppet module
+"""
+from __future__ import absolute_import
+
+# Standard modules
+import logging
+import re
+import copy
+import json
+import os
+
+from pathlib import Path
+
+# Third party modules
+import six
+from fb_tools.common import pp, to_str, is_sequence
+from fb_tools.obj import FbGenericBaseObject, FbBaseObject
+
+# Own modules
+from .errors import PuppetToolsError
+
+from .xlate import XLATOR
+
+__version__ = '0.6.0'
+
+LOG = logging.getLogger(__name__)
+
+_ = XLATOR.gettext
+ngettext = XLATOR.ngettext
+
+
+# =============================================================================
+class MetadataInitError(PuppetToolsError):
+ """Base exception class for all exceptions in this module."""
+ pass
+
+
+# =============================================================================
+class MetadataFileError(MetadataInitError):
+ """Exception class for all exceptions on reading or writing a metadata.json file.."""
+ pass
+
+
+# =============================================================================
+class ModuleDependency(FbGenericBaseObject):
+ """Class for encapsulating a dependency to another Puppet module."""
+
+ # -------------------------------------------------------------------------
+ def __init__(self, name, requirement):
+
+ self.name = str(name).strip().lower()
+ self.requirement = str(requirement).strip()
+
+ # -------------------------------------------------------------------------
+ @classmethod
+ def from_data(cls, data):
+
+ for prop in ('name', 'version_requirement'):
+ if prop not in data:
+ msg = _("{} not included in JSON data.").format(prop)
+ raise MetadataInitError(msg)
+
+ dep = cls(name=data['name'], requirement=data['version_requirement'])
+ return dep
+
+ # -------------------------------------------------------------------------
+ def __repr__(self):
+ """Typecasting into a string for reproduction."""
+ out = "<%s(" % (self.__class__.__name__)
+
+ fields = []
+ fields.append("name={!r}".format(self.name))
+ fields.append("requirement={!r}".format(self.requirement))
+
+ out += ", ".join(fields) + ")>"
+ return out
+
+ # -------------------------------------------------------------------------
+ def as_dict(self, short=True):
+
+ ret = super(ModuleDependency, self).as_dict(self, short=short)
+
+ ret['name'] = self.name
+ ret['requirement'] = self.requirement
+
+ return ret
+
+ # -------------------------------------------------------------------------
+ def to_data(self):
+
+ ret = {
+ 'name': self.name,
+ 'version_requirement': self.requirement,
+ }
+ return ret
+
+ # -------------------------------------------------------------------------
+ def __copy__(self):
+
+ new = self.__class__(name=self.name, requirement=self.requirement)
+ return new
+
+
+# =============================================================================
+class ModuleOsSupport(FbGenericBaseObject):
+ """Class for encapsulating the support of the module to a particular operating system."""
+
+ # -------------------------------------------------------------------------
+ def __init__(self, name, releases):
+
+ self.name = str(name).strip()
+ self.releases = copy.copy(releases)
+
+ # -------------------------------------------------------------------------
+ @classmethod
+ def from_data(cls, data):
+
+ if 'operatingsystem' not in data:
+ msg = _("{} not included in JSON data.").format('operatingsystem')
+ raise MetadataInitError(msg)
+
+ releases = []
+ if 'operatingsystemrelease' in data:
+ if not is_sequence(data['operatingsystemrelease']):
+ msg = _("Invalid item {} - not a list.").format('operatingsystemrelease')
+ raise MetadataInitError(msg)
+ for item in data['operatingsystemrelease']:
+ releases.append(str(item).strip())
+
+ os_supp = cls(name=data['operatingsystem'], releases=releases)
+ return os_supp
+
+ # -------------------------------------------------------------------------
+ def __repr__(self):
+ """Typecasting into a string for reproduction."""
+ out = "<%s(" % (self.__class__.__name__)
+
+ fields = []
+ fields.append("name={!r}".format(self.name))
+ fields.append("releases={!r}".format(self.releases))
+
+ out += ", ".join(fields) + ")>"
+ return out
+
+ # -------------------------------------------------------------------------
+ def as_dict(self):
+
+ ret = super(ModuleDependency, self).as_dict(self, short=short)
+
+ ret['name'] = self.name
+ ret['releases'] = copy.copy(self.releases)
+
+ return ret
+
+ # -------------------------------------------------------------------------
+ def to_data(self):
+
+ ret = {'operatingsystem': self.name}
+ if self.releases:
+ ret['operatingsystemrelease'] = copy.copy(self.releases)
+ return ret
+
+ # -------------------------------------------------------------------------
+ def __copy__(self):
+
+ new = self.__class__(name=self.name, releases=copy.copy(self.releases))
+ return new
+
+
+# =============================================================================
+class ModuleMetadata(FbBaseObject):
+ """Class for encapsulating information about a Puppet module."""
+
+ re_author_from_name = re.compile(r'^([^-]+)-')
+
+ open_args = {}
+ if six.PY3:
+ open_args = {
+ 'encoding': 'utf-8',
+ 'errors': 'surrogateescape',
+ }
+
+ # -------------------------------------------------------------------------
+ def __init__(
+ self, appname=None, verbose=0, version=__version__, base_dir=None,
+ initialized=None):
+
+ self._name = None
+ self._version = None
+ self._author = None
+ self._license = None
+ self._summary = None
+ self._source = None
+ self.dependencies = []
+ self.requirements = []
+ self._project_page = None
+ self._issues_url = None
+ self.operatingsystem_support = []
+ self.tags = []
+
+ super(ModuleMetadata, self).__init__(
+ appname=appname, verbose=verbose, version=version,
+ base_dir=base_dir, initialized=False,
+ )
+
+ # -------------------------------------------------------------------------
+ @property
+ def name(self):
+ """The name of the module."""
+ return self._name
+
+ @name.setter
+ def name(self, value):
+ if value is None:
+ self._name = None
+ return
+ self._name = to_str(value).strip().lower()
+
+ # -------------------------------------------------------------------------
+ @property
+ def version(self):
+ """The version of the module."""
+ return self._version
+
+ @version.setter
+ def version(self, value):
+ if value is None:
+ self._version = None
+ return
+ self._version = to_str(value).strip()
+
+ # -------------------------------------------------------------------------
+ @property
+ def author(self):
+ """The author of the module."""
+ if not self._author:
+ match = self.re_author_from_name.match(self.name)
+ if match:
+ return match.group(1)
+ return self._author
+
+ @author.setter
+ def author(self, value):
+ if value is None:
+ self._author = None
+ return
+ self._author = to_str(value).strip()
+
+ # -------------------------------------------------------------------------
+ @property
+ def license(self):
+ """The license of the module."""
+ return self._license
+
+ @license.setter
+ def license(self, value):
+ if value is None:
+ self._license = None
+ return
+ self._license = to_str(value).strip()
+
+ # -------------------------------------------------------------------------
+ @property
+ def summary(self):
+ """The summary of the module."""
+ return self._summary
+
+ @summary.setter
+ def summary(self, value):
+ if value is None:
+ self._summary = None
+ return
+ self._summary = to_str(value).strip()
+
+ # -------------------------------------------------------------------------
+ @property
+ def source(self):
+ """The source of the module."""
+ return self._source
+
+ @source.setter
+ def source(self, value):
+ if value is None:
+ self._source = None
+ return
+ self._source = to_str(value).strip()
+
+ # -------------------------------------------------------------------------
+ @property
+ def project_page(self):
+ """The project page of the module."""
+ return self._project_page
+
+ @project_page.setter
+ def project_page(self, value):
+ if value is None:
+ self._project_page = None
+ return
+ self._project_page = to_str(value).strip()
+
+ # -------------------------------------------------------------------------
+ @property
+ def issues_url(self):
+ """The link to the module's issue tracker."""
+ return self._issues_url
+
+ @issues_url.setter
+ def issues_url(self, value):
+ if value is None:
+ self._issues_url = None
+ return
+ self._issues_url = to_str(value).strip()
+
+ # -------------------------------------------------------------------------
+ def as_dict(self, short=True):
+ """
+ Transforms the elements of the object into a dict
+
+ @return: structure as dict
+ @rtype: dict
+ """
+
+ res = super(ModuleMetadata, self).as_dict(short=short)
+
+ res['name'] = self.name
+ res['version'] = self.version
+ res['author'] = self.author
+ res['license'] = self.license
+ res['summary'] = self.summary
+ res['source'] = self.source
+ res['dependencies'] = []
+ for dep in self.dependencies:
+ res['dependencies'].append(dep.as_dict())
+ res['requirements'] = []
+ for req in self.requirements:
+ res['requirements'].append(req.as_dict())
+ res['project_page'] = self.project_page
+ res['issues_url'] = self.issues_url
+ res['operatingsystem_support'] = []
+ for supp in self.operatingsystem_support:
+ res['operatingsystem_support'].append(supp.as_dict())
+
+ return res
+
+ # -------------------------------------------------------------------------
+ @classmethod
+ def from_file(cls, filename, appname=None, verbose=0, base_dir=None):
+ """Tries to read the given metadata.json file."""
+
+ file_path = Path(filename)
+
+ if not file_path.exists():
+ msg = _("Metadata file {!r} does not exists.").format(str(file_path))
+ raise MetadataFileError(msg)
+
+ if not file_path.is_file():
+ msg = _("Metadata file {!r} is not a regular file.").format(str(file_path))
+ raise MetadataFileError(msg)
+
+ if not os.access(str(file_path), os.R_OK)):
+ msg = _("Metadata file {!r} is readable.").format(str(file_path))
+ raise MetadataFileError(msg)
+
+ LOG.debug(_("Reading {!r} ...").format(str(file_path)))
+
+ json_data = None
+ with file_path.open('r', **cls.open_args) as fh:
+ try:
+ json_data = json.load(fh)
+ except ValueError as e:
+ msg = _("Could not interprete {f!r} as a regular JSON file: {e}").format(
+ f=str(file_path), e=e)
+ raise MetadataFileError(msg)
+
+ return cls.from_json_data(
+ json_data, appname=appname, verbose=verbose, base_dir=base_dir)
+
+ # -------------------------------------------------------------------------
+ @classmethod
+ def from_json_data(cls, json_data, appname=None, verbose=0, base_dir=None):
+ """ Tries to instantiate a new ModuleMetadata object from data
+ read from a metadata.json file.
+ See https://puppet.com/docs/puppet/5.2/modules_metadata.html for syntax.
+ """
+
+ if verbose > 3:
+ LOG.debug("Trying to create a {c} object from:\n{d}".format(
+ c=cls.__name__, d=pp(json_data)))
+
+ required_props = {
+ 'name': 'Name',
+ 'version': 'Version',
+ 'license': 'License',
+ 'summary': 'Summary',
+ 'source': 'Source',
+ 'dependencies': 'Dependencies',
+ }
+
+ for prop in required_props.keys():
+ desc = required_props[prop]
+ if prop not in json_data:
+ msg = _("{} not included in JSON data.").format(desc)
+ raise MetadataInitError(msg)
+
+ if not is_sequence(json_data['dependencies']):
+ msg = _("Invalid item {} - not a list.").format('dependencies')
+ raise MetadataInitError(msg)
+
+ if 'requirements' in json_data:
+ if not is_sequence(json_data['requirements']):
+ msg = _("Invalid item {} - not a list.").format('requirements')
+ raise MetadataInitError(msg)
+
+ if 'operatingsystem_support' in json_data:
+ if not is_sequence(json_data['operatingsystem_support']):
+ msg = _("Invalid item {} - not a list.").format('operatingsystem_support')
+ raise MetadataInitError(msg)
+
+ metadata = cls(appname=appname, verbose=verbose, base_dir=base_dir)
+ metadata._apply_json_data(json_data)
+
+ return metadata
+
+ # -------------------------------------------------------------------------
+ def _apply_json_data(self, json_data):
+
+ self.name = json_data['name']
+ self.version = json_data['version']
+ if 'author' in json_data:
+ self.author = json_data['author']
+ else:
+ self.author = self.author
+ self.license = json_data['license']
+ self.summary = json_data['summary']
+ self.source = json_data['source']
+ self.dependencies = []
+ for item in json_data['dependencies']:
+ dep = ModuleDependency.from_data(item)
+ self.dependencies.append(dep)
+ self.requirements = []
+ if 'requirements' in json_data:
+ for item in json_data['requirements']:
+ req = ModuleDependency.from_data(item)
+ self.requirements.append(req)
+ if 'project_page' in json_data:
+ self.project_page = json_data['project_page']
+ if 'issues_url' in json_data:
+ self.issues_url = json_data['issues_url']
+ self.operatingsystem_support = []
+ if 'operatingsystem_support' in json_data:
+ for item in json_data['operatingsystem_support']:
+ supp = ModuleOsSupport.from_data(item)
+ self.operatingsystem_support.append(supp)
+ self.tags = []
+ if 'tags' in json_data:
+ for tag in json_data['tags']:
+ tg = str(tag).strip()
+ if tg:
+ self.tags.append(tg)
+
+ self.initialized = True
+ if self.verbose > 3:
+ LOG.debug("ModuleMetadata:\n{}".format(pp(self.as_dict())))
+
+ # -------------------------------------------------------------------------
+ def __copy__(self):
+
+ new = self.__class__(appname=self.appname, verbose=self.verbose, base_dir=self.base_dir)
+
+ new.name = self.name
+ new.version = self.version
+ new.author = self.author
+ new.license = self.license
+ new.summary = self.summary
+ new.source = self.source
+ for dep in self.dependencies:
+ new.dependencies.append(copy.copy(dep))
+ for req in self.requirements:
+ new.requirements.append(copy.copy(req))
+ new.project_page = self.project_page
+ new.issues_url = self.issues_url
+ for supp in self.operatingsystem_support:
+ new.operatingsystem_support.append(copy.copy(supp))
+ for tag in self.tags:
+ new.tags.append(tag)
+
+ new.initialized = True
+
+ return new
+
+ # -------------------------------------------------------------------------
+ def to_data(self):
+
+ data = {}
+ data['name'] = self.name
+ data['version'] = self.version
+ data['author'] = self.author
+ data['license'] = self.license
+ data['summary'] = self.summary
+ data['source'] = self.source
+ data['dependencies'] = []
+ for dep in self.dependencies:
+ data['dependencies'].append(dep.to_data())
+ if self.requirements:
+ data['requirements'] = []
+ for req in self.requirements:
+ data['requirements'].append(req.to_data())
+ if self.project_page:
+ data['project_page'] = self.project_page
+ if self.issues_url:
+ data['issues_url'] = self.issues_url
+ if self.operatingsystem_support:
+ data['operatingsystem_support'] = []
+ for supp in self.operatingsystem_support:
+ data['operatingsystem_support'].append(supp.to_data())
+ if self.tags:
+ data['tags'] = []
+ for tag in self.tags:
+ data['tags'].append(tag)
+
+ if self.verbose > 4:
+ LOG.debug("ModuleMetadata:\n{}".format(pp(data)))
+
+ return data
+
+ # -------------------------------------------------------------------------
+ def to_json(self, indent=None):
+
+ data = self.to_data()
+ ret = json.dumps(data, indent=indent, sort_keys=True)
+ if self.verbose > 4:
+ LOG.debug("ModuleMetadata as JSON:\n{}".format(ret))
+ return ret
+
+ # -------------------------------------------------------------------------
+ def __str__(self):
+ return self.to_json(indent=4)
+
+
+# =============================================================================
+if __name__ == "__main__":
+
+ pass
+
+# =============================================================================
+
+# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 list