Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use host_vars method in molecule

Best Python code snippet using molecule_python

validate.py

Source:validate.py

1from ansible.plugins.action import ActionBase2from distutils.version import LooseVersion3from ansible.module_utils.six import string_types4from ansible.errors import AnsibleUndefinedVariable5try:6    from __main__ import display7except ImportError:8    from ansible.utils.display import Display9    display = Display()10try:11    import notario12except ImportError:13    msg = "The python-notario library is missing. Please install it on the node you are running ceph-ansible to continue."14    display.error(msg)15    raise SystemExit(msg)16if LooseVersion(notario.__version__) < LooseVersion("0.0.13"):17    msg = "The python-notario libary has an incompatible version. Version >= 0.0.13 is needed, current version: %s" % notario.__version__18    display.error(msg)19    raise SystemExit(msg)20from notario.exceptions import Invalid21from notario.validators import types, chainable, iterables22from notario.decorators import optional23from notario.store import store as notario_store24CEPH_RELEASES = ['jewel', 'kraken', 'luminous', 'mimic', 'nautilus']25class ActionModule(ActionBase):26    def run(self, tmp=None, task_vars=None):27        # we must use vars, since task_vars will have un-processed variables28        host_vars = self.expand_all_jinja2_templates(task_vars['vars'])29        host = host_vars['ansible_hostname']30        mode = self._task.args.get('mode', 'permissive')31        self._supports_check_mode = False # XXX ?32        self._supports_async = True33        result = {}34        result['_ansible_verbose_always'] = True35        try:36            notario_store["groups"] = host_vars["groups"]37            notario_store["containerized_deployment"] = host_vars["containerized_deployment"]38            notario.validate(host_vars, install_options, defined_keys=True)39            if host_vars["ceph_origin"] == "repository" and not host_vars["containerized_deployment"]:40                notario.validate(host_vars, ceph_origin_repository, defined_keys=True)41                if host_vars["ceph_repository"] == "community":42                    notario.validate(host_vars, ceph_repository_community, defined_keys=True)43                if host_vars["ceph_repository"] == "rhcs":44                    notario.validate(host_vars, ceph_repository_rhcs, defined_keys=True)45                if host_vars["ceph_repository"] == "dev":46                    notario.validate(host_vars, ceph_repository_dev, defined_keys=True)47            # store these values because one must be defined and the validation method48            # will need access to all three through the store49            notario_store["monitor_address"] = host_vars.get("monitor_address", None)50            notario_store["monitor_address_block"] = host_vars.get("monitor_address_block", None)51            notario_store["monitor_interface"] = host_vars.get("monitor_interface", None)52            if host_vars["mon_group_name"] in host_vars["group_names"]:53                notario.validate(host_vars, monitor_options, defined_keys=True)54            notario_store["radosgw_address"] = host_vars.get("radosgw_address", None)55            notario_store["radosgw_address_block"] = host_vars.get("radosgw_address_block", None)56            notario_store["radosgw_interface"] = host_vars.get("radosgw_interface", None)57            if host_vars["rgw_group_name"] in host_vars["group_names"]:58                notario.validate(host_vars, rados_options, defined_keys=True)59            # validate osd scenario setup60            if host_vars["osd_group_name"] in host_vars["group_names"]:61                notario.validate(host_vars, osd_options, defined_keys=True)62                notario_store['osd_objectstore'] = host_vars["osd_objectstore"]63                if host_vars["osd_scenario"] == "collocated":64                    if not host_vars.get("osd_auto_discovery", False):65                        notario.validate(host_vars, collocated_osd_scenario, defined_keys=True)66                if host_vars["osd_scenario"] == "non-collocated":67                    notario.validate(host_vars, non_collocated_osd_scenario, defined_keys=True)68                if host_vars["osd_scenario"] == "lvm":69                    if not host_vars.get("osd_auto_discovery", False):70                        if host_vars.get("devices"):71                            notario.validate(host_vars, lvm_batch_scenario, defined_keys=True)72                        elif notario_store['osd_objectstore'] == 'filestore':73                            notario.validate(host_vars, lvm_filestore_scenario, defined_keys=True)74                        elif notario_store['osd_objectstore'] == 'bluestore':75                            notario.validate(host_vars, lvm_bluestore_scenario, defined_keys=True)76        except Invalid as error:77            display.vvv("Notario Failure: %s" % str(error))78            msg = ""79            if error.path:80                msg = "[{}] Validation failed for variable: {}".format(host, error.path[0])81                display.error(msg)82                reason = "[{}] Reason: {}".format(host, error.reason)83            else:84                reason = "[{}] Reason: {}".format(host, str(error))85            given = ""86            try:87                if "schema is missing" not in error.message:88                    for i in range(0, len(error.path)):89                        if i == 0:90                            given = "[{}] Given value for {}".format(91                                    host, error.path[0])92                        else:93                            given = given + ": {}".format(error.path[i])94                    if given:95                        display.error(given)96                else:97                    given = ""98                    reason = "[{}] Reason: {}".format(host, error.message)99            except KeyError:100                given = ""101            display.error(reason)102            result['failed'] = mode == 'strict'103            result['msg'] = "\n".join([s for s in (msg, reason, given) if len(s) > 0])104            result['stderr_lines'] = result['msg'].split('\n')105        return result106    def expand_all_jinja2_templates(self, variables):107        for k, v in variables.items():108            try:109                if self._templar.is_template(v):110                    variables[k] = self.expand_jinja2_template(v)111            except AnsibleUndefinedVariable as e:112                variables[k] = u"VARIABLE IS NOT DEFINED!"113        return variables114    def expand_jinja2_template(self, var):115        expanded_var = self._templar.template(var, convert_bare=True,116                                              fail_on_undefined=True)117        if expanded_var == var:118            if not isinstance(expanded_var, string_types):119                raise AnsibleUndefinedVariable120            expanded_var = self._templar.template("{{%s}}" % expanded_var,121                                                  convert_bare=True,122                                                  fail_on_undefined=True)123        return expanded_var124# Schemas125def osd_objectstore_choices(value):126    assert value in ['bluestore', 'filestore'], "osd_objectstore must be either 'bluestore' or 'filestore'"127def ceph_origin_choices(value):128    if not notario_store["containerized_deployment"]:129        assert value in ['repository', 'distro', 'local'], "ceph_origin must be either 'repository', 'distro' or 'local'"130def ceph_repository_choices(value):131    assert value in ['community', 'rhcs', 'dev', 'custom'], "ceph_repository must be either 'community', 'rhcs', 'dev', or 'custom'"132def ceph_repository_type_choices(value):133    assert value in ['cdn', 'iso'], "ceph_repository_type must be either 'cdn' or 'iso'"134def validate_monitor_options(value):135    """136    Either monitor_address, monitor_address_block or monitor_interface must137    be defined.138    """139    monitor_address_given = notario_store["monitor_address"] != "0.0.0.0"140    monitor_address_block_given = notario_store["monitor_address_block"] != "subnet"141    monitor_interface_given = notario_store["monitor_interface"] != "interface"142    msg = "Either monitor_address, monitor_address_block or monitor_interface must be provided"143    assert any([monitor_address_given, monitor_address_block_given, monitor_interface_given]), msg144def validate_dmcrypt_bool_value(value):145    assert value in ["true", True, "false", False], "dmcrypt can be set to true/True or false/False (default)"146def validate_osd_auto_discovery_bool_value(value):147    assert value in ["true", True, "false", False], "osd_auto_discovery can be set to true/True or false/False (default)"148def validate_osd_scenarios(value):149    assert value in ["collocated", "non-collocated", "lvm"], "osd_scenario must be set to 'collocated', 'non-collocated' or 'lvm'"150def validate_objectstore(value):151    assert value in ["filestore", "bluestore"], "objectstore must be set to 'filestore' or 'bluestore'"152def validate_ceph_stable_release(value):153    assert value in CEPH_RELEASES, "ceph_stable_release must be set to one of the following: %s" % ", ".join(CEPH_RELEASES)154def validate_rados_options(value):155    """156    Either radosgw_interface, radosgw_address or radosgw_address_block must157    be defined.158    """159    radosgw_address_given = notario_store["radosgw_address"] != "0.0.0.0"160    radosgw_address_block_given = notario_store["radosgw_address_block"] != "subnet"161    radosgw_interface_given = notario_store["radosgw_interface"] != "interface"162    msg = "Either radosgw_address, radosgw_address_block or radosgw_interface must be provided"163    assert any([radosgw_address_given, radosgw_address_block_given, radosgw_interface_given]), msg164install_options = (165    ("ceph_origin", ceph_origin_choices),166    ("containerized_deployment", types.boolean),167    ('osd_objectstore', osd_objectstore_choices),168)169ceph_origin_repository = ("ceph_repository", ceph_repository_choices)170ceph_repository_community = (171    ("ceph_mirror", types.string),172    ("ceph_stable_key", types.string),173    ("ceph_stable_release", validate_ceph_stable_release),174    ("ceph_stable_repo", types.string),175)176ceph_repository_rhcs = (177    ("ceph_repository_type", ceph_repository_type_choices),178    ("ceph_rhcs_version", chainable.AnyIn(types.string, types.integer)),179)180ceph_repository_dev = (181    ("ceph_dev_branch", types.string),182    ("ceph_dev_sha1", types.string),183)184monitor_options = (185    ("cluster_network", types.string),186    ("fsid", types.string),187    ("monitor_address", validate_monitor_options),188    ("monitor_address_block", validate_monitor_options),189    ("monitor_interface", validate_monitor_options),190    ("public_network", types.string),191)192rados_options = (193    ("radosgw_address", validate_rados_options),194    ("radosgw_address_block", validate_rados_options),195    ("radosgw_interface", validate_rados_options),196)197osd_options = (198    (optional("dmcrypt"), validate_dmcrypt_bool_value),199    (optional("osd_auto_discovery"), validate_osd_auto_discovery_bool_value),200    ("osd_scenario", validate_osd_scenarios),201)202collocated_osd_scenario = ("devices", iterables.AllItems(types.string))203non_collocated_osd_scenario = (204    (optional("bluestore_wal_devices"), iterables.AllItems(types.string)),205    (optional("dedicated_devices"), iterables.AllItems(types.string)),206    ("devices", iterables.AllItems(types.string)),207)208lvm_batch_scenario = ("devices", iterables.AllItems(types.string))209lvm_filestore_scenario = ("lvm_volumes", iterables.AllItems((210    (optional('crush_device_class'), types.string),211    ('data', types.string),212    (optional('data_vg'), types.string),213    ('journal', types.string),214    (optional('journal_vg'), types.string),215)))216lvm_bluestore_scenario = ("lvm_volumes", iterables.AllItems((217    (optional('crush_device_class'), types.string),218    ('data', types.string),219    (optional('data_vg'), types.string),220    (optional('db'), types.string),221    (optional('db_vg'), types.string),222    (optional('wal'), types.string),223    (optional('wal_vg'), types.string),...

