From 05b45035312f9b6bf97621d3b04f21166b2a6bfd Mon Sep 17 00:00:00 2001 From: niphlod Date: Mon, 2 Jun 2014 15:59:59 +0200 Subject: [PATCH] fix multiprocessing tests under windows --- gluon/tests/test_languages.py | 178 +++++++++++++++++++--------------- 1 file changed, 100 insertions(+), 78 deletions(-) diff --git a/gluon/tests/test_languages.py b/gluon/tests/test_languages.py index 682f3cd4..91bc3762 100644 --- a/gluon/tests/test_languages.py +++ b/gluon/tests/test_languages.py @@ -12,10 +12,6 @@ import tempfile import threading import logging -def is_gae(): - import httplib - return 'appengine' in str(httplib.HTTP) - def fix_sys_path(): """ logic to have always the correct sys.path @@ -41,88 +37,114 @@ def fix_sys_path(): os.path.abspath(os.path.join(path, 'gluon')), ''] [add_path_first(path) for path in paths] + fix_sys_path() +#support skipif also in python 2.6 +def _skipIf(cond, message=''): + def _decorator(testcase): + if cond: + return lambda *a, **kw: None + else: + return testcase + return _decorator + +if hasattr(unittest, 'skipIf'): + skipIf = unittest.skipIf +else: + skipIf = _skipIf import languages from storage import Storage - +MP_WORKING = 0 try: import multiprocessing - - def read_write(args): - (filename, iterations) = args - for i in range(0, iterations): - content = languages.read_dict(filename) - if not len(content): - return False - languages.write_dict(filename, content) - return True - - class TestLanguagesParallel(unittest.TestCase): - - def setUp(self): - self.filename = tempfile.mktemp() - contents = dict() - for i in range(1000): - contents["key%d" % i] = "value%d" % i - languages.write_dict(self.filename, contents) - languages.read_dict(self.filename) - - def tearDown(self): - try: - os.remove(self.filename) - except: - pass - - def test_reads_and_writes(self): - if is_gae(): - return # do not check since on GAE not write of filesystem - readwriters = 10 - pool = multiprocessing.Pool(processes=readwriters) - results = pool.map(read_write, [[self.filename, 10]] * readwriters) - for result in results: - self.assertTrue(result) - - class TestTranslations(unittest.TestCase): - - def setUp(self): - if os.path.isdir('gluon'): - self.langpath = 'applications/welcome/languages' - else: - self.langpath = os.path.realpath( - '../../applications/welcome/languages') - self.http_accept_language = 'en' - - def tearDown(self): - pass - - def test_plain(self): - T = languages.translator(self.langpath, self.http_accept_language) - self.assertEqual(str(T('Hello World')), - 'Hello World') - self.assertEqual(str(T('Hello World## comment')), - 'Hello World') - self.assertEqual(str(T('%s %%{shop}', 1)), - '1 shop') - self.assertEqual(str(T('%s %%{shop}', 2)), - '2 shops') - self.assertEqual(str(T('%s %%{shop[0]}', 1)), - '1 shop') - self.assertEqual(str(T('%s %%{shop[0]}', 2)), - '2 shops') - self.assertEqual(str(T('%s %%{quark[0]}', 1)), - '1 quark') - self.assertEqual(str(T('%s %%{quark[0]}', 2)), - '2 quarks') - self.assertEqual(str(T.M('**Hello World**')), - 'Hello World') - T.force('it') - self.assertEqual(str(T('Hello World')), - 'Salve Mondo') - + MP_WORKING = 1 + #due to http://bugs.python.org/issue10845, testing multiprocessing in python is impossible + if sys.platform.startswith('win'): + MP_WORKING = 0 except ImportError: - logging.warning("Skipped test case, no multiprocessing module.") + pass + +def read_write(args): + (filename, iterations) = args + for i in range(0, iterations): + content = languages.read_dict(filename) + if not len(content): + return False + languages.write_dict(filename, content) + return True + +class TestLanguagesParallel(unittest.TestCase): + + def setUp(self): + self.filename = tempfile.mktemp() + contents = dict() + for i in range(1000): + contents["key%d" % i] = "value%d" % i + languages.write_dict(self.filename, contents) + languages.read_dict(self.filename) + + def tearDown(self): + try: + os.remove(self.filename) + except: + pass + + @skipIf(MP_WORKING == 0, 'multiprocessing tests unavailable') + def test_reads_and_writes(self): + readwriters = 10 + pool = multiprocessing.Pool(processes=readwriters) + results = pool.map(read_write, [[self.filename, 10]] * readwriters) + for result in results: + self.assertTrue(result) + + @skipIf(MP_WORKING == 1, 'multiprocessing tests available') + def test_reads_and_writes_no_mp(self): + results = [] + for i in range(10): + results.append(read_write([self.filename, 10])) + for result in results: + self.assertTrue(result) + +class TestTranslations(unittest.TestCase): + + def setUp(self): + if os.path.isdir('gluon'): + self.langpath = 'applications/welcome/languages' + else: + self.langpath = os.path.realpath( + '../../applications/welcome/languages') + self.http_accept_language = 'en' + + def tearDown(self): + pass + + def test_plain(self): + T = languages.translator(self.langpath, self.http_accept_language) + self.assertEqual(str(T('Hello World')), + 'Hello World') + self.assertEqual(str(T('Hello World## comment')), + 'Hello World') + self.assertEqual(str(T('%s %%{shop}', 1)), + '1 shop') + self.assertEqual(str(T('%s %%{shop}', 2)), + '2 shops') + self.assertEqual(str(T('%s %%{shop[0]}', 1)), + '1 shop') + self.assertEqual(str(T('%s %%{shop[0]}', 2)), + '2 shops') + self.assertEqual(str(T('%s %%{quark[0]}', 1)), + '1 quark') + self.assertEqual(str(T('%s %%{quark[0]}', 2)), + '2 quarks') + self.assertEqual(str(T.M('**Hello World**')), + 'Hello World') + T.force('it') + self.assertEqual(str(T('Hello World')), + 'Salve Mondo') + if __name__ == '__main__': unittest.main() + \ No newline at end of file