strip_file_handles() - Code Metrics - IDSIA/sacred - Measure and Improve Code Quality continuously with Scrutinizer

strip_file_handles() D
last analyzed 2018-07-01 19:23 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	12
c	0
b	0
f	0
dl	0
loc	30
rs	4.8

How to fix Complexity

#!/usr/bin/env python
# coding=utf-8
from __future__ import (division, print_function, unicode_literals,
                        absolute_import)

import datetime
import os
import tempfile
import io

import pytest

tinydb = pytest.importorskip("tinydb")
hashfs = pytest.importorskip("hashfs")

from tinydb import Query

from sacred.dependencies import get_digest
from sacred.observers.tinydb_hashfs import TinyDbObserver, TinyDbReader


# Utilities and fixtures
@pytest.fixture
def sample_run():

    T1 = datetime.datetime(1999, 5, 4, 3, 2, 1, 0)

    exp = {
        'name': 'test_exp',
        'sources': [],
        'doc': '',
        'base_dir': os.path.join(os.path.dirname(__file__), '..', '..'),
        'dependencies': ['sacred==0.7b0']
    }
    host = {'hostname': 'test_host', 'cpu_count': 1, 'python_version': '3.4'}
    config = {'config': 'True', 'foo': 'bar', 'answer': 42}
    command = 'run'
    meta_info = {'comment': 'test run'}
    sample_run = {
        '_id': 'FED235DA13',
        'ex_info': exp,
        'command': command,
        'host_info': host,
        'start_time': T1,
        'config': config,
        'meta_info': meta_info,
    }

    filename = 'setup.py'
    md5 = get_digest(filename)
    sample_run['ex_info']['sources'] = [[filename, md5]]

    return sample_run


def run_test_experiment(exp_name, exp_id, root_dir):

    T2 = datetime.datetime(1999, 5, 5, 5, 5, 5, 5)
    T3 = datetime.datetime(1999, 5, 5, 6, 6, 6, 6)

    run_date = sample_run()
    run_date['ex_info']['name'] = exp_name
    run_date['_id'] = exp_id

    # Create
    tinydb_obs = TinyDbObserver.create(path=root_dir)

    # Start exp 1
    tinydb_obs.started_event(**run_date)

    # Heartbeat
    info = {'my_info': [1, 2, 3], 'nr': 7}
    outp = 'some output'
    tinydb_obs.heartbeat_event(info=info, captured_out=outp, beat_time=T2,
                               result=7)
    # Add Artifact
    filename = "sacred/__about__.py"
    name = 'about'
    tinydb_obs.artifact_event(name, filename)

    # Add Resource
    filename = "sacred/__init__.py"
    tinydb_obs.resource_event(filename)

    # Complete
    tinydb_obs.completed_event(stop_time=T3, result=42)

    return tinydb_obs


def strip_file_handles(results):
    """Return a database result set with all file handle objects removed.abs

    Utility function to aid comparison of database entries. As file handles are
    created newly each object, these are always different so can be excluded.
    """

    if not isinstance(results, (list, tuple)):
        results = [results]

    cleaned_results = []
    for result in results:
        sources = result['experiment']['sources']
        artifacts = result['artifacts']
        resources = result['resources']
        if sources:
            for src in sources:
                if isinstance(src[-1], io.BufferedReader):
                    del src[-1]
        if artifacts:
            for art in artifacts:
                if isinstance(art[-1], io.BufferedReader):
                    del art[-1]
        if resources:
            for res in resources:
                if isinstance(res[-1], io.BufferedReader):
                    del res[-1]
        cleaned_results.append(result)

    return cleaned_results


# TinyDbReader Tests
def test_tinydb_reader_loads_db_and_fs(tmpdir):

    root = tmpdir.strpath
    tinydb_obs = run_test_experiment(exp_name='exp1', exp_id='1234', root_dir=root)
    tinydb_reader = TinyDbReader(root)

    assert tinydb_obs.fs.root == tinydb_reader.fs.root
    # Different file handles are used in each object so compar based on str
    # representation
    assert str(tinydb_obs.runs.all()[0]) == str(tinydb_reader.runs.all()[0])


