coalib.misc.prepare_file() - Code Metrics - Inspection of "Revamp LocalBearTestHelper" - coala-analyzer/coala - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#1431)

by Abdeali

created 2016-02-12 12:19 UTC

coalib.misc.prepare_file() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	10
dl	0
loc	30
rs	3.1304

How to fix Complexity

from contextlib import contextmanager, closing
import sys
import os
from io import StringIO
import builtins
import signal
import threading
import platform
import tempfile

from coalib.misc.MutableValue import MutableValue


@contextmanager
def subprocess_timeout(sub_process, seconds, kill_pg=False):
    """
    Kill subprocess if the sub process takes more the than the timeout.

    :param sub_process: The sub process to run.
    :param seconds:     The number of seconds to allow the test to run for. If
                        set to 0 or a negative value, it waits indefinitely.
                        Floats can be used to specify units smaller than
                        seconds.
    :param kill_pg:     Boolean whether to kill the process group or only this
                        process. (not applicable for windows)
    """
    timedout = MutableValue(False)

    if seconds <= 0:
        yield timedout
        return

    finished = threading.Event()

    if platform.system() == "Windows":  # pragma: no cover
        kill_pg = False

    def kill_it():
        finished.wait(seconds)
        if not finished.is_set():
            timedout.value = True
            if kill_pg:
                pgid = os.getpgid(sub_process.pid)
            os.kill(sub_process.pid, signal.SIGINT)
            if kill_pg:
                os.killpg(pgid, signal.SIGINT)

    thread = threading.Thread(name='timeout-killer', target=kill_it)
    try:
        thread.start()
        yield timedout
    finally:
        finished.set()
        thread.join()


@contextmanager
def replace_stdout(replacement):
    """
    Replaces stdout with the replacement, yields back to the caller and then
    reverts everything back.
    """
    _stdout = sys.stdout
    sys.stdout = replacement
    try:
        yield
    finally:
        sys.stdout = _stdout


@contextmanager
def suppress_stdout():
    """
    Suppresses everything going to stdout.
    """
    with open(os.devnull, "w") as devnull, replace_stdout(devnull):
        yield


@contextmanager
def retrieve_stdout():
    """
    Yields a StringIO object from which one can read everything that was
    printed to stdout. (It won't be printed to the real stdout!)

    Example usage:

    with retrieve_stdout() as stdout:
        print("something")  # Won't print to the console
        what_was_printed = stdout.getvalue()  # Save the value
    """
    with closing(StringIO()) as sio, replace_stdout(sio):
        oldprint = builtins.print
        try:
            # Overriding stdout doesn't work with libraries, this ensures even
            # cached variables take this up. Well... it works.
            def newprint(*args, **kwargs):
                kwargs['file'] = sio
                oldprint(*args, **kwargs)

            builtins.print = newprint
            yield sio
        finally:
            builtins.print = oldprint


@contextmanager
def simulate_console_inputs(*inputs):
    """
    Does some magic to simulate the given inputs to any calls to the `input`
    builtin. This yields back an InputGenerator object so you can check
    which input was already used and append any additional inputs you want.
    Example:

        with simulate_console_inputs(0, 1, 2) as generator:
            assert(input() == 0)
            assert(generator.last_input == 0)
            generator.inputs.append(3)
            assert(input() == 1)
            assert(input() == 2)
            assert(input() == 3)
            assert(generator.last_input == 3)

    :param inputs:      Any inputs to simulate.
    :raises ValueError: Raised when was asked for more input but there's no
                        more provided.
    """
    class InputGenerator:

        def __init__(self, inputs):
            self.last_input = -1
            self.inputs = inputs

        def generate_input(self, prompt=''):
            print(prompt, end="")
            self.last_input += 1
            try:
                return self.inputs[self.last_input]
            except IndexError:
                raise ValueError("Asked for more input, but no more was "
                                 "provided from `simulate_console_inputs`.")

    input_generator = InputGenerator(list(inputs))
    _input = builtins.input
    builtins.input = input_generator.generate_input
    try:
        yield input_generator
    finally:
        builtins.input = _input


@contextmanager
def make_temp(suffix="", prefix="tmp", dir=None):
    """
    Creates a temporary file with a closed stream and deletes it when done.

    :return: A contextmanager retrieving the file path.
    """
    temporary = tempfile.mkstemp(suffix=suffix, prefix=prefix, dir=dir)
    os.close(temporary[0])
    try:
        yield temporary[1]
    finally:
        os.remove(temporary[1])


@contextmanager
def prepare_file(lines,
                 filename,
                 force_linebreaks=True,
                 create_tempfile=True,
                 tempfile_kwargs={}):
    """
    Can creates a temporary file (if filename is None) with the lines.
    Can also add a trailing newline to each line specified if needed.

    :param lines:            The lines from the file. (list of strings)
    :param filename:         The filename to be prepared.
    :param force_linebreaks: Whether to append newlines at each line if needed.
    :param create_tempfile:  Whether to save lines in tempfile if needed.
    :param tempfile_kwargs:  Kwargs passed to tempfile.mkstemp().
    """
    if force_linebreaks:
        for i, line in enumerate(lines):
            lines[i] = line if line.endswith("\n") else line + "\n"

    if not create_tempfile and filename is None:
        filename = "dummy_file_name"

    if not isinstance(filename, str) and create_tempfile:
        with make_temp(**tempfile_kwargs) as filename:
            with open(filename, 'w') as file:
                file.writelines(lines)
            yield lines, filename
    else:
        yield lines, filename


