__do_run_script() - Code Metrics - Inspection of "WIP - Hooks enhancement: running real file in plac..." - audreyr/cookiecutter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#694)

by Eric

created 2016-06-09 22:48 UTC

__do_run_script() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	1

Metric	Value
cc	5
c	2
b	0
f	1
dl	0
loc	43
rs	8.0894

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
cookiecutter.hooks
------------------

Functions for discovering and executing various cookiecutter hooks.
"""

import io
import logging
import os
import subprocess
import sys
import tempfile
import errno

from jinja2 import Template

from pydoc import locate
from cookiecutter import utils
from .exceptions import FailedHookException, BadSerializedStringFormat
from .serialization import SerializationFacade, make_persistent
from .config import get_from_context


_HOOKS = [
    'pre_gen_project',
    'post_gen_project',
    'pre_user_prompt',
    # TODO: other hooks should be listed here
]
EXIT_SUCCESS = 0


def find_hooks():
    """
    Must be called with the project template as the current working directory.
    Returns a dict of all hook scripts provided.
    Dict's key will be the hook/script's name, without extension, while
    values will be the absolute path to the script.
    Missing scripts will not be included in the returned dict.
    """
    hooks_dir = 'hooks'
    r = {}
    logging.debug('hooks_dir is {0}'.format(hooks_dir))
    if not os.path.isdir(hooks_dir):
        logging.debug('No hooks/ dir in template_dir')
        return r
    for f in os.listdir(hooks_dir):
        basename = os.path.splitext(os.path.basename(f))[0]
        if basename in _HOOKS:
            r[basename] = os.path.abspath(os.path.join(hooks_dir, f))
    return r


def run_script_with_context(script_path, cwd, context):
    """
    Executes a script either after rendering with it Jinja or in place without
    template rendering.

    :param script_path: Absolute path to the script to run.
    :param cwd: The directory to run the script from.
    :param context: Cookiecutter project template context.
    """
    lazy_load_from_extra_dir(os.path.dirname(os.path.dirname(script_path)))

    script = __new_script(context, script_path)

    try:
        serializer = __new_serialization_facade(context)
        make_persistent(serializer)

        result = __do_run_script(
            script, cwd, serializer.serialize(context).encode())

        return serializer.deserialize(result[0].decode())

    except BadSerializedStringFormat:
        return context


def run_hook(hook_name, project_dir, context):
    """
    Try to find and execute a hook from the specified project directory.

    :param hook_name: The hook to execute.
    :param project_dir: The directory to execute the script from.
    :param context: Cookiecutter project context.
    """
    script = find_hooks().get(hook_name)
    if script is None:
        logging.debug('No hooks found')
        return context

    return run_script_with_context(script, project_dir, context)


def lazy_load_from_extra_dir(template_dir):
    """
    permit lazy load from the 'extra' directory
    :param template_dir: the project template directory
    """
    extra_dir = os.path.abspath(os.path.join(template_dir, 'extra'))
    if os.path.exists(extra_dir) and extra_dir not in sys.path:
        sys.path.insert(1, extra_dir)


def __create_renderable_hook(script_path, context):
    """
    Create a renderable hook by copying the real hook and applying the template

    :param script_path: Absolute path to the base hook.
    :param context: Cookiecutter project template context.
    """
    _, extension = os.path.splitext(script_path)
    contents = io.open(script_path, 'r', encoding='utf-8').read()
    with tempfile.NamedTemporaryFile(
        delete=False,
        mode='wb',
        suffix=extension
    ) as temp:
        output = Template(contents).render(**context)
        temp.write(output.encode('utf-8'))

    return temp.name


def __get_script_command(script_path):
    """
    Get the executable command of a given script

    :param script_path: Absolute path to the script to run.
    """
    if script_path.endswith('.py'):
        script_command = [sys.executable, script_path]
    else:
        script_command = [script_path]

    utils.make_executable(script_path)

    return script_command


def __do_run_script(script_path, cwd, serialized_context):
    """
    Executes a script wrinting the given serialized context to its standard
    input stream.

    :param script_path: Absolute path to the script to run.
    :param cwd: The directory to run the script from.
    :param serialized_context: Serialized Cookiecutter project template
                               context.
    """
    result = (serialized_context, b'')
    run_thru_shell = sys.platform.startswith('win')

    os.environ['PYTHONPATH'] = os.pathsep.join(sys.path)

    proc = subprocess.Popen(
        __get_script_command(script_path),
        shell=run_thru_shell,
        cwd=cwd,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )

    try:
        result = proc.communicate(serialized_context)
    except OSError as e:
        if e.errno == errno.EINVAL and run_thru_shell:
            logging.warn(
                'Popen.communicate failed certainly ' +
                'because of the issue #19612'
            )
            pass
        else:
            raise e

    exit_status = proc.wait()
    if exit_status != EXIT_SUCCESS:
        raise FailedHookException(
            "Hook script failed (exit status: %d): \n%s" %
            (exit_status, result[1]))

    return result


def __get_from_context(context, key, default=None):
    """
    config.get_from_context wrapper

    :param context: context to search in
    :param key: key to look for
    :param default: default value to get back if key does not exist
    """
    result = get_from_context(context, key)

    return result if result is not None else get_from_context(
        context, 'cookiecutter.' + key, default)


def __new_serialization_facade(context):
    """
    serialization facade factory function

    :param context: current context
    """
    serializers = {}
    usetype = __get_from_context(context, '_serializers.use', 'json')
    classes = __get_from_context(context, '_serializers.classes', [])
    for type in classes:
        serializers[type] = locate(classes[type], 1)

    return SerializationFacade(serializers).use(usetype)


def __new_script(context, script_path):
    """
    script factory function

    :param context: current context
    :param script_path: absolute path to the script to run
    """
    return script_path if __get_from_context(
        context, '_run_hook_in_place', False) else __create_renderable_hook(
            script_path, context)


1			#!/usr/bin/env python
2			# -- coding: utf-8 --
3
4			"""
5			cookiecutter.hooks
6			------------------
7
8			Functions for discovering and executing various cookiecutter hooks.
9			"""
10
11			import io
12			import logging
13			import os
14			import subprocess
15			import sys
16			import tempfile
17			import errno
18
19			from jinja2 import Template
20
21			from pydoc import locate
22			from cookiecutter import utils
23			from .exceptions import FailedHookException, BadSerializedStringFormat
24			from .serialization import SerializationFacade, make_persistent
25			from .config import get_from_context
26
27
28			_HOOKS = [
29			'pre_gen_project',
30			'post_gen_project',
31			'pre_user_prompt',
32			# TODO: other hooks should be listed here
33			]
34			EXIT_SUCCESS = 0
35
36
37			def find_hooks():
38			"""
39			Must be called with the project template as the current working directory.
40			Returns a dict of all hook scripts provided.
41			Dict's key will be the hook/script's name, without extension, while
42			values will be the absolute path to the script.
43			Missing scripts will not be included in the returned dict.
44			"""
45			hooks_dir = 'hooks'
46			r = {}
47			logging.debug('hooks_dir is {0}'.format(hooks_dir))
48			if not os.path.isdir(hooks_dir):
49			logging.debug('No hooks/ dir in template_dir')
50			return r
51			for f in os.listdir(hooks_dir):
52			basename = os.path.splitext(os.path.basename(f))[0]
53			if basename in _HOOKS:
54			r[basename] = os.path.abspath(os.path.join(hooks_dir, f))
55			return r
56
57
58			def run_script_with_context(script_path, cwd, context):
59			"""
60			Executes a script either after rendering with it Jinja or in place without
61			template rendering.
62
63			:param script_path: Absolute path to the script to run.
64			:param cwd: The directory to run the script from.
65			:param context: Cookiecutter project template context.
66			"""
67			lazy_load_from_extra_dir(os.path.dirname(os.path.dirname(script_path)))
68
69			script = __new_script(context, script_path)
70
71			try:
72			serializer = __new_serialization_facade(context)
73			make_persistent(serializer)
74
75			result = __do_run_script(
76			script, cwd, serializer.serialize(context).encode())
77
78			return serializer.deserialize(result[0].decode())
79
80			except BadSerializedStringFormat:
81			return context
82
83
84			def run_hook(hook_name, project_dir, context):
85			"""
86			Try to find and execute a hook from the specified project directory.
87
88			:param hook_name: The hook to execute.
89			:param project_dir: The directory to execute the script from.
90			:param context: Cookiecutter project context.
91			"""
92			script = find_hooks().get(hook_name)
93			if script is None:
94			logging.debug('No hooks found')
95			return context
96
97			return run_script_with_context(script, project_dir, context)
98
99
100			def lazy_load_from_extra_dir(template_dir):
101			"""
102			permit lazy load from the 'extra' directory
103			:param template_dir: the project template directory
104			"""
105			extra_dir = os.path.abspath(os.path.join(template_dir, 'extra'))
106			if os.path.exists(extra_dir) and extra_dir not in sys.path:
107			sys.path.insert(1, extra_dir)
108
109
110			def __create_renderable_hook(script_path, context):
111			"""
112			Create a renderable hook by copying the real hook and applying the template
113
114			:param script_path: Absolute path to the base hook.
115			:param context: Cookiecutter project template context.
116			"""
117			_, extension = os.path.splitext(script_path)
118			contents = io.open(script_path, 'r', encoding='utf-8').read()
119			with tempfile.NamedTemporaryFile(
120			delete=False,
121			mode='wb',
122			suffix=extension
123			) as temp:
124			output = Template(contents).render(**context)
125			temp.write(output.encode('utf-8'))
126
127			return temp.name
128
129
130			def __get_script_command(script_path):
131			"""
132			Get the executable command of a given script
133
134			:param script_path: Absolute path to the script to run.
135			"""
136			if script_path.endswith('.py'):
137			script_command = [sys.executable, script_path]
138			else:
139			script_command = [script_path]
140
141			utils.make_executable(script_path)
142
143			return script_command
144
145
146			def __do_run_script(script_path, cwd, serialized_context):
147			"""
148			Executes a script wrinting the given serialized context to its standard
149			input stream.
150
151			:param script_path: Absolute path to the script to run.
152			:param cwd: The directory to run the script from.
153			:param serialized_context: Serialized Cookiecutter project template
154			context.
155			"""
156			result = (serialized_context, b'')
157			run_thru_shell = sys.platform.startswith('win')
158
159			os.environ['PYTHONPATH'] = os.pathsep.join(sys.path)
160
161			proc = subprocess.Popen(
162			__get_script_command(script_path),
163			shell=run_thru_shell,
164			cwd=cwd,
165			stdin=subprocess.PIPE,
166			stdout=subprocess.PIPE,
167			stderr=subprocess.PIPE
168			)
169
170			try:
171			result = proc.communicate(serialized_context)
172			except OSError as e:
173			if e.errno == errno.EINVAL and run_thru_shell:
174			logging.warn(
175			'Popen.communicate failed certainly ' +
176			'because of the issue #19612'
177			)
178			pass
179			else:
180			raise e
181
182			exit_status = proc.wait()
183			if exit_status != EXIT_SUCCESS:
184			raise FailedHookException(
185			"Hook script failed (exit status: %d): \n%s" %
186			(exit_status, result[1]))
187
188			return result
189
190
191			def __get_from_context(context, key, default=None):
192			"""
193			config.get_from_context wrapper
194
195			:param context: context to search in
196			:param key: key to look for
197			:param default: default value to get back if key does not exist
198			"""
199			result = get_from_context(context, key)
200
201			return result if result is not None else get_from_context(
202			context, 'cookiecutter.' + key, default)
203
204
205			def __new_serialization_facade(context):
206			"""
207			serialization facade factory function
208
209			:param context: current context
210			"""
211			serializers = {}
212			usetype = __get_from_context(context, '_serializers.use', 'json')
213			classes = __get_from_context(context, '_serializers.classes', [])
214			for type in classes:
215			serializers[type] = locate(classes[type], 1)
216
217			return SerializationFacade(serializers).use(usetype)
218
219
220			def __new_script(context, script_path):
221			"""
222			script factory function
223
224			:param context: current context
225			:param script_path: absolute path to the script to run
226			"""
227			return script_path if __get_from_context(
228			context, '_run_hook_in_place', False) else __create_renderable_hook(
229			script_path, context)
230

audreyr / cookiecutter

Pull Request — master (#694)

__do_run_script() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like