print_mock_progress() - Code Metrics - Inspection of "skip fd capture test under windows" - IDSIA/sacred - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 193896...29ec47 )

by Klaus

created 2017-05-07 21:40 UTC

print_mock_progress() A

↳ Parent: test_stdout_capturing()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	4
Ratio	100 %

Importance

Changes	2
Bugs	1	Features	1

Metric	Value
cc	2
c	2
b	1
f	1
dl	4
loc	4
rs	10

#!/usr/bin/env python
# coding=utf-8
from __future__ import division, print_function, unicode_literals
from datetime import datetime
import mock
import os
import pytest
import tempfile
import sys

from sacred.run import Run
from sacred.config.config_summary import ConfigSummary
from sacred.utils import (ObserverError, SacredInterrupt, TimeoutInterrupt,
                          apply_backspaces_and_linefeeds)


@pytest.fixture
def run():
    config = {'a': 17, 'foo': {'bar': True, 'baz': False}, 'seed': 1234}
    config_mod = ConfigSummary()
    signature = mock.Mock()
    signature.name = 'main_func'
    main_func = mock.Mock(return_value=123, prefix='', signature=signature)
    logger = mock.Mock()
    observer = [mock.Mock(priority=10)]
    return Run(config, config_mod, main_func, observer, logger, logger, {},
               {}, [], [])


def test_run_attributes(run):
    assert isinstance(run.config, dict)
    assert isinstance(run.config_modifications, ConfigSummary)
    assert isinstance(run.experiment_info, dict)
    assert isinstance(run.host_info, dict)
    assert isinstance(run.info, dict)


def test_run_state_attributes(run):
    assert run.start_time is None
    assert run.stop_time is None
    assert run.captured_out is None
    assert run.result is None


def test_run_run(run):
    assert run() == 123
    assert (run.start_time - datetime.utcnow()).total_seconds() < 1
    assert (run.stop_time - datetime.utcnow()).total_seconds() < 1
    assert run.result == 123
    assert run.captured_out == ''


def test_run_emits_events_if_successful(run):
    run()

    observer = run.observers[0]
    assert observer.started_event.called
    assert observer.heartbeat_event.called
    assert observer.completed_event.called
    assert not observer.interrupted_event.called
    assert not observer.failed_event.called


@pytest.mark.parametrize('exception,status', [
    (KeyboardInterrupt, 'INTERRUPTED'),
    (SacredInterrupt, 'INTERRUPTED'),
    (TimeoutInterrupt, 'TIMEOUT'),
])
def test_run_emits_events_if_interrupted(run, exception, status):
    observer = run.observers[0]
    run.main_function.side_effect = exception
    with pytest.raises(exception):
        run()
    assert observer.started_event.called
    assert observer.heartbeat_event.called
    assert not observer.completed_event.called
    assert observer.interrupted_event.called
    observer.interrupted_event.assert_called_with(
        interrupt_time=run.stop_time,
        status=status)
    assert not observer.failed_event.called


def test_run_emits_events_if_failed(run):

    observer = run.observers[0]
    run.main_function.side_effect = TypeError
    with pytest.raises(TypeError):
        run()
    assert observer.started_event.called
    assert observer.heartbeat_event.called
    assert not observer.completed_event.called
    assert not observer.interrupted_event.called
    assert observer.failed_event.called


def test_run_started_event(run):
    observer = run.observers[0]
    run()
    observer.started_event.assert_called_with(
        command='main_func',
        ex_info=run.experiment_info,
        host_info=run.host_info,
        start_time=run.start_time,
        config=run.config,
        meta_info={},
        _id=None
    )


def test_run_completed_event(run):
    observer = run.observers[0]
    run()
    observer.completed_event.assert_called_with(
        stop_time=run.stop_time,
        result=run.result
    )


def test_run_heartbeat_event(run):
    observer = run.observers[0]
    run.info['test'] = 321
    run()
    call_args, call_kwargs = observer.heartbeat_event.call_args_list[0]
    assert call_kwargs['info'] == run.info
    assert call_kwargs['captured_out'] == ""
    assert (call_kwargs['beat_time'] - datetime.utcnow()).total_seconds() < 1


def test_run_artifact_event(run):
    observer = run.observers[0]
    handle, f_name = tempfile.mkstemp()
    run.add_artifact(f_name, name='foobar')
    observer.artifact_event.assert_called_with(filename=f_name, name='foobar')
    os.close(handle)
    os.remove(f_name)


