Web   ·   Wiki   ·   Activities   ·   Blog   ·   Lists   ·   Chat   ·   Meeting   ·   Bugs   ·   Git   ·   Translate   ·   Archive   ·   People   ·   Donate
summaryrefslogtreecommitdiffstats
path: root/tests/__init__.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/__init__.py')
-rw-r--r--tests/__init__.py122
1 files changed, 0 insertions, 122 deletions
diff --git a/tests/__init__.py b/tests/__init__.py
deleted file mode 100644
index b801871..0000000
--- a/tests/__init__.py
+++ /dev/null
@@ -1,122 +0,0 @@
-# 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