def test_tinydb_reader_raises_exceptions(tmpdir):
    with pytest.raises(IOError):
        TinyDbReader('foo')


def test_fetch_metadata_function_with_indices(tmpdir, sample_run):

    # Setup and run three experiments
    root = tmpdir.strpath
    tinydb_obs = run_test_experiment(exp_name='experiment 1 alpha',
                                     exp_id='1234', root_dir=root)
    tinydb_obs = run_test_experiment(exp_name='experiment 2 beta',
                                     exp_id='5678', root_dir=root)
    tinydb_obs = run_test_experiment(exp_name='experiment 3 alpha',
                                     exp_id='9990', root_dir=root)

    tinydb_reader = TinyDbReader(root)

    # Test fetch by indices
    res = tinydb_reader.fetch_metadata(indices=-1)
    res2 = tinydb_reader.fetch_metadata(indices=[-1])
    assert strip_file_handles(res) == strip_file_handles(res2)
    res3 = tinydb_reader.fetch_metadata(indices=[0, -1])
    assert len(res3) == 2

    exp1_res = tinydb_reader.fetch_metadata(indices=0)
    assert len(exp1_res) == 1
    assert exp1_res[0]['experiment']['name'] == 'experiment 1 alpha'
    assert exp1_res[0]['_id'] == '1234'

    # Test Exception
    with pytest.raises(ValueError):
        tinydb_reader.fetch_metadata(indices=4)

    # Test returned values
    exp1 = strip_file_handles(exp1_res)[0]

    sample_run['ex_info']['name'] = 'experiment 1 alpha'
    sample_run['ex_info']['sources'] = [
        ['setup.py', get_digest('setup.py')]
    ]

    assert exp1 == {
        '_id': '1234',
        'experiment': sample_run['ex_info'],
        'format': tinydb_obs.VERSION,
        'command': sample_run['command'],
        'host': sample_run['host_info'],
        'start_time': sample_run['start_time'],
        'heartbeat': datetime.datetime(1999, 5, 5, 5, 5, 5, 5),
        'info': {'my_info': [1, 2, 3], 'nr': 7},
        'captured_out': 'some output',
        'artifacts': [
            ['about', 'sacred/__about__.py', get_digest('sacred/__about__.py')]
        ],
        'config': sample_run['config'],
        'meta': sample_run['meta_info'],
        'status': 'COMPLETED',
        'resources': [
            ['sacred/__init__.py', get_digest('sacred/__init__.py')]
        ],
        'result': 42,
        'stop_time': datetime.datetime(1999, 5, 5, 6, 6, 6, 6)
    }


def test_fetch_metadata_function_with_exp_name(tmpdir):

    # Setup and run three experiments
    root = tmpdir.strpath
    run_test_experiment(exp_name='experiment 1 alpha',
                        exp_id='1234', root_dir=root)
    run_test_experiment(exp_name='experiment 2 beta',
                        exp_id='5678', root_dir=root)
    run_test_experiment(exp_name='experiment 3 alpha',
                        exp_id='9990', root_dir=root)

    tinydb_reader = TinyDbReader(root)

    # Test Fetch by exp name
    res1 = tinydb_reader.fetch_metadata(exp_name='alpha')
    assert len(res1) == 2
    res2 = tinydb_reader.fetch_metadata(exp_name='experiment 1')
    assert len(res2) == 1
    assert res2[0]['experiment']['name'] == 'experiment 1 alpha'
    res2 = tinydb_reader.fetch_metadata(exp_name='foo')
    assert len(res2) == 0


def test_fetch_metadata_function_with_querry(tmpdir):

    # Setup and run three experiments
    root = tmpdir.strpath
    run_test_experiment(exp_name='experiment 1 alpha',
                        exp_id='1234', root_dir=root)
    run_test_experiment(exp_name='experiment 2 beta',
                        exp_id='5678', root_dir=root)
    run_test_experiment(exp_name='experiment 3 alpha beta',
                        exp_id='9990', root_dir=root)

    tinydb_reader = TinyDbReader(root)

    record = Query()

    exp1_query = record.experiment.name.matches('.*alpha$')

    exp3_query = (
        (record.experiment.name.search('alpha')) &
        (record._id == '9990')
    )

    # Test Fetch by Tinydb Query
    res1 = tinydb_reader.fetch_metadata(query=exp1_query)
    assert len(res1) == 1
    assert res1[0]['experiment']['name'] == 'experiment 1 alpha'

    res2 = tinydb_reader.fetch_metadata(
        query=record.experiment.name.search('experiment [23]'))
    assert len(res2) == 2

    res3 = tinydb_reader.fetch_metadata(query=exp3_query)
    assert len(res3) == 1
    assert res3[0]['experiment']['name'] == 'experiment 3 alpha beta'

    # Test Exception
    with pytest.raises(ValueError):
        tinydb_reader.fetch_metadata()


