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

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

↳ Parent: tests.test_watchers

Complexity

Conditions

Size

Total Lines	56
Code Lines	35

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	35
dl	0
loc	56
rs	8.5733
c	0
b	0
f	0
cc	5
nop	2

How to fix Long Method

# -*- 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 os.path import dirname, exists
import os
import shutil
import sys
import time

# Third party imports
import pytest

# Local imports
#Local imports
from qtsass import compile_filename
from qtsass.watchers import PollingWatcher, QtWatcher
from qtsass.watchers.api import retry

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


class CallCounter(object):

    def __init__(self):
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1


@pytest.mark.parametrize(
    'Watcher', (PollingWatcher, QtWatcher),
)
def test_watchers(Watcher, tmpdir):
    """Stress test Watcher implementations"""

    # Skip when QtWatcher is None - when Qt is not installed.
    if not Watcher:
        return

    watch_dir = tmpdir.join('src').strpath
    os.makedirs(watch_dir)
    shutil.copy2(example('dummy.scss'), watch_dir)
    input = tmpdir.join('src/dummy.scss').strpath
    output = tmpdir.join('build/dummy.css').strpath
    output_exists = lambda: exists(output)


    c = CallCounter()
    w = Watcher(
        watch_dir=watch_dir,
        compiler=compile_filename,
        args=(input, output),
    )
    w.connect(c)

    # Output should not yet exist
    assert not exists(output)

    w.start()

    touch(input)
    time.sleep(0.5)
    if not await_condition(output_exists):
        assert False, 'Output file not created...'

    # Removing the watch_dir should not kill the Watcher
    # simply stop dispatching callbacks
    shutil.rmtree(watch_dir)
    time.sleep(0.5)
    assert c.count == 1

    # Watcher should recover once the input file is there again
    os.makedirs(watch_dir)
    shutil.copy2(example('dummy.scss'), watch_dir)
    time.sleep(0.5)
    assert c.count == 2

    # Stop watcher
    w.stop()
    w.join()

    for _ in range(5):
        touch(input)

    # Count should not change
    assert c.count == 2


@pytest.mark.skipif(sys.platform.startswith('linux') or not QtWatcher,
                    reason="Fails on linux")
def test_qtwatcher(tmpdir):
    """Test QtWatcher implementation."""
    # Constructing a QApplication will cause the QtWatcher constructed
    # below to use a Signal to dispatch callbacks.
    from qtsass.watchers.qt import QApplication

    qt_app = QApplication.instance()
    if not qt_app:
        qt_app = QApplication([])

    watch_dir = tmpdir.join('src').strpath
    os.makedirs(watch_dir)
    shutil.copy2(example('dummy.scss'), watch_dir)
    input = tmpdir.join('src/dummy.scss').strpath
    output = tmpdir.join('build/dummy.css').strpath
    output_exists = lambda: exists(output)

    c = CallCounter()
    w = QtWatcher(
        watch_dir=watch_dir,
        compiler=compile_filename,
        args=(input, output),
    )
    # We connect a counter directly to the Watcher's Qt Signal in order to
    # verify that the Watcher is actually using a Qt Signal.
    w.qtdispatcher.signal.connect(c)
    w.start()

    touch(input)
    time.sleep(0.5)
    if not await_condition(output_exists, qt_app=qt_app):
        assert False, 'Output file not created...'
    assert c.count == 1

    # Stop watcher
    w.stop()
    w.join()


