ACIL FM

Nama	Ukuran	Aksi
__pycache__	-	Open
add_host.py	3.58 MB	View DL Edit
assemble.py	6.42 MB	View DL Edit
assert.py	5.03 MB	View DL Edit
async_status.py	1.78 MB	View DL Edit
command.py	1.13 MB	View DL Edit
copy.py	26.72 MB	View DL Edit
debug.py	3.28 MB	View DL Edit
fail.py	1.44 MB	View DL Edit
fetch.py	9.64 MB	View DL Edit
gather_facts.py	6.55 MB	View DL Edit
group_by.py	1.87 MB	View DL Edit
include_vars.py	11.19 MB	View DL Edit
normal.py	2.39 MB	View DL Edit
package.py	4.13 MB	View DL Edit
pause.py	11.31 MB	View DL Edit
raw.py	1.78 MB	View DL Edit
reboot.py	20.59 MB	View DL Edit
script.py	7.8 MB	View DL Edit
service.py	4.51 MB	View DL Edit
set_fact.py	2.75 MB	View DL Edit
set_stats.py	2.82 MB	View DL Edit
shell.py	1.22 MB	View DL Edit
template.py	9.74 MB	View DL Edit
unarchive.py	4.78 MB	View DL Edit
uri.py	3.96 MB	View DL Edit
validate_argument_spec.py	3.98 MB	View DL Edit
wait_for_connection.py	4.47 MB	View DL Edit
yum.py	4.6 MB	View DL Edit
__init__.py	67.24 MB	View DL Edit

Edit file: /usr/lib/python3.9/site-packages/ansible/plugins/action/template.py

# Copyright: (c) 2015, Michael DeHaan <michael.dehaan@gmail.com>
# Copyright: (c) 2018, Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)

from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

import os
import shutil
import stat
import tempfile

from ansible import constants as C
from ansible.config.manager import ensure_type
from ansible.errors import AnsibleError, AnsibleFileNotFound, AnsibleAction, AnsibleActionFail
from ansible.module_utils._text import to_bytes, to_text, to_native
from ansible.module_utils.parsing.convert_bool import boolean
from ansible.module_utils.six import string_types
from ansible.plugins.action import ActionBase
from ansible.template import generate_ansible_template_vars, AnsibleEnvironment

class ActionModule(ActionBase):

TRANSFERS_FILES = True
    DEFAULT_NEWLINE_SEQUENCE = "\n"

def run(self, tmp=None, task_vars=None):
        ''' handler for template operations '''

if task_vars is None:
            task_vars = dict()

result = super(ActionModule, self).run(tmp, task_vars)
        del tmp  # tmp no longer has any effect

# Options type validation
        # stings
        for s_type in ('src', 'dest', 'state', 'newline_sequence', 'variable_start_string', 'variable_end_string', 'block_start_string',
                       'block_end_string', 'comment_start_string', 'comment_end_string'):
            if s_type in self._task.args:
                value = ensure_type(self._task.args[s_type], 'string')
                if value is not None and not isinstance(value, string_types):
                    raise AnsibleActionFail("%s is expected to be a string, but got %s instead" % (s_type, type(value)))
                self._task.args[s_type] = value

# booleans
        try:
            follow = boolean(self._task.args.get('follow', False), strict=False)
            trim_blocks = boolean(self._task.args.get('trim_blocks', True), strict=False)
            lstrip_blocks = boolean(self._task.args.get('lstrip_blocks', False), strict=False)
        except TypeError as e:
            raise AnsibleActionFail(to_native(e))

# assign to local vars for ease of use
        source = self._task.args.get('src', None)
        dest = self._task.args.get('dest', None)
        state = self._task.args.get('state', None)
        newline_sequence = self._task.args.get('newline_sequence', self.DEFAULT_NEWLINE_SEQUENCE)
        variable_start_string = self._task.args.get('variable_start_string', None)
        variable_end_string = self._task.args.get('variable_end_string', None)
        block_start_string = self._task.args.get('block_start_string', None)
        block_end_string = self._task.args.get('block_end_string', None)
        comment_start_string = self._task.args.get('comment_start_string', None)
        comment_end_string = self._task.args.get('comment_end_string', None)
        output_encoding = self._task.args.get('output_encoding', 'utf-8') or 'utf-8'

wrong_sequences = ["\\n", "\\r", "\\r\\n"]
        allowed_sequences = ["\n", "\r", "\r\n"]

# We need to convert unescaped sequences to proper escaped sequences for Jinja2
        if newline_sequence in wrong_sequences:
            newline_sequence = allowed_sequences[wrong_sequences.index(newline_sequence)]