def test_run_resource_event(run):
    observer = run.observers[0]
    handle, f_name = tempfile.mkstemp()
    run.open_resource(f_name)
    observer.resource_event.assert_called_with(filename=f_name)
    os.close(handle)
    os.remove(f_name)


def test_run_cannot_be_started_twice(run):
    run()
    with pytest.raises(RuntimeError):
        run()


def test_run_observer_failure_on_startup_not_caught(run):
    observer = run.observers[0]
    observer.started_event.side_effect = ObserverError
    with pytest.raises(ObserverError):
        run()


def test_run_observer_error_in_heartbeat_is_caught(run):
    observer = run.observers[0]
    observer.heartbeat_event.side_effect = ObserverError
    run()
    assert observer in run._failed_observers
    assert observer.started_event.called
    assert observer.heartbeat_event.called
    assert observer.completed_event.called


def test_run_exception_in_heartbeat_is_not_caught(run):

    observer = run.observers[0]
    observer.heartbeat_event.side_effect = TypeError
    with pytest.raises(TypeError):
        run()
    assert observer in run._failed_observers
    assert observer.started_event.called
    assert observer.heartbeat_event.called
    assert not observer.completed_event.called
    assert not observer.interrupted_event.called
    # assert observer.failed_event.called  # TODO: make this happen


def test_run_exception_in_completed_event_is_caught(run):
    observer = run.observers[0]
    observer2 = mock.Mock(priority=20)
    run.observers.append(observer2)
    observer.completed_event.side_effect = TypeError
    run()
    assert observer.completed_event.called
    assert observer2.completed_event.called


def test_run_exception_in_interrupted_event_is_caught(run):

    observer = run.observers[0]
    observer2 = mock.Mock(priority=20)
    run.observers.append(observer2)
    observer.interrupted_event.side_effect = TypeError
    run.main_function.side_effect = KeyboardInterrupt
    with pytest.raises(KeyboardInterrupt):
        run()
    assert observer.interrupted_event.called
    assert observer2.interrupted_event.called


def test_run_exception_in_failed_event_is_caught(run):

    observer = run.observers[0]
    observer2 = mock.Mock(priority=20)
    run.observers.append(observer2)
    observer.failed_event.side_effect = TypeError
    run.main_function.side_effect = AttributeError
    with pytest.raises(AttributeError):
        run()
    assert observer.failed_event.called
    assert observer2.failed_event.called


def test_unobserved_run_doesnt_emit(run):
    observer = run.observers[0]
    run.unobserved = True
    run()
    assert not observer.started_event.called
    assert not observer.heartbeat_event.called
    assert not observer.completed_event.called
    assert not observer.interrupted_event.called
    assert not observer.failed_event.called


@pytest.mark.parametrize("capture_mode", ["no", "sys", "fd"])

def test_stdout_capturing(run, capsys, capture_mode):
    if sys.platform.startswith('win') and capture_mode == "fd":
        return  # skip fd mode under windows

    def print_mock_progress():
        for i in range(10):
            print(i, end="")
        sys.stdout.flush()

    run.main_function.side_effect = print_mock_progress
    run.capture_mode = capture_mode
    with capsys.disabled():
        run()
    if capture_mode != "no":
        assert run.captured_out == '0123456789'


def test_captured_out_filter(run, capsys):
    def print_mock_progress():
        sys.stdout.write('progress 0')
        sys.stdout.flush()
        for i in range(10):
            sys.stdout.write('\b')
            sys.stdout.write(str(i))
            sys.stdout.flush()

    run.captured_out_filter = apply_backspaces_and_linefeeds
    run.main_function.side_effect = print_mock_progress
    with capsys.disabled():
        run()
        assert run.captured_out == 'progress 9'


