tests.test_cli.test_watch_dummy() - Code Metrics - spyder-ide/qtsass - Measure and Improve Code Quality continuously with Scrutinizer

tests.test_cli.test_watch_dummy() B
last analyzed 2020-03-24 17:01 UTC

↳ Parent: tests.test_cli

Complexity

Conditions

Size

Total Lines	34
Code Lines	24

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	24
dl	0
loc	34
rs	8.8373
c	0
b	0
f	0
cc	5
nop	1

# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright (c) 2015 Yann Lanthony
# Copyright (c) 2017-2018 Spyder Project Contributors
#
# Licensed under the terms of the MIT License
# (See LICENSE.txt for details)
# -----------------------------------------------------------------------------
"""Test qtsass cli."""

from __future__ import absolute_import

# Standard library imports
from collections import namedtuple
from os.path import basename, exists
from subprocess import PIPE, Popen
import time

# Local imports
from . import PROJECT_DIR, await_condition, example, touch


SLEEP_INTERVAL = 1
Result = namedtuple('Result', "code stdout stderr")


def indent(text, prefix='    '):
    """Like textwrap.indent"""

    return ''.join([prefix + line for line in text.splitlines(True)])


def invoke(args):
    """Invoke qtsass cli with specified args"""

    kwargs = dict(
        stdout=PIPE,
        stderr=PIPE,
        cwd=PROJECT_DIR
    )
    proc = Popen(['python', '-m', 'qtsass'] + args, **kwargs)
    return proc


def invoke_with_result(args):
    """Invoke qtsass cli and return a Result obj"""

    proc = invoke(args)
    out, err = proc.communicate()
    out = out.decode('ascii', errors="ignore")
    err = err.decode('ascii', errors="ignore")
    return Result(proc.returncode, out, err)


def kill(proc, timeout=1):
    """Kill a subprocess and return a Result obj"""

    proc.kill()
    out, err = proc.communicate()
    out = out.decode('ascii', errors="ignore")
    err = err.decode('ascii', errors="ignore")
    return Result(proc.returncode, out, err)


def format_result(result):
    """Format a subprocess Result obj"""

    out = [
        'Subprocess Report...',
        'Exit code: %s' % result.code,
    ]
    if result.stdout:
        out.append('stdout:')
        out.append(indent(result.stdout, '    '))
    if result.stderr:
        out.append('stderr:')
        out.append(indent(result.stderr, '    '))
    return '\n'.join(out)


def test_compile_dummy_to_stdout():
    """CLI compile dummy example to stdout."""

    args = [example('dummy.scss')]
    result = invoke_with_result(args)

    assert result.code == 0
    assert result.stdout


def test_compile_dummy_to_file(tmpdir):
    """CLI compile dummy example to file."""

    input = example('dummy.scss')
    output = tmpdir.join('dummy.css')
    args = [input, '-o', output.strpath]
    result = invoke_with_result(args)

    assert result.code == 0
    assert exists(output.strpath)


def test_watch_dummy(tmpdir):
    """CLI watch dummy example."""

    input = example('dummy.scss')
    output = tmpdir.join('dummy.css')
    args = [input, '-o', output.strpath, '-w']
    proc = invoke(args)

    # Wait for initial compile
    output_exists = lambda: exists(output.strpath)
    if not await_condition(output_exists):
        result = kill(proc)
        report = format_result(result)
        err = "Failed to compile dummy.scss\n"
        err += report
        assert False, report

    # Ensure subprocess is still alive
    assert proc.poll() is None

    # Touch input file, triggering a recompile
    created = output.mtime()
    file_modified = lambda: output.mtime() > created
    time.sleep(SLEEP_INTERVAL)
    touch(input)

    if not await_condition(file_modified):
        result = kill(proc)
        report = format_result(result)
        err = 'Modifying %s did not trigger recompile.\n' % basename(input)
        err += report
        assert False, err

    kill(proc)


def test_compile_complex(tmpdir):
    """CLI compile complex example."""

    input = example('complex')
    output = tmpdir.mkdir('output')
    args = [input, '-o', output.strpath]
    result = invoke_with_result(args)

    assert result.code == 0

    expected_files = [output.join('light.css'), output.join('dark.css')]
    for file in expected_files:
        assert exists(file.strpath)