try:
            # logical validation
            if state is not None:
                raise AnsibleActionFail("'state' cannot be specified on a template")
            elif source is None or dest is None:
                raise AnsibleActionFail("src and dest are required")
            elif newline_sequence not in allowed_sequences:
                raise AnsibleActionFail("newline_sequence needs to be one of: \n, \r or \r\n")
            else:
                try:
                    source = self._find_needle('templates', source)
                except AnsibleError as e:
                    raise AnsibleActionFail(to_text(e))

mode = self._task.args.get('mode', None)
            if mode == 'preserve':
                mode = '0%03o' % stat.S_IMODE(os.stat(source).st_mode)

# Get vault decrypted tmp file
            try:
                tmp_source = self._loader.get_real_file(source)
            except AnsibleFileNotFound as e:
                raise AnsibleActionFail("could not find src=%s, %s" % (source, to_text(e)))
            b_tmp_source = to_bytes(tmp_source, errors='surrogate_or_strict')

# template the source data locally & get ready to transfer
            try:
                with open(b_tmp_source, 'rb') as f:
                    try:
                        template_data = to_text(f.read(), errors='surrogate_or_strict')
                    except UnicodeError:
                        raise AnsibleActionFail("Template source files must be utf-8 encoded")

# set jinja2 internal search path for includes
                searchpath = task_vars.get('ansible_search_path', [])
                searchpath.extend([self._loader._basedir, os.path.dirname(source)])

# We want to search into the 'templates' subdir of each search path in
                # addition to our original search paths.
                newsearchpath = []
                for p in searchpath:
                    newsearchpath.append(os.path.join(p, 'templates'))
                    newsearchpath.append(p)
                searchpath = newsearchpath

# add ansible 'template' vars
                temp_vars = task_vars.copy()
                # NOTE in the case of ANSIBLE_DEBUG=1 task_vars is VarsWithSources(MutableMapping)
                # so | operator cannot be used as it can be used only on dicts
                # https://peps.python.org/pep-0584/#what-about-mapping-and-mutablemapping
                temp_vars.update(generate_ansible_template_vars(self._task.args.get('src', None), source, dest))

# force templar to use AnsibleEnvironment to prevent issues with native types
                # https://github.com/ansible/ansible/issues/46169
                templar = self._templar.copy_with_new_env(environment_class=AnsibleEnvironment,
                                                          searchpath=searchpath,
                                                          newline_sequence=newline_sequence,
                                                          block_start_string=block_start_string,
                                                          block_end_string=block_end_string,
                                                          variable_start_string=variable_start_string,
                                                          variable_end_string=variable_end_string,
                                                          comment_start_string=comment_start_string,
                                                          comment_end_string=comment_end_string,
                                                          trim_blocks=trim_blocks,
                                                          lstrip_blocks=lstrip_blocks,
                                                          available_variables=temp_vars)
                resultant = templar.do_template(template_data, preserve_trailing_newlines=True, escape_backslashes=False)
            except AnsibleAction:
                raise
            except Exception as e:
                raise AnsibleActionFail("%s: %s" % (type(e).__name__, to_text(e)))
            finally:
                self._loader.cleanup_tmp_file(b_tmp_source)

new_task = self._task.copy()
            # mode is either the mode from task.args or the mode of the source file if the task.args
            # mode == 'preserve'
            new_task.args['mode'] = mode

# remove 'template only' options:
            for remove in ('newline_sequence', 'block_start_string', 'block_end_string', 'variable_start_string', 'variable_end_string',
                           'comment_start_string', 'comment_end_string', 'trim_blocks', 'lstrip_blocks', 'output_encoding'):
                new_task.args.pop(remove, None)

local_tempdir = tempfile.mkdtemp(dir=C.DEFAULT_LOCAL_TMP)

try:
                result_file = os.path.join(local_tempdir, os.path.basename(source))
                with open(to_bytes(result_file, errors='surrogate_or_strict'), 'wb') as f:
                    f.write(to_bytes(resultant, encoding=output_encoding, errors='surrogate_or_strict'))

new_task.args.update(
                    dict(
                        src=result_file,
                        dest=dest,
                        follow=follow,
                    ),
                )
                # call with ansible.legacy prefix to eliminate collisions with collections while still allowing local override
                copy_action = self._shared_loader_obj.action_loader.get('ansible.legacy.copy',
                                                                        task=new_task,
                                                                        connection=self._connection,
                                                                        play_context=self._play_context,
                                                                        loader=self._loader,
                                                                        templar=self._templar,
                                                                        shared_loader_obj=self._shared_loader_obj)
                result.update(copy_action.run(task_vars=task_vars))
            finally:
                shutil.rmtree(to_bytes(local_tempdir, errors='surrogate_or_strict'))

except AnsibleAction as e:
            result.update(e.result)
        finally:
            self._remove_tmp_path(self._connection._shell.tmpdir)

return result

Batal