# Third party modules
import pytz
-# Own modules
-
-from fb_tools.obj import FbBaseObject
-
from fb_tools.config import BaseConfiguration
-from fb_tools.common import to_bool, RE_FQDN, is_sequence, pp
+from fb_tools.common import to_bool, RE_FQDN, pp
from fb_pdnstools import DEFAULT_PORT as DEFAULT_PDNS_API_PORT
-from fb_pdnstools import DEFAULT_TIMEOUT as DEFAULT_PDNS_API_TIMEOUT # noqa
-from fb_pdnstools import DEFAULT_API_PREFIX as DEFAULT_PDNS_API_PREFIX # noqa
+from fb_pdnstools import DEFAULT_TIMEOUT as DEFAULT_PDNS_API_TIMEOUT # noqa: F401
+from fb_pdnstools import DEFAULT_API_PREFIX as DEFAULT_PDNS_API_PREFIX
from fb_pdnstools import DEFAULT_USE_HTTPS as DEFAULT_PDNS_API_USE_HTTPS
+# Own modules
+
from .errors import CrTfConfigError
+from .vs_config import VsphereConfig
+
from .xlate import XLATOR
-__version__ = '1.8.0'
+__version__ = '1.8.1'
LOG = logging.getLogger(__name__)
_ = XLATOR.gettext
ngettext = XLATOR.ngettext
-# =============================================================================
-class VsphereConfig(FbBaseObject):
- """Class for encapsulation of config data of a connection to a VSPhere center."""
-
- default_host = 'vcs01.ppbrln.internal'
- default_port = 443
- default_dc = 'vmcc'
- default_cluster = 'vmcc-l105-01'
- default_min_root_size_gb = 20.0
- default_guest_id = 'centos8_64Guest'
-
- default_template_name = 'rhel9-template'
-
- # -------------------------------------------------------------------------
- def __init__(
- self, appname=None, verbose=0, version=__version__, base_dir=None, name=None,
- host=None, port=None, user=None, password=None, dc=None, cluster=None,
- template_name=None, min_root_size_gb=None, excluded_ds=None, guest_id=None,
- rhsm_user=None, rhsm_password=None, initialized=False):
-
- self._name = None
- self._host = self.default_host
- self._port = self.default_port
- self._user = None
- self._password = None
- self._dc = self.default_dc
- self._cluster = self.default_cluster
- self._template_name = self.default_template_name
- self._min_root_size_gb = self.default_min_root_size_gb
- self._guest_id = self.default_guest_id
- self.excluded_ds = []
- self.used_templates = []
-
- super(VsphereConfig, self).__init__(
- appname=appname, verbose=verbose, version=version,
- base_dir=base_dir, initialized=False,
- )
-
- if name is not None:
- self.name = name
- if host is not None:
- self.host = host
- if port is not None:
- self.port = port
- if user is not None:
- self.user = user
- if password is not None:
- self.password = password
- if dc is not None:
- self.dc = dc
- if cluster is not None:
- self.cluster = cluster
- if template_name is not None:
- self.template_name = template_name
- if min_root_size_gb is not None:
- self.min_root_size_gb = min_root_size_gb
- if guest_id is not None:
- self.guest_id = guest_id
-
- if excluded_ds:
- if is_sequence(excluded_ds):
- for ds in excluded_ds:
- self.excluded_ds.append(str(ds))
- else:
- self.excluded_ds.append(str(excluded_ds))
-
- if initialized:
- self.initialized = True
-
- # -----------------------------------------------------------
- @property
- def name(self):
- """The name of the VSphere."""
- return self._name
-
- @name.setter
- def name(self, value):
- if value is None:
- self._name = None
- return
- val = str(value).strip().lower()
- if val == '':
- self._name = None
- else:
- self._name = val
-
- # -----------------------------------------------------------
- @property
- def host(self):
- """The host name or address of the VSphere server."""
- return self._host
-
- @host.setter
- def host(self, value):
- if value is None:
- self._host = self.default_host
- return
- val = str(value).strip().lower()
- if val == '':
- self._host = None
- else:
- self._host = val
-
- # -----------------------------------------------------------
- @property
- def port(self):
- """The TCP port number, where the API is listening on the VSphere server."""
- return self._port
-
- @port.setter
- def port(self, value):
- if value is None:
- self._port = self.default_port
- return
- val = self.default_port
- try:
- val = int(value)
- if val < 1:
- msg = _("a port may not be less than 1: {}.").format(val)
- raise CrTfConfigError(msg)
- max_val = (2 ** 16) - 1
- if val > max_val:
- msg = _("a port may not be greater than {m}: {v}.").format(
- m=max_val, v=val)
- raise CrTfConfigError(msg)
- except ValueError as e:
- msg = _("Wrong port number {v!r}: {e}").format(v=value, e=e)
- LOG.error(msg)
- else:
- self._port = val
-
- # -----------------------------------------------------------
- @property
- def user(self):
- """The user name to connect to the VSphere server."""
- return self._user
-
- @user.setter
- def user(self, value):
- if value is None:
- self._user = None
- return
- val = str(value).strip()
- if val == '':
- self._user = None
- else:
- self._user = val
-
- # -----------------------------------------------------------
- @property
- def password(self):
- """The password of the VSphere user."""
- return self._password
-
- @password.setter
- def password(self, value):
- if value is None:
- self._password = None
- return
- val = str(value)
- if val == '':
- self._password = None
- else:
- self._password = val
-
- # -----------------------------------------------------------
- @property
- def dc(self):
- """The name of the datacenter in VSphere."""
- return self._dc
-
- @dc.setter
- def dc(self, value):
- if value is None:
- self._dc = self.default_dc
- return
- val = str(value).strip()
- if val == '':
- self._dc = self.default_dc
- else:
- self._dc = val
-
- # -----------------------------------------------------------
- @property
- def cluster(self):
- """The name of the default cluster in VSphere."""
- return self._cluster
-
- @cluster.setter
- def cluster(self, value):
- if value is None:
- self._cluster = self.default_cluster
- return
- val = str(value).strip()
- if val == '':
- self._cluster = self.default_cluster
- else:
- self._cluster = val
-
- # -----------------------------------------------------------
- @property
- def template_name(self):
- """The name of the default cluster in VSphere."""
- return self._template_name
-
- @template_name.setter
- def template_name(self, value):
- if value is None:
- self._template_name = self.default_template_name
- return
- val = str(value).strip().lower()
- if val == '':
- self._template_name = self.default_template_name
- else:
- self._template_name = val
-
- # -----------------------------------------------------------
- @property
- def min_root_size_gb(self):
- """The minimum size of a root disk in GiB."""
- return self._min_root_size_gb
-
- @min_root_size_gb.setter
- def min_root_size_gb(self, value):
- if value is None:
- self._min_root_size_gb = self.default_min_root_size_gb
- return
- val = self.default_min_root_size_gb
- try:
- val = float(value)
- if val < 10:
- msg = _("may not be less than 10: {:0.1f}.").format(val)
- raise CrTfConfigError(msg)
- max_val = 4 * 1024
- if val > max_val:
- msg = _("may not be greater than {m}: {v:0.1f}.").format(
- m=max_val, v=val)
- raise CrTfConfigError(msg)
- except ValueError as e:
- msg = _("Wrong minimum root size in GiB {v!r}: {e}").format(v=value, e=e)
- LOG.error(msg)
- else:
- self._min_root_size_gb = val
-
- # -----------------------------------------------------------
- @property
- def guest_id(self):
- """The Id of the Guest OS in VSphere."""
- return self._guest_id
-
- @guest_id.setter
- def guest_id(self, value):
- if value is None:
- self._guest_id = self.default_guest_id
- return
- val = str(value).strip()
- if val == '':
- self._guest_id = self.default_guest_id
- else:
- self._guest_id = val
-
- # -------------------------------------------------------------------------
- def as_dict(self, short=True, show_secrets=False):
- """
- Transforms the elements of the object into a dict
-
- @param short: don't include local properties in resulting dict.
- @type short: bool
-
- @return: structure as dict
- @rtype: dict
- """
-
- res = super(VsphereConfig, self).as_dict(short=short)
-
- res['name'] = self.name
- res['host'] = self.host
- res['port'] = self.port
- res['user'] = self.user
- res['dc'] = self.dc
- res['cluster'] = self.cluster
- res['template_name'] = self.template_name
- res['min_root_size_gb'] = self.min_root_size_gb
- res['guest_id'] = self.guest_id
-
- if self.password:
- if show_secrets or self.verbose > 4:
- res['password'] = self.password
- else:
- res['password'] = '*******'
- else:
- res['password'] = None
-
- return res
-
- # -------------------------------------------------------------------------
- def __copy__(self):
-
- vsphere = self.__class__(
- appname=self.appname, verbose=self.verbose, base_dir=self.base_dir,
- initialized=self.initialized, host=self.host, port=self.port, user=self.user,
- password=self.password, dc=self.dc, cluster=self.cluster,
- template_name=self.template_name, excluded_ds=self.excluded_ds,
- min_root_size_gb=self.min_root_size_gb, guest_id=self.guest_id,
- )
- return vsphere
-
- # -------------------------------------------------------------------------
- def __eq__(self, other):
-
- if self.verbose > 4:
- LOG.debug(_("Comparing {} objects ...").format(self.__class__.__name__))
-
- if not isinstance(other, VsphereConfig):
- return False
-
- if self.name != other.name:
- return False
- if self.host != other.host:
- return False
- if self.port != other.port:
- return False
- if self.user != other.user:
- return False
- if self.password != other.password:
- return False
- if self.dc != other.dc:
- return False
- if self.cluster != other.cluster:
- return False
- if self.template_name != other.template_name:
- return False
- if self.min_root_size_gb != other.min_root_size_gb:
- return False
- if self.guest_id != other.guest_id:
- return False
- if self.excluded_ds != other.excluded_ds:
- return False
-
- return True
-
- # -------------------------------------------------------------------------
- def is_valid(self, raise_on_error=False):
-
- name = '<{}>'.format(_('unknown'))
- if self.name:
- name = self.name
- if self.verbose > 1:
- LOG.debug(_("Checking validity of {o}-object {n!r} ...").format(
- o=self.__class__.__name__, n=name))
-
- error_lst = []
-
- mandatory_attribs = ('name', 'host', 'dc', 'cluster')
- requested_attribs = ('user', 'password')
-
- for attrib in mandatory_attribs:
- cur_val = getattr(self, attrib, None)
- if not cur_val:
- msg = _("Attribute {a!r} of the {o}-object {n!r} is not set.").format(
- a=attrib, o=self.__class__.__name__, n=name)
- error_lst.append(msg)
- if not raise_on_error:
- LOG.error(msg)
-
- if error_lst:
- if raise_on_error:
- nr = len(error_lst)
- msg = ngettext(
- 'Found an error in VSPhere configuration',
- 'Found {} errors in VSPhere configuration', nr)
- msg = msg.format(nr) + '\n * ' + '\n * '.join(error_lst)
- raise CrTfConfigError(msg)
- return False
-
- for attrib in requested_attribs:
- cur_val = getattr(self, attrib, None)
- if not cur_val:
- msg = _(
- "Attribute {a!r} of the {o}-object {n!r} is not set, it "
- "will be requestet during this script and on starting terraform.").format(
- a=attrib, o=self.__class__.__name__, n=name)
- LOG.warn(msg)
-
- return True
-
# =============================================================================
class CrTfConfiguration(BaseConfiguration):
"""
elif self.re_template.search(key) and value.strip():
params['template_name'] = value.strip()
elif self.re_excl_ds.search(key) and value.strip():
- datastores = re_split_ds.split(value.strip())
+ datastores = self.re_split_ds.split(value.strip())
params['excluded_ds'] = datastores
elif self.re_min_root_size.search(key) and value.strip():
params['min_root_size_gb'] = value
elif self.re_template.search(key) and value.strip():
vsphere.template_name = value.strip()
elif self.re_excl_ds.search(key) and value.strip():
- datastores = re_split_ds.split(value.strip())
+ datastores = self.re_split_ds.split(value.strip())
vsphere.datastores = datastores
elif self.re_min_root_size.search(key) and value.strip():
vsphere.min_root_size_gb = value.strip()
match = re_puppet_env.match(env)
if not match:
msg = _("Invalid puppet environment {env!r} found in {k!r}.").format(
- env=env, k=key)
+ env=env, k=key)
LOG.warn(msg)
continue
sign = match.group(1)
--- /dev/null
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@author: Frank Brehm
+@contact: frank.brehm@pixelpark.com
+@copyright: © 2021 by Frank Brehm, Berlin
+@summary: A module for providing a configuration
+"""
+from __future__ import absolute_import
+
+# Standard module
+import logging
+
+# Third party modules
+from fb_tools.obj import FbBaseObject
+
+from fb_tools.common import is_sequence
+
+# Own modules
+from .errors import CrTfConfigError
+
+from .xlate import XLATOR
+
+__version__ = '1.8.1'
+LOG = logging.getLogger(__name__)
+
+_ = XLATOR.gettext
+ngettext = XLATOR.ngettext
+
+
+# =============================================================================
+class VsphereConfig(FbBaseObject):
+ """Class for encapsulation of config data of a connection to a VSPhere center."""
+
+ default_host = 'vcs01.ppbrln.internal'
+ default_port = 443
+ default_dc = 'vmcc'
+ default_cluster = 'vmcc-l105-01'
+ default_min_root_size_gb = 20.0
+ default_guest_id = 'centos8_64Guest'
+
+ default_template_name = 'rhel9-template'
+
+ # -------------------------------------------------------------------------
+ def __init__(
+ self, appname=None, verbose=0, version=__version__, base_dir=None, name=None,
+ host=None, port=None, user=None, password=None, dc=None, cluster=None,
+ template_name=None, min_root_size_gb=None, excluded_ds=None, guest_id=None,
+ rhsm_user=None, rhsm_password=None, initialized=False):
+
+ self._name = None
+ self._host = self.default_host
+ self._port = self.default_port
+ self._user = None
+ self._password = None
+ self._dc = self.default_dc
+ self._cluster = self.default_cluster
+ self._template_name = self.default_template_name
+ self._min_root_size_gb = self.default_min_root_size_gb
+ self._guest_id = self.default_guest_id
+ self.excluded_ds = []
+ self.used_templates = []
+
+ super(VsphereConfig, self).__init__(
+ appname=appname, verbose=verbose, version=version,
+ base_dir=base_dir, initialized=False,
+ )
+
+ if name is not None:
+ self.name = name
+ if host is not None:
+ self.host = host
+ if port is not None:
+ self.port = port
+ if user is not None:
+ self.user = user
+ if password is not None:
+ self.password = password
+ if dc is not None:
+ self.dc = dc
+ if cluster is not None:
+ self.cluster = cluster
+ if template_name is not None:
+ self.template_name = template_name
+ if min_root_size_gb is not None:
+ self.min_root_size_gb = min_root_size_gb
+ if guest_id is not None:
+ self.guest_id = guest_id
+
+ if excluded_ds:
+ if is_sequence(excluded_ds):
+ for ds in excluded_ds:
+ self.excluded_ds.append(str(ds))
+ else:
+ self.excluded_ds.append(str(excluded_ds))
+
+ if initialized:
+ self.initialized = True
+
+ # -----------------------------------------------------------
+ @property
+ def name(self):
+ """The name of the VSphere."""
+ return self._name
+
+ @name.setter
+ def name(self, value):
+ if value is None:
+ self._name = None
+ return
+ val = str(value).strip().lower()
+ if val == '':
+ self._name = None
+ else:
+ self._name = val
+
+ # -----------------------------------------------------------
+ @property
+ def host(self):
+ """The host name or address of the VSphere server."""
+ return self._host
+
+ @host.setter
+ def host(self, value):
+ if value is None:
+ self._host = self.default_host
+ return
+ val = str(value).strip().lower()
+ if val == '':
+ self._host = None
+ else:
+ self._host = val
+
+ # -----------------------------------------------------------
+ @property
+ def port(self):
+ """The TCP port number, where the API is listening on the VSphere server."""
+ return self._port
+
+ @port.setter
+ def port(self, value):
+ if value is None:
+ self._port = self.default_port
+ return
+ val = self.default_port
+ try:
+ val = int(value)
+ if val < 1:
+ msg = _("a port may not be less than 1: {}.").format(val)
+ raise CrTfConfigError(msg)
+ max_val = (2 ** 16) - 1
+ if val > max_val:
+ msg = _("a port may not be greater than {m}: {v}.").format(
+ m=max_val, v=val)
+ raise CrTfConfigError(msg)
+ except ValueError as e:
+ msg = _("Wrong port number {v!r}: {e}").format(v=value, e=e)
+ LOG.error(msg)
+ else:
+ self._port = val
+
+ # -----------------------------------------------------------
+ @property
+ def user(self):
+ """The user name to connect to the VSphere server."""
+ return self._user
+
+ @user.setter
+ def user(self, value):
+ if value is None:
+ self._user = None
+ return
+ val = str(value).strip()
+ if val == '':
+ self._user = None
+ else:
+ self._user = val
+
+ # -----------------------------------------------------------
+ @property
+ def password(self):
+ """The password of the VSphere user."""
+ return self._password
+
+ @password.setter
+ def password(self, value):
+ if value is None:
+ self._password = None
+ return
+ val = str(value)
+ if val == '':
+ self._password = None
+ else:
+ self._password = val
+
+ # -----------------------------------------------------------
+ @property
+ def dc(self):
+ """The name of the datacenter in VSphere."""
+ return self._dc
+
+ @dc.setter
+ def dc(self, value):
+ if value is None:
+ self._dc = self.default_dc
+ return
+ val = str(value).strip()
+ if val == '':
+ self._dc = self.default_dc
+ else:
+ self._dc = val
+
+ # -----------------------------------------------------------
+ @property
+ def cluster(self):
+ """The name of the default cluster in VSphere."""
+ return self._cluster
+
+ @cluster.setter
+ def cluster(self, value):
+ if value is None:
+ self._cluster = self.default_cluster
+ return
+ val = str(value).strip()
+ if val == '':
+ self._cluster = self.default_cluster
+ else:
+ self._cluster = val
+
+ # -----------------------------------------------------------
+ @property
+ def template_name(self):
+ """The name of the default cluster in VSphere."""
+ return self._template_name
+
+ @template_name.setter
+ def template_name(self, value):
+ if value is None:
+ self._template_name = self.default_template_name
+ return
+ val = str(value).strip().lower()
+ if val == '':
+ self._template_name = self.default_template_name
+ else:
+ self._template_name = val
+
+ # -----------------------------------------------------------
+ @property
+ def min_root_size_gb(self):
+ """The minimum size of a root disk in GiB."""
+ return self._min_root_size_gb
+
+ @min_root_size_gb.setter
+ def min_root_size_gb(self, value):
+ if value is None:
+ self._min_root_size_gb = self.default_min_root_size_gb
+ return
+ val = self.default_min_root_size_gb
+ try:
+ val = float(value)
+ if val < 10:
+ msg = _("may not be less than 10: {:0.1f}.").format(val)
+ raise CrTfConfigError(msg)
+ max_val = 4 * 1024
+ if val > max_val:
+ msg = _("may not be greater than {m}: {v:0.1f}.").format(
+ m=max_val, v=val)
+ raise CrTfConfigError(msg)
+ except ValueError as e:
+ msg = _("Wrong minimum root size in GiB {v!r}: {e}").format(v=value, e=e)
+ LOG.error(msg)
+ else:
+ self._min_root_size_gb = val
+
+ # -----------------------------------------------------------
+ @property
+ def guest_id(self):
+ """The Id of the Guest OS in VSphere."""
+ return self._guest_id
+
+ @guest_id.setter
+ def guest_id(self, value):
+ if value is None:
+ self._guest_id = self.default_guest_id
+ return
+ val = str(value).strip()
+ if val == '':
+ self._guest_id = self.default_guest_id
+ else:
+ self._guest_id = val
+
+ # -------------------------------------------------------------------------
+ def as_dict(self, short=True, show_secrets=False):
+ """
+ Transforms the elements of the object into a dict
+
+ @param short: don't include local properties in resulting dict.
+ @type short: bool
+
+ @return: structure as dict
+ @rtype: dict
+ """
+
+ res = super(VsphereConfig, self).as_dict(short=short)
+
+ res['name'] = self.name
+ res['host'] = self.host
+ res['port'] = self.port
+ res['user'] = self.user
+ res['dc'] = self.dc
+ res['cluster'] = self.cluster
+ res['template_name'] = self.template_name
+ res['min_root_size_gb'] = self.min_root_size_gb
+ res['guest_id'] = self.guest_id
+
+ if self.password:
+ if show_secrets or self.verbose > 4:
+ res['password'] = self.password
+ else:
+ res['password'] = '*******'
+ else:
+ res['password'] = None
+
+ return res
+
+ # -------------------------------------------------------------------------
+ def __copy__(self):
+
+ vsphere = self.__class__(
+ appname=self.appname, verbose=self.verbose, base_dir=self.base_dir,
+ initialized=self.initialized, host=self.host, port=self.port, user=self.user,
+ password=self.password, dc=self.dc, cluster=self.cluster,
+ template_name=self.template_name, excluded_ds=self.excluded_ds,
+ min_root_size_gb=self.min_root_size_gb, guest_id=self.guest_id,
+ )
+ return vsphere
+
+ # -------------------------------------------------------------------------
+ def __eq__(self, other):
+
+ if self.verbose > 4:
+ LOG.debug(_("Comparing {} objects ...").format(self.__class__.__name__))
+
+ if not isinstance(other, VsphereConfig):
+ return False
+
+ if self.name != other.name:
+ return False
+ if self.host != other.host:
+ return False
+ if self.port != other.port:
+ return False
+ if self.user != other.user:
+ return False
+ if self.password != other.password:
+ return False
+ if self.dc != other.dc:
+ return False
+ if self.cluster != other.cluster:
+ return False
+ if self.template_name != other.template_name:
+ return False
+ if self.min_root_size_gb != other.min_root_size_gb:
+ return False
+ if self.guest_id != other.guest_id:
+ return False
+ if self.excluded_ds != other.excluded_ds:
+ return False
+
+ return True
+
+ # -------------------------------------------------------------------------
+ def is_valid(self, raise_on_error=False):
+
+ name = '<{}>'.format(_('unknown'))
+ if self.name:
+ name = self.name
+ if self.verbose > 1:
+ LOG.debug(_("Checking validity of {o}-object {n!r} ...").format(
+ o=self.__class__.__name__, n=name))
+
+ error_lst = []
+
+ mandatory_attribs = ('name', 'host', 'dc', 'cluster')
+ requested_attribs = ('user', 'password')
+
+ for attrib in mandatory_attribs:
+ cur_val = getattr(self, attrib, None)
+ if not cur_val:
+ msg = _("Attribute {a!r} of the {o}-object {n!r} is not set.").format(
+ a=attrib, o=self.__class__.__name__, n=name)
+ error_lst.append(msg)
+ if not raise_on_error:
+ LOG.error(msg)
+
+ if error_lst:
+ if raise_on_error:
+ nr = len(error_lst)
+ msg = ngettext(
+ 'Found an error in VSPhere configuration',
+ 'Found {} errors in VSPhere configuration', nr)
+ msg = msg.format(nr) + '\n * ' + '\n * '.join(error_lst)
+ raise CrTfConfigError(msg)
+ return False
+
+ for attrib in requested_attribs:
+ cur_val = getattr(self, attrib, None)
+ if not cur_val:
+ msg = _(
+ "Attribute {a!r} of the {o}-object {n!r} is not set, it "
+ "will be requestet during this script and on starting terraform.").format(
+ a=attrib, o=self.__class__.__name__, n=name)
+ LOG.warn(msg)
+
+ return True
+
+
+# =============================================================================
+
+if __name__ == "__main__":
+
+ pass
+
+# =============================================================================
+
+# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 list
projects / pixelpark / create-terraform.git / commitdiff