-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_sqlite.py
135 lines (118 loc) · 7.12 KB
/
test_sqlite.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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
import unittest
import os
from datetime import datetime
import sqlite
class SqliteTest(unittest.TestCase):
try:
os.remove('tests/test_plot.db')
print(f'File tests/test_plot.db deleted successfully.')
except OSError:
print(f'tests/test_plot.db does not exist')
def test_get_all_destination_by_status(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='random.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool.update_plot_by_name(plot_name='random.plot', dest='/mnt/moon', status='in_progress')
clock2 = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='alpha.plot', source='/mnt/imaginarium', timestamp=clock2)
clock3 = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='beta.plot', source='/mnt/imaginarium', timestamp=clock3)
DBPool.update_plot_by_name(plot_name='beta.plot', dest='/mnt/jupyter', status='in_progress')
res = DBPool.get_all_destination_by_status('in_progress')
self.assertEqual([('/mnt/moon',), ('/mnt/jupyter',)], res)
DBPool.drop_table()
def test_doubleBDD(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='random.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool.insert_new_plot(plot_name='alpha.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool.insert_new_plot(plot_name='beta.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool2 = sqlite.DBPool('tests/test_plot.db')
res = DBPool2.get_all_plots()
self.assertEqual([('random.plot', '/mnt/imaginarium', None, 'to_process', clock),
('alpha.plot', '/mnt/imaginarium', None, 'to_process', clock),
('beta.plot', '/mnt/imaginarium', None, 'to_process', clock)], res)
DBPool.drop_table()
def test_count_raw(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='random.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool.insert_new_plot(plot_name='alpha.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool.insert_new_plot(plot_name='beta.plot', source='/mnt/imaginarium', timestamp=clock)
res = DBPool.count_raw()
self.assertEqual([(3,)], res)
DBPool.drop_table()
def test_ensure_db_has_not_in_progess_plot_at_start_up(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='random.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool.update_plot_by_name(plot_name='random.plot', dest='/mnt/moon', status='in_progress')
DBPool.insert_new_plot(plot_name='alpha.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool.insert_new_plot(plot_name='beta.plot', source='/mnt/imaginarium', timestamp=clock)
res = DBPool.get_all_plots()
self.assertEqual([('random.plot',
'/mnt/imaginarium',
'/mnt/moon',
'in_progress',
clock),
('alpha.plot', '/mnt/imaginarium', None, 'to_process', clock),
('beta.plot', '/mnt/imaginarium', None, 'to_process', clock)], res)
_ = DBPool.ensure_db_has_not_in_progess_plot_at_start_up()
res = DBPool.get_all_plots()
self.assertEqual([('random.plot', '/mnt/imaginarium', None, 'to_process', clock),
('alpha.plot', '/mnt/imaginarium', None, 'to_process', clock),
('beta.plot', '/mnt/imaginarium', None, 'to_process', clock)], res)
DBPool.drop_table()
def test_get_first_plot_without_status(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='random.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool.update_plot_by_name(plot_name='random.plot', dest='/mnt/moon', status='in_progress')
clock2 = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='alpha.plot', source='/mnt/imaginarium', timestamp=clock2)
clock3 = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='beta.plot', source='/mnt/imaginarium', timestamp=clock3)
res = DBPool.get_first_plot_without_status()
self.assertEqual([('alpha.plot', '/mnt/imaginarium', None, 'to_process', clock2)], res)
DBPool.drop_table()
def test_get_plot_status_by_name(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp()
DBPool.insert_new_plot(plot_name='random.plot', source='/mnt/imaginarium', timestamp=clock)
DBPool.update_plot_by_name(plot_name='random.plot', dest='/mnt/moon', status='in_progress')
_ = DBPool.get_plot_by_name('random.plot')
res = DBPool.get_plot_status_by_name(plot_name='random.plot')
self.assertEqual([('in_progress',)], res)
DBPool.drop_table()
def test_update_bdd(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp()
_ = DBPool.insert_new_plot(plot_name='random.plot', source='/mnt/imaginarium', timestamp=clock)
_ = DBPool.update_plot_by_name(plot_name='random.plot', dest='/mnt/moon', status='in_progress')
res = DBPool.get_plot_by_name('random.plot')
self.assertEqual([('random.plot', '/mnt/imaginarium', '/mnt/moon', 'in_progress', clock)], res)
DBPool.drop_table()
def test_insert_bdd(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp()
_ = DBPool.insert_new_plot(plot_name='random.plot', source='/mnt/imaginarium', timestamp=clock)
res = DBPool.get_plot_by_name('random.plot')
self.assertEqual([('random.plot', '/mnt/imaginarium', None, 'to_process', clock)], res)
DBPool.drop_table()
def test_get_first_plot_without_status_and_change_status(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp() - 3600 * 72
_ = DBPool.insert_new_plot(plot_name='safe_lock_get_update.plot', source='/mnt/olympus', timestamp=clock)
_ = DBPool.get_first_plot_without_status_and_change_status()
res = DBPool.get_plot_by_name('safe_lock_get_update.plot')
self.assertEqual([('safe_lock_get_update.plot', '/mnt/olympus', None, 'in_progress', clock)], res)
DBPool.drop_table()
def test_insert_new_plot_if_not_exist(self):
DBPool = sqlite.DBPool('tests/test_plot.db')
clock = datetime.now().timestamp()
res = DBPool.insert_new_plot_if_not_exist(plot_name='should_be_inserted.plot', source='/mnt/olympus', timestamp=clock)
self.assertEqual(True, res)
res = DBPool.insert_new_plot_if_not_exist(plot_name='should_be_inserted.plot', source='/mnt/olympus', timestamp=clock)
self.assertEqual(None, res)
if __name__ == '__main__':
unittest.main()