inventory.py

Source:inventory.py

1#!/usr/bin/python32import ipaddress3import json4import subprocess5from collections import defaultdict6from pathlib import Path7class InventoryBuilder:8    def __init__(self):9        self.vms = json.loads(10            subprocess.run(11                ["vml", "show", "--format-json", "--running"],12                stdout=subprocess.PIPE,13            ).stdout14        )15        self.names = [vm["name"] for vm in self.vms]16        self.inventory = {17            "all": {18                "hosts": self.names,19                "vars": {"ansible_python_interpreter": "/usr/bin/python3"},20            },21            "_meta": {"hostvars": {}},22        }23    def hostvars(self):24        for vm in self.vms:25            host_vars = {}26            if host := vm.get("ssh_host"):27                host_vars["ansible_host"] = host28            if port := vm.get("ssh_port"):29                host_vars["ansible_port"] = port30            if user := vm.get("ssh_user"):31                host_vars["ansible_user"] = user32            if key := vm.get("ssh_key"):33                host_vars["ansible_ssh_private_key_file"] = key34            if options := vm.get("ssh_options"):35                host_vars["ansible_ssh_common_args"] = options36            if cidr := vm.get("network_address"):37                host_vars["network_cidr"] = cidr38                address = str(ipaddress.ip_interface(cidr).ip)39                host_vars["network_address"] = address40            if host_vars:41                self.inventory["_meta"]["hostvars"][vm["name"]] = host_vars42    def groups(self):43        groups = defaultdict(set)44        for name in self.names:45            for group in [p.as_posix() for p in Path(name).parents][:-1]:46                groups[group].add(name)47        if groups:48            self.inventory["all"]["children"] = list(groups.keys())49            for group, hosts in groups.items():50                self.inventory[group] = {"hosts": list(hosts)}51def main():52    ib = InventoryBuilder()53    ib.hostvars()54    ib.groups()55    print(json.dumps(ib.inventory))56if __name__ == '__main__':...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.