Merge branch 'master' of github.com:web2py/web2py

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
+    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:
+    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 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
 
-        def setUp(self):
+class TestLanguagesParallel(unittest.TestCase):
-            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):
+    def setUp(self):
-            try:
+        self.filename = tempfile.mktemp()
-                os.remove(self.filename)
+        contents = dict()
-            except:
+        for i in range(1000):
-                pass
+            contents["key%d" % i] = "value%d" % i
+        languages.write_dict(self.filename, contents)
+        languages.read_dict(self.filename)
 
-        def test_reads_and_writes(self):
+    def tearDown(self):
-            if is_gae():
+        try:
-                return # do not check since on GAE not write of filesystem
+            os.remove(self.filename)
-            readwriters = 10
+        except:
-            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):
+    @skipIf(MP_WORKING == 0, 'multiprocessing tests unavailable')
-            T = languages.translator(self.langpath, self.http_accept_language)
+    def test_reads_and_writes(self):
-            self.assertEqual(str(T('Hello World')),
+        readwriters = 10
-                             'Hello World')
+        pool = multiprocessing.Pool(processes=readwriters)
-            self.assertEqual(str(T('Hello World## comment')),
+        results = pool.map(read_write, [[self.filename, 10]] * readwriters)
-                             'Hello World')
+        for result in results:
-            self.assertEqual(str(T('%s %%{shop}', 1)),
+            self.assertTrue(result)
-                             '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**')),
-                             '<strong>Hello World</strong>')
-            T.force('it')
-            self.assertEqual(str(T('Hello World')),
-                             'Salve Mondo')
     
-except ImportError:
+    @skipIf(MP_WORKING == 1, 'multiprocessing tests available')
-    logging.warning("Skipped test case, no multiprocessing module.")
+    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**')),
+                         '<strong>Hello World</strong>')
+        T.force('it')
+        self.assertEqual(str(T('Hello World')),
+                         'Salve Mondo')
       
 if __name__ == '__main__':
     unittest.main()
+