tests_multitenant: test threading.Timer (#44021)

tests_multitenant/test_settings.py

@@ -17,13 +17,11 @@
 import datetime
 import json
 import os
-import threading
 import random
 import pytest
 
 import django.conf
 import django.db
-from django.core.cache import cache, caches
 
 from tenant_schemas.utils import tenant_context
 
@@ -159,65 +157,6 @@ def test_tenant_theme_settings(tenants, settings, client):
         os.environ['DJANGO_SETTINGS_MODULE'] = old_value
 
 
-def test_multithreading(tenants, settings, client):
-
-    settings.clear_tenants_settings()
-    with utilities.patch_default_settings(settings,
-                                          TENANT_SETTINGS_LOADERS=('hobo.multitenant.settings_loaders.TemplateVars',)):
-
-        def f(tenant=None):
-            if tenant is not None:
-                assert hasattr(settings, 'TEMPLATE_VARS')
-                assert settings.TEMPLATE_VARS['test_url'] == tenant.get_base_url()
-            else:
-                assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')
-
-        assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')
-        t1 = threading.Thread(target=f)
-        t1.start()
-        t1.join()
-
-        for tenant in tenants:
-            with tenant_context(tenant):
-                assert hasattr(settings, 'TEMPLATE_VARS')
-                t2 = threading.Thread(target=f, args=(tenant,))
-                t2.start()
-                t2.join()
-
-        assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')
-        t3 = threading.Thread(target=f)
-        t3.start()
-        t3.join()
-
-
-def test_cache(tenants, client):
-    # Clear caches
-    caches._caches.caches = {}
-
-    cache.set('coin', 1)
-
-    for tenant in tenants:
-        with tenant_context(tenant):
-            assert cache.get('coin') is None
-            cache.set('coin', tenant.domain_url)
-
-    assert cache.get('coin') == 1
-
-    for tenant in tenants:
-        with tenant_context(tenant):
-            def f():
-                assert cache.get('coin') == tenant.domain_url
-            t1 = threading.Thread(target=f)
-            t1.start()
-            t1.join()
-
-    def g():
-        assert cache.get('coin') == 1
-    t2 = threading.Thread(target=g)
-    t2.start()
-    t2.join()
-
-
 def test_shared_secret():
     from hobo.multitenant.settings_loaders import KnownServices
 

tests_multitenant/test_threading.py

@@ -0,0 +1,113 @@
+# hobo - portal to configure and deploy applications
+# Copyright (C) 2015-2016 Entr'ouvert
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import threading
+
+import django.conf
+from django.core.cache import cache, caches
+
+from tenant_schemas.utils import tenant_context
+
+import utilities
+
+
+def test_thread(tenants, settings, client):
+
+    settings.clear_tenants_settings()
+    with utilities.patch_default_settings(settings,
+                                          TENANT_SETTINGS_LOADERS=('hobo.multitenant.settings_loaders.TemplateVars',)):
+
+        def f(tenant=None):
+            if tenant is not None:
+                assert hasattr(settings, 'TEMPLATE_VARS')
+                assert settings.TEMPLATE_VARS['test_url'] == tenant.get_base_url()
+            else:
+                assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')
+
+        assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')
+        t1 = threading.Thread(target=f)
+        t1.start()
+        t1.join()
+
+        for tenant in tenants:
+            with tenant_context(tenant):
+                assert hasattr(settings, 'TEMPLATE_VARS')
+                t2 = threading.Thread(target=f, args=(tenant,))
+                t2.start()
+                t2.join()
+
+        assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')
+        t3 = threading.Thread(target=f)
+        t3.start()
+        t3.join()
+
+
+def test_cache(tenants, client):
+    # Clear caches
+    caches._caches.caches = {}
+
+    cache.set('coin', 1)
+
+    for tenant in tenants:
+        with tenant_context(tenant):
+            assert cache.get('coin') is None
+            cache.set('coin', tenant.domain_url)
+
+    assert cache.get('coin') == 1
+
+    for tenant in tenants:
+        with tenant_context(tenant):
+            def f():
+                assert cache.get('coin') == tenant.domain_url
+            t1 = threading.Thread(target=f)
+            t1.start()
+            t1.join()
+
+    def g():
+        assert cache.get('coin') == 1
+    t2 = threading.Thread(target=g)
+    t2.start()
+    t2.join()
+
+
+def test_timer_thread(tenants, settings, client):
+    settings.clear_tenants_settings()
+    with utilities.patch_default_settings(settings,
+                                          TENANT_SETTINGS_LOADERS=('hobo.multitenant.settings_loaders.TemplateVars',)):
+
+        def f(tenant=None):
+            if tenant is not None:
+                assert hasattr(settings, 'TEMPLATE_VARS')
+                assert settings.TEMPLATE_VARS['test_url'] == tenant.get_base_url()
+            else:
+                assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')
+
+        assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')
+        t1 = threading.Timer(0.0, f)
+        t1.start()
+        t1.join()
+
+        for tenant in tenants:
+            with tenant_context(tenant):
+                assert hasattr(settings, 'TEMPLATE_VARS')
+                t2 = threading.Timer(0.0, f, args=(tenant,))
+                t2.start()
+                t2.join()
+
+        assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')
+        t3 = threading.Timer(0.0, f)
+        t3.start()
+        t3.join()