1			from contextlib import contextmanager, closing
2			import sys
3			import os
4			from io import StringIO
5			import builtins
6			import signal
7			import threading
8			import platform
9			import tempfile
10
11			from coalib.misc.MutableValue import MutableValue
12
13
14			@contextmanager
15			def subprocess_timeout(sub_process, seconds, kill_pg=False):
16			"""
17			Kill subprocess if the sub process takes more the than the timeout.
18
19			:param sub_process: The sub process to run.
20			:param seconds: The number of seconds to allow the test to run for. If
21			set to 0 or a negative value, it waits indefinitely.
22			Floats can be used to specify units smaller than
23			seconds.
24			:param kill_pg: Boolean whether to kill the process group or only this
25			process. (not applicable for windows)
26			"""
27			timedout = MutableValue(False)
28
29			if seconds <= 0:
30			yield timedout
31			return
32
33			finished = threading.Event()
34
35			if platform.system() == "Windows": # pragma: no cover
36			kill_pg = False
37
38			def kill_it():
39			finished.wait(seconds)
40			if not finished.is_set():
41			timedout.value = True
42			if kill_pg:
43			pgid = os.getpgid(sub_process.pid)
44			os.kill(sub_process.pid, signal.SIGINT)
45			if kill_pg:
46			os.killpg(pgid, signal.SIGINT)
47
48			thread = threading.Thread(name='timeout-killer', target=kill_it)
49			try:
50			thread.start()
51			yield timedout
52			finally:
53			finished.set()
54			thread.join()
55
56
57			@contextmanager
58			def replace_stdout(replacement):
59			"""
60			Replaces stdout with the replacement, yields back to the caller and then
61			reverts everything back.
62			"""
63			_stdout = sys.stdout
64			sys.stdout = replacement
65			try:
66			yield
67			finally:
68			sys.stdout = _stdout
69
70
71			@contextmanager
72			def suppress_stdout():
73			"""
74			Suppresses everything going to stdout.
75			"""
76			with open(os.devnull, "w") as devnull, replace_stdout(devnull):
77			yield
78
79
80			@contextmanager
81			def retrieve_stdout():
82			"""
83			Yields a StringIO object from which one can read everything that was
84			printed to stdout. (It won't be printed to the real stdout!)
85
86			Example usage:
87
88			with retrieve_stdout() as stdout:
89			print("something") # Won't print to the console
90			what_was_printed = stdout.getvalue() # Save the value
91			"""
92			with closing(StringIO()) as sio, replace_stdout(sio):
93			oldprint = builtins.print
94			try:
95			# Overriding stdout doesn't work with libraries, this ensures even
96			# cached variables take this up. Well... it works.
97			def newprint(args, *kwargs):
98			kwargs['file'] = sio
99			oldprint(args, *kwargs)
100
101			builtins.print = newprint
102			yield sio
103			finally:
104			builtins.print = oldprint
105
106
107			@contextmanager
108			def simulate_console_inputs(*inputs):
109			"""
110			Does some magic to simulate the given inputs to any calls to the `input`
111			builtin. This yields back an InputGenerator object so you can check
112			which input was already used and append any additional inputs you want.
113			Example:
114
115			with simulate_console_inputs(0, 1, 2) as generator:
116			assert(input() == 0)
117			assert(generator.last_input == 0)
118			generator.inputs.append(3)
119			assert(input() == 1)
120			assert(input() == 2)
121			assert(input() == 3)
122			assert(generator.last_input == 3)
123
124			:param inputs: Any inputs to simulate.
125			:raises ValueError: Raised when was asked for more input but there's no
126			more provided.
127			"""
128			class InputGenerator:
129
130			def __init__(self, inputs):
131			self.last_input = -1
132			self.inputs = inputs
133
134			def generate_input(self, prompt=''):
135			print(prompt, end="")
136			self.last_input += 1
137			try:
138			return self.inputs[self.last_input]
139			except IndexError:
140			raise ValueError("Asked for more input, but no more was "
141			"provided from `simulate_console_inputs`.")
142
143			input_generator = InputGenerator(list(inputs))
144			_input = builtins.input
145			builtins.input = input_generator.generate_input
146			try:
147			yield input_generator
148			finally:
149			builtins.input = _input
150
151
152			@contextmanager
153			def make_temp(suffix="", prefix="tmp", dir=None):
154			"""
155			Creates a temporary file with a closed stream and deletes it when done.
156
157			:return: A contextmanager retrieving the file path.
158			"""
159			temporary = tempfile.mkstemp(suffix=suffix, prefix=prefix, dir=dir)
160			os.close(temporary[0])
161			try:
162			yield temporary[1]
163			finally:
164			os.remove(temporary[1])
165
166
167			@contextmanager
168			def prepare_file(lines,
169			filename,
170			force_linebreaks=True,
171			create_tempfile=True,
172			tempfile_kwargs={}):
173			"""
174			Can creates a temporary file (if filename is None) with the lines.
175			Can also add a trailing newline to each line specified if needed.
176
177			:param lines: The lines from the file. (list of strings)
178			:param filename: The filename to be prepared.
179			:param force_linebreaks: Whether to append newlines at each line if needed.
180			:param create_tempfile: Whether to save lines in tempfile if needed.
181			:param tempfile_kwargs: Kwargs passed to tempfile.mkstemp().
182			"""
183			if force_linebreaks:
184			for i, line in enumerate(lines):
185			lines[i] = line if line.endswith("\n") else line + "\n"
186
187			if not create_tempfile and filename is None:
188			filename = "dummy_file_name"
189
190			if not isinstance(filename, str) and create_tempfile:
191			with make_temp(**tempfile_kwargs) as filename:
192			with open(filename, 'w') as file:
193			file.writelines(lines)
194			yield lines, filename
195			else:
196			yield lines, filename
197

coala-analyzer / coala

Pull Request — master (#1431)

coalib.misc.prepare_file() F

Complexity

Size

Duplication

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like