def test_search_function(tmpdir):

    # Setup and run three experiments
    root = tmpdir.strpath
    run_test_experiment(exp_name='experiment 1 alpha',
                        exp_id='1234', root_dir=root)
    run_test_experiment(exp_name='experiment 2 beta',
                        exp_id='5678', root_dir=root)
    run_test_experiment(exp_name='experiment 3 alpha beta',
                        exp_id='9990', root_dir=root)

    tinydb_reader = TinyDbReader(root)

    # Test Fetch by Tinydb Query in search function
    record = Query()
    q = record.experiment.name.search('experiment [23]')

    res = tinydb_reader.search(q)
    assert len(res) == 2
    res2 = tinydb_reader.fetch_metadata(query=q)
    assert strip_file_handles(res) == strip_file_handles(res2)


def test_fetch_files_function(tmpdir):
    # Setup and run three experiments
    root = tmpdir.strpath
    run_test_experiment(exp_name='experiment 1 alpha',
                        exp_id='1234', root_dir=root)
    run_test_experiment(exp_name='experiment 2 beta',
                        exp_id='5678', root_dir=root)
    run_test_experiment(exp_name='experiment 3 alpha beta',
                        exp_id='9990', root_dir=root)

    tinydb_reader = TinyDbReader(root)

    res = tinydb_reader.fetch_files(indices=0)
    assert len(res) == 1
    assert list(res[0]['artifacts'].keys()) == ['about']
    assert isinstance(res[0]['artifacts']['about'], io.BufferedReader)
    assert res[0]['date'] == datetime.datetime(1999, 5, 4, 3, 2, 1)
    assert res[0]['exp_id'] == '1234'
    assert res[0]['exp_name'] == 'experiment 1 alpha'
    assert list(res[0]['resources'].keys()) == ['sacred/__init__.py']
    assert isinstance(res[0]['resources']['sacred/__init__.py'], io.BufferedReader)
    assert list(res[0]['sources'].keys()) == ['setup.py']
    assert isinstance(res[0]['sources']['setup.py'], io.BufferedReader)


def test_fetch_report_function(tmpdir):

    # Setup and run three experiments
    root = tmpdir.strpath
    run_test_experiment(exp_name='experiment 1 alpha',
                        exp_id='1234', root_dir=root)
    run_test_experiment(exp_name='experiment 2 beta',
                        exp_id='5678', root_dir=root)
    run_test_experiment(exp_name='experiment 3 alpha beta',
                        exp_id='9990', root_dir=root)

    tinydb_reader = TinyDbReader(root)

    res = tinydb_reader.fetch_report(indices=0)

    target = """
-------------------------------------------------
Experiment: experiment 1 alpha
-------------------------------------------------
ID: 1234
Date: Tue 04 May 1999    Duration: 27:04:05.0

Parameters:
    answer: 42
    config: True
    foo: bar

Result:
    42

Dependencies:
    sacred==0.7b0

Resources:
    sacred/__init__.py

Source Files:
    setup.py

Outputs:
    about
"""

    assert res[0] == target


