forked from mxm/yoka
-
Notifications
You must be signed in to change notification settings - Fork 0
/
tests.py
107 lines (86 loc) · 2.54 KB
/
tests.py
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
import unittest
import os
from core.lib import System, Experiment, Benchmark, ClusterSuite
from core.clusters import ComputeEngine
#from systems import Hadoop, Flink
from configs import compute_engine_config, hadoop_config, flink_config
import core.results as results
results.DB_FILE = 'test.db'
cluster = ComputeEngine(compute_engine_config)
#hadoop = Hadoop(hadoop_config)
#flink = Flink(flink_config)
class StupidSystem(System):
def __init__(self, name):
self.name = name
def set_config(self):
pass
def install(self):
pass
def configure(self):
pass
def reset(self):
pass
def start(self):
pass
def stop(self):
pass
def save_log(self, unique_full_path):
pass
def __str__(self):
return self.name
systems = [StupidSystem("stupid1"), StupidSystem("stupid2")]
class TestExperiment(Experiment):
def setup(self):
pass
def run(self):
print "running experiment"
def shutdown(self):
pass
benchmarks = [
Benchmark("Test1",
systems=systems,
experiment = TestExperiment(),
times = 1
),
Benchmark("Test2",
systems=systems,
experiment = TestExperiment(),
times = 10
),
Benchmark("Test3",
systems=systems,
experiment = TestExperiment(),
times = 3
),
]
class TestResults(unittest.TestCase):
def setup(self):
try:
os.remove(results.DB_FILE)
except:
pass
def test_save(self):
suite = ClusterSuite("SuiteTest", cluster, systems, [], benchmarks)
# skip setup and shutdown, just run the tests
suite.run()
#check corresponding db entries
suite_id = suite.id
suite_uid = suite.uid
for b in suite.benchmarks:
data = (suite_uid, suite_id, b.id)
with results.DB() as db:
c = db.cursor()
c.execute("""
SELECT * FROM results
WHERE suite_uid = ? and suite_id = ? and
bench_id = ?
""", data)
self.assertEquals(c.fetchall().__len__(), b.times)
def test_email_plot(self):
from pprint import pformat
filename = results.gen_plot("TestSuiteId")
text = "Cluster config:\n%s" % (pformat(compute_engine_config))
results.send_email(filename, additional_text=text)
def tearDown(self):
pass
unittest.main()