KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _vivify method in molecule

Best Python code snippet using molecule_python

ansible.py

Source:ansible.py

...355                instance-2:356                  ansible_connection: docker357        :return: str358        """359        dd = self._vivify()360        for platform in self._config.platforms.instances:361            for group in platform.get('groups', ['ungrouped']):362                instance_name = platform['name']363                connection_options = self.connection_options(instance_name)364                molecule_vars = {365                    'molecule_file':366                    "{{ lookup('env', 'MOLECULE_FILE') }}",367                    'molecule_ephemeral_directory':368                    "{{ lookup('env', 'MOLECULE_EPHEMERAL_DIRECTORY') }}",369                    'molecule_scenario_directory':370                    "{{ lookup('env', 'MOLECULE_SCENARIO_DIRECTORY') }}",371                    'molecule_yml':372                    "{{ lookup('file', molecule_file) | molecule_from_yaml }}",373                    'molecule_instance_config':374                    "{{ lookup('env', 'MOLECULE_INSTANCE_CONFIG') }}",375                }376                # All group377                dd['all']['hosts'][instance_name] = connection_options378                dd['all']['vars'] = molecule_vars379                # Named group380                dd[group]['hosts'][instance_name] = connection_options381                dd[group]['vars'] = molecule_vars382                # Ungrouped383                dd['ungrouped']['vars'] = {}384                # Children385                for child_group in platform.get('children', []):386                    dd[group]['children'][child_group]['hosts'][387                        instance_name] = connection_options388        return self._default_to_regular(dd)389    @property390    def inventory_file(self):391        return os.path.join(self._config.scenario.ephemeral_directory,392                            'ansible_inventory.yml')393    @property394    def config_file(self):395        return os.path.join(self._config.scenario.ephemeral_directory,396                            'ansible.cfg')397    @property398    @util.memoize399    def playbooks(self):400        return ansible_playbooks.AnsiblePlaybooks(self._config)401    @property402    def directory(self):403        return os.path.join(404            os.path.dirname(__file__), os.path.pardir, os.path.pardir,405            'molecule', 'provisioner', 'ansible')406    def connection_options(self, instance_name):407        d = self._config.driver.ansible_connection_options(instance_name)408        return util.merge_dicts(409            d, self._config.config['provisioner']['connection_options'])410    def check(self):411        """412        Executes `ansible-playbook` against the converge playbook with the413        ``--check`` flag and returns None.414        :return: None415        """416        pb = self._get_ansible_playbook(self.playbooks.converge)417        pb.add_cli_arg('check', True)418        pb.execute()419    def converge(self, playbook=None, **kwargs):420        """421        Executes `ansible-playbook` against the converge playbook unless422        specified otherwise and returns a string.423        :param playbook: An optional string containing an absolute path to a424         playbook.425        :param kwargs: An optional keyword arguments.426        :return: str427        """428        if playbook is None:429            pb = self._get_ansible_playbook(self.playbooks.converge, **kwargs)430        else:431            pb = self._get_ansible_playbook(playbook, **kwargs)432        return pb.execute()433    def destroy(self):434        """435        Executes `ansible-playbook` against the destroy playbook and returns436        None.437        :return: None438        """439        pb = self._get_ansible_playbook(self.playbooks.destroy)440        pb.execute()441    def side_effect(self):442        """443        Executes `ansible-playbook` against the side_effect playbook and444        returns None.445        :return: None446        """447        pb = self._get_ansible_playbook(self.playbooks.side_effect)448        pb.execute()449    def create(self):450        """451        Executes `ansible-playbook` against the create playbook and returns452        None.453        :return: None454        """455        pb = self._get_ansible_playbook(self.playbooks.create)456        pb.execute()457    def prepare(self):458        """459        Executes `ansible-playbook` against the prepare playbook and returns460        None.461        :return: None462        """463        pb = self._get_ansible_playbook(self.playbooks.prepare)464        pb.execute()465    def syntax(self):466        """467        Executes `ansible-playbook` against the converge playbook with the468        ``-syntax-check`` flag and returns None.469        :return: None470        """471        pb = self._get_ansible_playbook(self.playbooks.converge)472        pb.add_cli_arg('syntax-check', True)473        pb.execute()474    def verify(self):475        """476        Executes `ansible-playbook` against the verify playbook and returns477        None.478        :return: None479        """480        pb = self._get_ansible_playbook(self.playbooks.verify)481        pb.execute()482    def write_config(self):483        """484        Writes the provisioner's config file to disk and returns None.485        :return: None486        """487        template = util.render_template(488            self._get_config_template(), config_options=self.config_options)489        util.write_file(self.config_file, template)490    def manage_inventory(self):491        """492        Manages inventory for Ansible and returns None.493        :returns: None494        """495        self._write_inventory()496        self._remove_vars()497        if not self.links:498            self._add_or_update_vars()499        else:500            self._link_or_update_vars()501    def abs_path(self, path):502        return util.abs_path(503            os.path.join(self._config.scenario.directory, path))504    def _add_or_update_vars(self):505        """506        Creates host and/or group vars and returns None.507        :returns: None508        """509        for target in ['host_vars', 'group_vars']:510            if target == 'host_vars':511                vars_target = copy.deepcopy(self.host_vars)512                for instance_name, _ in self.host_vars.items():513                    if instance_name == 'localhost':514                        instance_key = instance_name515                    else:516                        instance_key = instance_name517                    vars_target[instance_key] = vars_target.pop(instance_name)518            elif target == 'group_vars':519                vars_target = self.group_vars520            if vars_target:521                ephemeral_directory = self._config.scenario.ephemeral_directory522                target_vars_directory = os.path.join(ephemeral_directory,523                                                     target)524                if not os.path.isdir(util.abs_path(target_vars_directory)):525                    os.mkdir(util.abs_path(target_vars_directory))526                for target in vars_target.keys():527                    target_var_content = vars_target[target]528                    path = os.path.join(529                        util.abs_path(target_vars_directory), target)530                    util.write_file(path, util.safe_dump(target_var_content))531    def _write_inventory(self):532        """533        Writes the provisioner's inventory file to disk and returns None.534        :return: None535        """536        self._verify_inventory()537        util.write_file(self.inventory_file, util.safe_dump(self.inventory))538    def _remove_vars(self):539        """540        Remove host and/or group vars and returns None.541        :returns: None542        """543        dirs = [544            os.path.join(self._config.scenario.ephemeral_directory,545                         'group_vars'),546            os.path.join(self._config.scenario.ephemeral_directory,547                         'host_vars'),548        ]549        for d in dirs:550            if os.path.islink(d):551                os.unlink(d)552            else:553                if os.path.exists(d):554                    shutil.rmtree(d)555    def _link_or_update_vars(self):556        """557        Creates or updates the symlink to group_vars and returns None.558        :returns: None559        """560        for d, source in self.links.items():561            target = os.path.join(self._config.scenario.ephemeral_directory, d)562            source = os.path.join(self._config.scenario.directory, source)563            if not os.path.exists(source):564                msg = "The source path '{}' does not exist.".format(source)565                util.sysexit_with_message(msg)566            msg = "Inventory {} linked to {}".format(source, target)567            LOG.info(msg)568            os.symlink(source, target)569    def _get_ansible_playbook(self, playbook, **kwargs):570        """571        Get an instance of AnsiblePlaybook and returns it.572        :param playbook: A string containing an absolute path to a573         provisioner's playbook.574        :param kwargs: An optional keyword arguments.575        :return: object576        """577        return ansible_playbook.AnsiblePlaybook(playbook, self._config,578                                                **kwargs)579    def _verify_inventory(self):580        """581        Verify the inventory is valid and returns None.582        :return: None583        """584        if not self.inventory:585            msg = ("Instances missing from the 'platform' "586                   "section of molecule.yml.")587            util.sysexit_with_message(msg)588    def _get_config_template(self):589        """590        Returns a config template string.591        :return: str592        """593        return """594{% for section, section_dict in config_options.items() -%}595[{{ section }}]596{% for k, v in section_dict.items() -%}597{{ k }} = {{ v }}598{% endfor -%}599{% endfor -%}600""".strip()601    def _vivify(self):602        """603        Returns an autovivification default dict.604        :return: dict605        """606        return collections.defaultdict(self._vivify)607    def _default_to_regular(self, d):608        if isinstance(d, collections.defaultdict):609            d = {k: self._default_to_regular(v) for k, v in d.items()}610        return d611    def _get_plugin_directory(self):612        return os.path.join(self.directory, 'plugins')613    def _get_libraries_directory(self):614        return util.abs_path(615            os.path.join(self._get_plugin_directory(), 'libraries'))...

