Repository navigation
Expand file tree
/
Copy pathtest_task_functions.py
More file actions
119 lines (102 loc) · 4.48 KB
/
Copy pathtest_task_functions.py
File metadata and controls
119 lines (102 loc) · 4.48 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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
import os
import sys
if sys.version_info[0] >= 3:
from unittest import mock
else:
import mock
from openml.util import is_string
from openml.testing import TestBase
from openml import OpenMLSplit, OpenMLTask
from openml.exceptions import OpenMLCacheException
import openml
class TestTask(TestBase):
def test__get_cached_tasks(self):
openml.config.set_cache_directory(self.static_cache_dir)
tasks = openml.tasks.functions._get_cached_tasks()
self.assertIsInstance(tasks, dict)
self.assertEqual(len(tasks), 3)
self.assertIsInstance(list(tasks.values())[0], OpenMLTask)
def test__get_cached_task(self):
openml.config.set_cache_directory(self.static_cache_dir)
task = openml.tasks.functions._get_cached_task(1)
self.assertIsInstance(task, OpenMLTask)
def test__get_cached_task_not_cached(self):
openml.config.set_cache_directory(self.static_cache_dir)
self.assertRaisesRegexp(OpenMLCacheException,
'Task file for tid 2 not cached',
openml.tasks.functions._get_cached_task, 2)
def test__get_estimation_procedure_list(self):
estimation_procedures = openml.tasks.functions.\
_get_estimation_procedure_list()
self.assertIsInstance(estimation_procedures, list)
self.assertIsInstance(estimation_procedures[0], dict)
self.assertEqual(estimation_procedures[0]['task_type_id'], 1)
def _check_task(self, task):
self.assertEqual(type(task), dict)
self.assertGreaterEqual(len(task), 2)
self.assertIn('did', task)
self.assertIsInstance(task['did'], int)
self.assertIn('status', task)
self.assertTrue(is_string(task['status']))
self.assertIn(task['status'],
['in_preparation', 'active', 'deactivated'])
def test_list_tasks_by_type(self):
num_curves_tasks = 200 # number is flexible, check server if fails
ttid=3
tasks = openml.tasks.list_tasks(task_type_id=ttid)
self.assertGreaterEqual(len(tasks), num_curves_tasks)
for tid in tasks:
self.assertEquals(ttid, tasks[tid]["ttid"])
self._check_task(tasks[tid])
def test_list_tasks_by_tag(self):
num_basic_tasks = 54 # number is flexible, check server if fails
tasks = openml.tasks.list_tasks(tag='basic')
self.assertGreaterEqual(len(tasks), num_basic_tasks)
for tid in tasks:
self._check_task(tasks[tid])
def test_list_tasks(self):
tasks = openml.tasks.list_tasks()
self.assertGreaterEqual(len(tasks), 2000)
for tid in tasks:
self._check_task(tasks[tid])
def test_list_tasks_paginate(self):
size = 10
max = 100
for i in range(0, max, size):
tasks = openml.tasks.list_tasks(offset=i, size=size)
self.assertGreaterEqual(size, len(tasks))
for tid in tasks:
self._check_task(tasks[tid])
def test_list_tasks_per_type_paginate(self):
size = 10
max = 100
task_types = 5
for j in range(1,task_types):
for i in range(0, max, size):
tasks = openml.tasks.list_tasks(task_type_id=j, offset=i, size=size)
self.assertGreaterEqual(size, len(tasks))
for tid in tasks:
self.assertEquals(j, tasks[tid]["ttid"])
self._check_task(tasks[tid])
def test__get_task(self):
openml.config.set_cache_directory(self.static_cache_dir)
task = openml.tasks.get_task(1882)
def test_get_task(self):
task = openml.tasks.get_task(1)
self.assertIsInstance(task, OpenMLTask)
self.assertTrue(os.path.exists(
os.path.join(os.getcwd(), "tasks", "1", "task.xml")))
self.assertTrue(os.path.exists(
os.path.join(os.getcwd(), "tasks", "1", "datasplits.arff")))
self.assertTrue(os.path.exists(
os.path.join(os.getcwd(), "datasets", "1", "dataset.arff")))
def test_get_task_with_cache(self):
openml.config.set_cache_directory(self.static_cache_dir)
task = openml.tasks.get_task(1)
self.assertIsInstance(task, OpenMLTask)
def test_download_split(self):
task = openml.tasks.get_task(1)
split = task.download_split()
self.assertEqual(type(split), OpenMLSplit)
self.assertTrue(os.path.exists(
os.path.join(os.getcwd(), "tasks", "1", "datasplits.arff")))