]> Frank Brehm's Git Trees - pixelpark/puppet-tools.git/commitdiff
Adding module dpx_puppettools.module_metadata for classes ModuleDependency, ModuleOsS...
authorFrank Brehm <frank.brehm@pixelpark.com>
Thu, 9 Feb 2023 16:00:06 +0000 (17:00 +0100)
committerFrank Brehm <frank.brehm@pixelpark.com>
Thu, 9 Feb 2023 16:00:06 +0000 (17:00 +0100)
lib/dpx_puppettools/module_metadata.py [new file with mode: 0644]
lib/dpx_puppettools/puppetfile.py

diff --git a/lib/dpx_puppettools/module_metadata.py b/lib/dpx_puppettools/module_metadata.py
new file mode 100644 (file)
index 0000000..cfa1c0b
--- /dev/null
@@ -0,0 +1,554 @@
+#!/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
index c8811b1384ba980adfdf6cdf74bb3fed0966a5f9..1e31c4920770bb18a0cdb02abc0431a88732bee1 100644 (file)
@@ -302,7 +302,7 @@ class Puppetfile(FbBaseObject):
         self.modules = ModuleInfoDict(
             appname=self.appname, verbose=self.verbose, base_dir=self.base_dir)
 
-        LOG.debug("Reading {!r} ...".format(self.filename))
+        LOG.debug(_("Reading {!r} ...").format(self.filename))
         line_nr = 0
         prev_line = ''