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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
|
# -*- coding: utf-8 -*-
import os
import shutil
import sys
import tempfile
import time
import unittest
from dirq import queue
from dirq.queue import Queue, QueueError
__all__ = ['TestQueue', 'TestQueueModuleFunctions']
class TestDirQueue(unittest.TestCase):
def setUp(self):
self.tempdir = tempfile.mkdtemp(prefix='dirq')
self.qdir = self.tempdir + '/dirq'
def tearDown(self):
shutil.rmtree(self.tempdir, ignore_errors=True)
class TestQueue(TestDirQueue):
def test1init(self):
""" Queue.__init__() """
path = self.tempdir + '/aaa/bbb/ccc'
umask = None
maxelts = 10
good_schemas = [
{'a': 'binary'},
{'a': 'binary*'},
{'a': 'string'},
{'a': 'string*'},
{'a': 'table'},
{'a': 'binary?', 'b': 'binary'},
{'a': 'binary?', 'b': 'binary*'},
{'a': 'string', 'b': 'binary?*'}, ]
for schema in good_schemas:
try:
Queue(path, umask=umask, maxelts=maxelts, schema=schema)
except Exception:
error = sys.exc_info()[1]
self.fail("Shouldn't have failed. Exception: %s" % error)
bad_schemas = [['foo'],
{'foo': 1},
{'a': 'binary?'}]
for schema in bad_schemas:
self.assertRaises(
QueueError, Queue, path,
umask=umask, maxelts=maxelts, schema=schema)
bad_schemas = [{'a': 'strings'}, {'a': 'table??'}]
for schema in bad_schemas:
self.assertRaises(
QueueError, Queue, path,
umask=umask, maxelts=maxelts, schema=schema)
def test2copy(self):
""" Queue.copy(). """
q = Queue(self.qdir, schema={'a': 'string'})
q1 = q.copy()
q.foo = 1
try:
q1.foo
except AttributeError:
pass
else:
self.fail('Not a copy, but reference.')
def test3_is_locked(self):
""" Queue._is_locked_*() """
q = Queue(self.qdir, schema={'a': 'string'})
assert q._is_locked_nlink('') is False
assert q._is_locked_nlink('not_there') is False
os.mkdir(self.qdir + '/a')
assert q._is_locked_nlink('a') is False
os.mkdir(self.qdir + '/a/%s' % queue.LOCKED_DIRECTORY)
assert q._is_locked_nlink('a') is True
time.sleep(1)
assert q._is_locked_nlink('a', time.time()) is True
assert q._is_locked_nonlink('') is False
assert q._is_locked_nonlink('not_there') is False
os.mkdir(self.qdir + '/b')
assert q._is_locked_nonlink('b') is False
os.mkdir(self.qdir + '/b/%s' % queue.LOCKED_DIRECTORY)
assert q._is_locked_nonlink('b') is True
time.sleep(1)
assert q._is_locked_nonlink('b', time.time()) is True
def test4_insertion_directory(self):
""" Queue._insertion_directory() """
q = queue.Queue(self.qdir, schema={'a': 'string'})
q.maxelts = 1
name0 = '%08x' % 0
assert q._insertion_directory() == name0
assert os.path.exists(self.qdir + '/' + name0)
os.mkdir('%s/%s/%s' % (self.qdir, name0, queue._name(q.rndhex)))
name1 = '%08x' % 1
assert q._insertion_directory() == name1
assert os.path.exists(self.qdir + '/' + name1)
def test5add(self):
""" Queue.add() """
q = queue.Queue(self.qdir, schema={'a': 'string'})
q.add({'a': 'a\n'})
assert os.listdir(self.qdir + '/' + queue.TEMPORARY_DIRECTORY) == []
data_file = '%s/%s/%s/a' % (self.qdir, '%08x' % 0,
os.listdir('%s/%08x' % (self.qdir, 0))[-1])
assert os.path.exists(data_file)
assert open(data_file).read() == 'a\n'
def test6touch(self):
""" Queue.touch() """
q = queue.Queue(self.qdir, schema={'a': 'string'})
q.add({'a': 'a\n'})
e = q.first()
element_dir = q.path + '/' + e
old_time = time.time() - 10
os.utime(element_dir, (old_time, old_time))
mtime = os.stat(element_dir).st_mtime
q.touch(e)
assert os.stat(element_dir).st_mtime > mtime
class TestQueueModuleFunctions(TestDirQueue):
def test2_check_element(self):
""" queue._check_element() """
queue._check_element('0' * 8 + '/' + '0' * 14)
queue._check_element('f' * 8 + '/' + 'f' * 14)
for e in ['f' * 7 + '/' + 'f' * 14,
'f' * 9 + '/' + 'f' * 14,
'f' * 8 + '/' + 'f' * 13,
'f' * 8 + '/' + 'f' * 15,
'f' * 8 + '/' + 'g' * 14,
'g' * 8 + '/' + 'f' * 14,
]:
self.assertRaises(queue.QueueError, queue._check_element, (e))
def test3_hash2string(self):
""" queue._hash2string() """
assert queue._hash2string({'a1': 'a2'}) == 'a1\ta2\n'
assert queue._hash2string({'a1\\': 'a2'}) == 'a1\\\\\ta2\n'
assert queue._hash2string({'a1 a2': 'a3 a4'}) == 'a1\\ta2\ta3\\ta4\n'
assert queue._hash2string({'a1 \na2': 'a3 \na4'}) == \
'a1\\t\\na2\ta3\\t\\na4\n'
def test3_string2hash(self):
""" queue._string2hash() """
assert queue._string2hash('a1\ta2\nb1\tb2') == {'a1': 'a2', 'b1': 'b2'}
assert queue._string2hash('a1\x5c\ta2\nb1\tb2\\') == \
{'a1\x5c': 'a2', 'b1': 'b2\\'}
for value in ['a', ]:
self.assertRaises(queue.QueueError, queue._string2hash, (value))
def test3_hash2string2hash(self):
""" queue._hash2string()+queue._string2hash() """
example = {"hi\\t\th\nere": "h\\ello\twor\nld"}
converted = queue._hash2string(example)
assert converted == "hi\\\\t\\th\\nere\th\\\\ello\\twor\\nld\n"
back = queue._string2hash(converted)
assert back == example
def main():
""" Main. """
testcases = [TestQueue,
TestQueueModuleFunctions]
for tc in testcases:
unittest.TextTestRunner(verbosity=2).\
run(unittest.TestLoader().loadTestsFromTestCase(tc))
if __name__ == "__main__":
main()
|
