1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
|
import unittest
import pytest
from django.test import override_settings
from channels import DEFAULT_CHANNEL_LAYER
from channels.exceptions import InvalidChannelLayerError
from channels.layers import (
BaseChannelLayer,
InMemoryChannelLayer,
channel_layers,
get_channel_layer,
)
class TestChannelLayerManager(unittest.TestCase):
@override_settings(
CHANNEL_LAYERS={"default": {"BACKEND": "channels.layers.InMemoryChannelLayer"}}
)
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": "channels.layers.InMemoryChannelLayer",
"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.expiry, 100500)
def test_override_settings(self):
"""
The channel layers cache is reset when the CHANNEL_LAYERS setting
changes.
"""
with override_settings(
CHANNEL_LAYERS={
"default": {"BACKEND": "channels.layers.InMemoryChannelLayer"}
}
):
self.assertEqual(channel_layers.backends, {})
get_channel_layer()
self.assertNotEqual(channel_layers.backends, {})
self.assertEqual(channel_layers.backends, {})
@override_settings(
CHANNEL_LAYERS={
"default": {
"BACKEND": "tests.test_layers.BrokenBackend",
}
}
)
def test_backend_import_error_not_hidden(self):
"""
Test that KeyError exceptions within the backend import are not hidden.
This test ensures that the PR #2146 fix works correctly.
"""
# This should raise a KeyError from the backend, not an InvalidChannelLayerError
with self.assertRaises(KeyError):
channel_layers.make_backend(DEFAULT_CHANNEL_LAYER)
# Mock backend that raises KeyError during import
class BrokenBackend:
def __init__(self, **kwargs):
# This will be called during backend initialization
# and should raise a KeyError that should not be caught
raise KeyError("This is a deliberate KeyError from the backend")
# In-memory layer tests
@pytest.mark.asyncio
async def test_send_receive():
layer = InMemoryChannelLayer()
message = {"type": "test.message"}
await layer.send("test.channel", message)
assert message == await layer.receive("test.channel")
@pytest.mark.parametrize(
"method",
[
BaseChannelLayer().require_valid_channel_name,
BaseChannelLayer().require_valid_group_name,
],
)
@pytest.mark.parametrize(
"channel_name,expected_valid",
[("¯\\_(ツ)_/¯", False), ("chat", True), ("chat" * 100, False)],
)
def test_channel_and_group_name_validation(method, channel_name, expected_valid):
if expected_valid:
method(channel_name)
else:
with pytest.raises(TypeError):
method(channel_name)
@pytest.mark.parametrize(
"name",
[
"a" * 101, # Group name too long
],
)
def test_group_name_length_error_message(name):
"""
Ensure the correct error message is raised when group names
exceed the character limit or contain invalid characters.
"""
layer = BaseChannelLayer()
expected_error_message = layer.invalid_name_error.format("Group")
with pytest.raises(TypeError, match=expected_error_message):
layer.require_valid_group_name(name)
@pytest.mark.parametrize(
"name",
[
"a" * 101, # Channel name too long
],
)
def test_channel_name_length_error_message(name):
"""
Ensure the correct error message is raised when group names
exceed the character limit or contain invalid characters.
"""
layer = BaseChannelLayer()
expected_error_message = layer.invalid_name_error.format("Channel")
with pytest.raises(TypeError, match=expected_error_message):
layer.require_valid_channel_name(name)
def test_deprecated_valid_channel_name():
"""
Test that the deprecated valid_channel_name method works
but raises a deprecation warning.
"""
layer = BaseChannelLayer()
# Should work with valid name but raise warning
with pytest.warns(DeprecationWarning, match="valid_channel_name is deprecated"):
assert layer.valid_channel_name("valid-channel")
# Should raise TypeError for invalid names
with pytest.warns(DeprecationWarning):
with pytest.raises(TypeError):
layer.valid_channel_name("¯\\_(ツ)_/¯")
def test_deprecated_valid_group_name():
"""
Test that the deprecated valid_group_name method works
but raises a deprecation warning.
"""
layer = BaseChannelLayer()
# Should work with valid name but raise warning
with pytest.warns(DeprecationWarning, match="valid_group_name is deprecated"):
assert layer.valid_group_name("valid-group")
# Should raise TypeError for invalid names
with pytest.warns(DeprecationWarning):
with pytest.raises(TypeError):
layer.valid_group_name("¯\\_(ツ)_/¯")
|
