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

How to use _absolute_path_for method in molecule

Best Python code snippet using molecule_python

ansible.py

Source:ansible.py

...304        roles_path = default_env['ANSIBLE_ROLES_PATH']305        library_path = default_env['ANSIBLE_LIBRARY']306        filter_plugins_path = default_env['ANSIBLE_FILTER_PLUGINS']307        try:308            path = self._absolute_path_for(env, 'ANSIBLE_ROLES_PATH')309            roles_path = '{}:{}'.format(roles_path, path)310        except KeyError:311            pass312        try:313            path = self._absolute_path_for(env, 'ANSIBLE_LIBRARY')314            library_path = '{}:{}'.format(library_path, path)315        except KeyError:316            pass317        try:318            path = self._absolute_path_for(env, 'ANSIBLE_FILTER_PLUGINS')319            filter_plugins_path = '{}:{}'.format(filter_plugins_path, path)320        except KeyError:321            pass322        env['ANSIBLE_ROLES_PATH'] = roles_path323        env['ANSIBLE_LIBRARY'] = library_path324        env['ANSIBLE_FILTER_PLUGINS'] = filter_plugins_path325        return util.merge_dicts(default_env, env)326    @property327    def host_vars(self):328        return self._config.config['provisioner']['inventory']['host_vars']329    @property330    def group_vars(self):331        return self._config.config['provisioner']['inventory']['group_vars']332    @property333    def links(self):334        return self._config.config['provisioner']['inventory']['links']335    @property336    def inventory(self):337        """338        Create an inventory structure and returns a dict.339        .. code-block:: yaml340            ungrouped:341              vars:342                foo: bar343              hosts:344                instance-1:345                instance-2:346              children:347                $child_group_name:348                  hosts:349                    instance-1:350                    instance-2:351            $group_name:352              hosts:353                instance-1:354                  ansible_connection: docker355                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'))616    def _get_filter_plugin_directory(self):617        return util.abs_path(618            os.path.join(self._get_plugin_directory(), 'filters'))619    def _absolute_path_for(self, env, key):...

ansible_collection.py

Source:ansible_collection.py

...131        roles_path = default_env['ANSIBLE_ROLES_PATH']132        library_path = default_env['ANSIBLE_LIBRARY']133        filter_plugins_path = default_env['ANSIBLE_FILTER_PLUGINS']134        try:135            path = self._absolute_path_for(env, 'ANSIBLE_ROLES_PATH')136            roles_path = '{}:{}'.format(roles_path, path)137        except KeyError:138            pass139        try:140            path = self._absolute_path_for(env, 'ANSIBLE_LIBRARY')141            library_path = '{}:{}'.format(library_path, path)142        except KeyError:143            pass144        try:145            path = self._absolute_path_for(env, 'ANSIBLE_FILTER_PLUGINS')146            filter_plugins_path = '{}:{}'.format(filter_plugins_path, path)147        except KeyError:148            pass149        env['ANSIBLE_ROLES_PATH'] = roles_path150        env['ANSIBLE_LIBRARY'] = library_path151        env['ANSIBLE_FILTER_PLUGINS'] = filter_plugins_path152        return util.merge_dicts(default_env, env)153    @property154    def hosts(self):155        return self._config.config['provisioner']['inventory']['hosts']156    @property157    def host_vars(self):158        return self._config.config['provisioner']['inventory']['host_vars']159    @property160    def group_vars(self):161        return self._config.config['provisioner']['inventory']['group_vars']162    @property163    def links(self):164        return self._config.config['provisioner']['inventory']['links']165    @property166    def inventory(self):167        """168        Create an inventory structure and returns a dict.169        .. code-block:: yaml170            ungrouped:171              vars:172                foo: bar173              hosts:174                instance-1:175                instance-2:176              children:177                $child_group_name:178                  hosts:179                    instance-1:180                    instance-2:181            $group_name:182              hosts:183                instance-1:184                  ansible_connection: docker185                instance-2:186                  ansible_connection: docker187        :return: str188        """189        dd = self._vivify()190        for platform in self._config.platforms.instances:191            for group in platform.get('groups', ['ungrouped']):192                instance_name = platform['name']193                connection_options = self.connection_options(instance_name)194                molecule_vars = {195                    'molecule_file':196                    "{{ lookup('env', 'MOLECULE_FILE') }}",197                    'molecule_ephemeral_directory':198                    "{{ lookup('env', 'MOLECULE_EPHEMERAL_DIRECTORY') }}",199                    'molecule_scenario_directory':200                    "{{ lookup('env', 'MOLECULE_SCENARIO_DIRECTORY') }}",201                    'molecule_yml':202                    "{{ lookup('file', molecule_file) | molecule_from_yaml }}",203                    'molecule_instance_config':204                    "{{ lookup('env', 'MOLECULE_INSTANCE_CONFIG') }}",205                    'molecule_no_log':206                    "{{ lookup('env', 'MOLECULE_NO_LOG') or not "207                    "molecule_yml.provisioner.log|default(False) | bool }}"208                }209                # All group210                dd['all']['hosts'][instance_name] = connection_options211                dd['all']['vars'] = molecule_vars212                # Named group213                dd[group]['hosts'][instance_name] = connection_options214                dd[group]['vars'] = molecule_vars215                # Ungrouped216                dd['ungrouped']['vars'] = {}217                # Children218                for child_group in platform.get('children', []):219                    dd[group]['children'][child_group]['hosts'][220                        instance_name] = connection_options221        return self._default_to_regular(dd)222    @property223    def inventory_directory(self):224        return self._config.scenario.inventory_directory225    @property226    def inventory_file(self):227        return os.path.join(self.inventory_directory, 'ansible_inventory.yml')228    @property229    def config_file(self):230        return os.path.join(self._config.scenario.ephemeral_directory,231                            'ansible.cfg')232    @property233    @util.memoize234    def playbooks(self):235        return ansible_playbooks.AnsiblePlaybooks(self._config)236    @property237    def directory(self):238        return os.path.join(239            os.path.dirname(__file__), os.path.pardir, os.path.pardir,240            'molecule', 'provisioner', 'ansible')241    def cleanup(self):242        """243        Executes `ansible-playbook` against the cleanup playbook and returns244        None.245        :return: None246        """247        pb = self._get_ansible_playbook(self.playbooks.cleanup)248        pb.execute()249    def connection_options(self, instance_name):250        d = self._config.driver.ansible_connection_options(instance_name)251        return util.merge_dicts(252            d, self._config.config['provisioner']['connection_options'])253    def check(self):254        """255        Executes ``ansible-playbook`` against the converge playbook with the256        ``--check`` flag and returns None.257        :return: None258        """259        pb = self._get_ansible_playbook(self.playbooks.converge)260        pb.add_cli_arg('check', True)261        pb.execute()262    def converge(self, playbook=None, **kwargs):263        """264        Executes ``ansible-playbook`` against the converge playbook unless265        specified otherwise and returns a string.266        :param playbook: An optional string containing an absolute path to a267         playbook.268        :param kwargs: An optional keyword arguments.269        :return: str270        """271        if playbook is None:272            pb = self._get_ansible_playbook(self.playbooks.converge, **kwargs)273        else:274            pb = self._get_ansible_playbook(playbook, **kwargs)275        return pb.execute()276    def destroy(self):277        """278        Executes ``ansible-playbook`` against the destroy playbook and returns279        None.280        :return: None281        """282        pb = self._get_ansible_playbook(self.playbooks.destroy)283        pb.execute()284    def side_effect(self):285        """286        Executes ``ansible-playbook`` against the side_effect playbook and287        returns None.288        :return: None289        """290        pb = self._get_ansible_playbook(self.playbooks.side_effect)291        pb.execute()292    def create(self):293        """294        Executes ``ansible-playbook`` against the create playbook and returns295        None.296        :return: None297        """298        pb = self._get_ansible_playbook(self.playbooks.create)299        pb.execute()300    def prepare(self):301        """302        Executes ``ansible-playbook`` against the prepare playbook and returns303        None.304        :return: None305        """306        pb = self._get_ansible_playbook(self.playbooks.prepare)307        pb.execute()308    def syntax(self):309        """310        Executes ``ansible-playbook`` against the converge playbook with the311        ``-syntax-check`` flag and returns None.312        :return: None313        """314        pb = self._get_ansible_playbook(self.playbooks.converge)315        pb.add_cli_arg('syntax-check', True)316        pb.execute()317    def verify(self):318        """319        Executes ``ansible-playbook`` against the verify playbook and returns320        None.321        :return: None322        """323        pb = self._get_ansible_playbook(self.playbooks.verify)324        pb.execute()325    def write_config(self):326        """327        Writes the provisioner's config file to disk and returns None.328        :return: None329        """330        template = util.render_template(331            self._get_config_template(), config_options=self.config_options)332        util.write_file(self.config_file, template)333    def manage_inventory(self):334        """335        Manages inventory for Ansible and returns None.336        :returns: None337        """338        self._write_inventory()339        self._remove_vars()340        if not self.links:341            self._add_or_update_vars()342        else:343            self._link_or_update_vars()344    def abs_path(self, path):345        return util.abs_path(346            os.path.join(self._config.scenario.directory, path))347    def _add_or_update_vars(self):348        """349        Creates host and/or group vars and returns None.350        :returns: None351        """352        # Create the hosts extra inventory source (only if not empty)353        hosts_file = os.path.join(self.inventory_directory, 'hosts')354        if self.hosts:355            util.write_file(hosts_file, util.safe_dump(self.hosts))356        # Create the host_vars and group_vars directories357        for target in ['host_vars', 'group_vars']:358            if target == 'host_vars':359                vars_target = copy.deepcopy(self.host_vars)360                for instance_name, _ in self.host_vars.items():361                    if instance_name == 'localhost':362                        instance_key = instance_name363                    else:364                        instance_key = instance_name365                    vars_target[instance_key] = vars_target.pop(instance_name)366            elif target == 'group_vars':367                vars_target = self.group_vars368            if vars_target:369                target_vars_directory = os.path.join(self.inventory_directory,370                                                     target)371                if not os.path.isdir(util.abs_path(target_vars_directory)):372                    os.mkdir(util.abs_path(target_vars_directory))373                for target in vars_target.keys():374                    target_var_content = vars_target[target]375                    path = os.path.join(376                        util.abs_path(target_vars_directory), target)377                    util.write_file(path, util.safe_dump(target_var_content))378    def _write_inventory(self):379        """380        Writes the provisioner's inventory file to disk and returns None.381        :return: None382        """383        self._verify_inventory()384        util.write_file(self.inventory_file, util.safe_dump(self.inventory))385    def _remove_vars(self):386        """387        Remove hosts/host_vars/group_vars and returns None.388        :returns: None389        """390        for name in ("hosts", "group_vars", "host_vars"):391            d = os.path.join(self.inventory_directory, name)392            if os.path.islink(d) or os.path.isfile(d):393                os.unlink(d)394            elif os.path.isdir(d):395                shutil.rmtree(d)396    def _link_or_update_vars(self):397        """398        Creates or updates the symlink to group_vars and returns None.399        :returns: None400        """401        for d, source in self.links.items():402            target = os.path.join(self.inventory_directory, d)403            source = os.path.join(self._config.scenario.directory, source)404            if not os.path.exists(source):405                msg = "The source path '{}' does not exist.".format(source)406                util.sysexit_with_message(msg)407            msg = "Inventory {} linked to {}".format(source, target)408            LOG.info(msg)409            os.symlink(source, target)410    def _get_ansible_playbook(self, playbook, **kwargs):411        """412        Get an instance of AnsiblePlaybook and returns it.413        :param playbook: A string containing an absolute path to a414         provisioner's playbook.415        :param kwargs: An optional keyword arguments.416        :return: object417        """418        return ansible_playbook.AnsiblePlaybook(playbook, self._config,419                                                **kwargs)420    def _verify_inventory(self):421        """422        Verify the inventory is valid and returns None.423        :return: None424        """425        if not self.inventory:426            msg = ("Instances missing from the 'platform' "427                   "section of molecule.yml.")428            util.sysexit_with_message(msg)429    def _get_config_template(self):430        """431        Returns a config template string.432        :return: str433        """434        return """435{% for section, section_dict in config_options.items() -%}436[{{ section }}]437{% for k, v in section_dict.items() -%}438{{ k }} = {{ v }}439{% endfor -%}440{% endfor -%}441""".strip()442    def _vivify(self):443        """444        Returns an autovivification default dict.445        :return: dict446        """447        return collections.defaultdict(self._vivify)448    def _default_to_regular(self, d):449        if isinstance(d, collections.defaultdict):450            d = {k: self._default_to_regular(v) for k, v in d.items()}451        return d452    def _get_plugin_directory(self):453        return os.path.join(self.directory, 'plugins')454    def _get_libraries_directory(self):455        return util.abs_path(456            os.path.join(self._get_plugin_directory(), 'libraries'))457    def _get_filter_plugin_directory(self):458        return util.abs_path(459            os.path.join(self._get_plugin_directory(), 'filters'))460    def _absolute_path_for(self, env, key):...

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.