def test_watch_complex(tmpdir):
    """CLI watch complex example."""

    input = example('complex')
    output = tmpdir.mkdir('output')
    args = [input, '-o', output.strpath, '-w']
    proc = invoke(args)

    expected_files = [output.join('light.css'), output.join('dark.css')]

    # Wait for initial compile
    files_created = lambda: all([exists(f.strpath) for f in expected_files])
    if not await_condition(files_created):
        result = kill(proc)
        report = format_result(result)
        err = 'All expected files have not been created...'
        err += report
        assert False, err

    # Ensure subprocess is still alive
    assert proc.poll() is None

    # Input files to touch
    input_full = example('complex', 'light.scss')
    input_partial = example('complex', '_base.scss')
    input_nested = example('complex', 'widgets', '_qwidget.scss')

    def touch_and_wait(input_file, timeout=2000):
        """Touch a file, triggering a recompile"""

        filename = basename(input_file)
        old_mtimes = [f.mtime() for f in expected_files]
        files_modified = lambda: all(
            [f.mtime() > old_mtimes[i] for i, f in enumerate(expected_files)]
        )
        time.sleep(SLEEP_INTERVAL)
        touch(input_file)

        if not await_condition(files_modified, timeout):
            result = kill(proc)
            report = format_result(result)
            err = 'Modifying %s did not trigger recompile.\n' % filename
            err += report
            for i, f in enumerate(expected_files):
                err += str(f) + '\n'
                err += str(old_mtimes[i]) + '\n'
                err += str(f.mtime()) + '\n'
                err += str(bool(f.mtime() > old_mtimes[i])) + '\n'
            assert False, err

        return True

    assert touch_and_wait(input_full)
    assert touch_and_wait(input_partial)
    assert touch_and_wait(input_nested)

    kill(proc)


def test_invalid_input():
    """CLI input is not a file or dir."""

    proc = invoke_with_result(['file_does_not_exist.scss'])
    assert proc.code == 1
    assert 'Error: input must be' in proc.stdout

    proc = invoke_with_result(['./dir/does/not/exist'])
    assert proc.code == 1
    assert 'Error: input must be' in proc.stdout


def test_dir_missing_output():
    """CLI dir missing output option"""

    proc = invoke_with_result([example('complex')])
    assert proc.code == 1
    assert 'Error: missing required option' in proc.stdout