1			#!/usr/bin/env python
2			# coding=utf-8
3			from __future__ import (division, print_function, unicode_literals,
4			absolute_import)
5
6			import datetime
7			import os
8			import tempfile
9			import io
10
11			import pytest
12
13			tinydb = pytest.importorskip("tinydb")
14			hashfs = pytest.importorskip("hashfs")
15
16			from tinydb import Query
17
18			from sacred.dependencies import get_digest
19			from sacred.observers.tinydb_hashfs import TinyDbObserver, TinyDbReader
20
21
22			# Utilities and fixtures
23			@pytest.fixture
24			def sample_run():
25
26			T1 = datetime.datetime(1999, 5, 4, 3, 2, 1, 0)
27
28			exp = {
29			'name': 'test_exp',
30			'sources': [],
31			'doc': '',
32			'base_dir': os.path.join(os.path.dirname(__file__), '..', '..'),
33			'dependencies': ['sacred==0.7b0']
34			}
35			host = {'hostname': 'test_host', 'cpu_count': 1, 'python_version': '3.4'}
36			config = {'config': 'True', 'foo': 'bar', 'answer': 42}
37			command = 'run'
38			meta_info = {'comment': 'test run'}
39			sample_run = {
40			'_id': 'FED235DA13',
41			'ex_info': exp,
42			'command': command,
43			'host_info': host,
44			'start_time': T1,
45			'config': config,
46			'meta_info': meta_info,
47			}
48
49			filename = 'setup.py'
50			md5 = get_digest(filename)
51			sample_run['ex_info']['sources'] = [[filename, md5]]
52
53			return sample_run
54
55
56			def run_test_experiment(exp_name, exp_id, root_dir):
57
58			T2 = datetime.datetime(1999, 5, 5, 5, 5, 5, 5)
59			T3 = datetime.datetime(1999, 5, 5, 6, 6, 6, 6)
60
61			run_date = sample_run()
62			run_date['ex_info']['name'] = exp_name
63			run_date['_id'] = exp_id
64
65			# Create
66			tinydb_obs = TinyDbObserver.create(path=root_dir)
67
68			# Start exp 1
69			tinydb_obs.started_event(**run_date)
70
71			# Heartbeat
72			info = {'my_info': [1, 2, 3], 'nr': 7}
73			outp = 'some output'
74			tinydb_obs.heartbeat_event(info=info, captured_out=outp, beat_time=T2,
75			result=7)
76			# Add Artifact
77			filename = "sacred/__about__.py"
78			name = 'about'
79			tinydb_obs.artifact_event(name, filename)
80
81			# Add Resource
82			filename = "sacred/__init__.py"
83			tinydb_obs.resource_event(filename)
84
85			# Complete
86			tinydb_obs.completed_event(stop_time=T3, result=42)
87
88			return tinydb_obs
89
90
91			def strip_file_handles(results):
92			"""Return a database result set with all file handle objects removed.abs
93
94			Utility function to aid comparison of database entries. As file handles are
95			created newly each object, these are always different so can be excluded.
96			"""
97
98			if not isinstance(results, (list, tuple)):
99			results = [results]
100
101			cleaned_results = []
102			for result in results:
103			sources = result['experiment']['sources']
104			artifacts = result['artifacts']
105			resources = result['resources']
106			if sources:
107			for src in sources:
108			if isinstance(src[-1], io.BufferedReader):
109			del src[-1]
110			if artifacts:
111			for art in artifacts:
112			if isinstance(art[-1], io.BufferedReader):
113			del art[-1]
114			if resources:
115			for res in resources:
116			if isinstance(res[-1], io.BufferedReader):
117			del res[-1]
118			cleaned_results.append(result)
119
120			return cleaned_results
121
122
123			# TinyDbReader Tests
124			def test_tinydb_reader_loads_db_and_fs(tmpdir):
125
126			root = tmpdir.strpath
127			tinydb_obs = run_test_experiment(exp_name='exp1', exp_id='1234', root_dir=root)
128			tinydb_reader = TinyDbReader(root)
129
130			assert tinydb_obs.fs.root == tinydb_reader.fs.root
131			# Different file handles are used in each object so compar based on str
132			# representation
133			assert str(tinydb_obs.runs.all()[0]) == str(tinydb_reader.runs.all()[0])
134
135
136			def test_tinydb_reader_raises_exceptions(tmpdir):
137			with pytest.raises(IOError):
138			TinyDbReader('foo')
139
140
141			def test_fetch_metadata_function_with_indices(tmpdir, sample_run):
142
143			# Setup and run three experiments
144			root = tmpdir.strpath
145			tinydb_obs = run_test_experiment(exp_name='experiment 1 alpha',
146			exp_id='1234', root_dir=root)
147			tinydb_obs = run_test_experiment(exp_name='experiment 2 beta',
148			exp_id='5678', root_dir=root)
149			tinydb_obs = run_test_experiment(exp_name='experiment 3 alpha',
150			exp_id='9990', root_dir=root)
151
152			tinydb_reader = TinyDbReader(root)
153
154			# Test fetch by indices
155			res = tinydb_reader.fetch_metadata(indices=-1)
156			res2 = tinydb_reader.fetch_metadata(indices=[-1])
157			assert strip_file_handles(res) == strip_file_handles(res2)
158			res3 = tinydb_reader.fetch_metadata(indices=[0, -1])
159			assert len(res3) == 2
160
161			exp1_res = tinydb_reader.fetch_metadata(indices=0)
162			assert len(exp1_res) == 1
163			assert exp1_res[0]['experiment']['name'] == 'experiment 1 alpha'
164			assert exp1_res[0]['_id'] == '1234'
165
166			# Test Exception
167			with pytest.raises(ValueError):
168			tinydb_reader.fetch_metadata(indices=4)
169
170			# Test returned values
171			exp1 = strip_file_handles(exp1_res)[0]
172
173			sample_run['ex_info']['name'] = 'experiment 1 alpha'
174			sample_run['ex_info']['sources'] = [
175			['setup.py', get_digest('setup.py')]
176			]
177
178			assert exp1 == {
179			'_id': '1234',
180			'experiment': sample_run['ex_info'],
181			'format': tinydb_obs.VERSION,
182			'command': sample_run['command'],
183			'host': sample_run['host_info'],
184			'start_time': sample_run['start_time'],
185			'heartbeat': datetime.datetime(1999, 5, 5, 5, 5, 5, 5),
186			'info': {'my_info': [1, 2, 3], 'nr': 7},
187			'captured_out': 'some output',
188			'artifacts': [
189			['about', 'sacred/__about__.py', get_digest('sacred/__about__.py')]
190			],
191			'config': sample_run['config'],
192			'meta': sample_run['meta_info'],
193			'status': 'COMPLETED',
194			'resources': [
195			['sacred/__init__.py', get_digest('sacred/__init__.py')]
196			],
197			'result': 42,
198			'stop_time': datetime.datetime(1999, 5, 5, 6, 6, 6, 6)
199			}
200
201
202			def test_fetch_metadata_function_with_exp_name(tmpdir):
203
204			# Setup and run three experiments
205			root = tmpdir.strpath
206			run_test_experiment(exp_name='experiment 1 alpha',
207			exp_id='1234', root_dir=root)
208			run_test_experiment(exp_name='experiment 2 beta',
209			exp_id='5678', root_dir=root)
210			run_test_experiment(exp_name='experiment 3 alpha',
211			exp_id='9990', root_dir=root)
212
213			tinydb_reader = TinyDbReader(root)
214
215			# Test Fetch by exp name
216			res1 = tinydb_reader.fetch_metadata(exp_name='alpha')
217			assert len(res1) == 2
218			res2 = tinydb_reader.fetch_metadata(exp_name='experiment 1')
219			assert len(res2) == 1
220			assert res2[0]['experiment']['name'] == 'experiment 1 alpha'
221			res2 = tinydb_reader.fetch_metadata(exp_name='foo')
222			assert len(res2) == 0
223
224
225			def test_fetch_metadata_function_with_querry(tmpdir):
226
227			# Setup and run three experiments
228			root = tmpdir.strpath
229			run_test_experiment(exp_name='experiment 1 alpha',
230			exp_id='1234', root_dir=root)
231			run_test_experiment(exp_name='experiment 2 beta',
232			exp_id='5678', root_dir=root)
233			run_test_experiment(exp_name='experiment 3 alpha beta',
234			exp_id='9990', root_dir=root)
235
236			tinydb_reader = TinyDbReader(root)
237
238			record = Query()
239
240			exp1_query = record.experiment.name.matches('.*alpha$')
241
242			exp3_query = (
243			(record.experiment.name.search('alpha')) &
244			(record._id == '9990')
245			)
246
247			# Test Fetch by Tinydb Query
248			res1 = tinydb_reader.fetch_metadata(query=exp1_query)
249			assert len(res1) == 1
250			assert res1[0]['experiment']['name'] == 'experiment 1 alpha'
251
252			res2 = tinydb_reader.fetch_metadata(
253			query=record.experiment.name.search('experiment [23]'))
254			assert len(res2) == 2
255
256			res3 = tinydb_reader.fetch_metadata(query=exp3_query)
257			assert len(res3) == 1
258			assert res3[0]['experiment']['name'] == 'experiment 3 alpha beta'
259
260			# Test Exception
261			with pytest.raises(ValueError):
262			tinydb_reader.fetch_metadata()
263
264
265			def test_search_function(tmpdir):
266
267			# Setup and run three experiments
268			root = tmpdir.strpath
269			run_test_experiment(exp_name='experiment 1 alpha',
270			exp_id='1234', root_dir=root)
271			run_test_experiment(exp_name='experiment 2 beta',
272			exp_id='5678', root_dir=root)
273			run_test_experiment(exp_name='experiment 3 alpha beta',
274			exp_id='9990', root_dir=root)
275
276			tinydb_reader = TinyDbReader(root)
277
278			# Test Fetch by Tinydb Query in search function
279			record = Query()
280			q = record.experiment.name.search('experiment [23]')
281
282			res = tinydb_reader.search(q)
283			assert len(res) == 2
284			res2 = tinydb_reader.fetch_metadata(query=q)
285			assert strip_file_handles(res) == strip_file_handles(res2)
286
287
288			def test_fetch_files_function(tmpdir):
289			# Setup and run three experiments
290			root = tmpdir.strpath
291			run_test_experiment(exp_name='experiment 1 alpha',
292			exp_id='1234', root_dir=root)
293			run_test_experiment(exp_name='experiment 2 beta',
294			exp_id='5678', root_dir=root)
295			run_test_experiment(exp_name='experiment 3 alpha beta',
296			exp_id='9990', root_dir=root)
297
298			tinydb_reader = TinyDbReader(root)
299
300			res = tinydb_reader.fetch_files(indices=0)
301			assert len(res) == 1
302			assert list(res[0]['artifacts'].keys()) == ['about']
303			assert isinstance(res[0]['artifacts']['about'], io.BufferedReader)
304			assert res[0]['date'] == datetime.datetime(1999, 5, 4, 3, 2, 1)
305			assert res[0]['exp_id'] == '1234'
306			assert res[0]['exp_name'] == 'experiment 1 alpha'
307			assert list(res[0]['resources'].keys()) == ['sacred/__init__.py']
308			assert isinstance(res[0]['resources']['sacred/__init__.py'], io.BufferedReader)
309			assert list(res[0]['sources'].keys()) == ['setup.py']
310			assert isinstance(res[0]['sources']['setup.py'], io.BufferedReader)
311
312
313			def test_fetch_report_function(tmpdir):
314
315			# Setup and run three experiments
316			root = tmpdir.strpath
317			run_test_experiment(exp_name='experiment 1 alpha',
318			exp_id='1234', root_dir=root)
319			run_test_experiment(exp_name='experiment 2 beta',
320			exp_id='5678', root_dir=root)
321			run_test_experiment(exp_name='experiment 3 alpha beta',
322			exp_id='9990', root_dir=root)
323
324			tinydb_reader = TinyDbReader(root)
325
326			res = tinydb_reader.fetch_report(indices=0)
327
328			target = """
329			-------------------------------------------------
330			Experiment: experiment 1 alpha
331			-------------------------------------------------
332			ID: 1234
333			Date: Tue 04 May 1999 Duration: 27:04:05.0
334
335			Parameters:
336			answer: 42
337			config: True
338			foo: bar
339
340			Result:
341			42
342
343			Dependencies:
344			sacred==0.7b0
345
346			Resources:
347			sacred/__init__.py
348
349			Source Files:
350			setup.py
351
352			Outputs:
353			about
354			"""
355
356			assert res[0] == target
357

IDSIA / sacred

strip_file_handles() D last analyzed 2018-07-01 19:23 UTC

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

strip_file_handles() D
last analyzed 2018-07-01 19:23 UTC