def test_retry():
    """Test retry decorator"""

    @retry(5, interval=0)
    def succeeds_after(n, counter):
        counter()
        if n <= counter.count:
            return True
        raise ValueError

    # Succeed when attempts < retries
    assert succeeds_after(4, CallCounter())

    # Fails when retries < attemps
    with pytest.raises(ValueError):
        assert succeeds_after(6, CallCounter())

    @retry(5, interval=0)
    def fails():
        raise ValueError

    # Most obvious case
    with pytest.raises(ValueError):
        fails()


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 os.path import dirname, exists
15			import os
16			import shutil
17			import sys
18			import time
19
20			# Third party imports
21			import pytest
22
23			# Local imports
24			#Local imports
25			from qtsass import compile_filename
26			from qtsass.watchers import PollingWatcher, QtWatcher
27			from qtsass.watchers.api import retry
28
29			# Local imports
30			from . import EXAMPLES_DIR, await_condition, example, touch
31
32
33			class CallCounter(object):
34
35			def __init__(self):
36			self.count = 0
37
38			def __call__(self, args, *kwargs):
39			self.count += 1
40
41
42			@pytest.mark.parametrize(
43			'Watcher', (PollingWatcher, QtWatcher),
44			)
45			def test_watchers(Watcher, tmpdir):
46			"""Stress test Watcher implementations"""
47
48			# Skip when QtWatcher is None - when Qt is not installed.
49			if not Watcher:
50			return
51
52			watch_dir = tmpdir.join('src').strpath
53			os.makedirs(watch_dir)
54			shutil.copy2(example('dummy.scss'), watch_dir)
55			input = tmpdir.join('src/dummy.scss').strpath
56			output = tmpdir.join('build/dummy.css').strpath
57			output_exists = lambda: exists(output)
			0 ignored issues – show introduced 2019-09-11 23:41 UTC by Report Bug Copy Issue Report The variable `output` does not seem to be defined for all execution paths. Loading history...
58
59			c = CallCounter()
60			w = Watcher(
61			watch_dir=watch_dir,
62			compiler=compile_filename,
63			args=(input, output),
64			)
65			w.connect(c)
66
67			# Output should not yet exist
68			assert not exists(output)
69
70			w.start()
71
72			touch(input)
73			time.sleep(0.5)
74			if not await_condition(output_exists):
75			assert False, 'Output file not created...'
76
77			# Removing the watch_dir should not kill the Watcher
78			# simply stop dispatching callbacks
79			shutil.rmtree(watch_dir)
80			time.sleep(0.5)
81			assert c.count == 1
82
83			# Watcher should recover once the input file is there again
84			os.makedirs(watch_dir)
85			shutil.copy2(example('dummy.scss'), watch_dir)
86			time.sleep(0.5)
87			assert c.count == 2
88
89			# Stop watcher
90			w.stop()
91			w.join()
92
93			for _ in range(5):
94			touch(input)
95
96			# Count should not change
97			assert c.count == 2
98
99
100			@pytest.mark.skipif(sys.platform.startswith('linux') or not QtWatcher,
101			reason="Fails on linux")
102			def test_qtwatcher(tmpdir):
103			"""Test QtWatcher implementation."""
104			# Constructing a QApplication will cause the QtWatcher constructed
105			# below to use a Signal to dispatch callbacks.
106			from qtsass.watchers.qt import QApplication
107
108			qt_app = QApplication.instance()
109			if not qt_app:
110			qt_app = QApplication([])
111
112			watch_dir = tmpdir.join('src').strpath
113			os.makedirs(watch_dir)
114			shutil.copy2(example('dummy.scss'), watch_dir)
115			input = tmpdir.join('src/dummy.scss').strpath
116			output = tmpdir.join('build/dummy.css').strpath
117			output_exists = lambda: exists(output)
118
119			c = CallCounter()
120			w = QtWatcher(
121			watch_dir=watch_dir,
122			compiler=compile_filename,
123			args=(input, output),
124			)
125			# We connect a counter directly to the Watcher's Qt Signal in order to
126			# verify that the Watcher is actually using a Qt Signal.
127			w.qtdispatcher.signal.connect(c)
128			w.start()
129
130			touch(input)
131			time.sleep(0.5)
132			if not await_condition(output_exists, qt_app=qt_app):
133			assert False, 'Output file not created...'
134			assert c.count == 1
135
136			# Stop watcher
137			w.stop()
138			w.join()
139
140
141			def test_retry():
142			"""Test retry decorator"""
143
144			@retry(5, interval=0)
145			def succeeds_after(n, counter):
146			counter()
147			if n <= counter.count:
148			return True
149			raise ValueError
150
151			# Succeed when attempts < retries
152			assert succeeds_after(4, CallCounter())
153
154			# Fails when retries < attemps
155			with pytest.raises(ValueError):
156			assert succeeds_after(6, CallCounter())
157
158			@retry(5, interval=0)
159			def fails():
160			raise ValueError
161
162			# Most obvious case
163			with pytest.raises(ValueError):
164			fails()
165

spyder-ide / qtsass

GitHub Access Token became invalid

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

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

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