vivify.py

Source:vivify.py

...240                " objects. Attribute types may not be suitable for vivification"241            )242        return setters243    @entry_exit_logging244    def _vivify(245        self,246        instances: Union[Mapping[str, str], str],247        config: Config,248        defaults: Optional[str] = DEFAULTSECT,249    ) -> Mapping[str, Any]:250        """Internal handler for vivifying objects.251        Args:252            instances (Union[Mapping[str, str], str]): Instances name type253                mapping or the name of the instances section in the supplied254                config.255            config (Config): Configuration data for attributes and optionally256                also instances.257            defaults (Optional[str], optional): Name of the default section in258                the config, if used. Defaults to DEFAULTSECT.259        Returns:260            Mapping[str, Any]: All vivified objects.261        """262        # store for vivified objects263        vivified: Mapping[str, Any] = {}264        # store for instances name and type data265        instances_config: Mapping[str, str]266        instances_is_str: bool = isinstance(instances, str)267        # get the instances data268        if instances_is_str:269            if instances in config:270                instances_config = config[instances]271            else:272                instances_config = {}273        else:274            instances_config = instances275        # extract just the attribute configuration data276        if defaults or instances_is_str:277            config = self._extract_configs(278                config=config,279                instances=instances if instances_is_str else None,280                defaults=defaults,281            )282        # create all objects and set the attributes283        vivified = {284            v: self.types[instances_config[v]]()285            for v in config286            if v in instances_config287        }288        seen: set[tuple[str, str]] = set()289        for instance, attribute, value in self._process_attributes(290            instances_config, config, vivified291        ):292            if (instance, attribute) not in seen:293                seen.add((instance, attribute))294                setattr(vivified[instance], attribute, value)295        return vivified296    @entry_exit_logging297    def vivify(298        self,299        instances: Union[Mapping[str, str], str],300        config: Config,301        defaults: Optional[str] = DEFAULTSECT,302    ) -> Mapping[str, Any]:303        """Vivify the specified objects.304        Args:305            instances (Union[Mapping[str, str], str]): Instances name type306                mapping or the name of the instances section in the supplied307                config.308            config (Config): Configuration data for attributes and optionally309                also instances.310            defaults (Optional[str], optional): Name of the default section in311                the config, if used. Defaults to DEFAULTSECT.312        Returns:313            Mapping[str, Any]: All vivified objects.314        """315        try:316            return self._vivify(317                instances=instances, config=config, defaults=defaults318            )319        except VivificationError:320            raise321        except Exception as error:322            raise VivificationError(323                "Error occurred during vivification."...