1		#!/usr/bin/env python
2		# coding=utf-8
3		from __future__ import division, print_function, unicode_literals
4		from datetime import datetime
5		import mock
6		import os
7		import pytest
8		import tempfile
9		import sys
10
11		from sacred.run import Run
12		from sacred.config.config_summary import ConfigSummary
13		from sacred.utils import (ObserverError, SacredInterrupt, TimeoutInterrupt,
14		apply_backspaces_and_linefeeds)
15
16
17		@pytest.fixture
18		def run():
19		config = {'a': 17, 'foo': {'bar': True, 'baz': False}, 'seed': 1234}
20		config_mod = ConfigSummary()
21		signature = mock.Mock()
22		signature.name = 'main_func'
23		main_func = mock.Mock(return_value=123, prefix='', signature=signature)
24		logger = mock.Mock()
25		observer = [mock.Mock(priority=10)]
26		return Run(config, config_mod, main_func, observer, logger, logger, {},
27		{}, [], [])
28
29
30		def test_run_attributes(run):
31		assert isinstance(run.config, dict)
32		assert isinstance(run.config_modifications, ConfigSummary)
33		assert isinstance(run.experiment_info, dict)
34		assert isinstance(run.host_info, dict)
35		assert isinstance(run.info, dict)
36
37
38		def test_run_state_attributes(run):
39		assert run.start_time is None
40		assert run.stop_time is None
41		assert run.captured_out is None
42		assert run.result is None
43
44
45		def test_run_run(run):
46		assert run() == 123
47		assert (run.start_time - datetime.utcnow()).total_seconds() < 1
48		assert (run.stop_time - datetime.utcnow()).total_seconds() < 1
49		assert run.result == 123
50		assert run.captured_out == ''
51
52
53		def test_run_emits_events_if_successful(run):
54		run()
55
56		observer = run.observers[0]
57		assert observer.started_event.called
58		assert observer.heartbeat_event.called
59		assert observer.completed_event.called
60		assert not observer.interrupted_event.called
61		assert not observer.failed_event.called
62
63
64		@pytest.mark.parametrize('exception,status', [
65		(KeyboardInterrupt, 'INTERRUPTED'),
66		(SacredInterrupt, 'INTERRUPTED'),
67		(TimeoutInterrupt, 'TIMEOUT'),
68		])
69		def test_run_emits_events_if_interrupted(run, exception, status):
70		observer = run.observers[0]
71		run.main_function.side_effect = exception
72		with pytest.raises(exception):
73		run()
74		assert observer.started_event.called
75		assert observer.heartbeat_event.called
76		assert not observer.completed_event.called
77		assert observer.interrupted_event.called
78		observer.interrupted_event.assert_called_with(
79		interrupt_time=run.stop_time,
80		status=status)
81		assert not observer.failed_event.called
82
83
84	View Code Duplication	def test_run_emits_events_if_failed(run):
		0 ignored issues – show Duplication introduced 2016-07-12 19:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
85		observer = run.observers[0]
86		run.main_function.side_effect = TypeError
87		with pytest.raises(TypeError):
88		run()
89		assert observer.started_event.called
90		assert observer.heartbeat_event.called
91		assert not observer.completed_event.called
92		assert not observer.interrupted_event.called
93		assert observer.failed_event.called
94
95
96		def test_run_started_event(run):
97		observer = run.observers[0]
98		run()
99		observer.started_event.assert_called_with(
100		command='main_func',
101		ex_info=run.experiment_info,
102		host_info=run.host_info,
103		start_time=run.start_time,
104		config=run.config,
105		meta_info={},
106		_id=None
107		)
108
109
110		def test_run_completed_event(run):
111		observer = run.observers[0]
112		run()
113		observer.completed_event.assert_called_with(
114		stop_time=run.stop_time,
115		result=run.result
116		)
117
118
119		def test_run_heartbeat_event(run):
120		observer = run.observers[0]
121		run.info['test'] = 321
122		run()
123		call_args, call_kwargs = observer.heartbeat_event.call_args_list[0]
124		assert call_kwargs['info'] == run.info
125		assert call_kwargs['captured_out'] == ""
126		assert (call_kwargs['beat_time'] - datetime.utcnow()).total_seconds() < 1
127
128
129		def test_run_artifact_event(run):
130		observer = run.observers[0]
131		handle, f_name = tempfile.mkstemp()
132		run.add_artifact(f_name, name='foobar')
133		observer.artifact_event.assert_called_with(filename=f_name, name='foobar')
134		os.close(handle)
135		os.remove(f_name)
136
137
138		def test_run_resource_event(run):
139		observer = run.observers[0]
140		handle, f_name = tempfile.mkstemp()
141		run.open_resource(f_name)
142		observer.resource_event.assert_called_with(filename=f_name)
143		os.close(handle)
144		os.remove(f_name)
145
146
147		def test_run_cannot_be_started_twice(run):
148		run()
149		with pytest.raises(RuntimeError):
150		run()
151
152
153		def test_run_observer_failure_on_startup_not_caught(run):
154		observer = run.observers[0]
155		observer.started_event.side_effect = ObserverError
156		with pytest.raises(ObserverError):
157		run()
158
159
160		def test_run_observer_error_in_heartbeat_is_caught(run):
161		observer = run.observers[0]
162		observer.heartbeat_event.side_effect = ObserverError
163		run()
164		assert observer in run._failed_observers
165		assert observer.started_event.called
166		assert observer.heartbeat_event.called
167		assert observer.completed_event.called
168
169
170	View Code Duplication	def test_run_exception_in_heartbeat_is_not_caught(run):
		0 ignored issues – show Duplication introduced 2016-08-02 17:07 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
