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
|
# sugar-lint: disable
import os
import sys
import time
import json
import shutil
import signal
import hashlib
import logging
import tempfile
import unittest
from os.path import dirname, join, exists, abspath
import active_document as ad
from active_toolkit import coroutine, util
from active_document import env as _env, index_queue as _index_queue
from active_document import storage as _storage
from active_document import directory as _directory
root = abspath(dirname(__file__))
tmproot = join(root, '.tmp')
tmpdir = None
def main():
shutil.rmtree(tmproot, ignore_errors=True)
unittest.main()
class Test(unittest.TestCase):
def setUp(self):
self._overriden = []
global tmpdir
tmpdir = join(tmproot, '.'.join(self.id().split('.')[1:]))
shutil.rmtree(tmpdir, ignore_errors=True)
os.makedirs(tmpdir)
os.chdir(tmpdir)
logfile = tmpdir + '.log'
if exists(logfile):
os.unlink(logfile)
self._logfile = file(logfile + '.out', 'a')
sys.stdout = sys.stderr = self._logfile
for handler in logging.getLogger().handlers:
logging.getLogger().removeHandler(handler)
logging.basicConfig(level=logging.DEBUG, filename=logfile)
_env.index_flush_timeout.value = 0
_env.index_flush_threshold.value = 1
_directory._DIFF_PAGE_SIZE = 256
_index_queue.errnum = 0
_storage._ensure_path_locker = _FakeLocker()
def tearDown(self):
self.assertEqual(0, _index_queue.errnum)
while self._overriden:
mod, name, old_handler = self._overriden.pop()
setattr(mod, name, old_handler)
_index_queue.stop()
sys.stdout.flush()
def override(self, mod, name, new_handler):
self._overriden.append((mod, name, getattr(mod, name)))
setattr(mod, name, new_handler)
def fork(self, cb):
pid = os.fork()
if not pid:
try:
cb()
result = 0
except Exception:
logging.exception('Child failed')
result = 1
sys.stdout.flush()
sys.stderr.flush()
os._exit(0)
else:
__, status = os.waitpid(pid, 0)
self.assertEqual(0, os.WEXITSTATUS(status))
def touch(self, *files):
for i in files:
if isinstance(i, str):
if i.endswith(os.sep):
i = i + '.stamp'
path = i
if exists(path):
content = file(path).read()
else:
content = i
else:
path, content = i
if isinstance(content, list):
content = '\n'.join(content)
path = join(tmpdir, path)
if not exists(dirname(path)):
os.makedirs(dirname(path))
if exists(path):
os.unlink(path)
f = file(path, 'w')
f.write(str(content))
f.close()
class _FakeLocker(object):
def __enter__(self, *args):
return self
def __exit__(self, *args):
pass
|
