Repository navigation
Expand file tree
/
Copy pathtesting.py
More file actions
82 lines (65 loc) · 2.62 KB
/
Copy pathtesting.py
File metadata and controls
82 lines (65 loc) · 2.62 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import hashlib
import inspect
import os
import shutil
import time
import unittest
import openml
class TestBase(unittest.TestCase):
"""Base class for tests
Note
----
Currently hard-codes a read-write key.
Hopefully soon allows using a test server, not the production server.
"""
def setUp(self):
# This cache directory is checked in to git to simulate a populated
# cache
self.maxDiff = None
self.static_cache_dir = None
static_cache_dir = os.path.dirname(os.path.abspath(inspect.getfile(self.__class__)))
static_cache_dir = os.path.abspath(os.path.join(static_cache_dir, '..'))
content = os.listdir(static_cache_dir)
if 'files' in content:
self.static_cache_dir = os.path.join(static_cache_dir, 'files')
if self.static_cache_dir is None:
raise ValueError('Cannot find test cache dir!')
self.cwd = os.getcwd()
workdir = os.path.dirname(os.path.abspath(__file__))
self.workdir = os.path.join(workdir, "tmp")
try:
shutil.rmtree(self.workdir)
except:
pass
os.mkdir(self.workdir)
os.chdir(self.workdir)
self.cached = True
# amueller's read/write key that he will throw away later
openml.config.apikey = "610344db6388d9ba34f6db45a3cf71de"
self.production_server = openml.config.server
self.test_server = "https://test.openml.org/api/v1/xml"
openml.config.server = self.test_server
openml.config.avoid_duplicate_runs = False
openml.config.set_cache_directory(self.workdir)
def tearDown(self):
os.chdir(self.cwd)
shutil.rmtree(self.workdir)
openml.config.server = self.production_server
def _add_sentinel_to_flow_name(self, flow, sentinel=None):
if sentinel is None:
# Create a unique prefix for the flow. Necessary because the flow is
# identified by its name and external version online. Having a unique
# name allows us to publish the same flow in each test run
md5 = hashlib.md5()
md5.update(str(time.time()).encode('utf-8'))
sentinel = md5.hexdigest()[:10]
sentinel = 'TEST%s' % sentinel
flows_to_visit = list()
flows_to_visit.append(flow)
while len(flows_to_visit) > 0:
current_flow = flows_to_visit.pop()
current_flow.name = '%s%s' % (sentinel, current_flow.name)
for subflow in current_flow.components.values():
flows_to_visit.append(subflow)
return flow, sentinel
__all__ = ['TestBase']