test_vivify.py

Source:test_vivify.py

...190    string: str191    integer: int192    vivifiable: TestVivifiableTuple193    reference: TestVivify194def test_vivify__vivify():195    """Test that the internal vivification function returns correct values."""196    # check for empty data197    assert (198        Vivifier(types=[])._vivify(instances={}, config={}, defaults="") == {}199    )200    assert (201        Vivifier(types=[])._vivify(202            instances="instances", config={}, defaults="DEFAULT"203        )204        == {}205    )206    # check for dict args207    vivified: Mapping[str, Any] = Vivifier(types=[TestVivify])._vivify(208        instances={"a": "TestVivify", "b": "TestVivify"},209        config={210            "a": {211                "string": "test",212                "integer": "1",213                "vivifiable": "1, 2, 3",214                "reference": "b",215            },216            "b": {217                "string": "test",218                "integer": "2",219                "vivifiable": "2",220                "reference": "a",221            },222        },223    )224    assert isinstance(vivified["a"], TestVivify)225    assert vivified["a"].string == "test"226    assert vivified["a"].integer == 1227    assert vivified["a"].vivifiable == (1, 2, 3)228    assert vivified["a"].reference == vivified["b"]229    assert isinstance(vivified["b"], TestVivify)230    assert vivified["b"].string == "test"231    assert vivified["b"].integer == 2232    assert vivified["b"].vivifiable == (2,)233    assert vivified["b"].reference == vivified["a"]234    # check for config parser style args235    config: ConfigParser = ConfigParser()236    config.read_string(237        """238        [main]239        a = TestVivify240        b = TestVivify241        [DEFAULT]242        string = test243        [a]244        integer = 1245        vivifiable = 1, 2, 3246        reference = b247        [b]248        integer = 2249        vivifiable = 2250        reference = a251        """252    )253    vivified = Vivifier(types=[TestVivify])._vivify(254        instances="main", config=config, defaults="DEFAULT"255    )256def test_vivify_vivify():257    """Test vivification."""258    # simple vivification function call test259    assert Vivifier(types=[]).vivify(instances={}, config={}, defaults="") == {}260    # test invalid missing instance for config261    assert (262        Vivifier(types=[])._vivify(263            instances={}, config={"a": {"integer": "A"}}, defaults=""264        )265        == {}266    )267    # test invalid missing config for instance268    assert (269        Vivifier(types=[])._vivify(instances={"a": "A"}, config={}, defaults="")270        == {}271    )272    # check unsupported type273    class TestUnsupportedList(object):274        unsupported: list[int]275    with pytest.raises(VivificationError):276        Vivifier([TestUnsupportedList]).vivify(277            instances={"test_unsupported_list": "TestUnsupportedList"},278            config={"test_unsupported_list": {"unsupported": "test"}},279            defaults="",280        )281    # check unsupported class282    class TestUnsupportedInteger(object):283        def __init__(self) -> None:...

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.