1			# -- coding: utf-8 --
2			# -----------------------------------------------------------------------------
3			# Copyright (c) 2015 Yann Lanthony
4			# Copyright (c) 2017-2018 Spyder Project Contributors
5			#
6			# Licensed under the terms of the MIT License
7			# (See LICENSE.txt for details)
8			# -----------------------------------------------------------------------------
9			"""Test qtsass cli."""
10
11			from __future__ import absolute_import
12
13			# Standard library imports
14			from collections import namedtuple
15			from os.path import basename, exists
16			from subprocess import PIPE, Popen
17			import time
18
19			# Local imports
20			from . import PROJECT_DIR, await_condition, example, touch
21
22
23			SLEEP_INTERVAL = 1
24			Result = namedtuple('Result', "code stdout stderr")
25
26
27			def indent(text, prefix=' '):
28			"""Like textwrap.indent"""
29
30			return ''.join([prefix + line for line in text.splitlines(True)])
31
32
33			def invoke(args):
34			"""Invoke qtsass cli with specified args"""
35
36			kwargs = dict(
37			stdout=PIPE,
38			stderr=PIPE,
39			cwd=PROJECT_DIR
40			)
41			proc = Popen(['python', '-m', 'qtsass'] + args, **kwargs)
42			return proc
43
44
45			def invoke_with_result(args):
46			"""Invoke qtsass cli and return a Result obj"""
47
48			proc = invoke(args)
49			out, err = proc.communicate()
50			out = out.decode('ascii', errors="ignore")
51			err = err.decode('ascii', errors="ignore")
52			return Result(proc.returncode, out, err)
53
54
55			def kill(proc, timeout=1):
56			"""Kill a subprocess and return a Result obj"""
57
58			proc.kill()
59			out, err = proc.communicate()
60			out = out.decode('ascii', errors="ignore")
61			err = err.decode('ascii', errors="ignore")
62			return Result(proc.returncode, out, err)
63
64
65			def format_result(result):
66			"""Format a subprocess Result obj"""
67
68			out = [
69			'Subprocess Report...',
70			'Exit code: %s' % result.code,
71			]
72			if result.stdout:
73			out.append('stdout:')
74			out.append(indent(result.stdout, ' '))
75			if result.stderr:
76			out.append('stderr:')
77			out.append(indent(result.stderr, ' '))
78			return '\n'.join(out)
79
80
81			def test_compile_dummy_to_stdout():
82			"""CLI compile dummy example to stdout."""
83
84			args = [example('dummy.scss')]
85			result = invoke_with_result(args)
86
87			assert result.code == 0
88			assert result.stdout
89
90
91			def test_compile_dummy_to_file(tmpdir):
92			"""CLI compile dummy example to file."""
93
94			input = example('dummy.scss')
95			output = tmpdir.join('dummy.css')
96			args = [input, '-o', output.strpath]
97			result = invoke_with_result(args)
98
99			assert result.code == 0
100			assert exists(output.strpath)
101
102
103			def test_watch_dummy(tmpdir):
104			"""CLI watch dummy example."""
105
106			input = example('dummy.scss')
107			output = tmpdir.join('dummy.css')
108			args = [input, '-o', output.strpath, '-w']
109			proc = invoke(args)
110
111			# Wait for initial compile
112			output_exists = lambda: exists(output.strpath)
113			if not await_condition(output_exists):
114			result = kill(proc)
115			report = format_result(result)
116			err = "Failed to compile dummy.scss\n"
117			err += report
118			assert False, report
119
120			# Ensure subprocess is still alive
121			assert proc.poll() is None
122
123			# Touch input file, triggering a recompile
124			created = output.mtime()
125			file_modified = lambda: output.mtime() > created
126			time.sleep(SLEEP_INTERVAL)
127			touch(input)
128
129			if not await_condition(file_modified):
130			result = kill(proc)
131			report = format_result(result)
132			err = 'Modifying %s did not trigger recompile.\n' % basename(input)
133			err += report
134			assert False, err
135
136			kill(proc)
137
138
139			def test_compile_complex(tmpdir):
140			"""CLI compile complex example."""
141
142			input = example('complex')
143			output = tmpdir.mkdir('output')
144			args = [input, '-o', output.strpath]
145			result = invoke_with_result(args)
146
147			assert result.code == 0
148
149			expected_files = [output.join('light.css'), output.join('dark.css')]
150			for file in expected_files:
151			assert exists(file.strpath)
152
153
154			def test_watch_complex(tmpdir):
155			"""CLI watch complex example."""
156
157			input = example('complex')
158			output = tmpdir.mkdir('output')
159			args = [input, '-o', output.strpath, '-w']
160			proc = invoke(args)
161
162			expected_files = [output.join('light.css'), output.join('dark.css')]
163
164			# Wait for initial compile
165			files_created = lambda: all([exists(f.strpath) for f in expected_files])
166			if not await_condition(files_created):
167			result = kill(proc)
168			report = format_result(result)
169			err = 'All expected files have not been created...'
170			err += report
171			assert False, err
172
173			# Ensure subprocess is still alive
174			assert proc.poll() is None
175
176			# Input files to touch
177			input_full = example('complex', 'light.scss')
178			input_partial = example('complex', '_base.scss')
179			input_nested = example('complex', 'widgets', '_qwidget.scss')
180
181			def touch_and_wait(input_file, timeout=2000):
182			"""Touch a file, triggering a recompile"""
183
184			filename = basename(input_file)
185			old_mtimes = [f.mtime() for f in expected_files]
186			files_modified = lambda: all(
187			[f.mtime() > old_mtimes[i] for i, f in enumerate(expected_files)]
188			)
189			time.sleep(SLEEP_INTERVAL)
190			touch(input_file)
191
192			if not await_condition(files_modified, timeout):
193			result = kill(proc)
194			report = format_result(result)
195			err = 'Modifying %s did not trigger recompile.\n' % filename
196			err += report
197			for i, f in enumerate(expected_files):
198			err += str(f) + '\n'
199			err += str(old_mtimes[i]) + '\n'
200			err += str(f.mtime()) + '\n'
201			err += str(bool(f.mtime() > old_mtimes[i])) + '\n'
202			assert False, err
203
204			return True
205
206			assert touch_and_wait(input_full)
207			assert touch_and_wait(input_partial)
208			assert touch_and_wait(input_nested)
209
210			kill(proc)
211
212
213			def test_invalid_input():
214			"""CLI input is not a file or dir."""
215
216			proc = invoke_with_result(['file_does_not_exist.scss'])
217			assert proc.code == 1
218			assert 'Error: input must be' in proc.stdout
219
220			proc = invoke_with_result(['./dir/does/not/exist'])
221			assert proc.code == 1
222			assert 'Error: input must be' in proc.stdout
223
224
225			def test_dir_missing_output():
226			"""CLI dir missing output option"""
227
228			proc = invoke_with_result([example('complex')])
229			assert proc.code == 1
230			assert 'Error: missing required option' in proc.stdout
231

spyder-ide / qtsass

GitHub Access Token became invalid

tests.test_cli.test_watch_dummy() B last analyzed 2020-03-24 17:01 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

tests.test_cli.test_watch_dummy() B
last analyzed 2020-03-24 17:01 UTC