Provide TEST_CONFIG alias. (#604)

Allows (well, forces) a different channel layer configuration in testing.

channels/asgi.py

@@ -26,6 +26,23 @@ class ChannelLayerManager(object):
         return getattr(settings, "CHANNEL_LAYERS", {})
 
     def make_backend(self, name):
+        """
+        Instantiate channel layer.
+        """
+        config = self.configs[name].get("CONFIG", {})
+        return self._make_backend(name, config)
+
+    def make_test_backend(self, name):
+        """
+        Instantiate channel layer using its test config.
+        """
+        try:
+            config = self.configs[name]["TEST_CONFIG"]
+        except KeyError:
+            raise InvalidChannelLayerError("No TEST_CONFIG specified for %s" % name)
+        return self._make_backend(name, config)
+
+    def _make_backend(self, name, config):
         # Load the backend class
         try:
             backend_class = import_string(self.configs[name]['BACKEND'])
@@ -41,7 +58,7 @@ class ChannelLayerManager(object):
         except KeyError:
             raise InvalidChannelLayerError("No ROUTING specified for %s" % name)
         # Initialise and pass config
-        asgi_layer = backend_class(**self.configs[name].get("CONFIG", {}))
+        asgi_layer = backend_class(**config)
         return ChannelLayerWrapper(
             channel_layer=asgi_layer,
             alias=name,

channels/test/liveserver.py

@@ -84,19 +84,19 @@ class WorkerProcess(ProcessSetup):
         try:
             self.common_setup()
             channel_layers = ChannelLayerManager()
-            check_default = channel_layers[DEFAULT_CHANNEL_LAYER].router.check_default
+            channel_layer = channel_layers.make_test_backend(DEFAULT_CHANNEL_LAYER)
             if self.serve_static and apps.is_installed('django.contrib.staticfiles'):
-                check_default(http_consumer=StaticFilesConsumer())
+                channel_layer.router.check_default(http_consumer=StaticFilesConsumer())
             else:
-                check_default()
+                channel_layer.router.check_default()
             if self.n_threads == 1:
                 self.worker = Worker(
-                    channel_layer=channel_layers[DEFAULT_CHANNEL_LAYER],
+                    channel_layer=channel_layer,
                     signal_handlers=False,
                 )
             else:
                 self.worker = WorkerGroup(
-                    channel_layer=channel_layers[DEFAULT_CHANNEL_LAYER],
+                    channel_layer=channel_layer,
                     signal_handlers=False,
                     n_threads=self.n_threads,
                 )
@@ -127,8 +127,9 @@ class DaphneProcess(ProcessSetup):
         try:
             self.common_setup()
             channel_layers = ChannelLayerManager()
+            channel_layer = channel_layers.make_test_backend(DEFAULT_CHANNEL_LAYER)
             self.server = Server(
-                channel_layer=channel_layers[DEFAULT_CHANNEL_LAYER],
+                channel_layer=channel_layer,
                 endpoints=['tcp:interface=%s:port=0' % (self.host)],
                 signal_handlers=False,
             )
@@ -183,7 +184,7 @@ class ChannelLiveServerTestCase(TransactionTestCase):
                 'ChannelLiveServerTestCase does not support multiple CHANNEL_LAYERS at this time'
             )
 
-        channel_layer = channel_layers[DEFAULT_CHANNEL_LAYER]
+        channel_layer = channel_layers.make_test_backend(DEFAULT_CHANNEL_LAYER)
         if 'flush' in channel_layer.extensions:
             channel_layer.flush()
 

docs/testing.rst

@@ -299,8 +299,26 @@ Live Server Test Case
 ---------------------
 
 You can use browser automation libraries like Selenium or Splinter to
-check your application against real layer installation.  Use
-``ChannelLiveServerTestCase`` for your acceptance tests.
+check your application against real layer installation.  First of all
+provide ``TEST_CONFIG`` setting to prevent overlapping with running
+dev environment.
+
+.. code:: python
+
+    CHANNEL_LAYERS = {
+        "default": {
+            "BACKEND": "asgi_redis.RedisChannelLayer",
+            "ROUTING": "my_project.routing.channel_routing",
+            "CONFIG": {
+                "hosts": [("redis-server-name", 6379)],
+            },
+            "TEST_CONFIG": {
+                "hosts": [("localhost", 6379)],
+            },
+        },
+    }
+
+Now use ``ChannelLiveServerTestCase`` for your acceptance tests.
 
 .. code:: python
 

tests/test_asgi.py

@@ -0,0 +1,34 @@
+from channels import DEFAULT_CHANNEL_LAYER
+from channels.asgi import InvalidChannelLayerError, channel_layers
+from channels.test import ChannelTestCase
+from django.test import override_settings
+
+
+class TestChannelLayerManager(ChannelTestCase):
+
+    def test_config_error(self):
+        """
+        If channel layer doesn't specify TEST_CONFIG, `make_test_backend`
+        should result into error.
+        """
+
+        with self.assertRaises(InvalidChannelLayerError):
+            channel_layers.make_test_backend(DEFAULT_CHANNEL_LAYER)
+
+    @override_settings(CHANNEL_LAYERS={
+        'default': {
+            'BACKEND': 'asgiref.inmemory.ChannelLayer',
+            'ROUTING': [],
+            'TEST_CONFIG': {
+                'expiry': 100500,
+            },
+        },
+    })
+    def test_config_instance(self):
+        """
+        If channel layer provides TEST_CONFIG, `make_test_backend` should
+        return channel layer instance appropriate for testing.
+        """
+
+        layer = channel_layers.make_test_backend(DEFAULT_CHANNEL_LAYER)
+        self.assertEqual(layer.channel_layer.expiry, 100500)