171		observer = run.observers[0]
172		observer.heartbeat_event.side_effect = TypeError
173		with pytest.raises(TypeError):
174		run()
175		assert observer in run._failed_observers
176		assert observer.started_event.called
177		assert observer.heartbeat_event.called
178		assert not observer.completed_event.called
179		assert not observer.interrupted_event.called
180		# assert observer.failed_event.called # TODO: make this happen
181
182
183		def test_run_exception_in_completed_event_is_caught(run):
184		observer = run.observers[0]
185		observer2 = mock.Mock(priority=20)
186		run.observers.append(observer2)
187		observer.completed_event.side_effect = TypeError
188		run()
189		assert observer.completed_event.called
190		assert observer2.completed_event.called
191
192
193	View Code Duplication	def test_run_exception_in_interrupted_event_is_caught(run):
		0 ignored issues – show Duplication introduced 2017-01-16 09:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
194		observer = run.observers[0]
195		observer2 = mock.Mock(priority=20)
196		run.observers.append(observer2)
197		observer.interrupted_event.side_effect = TypeError
198		run.main_function.side_effect = KeyboardInterrupt
199		with pytest.raises(KeyboardInterrupt):
200		run()
201		assert observer.interrupted_event.called
202		assert observer2.interrupted_event.called
203
204
205	View Code Duplication	def test_run_exception_in_failed_event_is_caught(run):
		0 ignored issues – show Duplication introduced 2017-01-16 09:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
206		observer = run.observers[0]
207		observer2 = mock.Mock(priority=20)
208		run.observers.append(observer2)
209		observer.failed_event.side_effect = TypeError
210		run.main_function.side_effect = AttributeError
211		with pytest.raises(AttributeError):
212		run()
213		assert observer.failed_event.called
214		assert observer2.failed_event.called
215
216
217		def test_unobserved_run_doesnt_emit(run):
218		observer = run.observers[0]
219		run.unobserved = True
220		run()
221		assert not observer.started_event.called
222		assert not observer.heartbeat_event.called
223		assert not observer.completed_event.called
224		assert not observer.interrupted_event.called
225		assert not observer.failed_event.called
226
227
228	View Code Duplication	@pytest.mark.parametrize("capture_mode", ["no", "sys", "fd"])
		0 ignored issues – show Duplication introduced 2017-05-07 21:10 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
229		def test_stdout_capturing(run, capsys, capture_mode):
230		if sys.platform.startswith('win') and capture_mode == "fd":
231		return # skip fd mode under windows
232
233		def print_mock_progress():
234		for i in range(10):
235		print(i, end="")
236		sys.stdout.flush()
237
238		run.main_function.side_effect = print_mock_progress
239		run.capture_mode = capture_mode
240		with capsys.disabled():
241		run()
242		if capture_mode != "no":
243		assert run.captured_out == '0123456789'
244
245
246		def test_captured_out_filter(run, capsys):
247		def print_mock_progress():
248		sys.stdout.write('progress 0')
249		sys.stdout.flush()
250		for i in range(10):
251		sys.stdout.write('\b')
252		sys.stdout.write(str(i))
253		sys.stdout.flush()
254
255		run.captured_out_filter = apply_backspaces_and_linefeeds
256		run.main_function.side_effect = print_mock_progress
257		with capsys.disabled():
258		run()
259		assert run.captured_out == 'progress 9'
260

IDSIA / sacred

Push — master ( 193896...29ec47 )

print_mock_progress() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like