带有硒测试的 Django 通道 2 失败
Django channels 2 with selenium test failed
我正在尝试学习 Django 频道教程。我能够实现所描述的聊天功能 here. But unittests completly copy pasted from this page 失败，出现以下错误 AttributeError: Can't pickle local object 'DaphneProcess.__init__.<locals>.<lambda>'。
完整追溯：
Traceback (most recent call last):
File "C:\Users\user\PycharmProjects\django_channels_test\venv35\lib\site-packages\django\test\testcases.py", line 202, in __call__
self._pre_setup()
File "C:\Users\user\PycharmProjects\django_channels_test\venv35\lib\site-packages\channels\testing\live.py", line 42, in _pre_setup
self._server_process.start()
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\process.py", line 105, in start
self._popen = self._Popen(self)
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\context.py", line 212, in _Popen
return _default_context.get_context().Process._Popen(process_obj)
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\context.py", line 313, in _Popen
return Popen(process_obj)
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\popen_spawn_win32.py", line 66, in __init__
reduction.dump(process_obj, to_child)
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\reduction.py", line 59, in dump
ForkingPickler(file, protocol).dump(obj)
AttributeError: Can't pickle local object 'DaphneProcess.__init__.<locals>.<lambda>'
我的消费者class：
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.room_name = self.scope['url_route']['kwargs']['room_name']
self.room_group_name = 'chat_%s' % self.room_name
# Join room group
await self.channel_layer.group_add(
self.room_group_name,
self.channel_name
)
await self.accept()
async def disconnect(self, close_code):
# Leave room group
await self.channel_layer.group_discard(
self.room_group_name,
self.channel_name
)
# Receive message from WebSocket
async def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']
# Send message to room group
await self.channel_layer.group_send(
self.room_group_name,
{
'type': 'chat_message',
'message': message
}
)
# Receive message from room group
async def chat_message(self, event):
message = event['message']
# Send message to WebSocket
await self.send(text_data=json.dumps({
'message': message
}))
我的测试模块：
from channels.testing import ChannelsLiveServerTestCase
from selenium import webdriver
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.support.wait import WebDriverWait
class ChatTests(ChannelsLiveServerTestCase):
serve_static = True # emulate StaticLiveServerTestCase
@classmethod
def setUpClass(cls):
super().setUpClass()
try:
# NOTE: Requires "chromedriver" binary to be installed in $PATH
cls.driver = webdriver.Chrome()
except:
super().tearDownClass()
raise
@classmethod
def tearDownClass(cls):
cls.driver.quit()
super().tearDownClass()
def test_when_chat_message_posted_then_seen_by_everyone_in_same_room(self):
try:
self._enter_chat_room('room_1')
self._open_new_window()
self._enter_chat_room('room_1')
self._switch_to_window(0)
self._post_message('hello')
WebDriverWait(self.driver, 2).until(lambda _:
'hello' in self._chat_log_value,
'Message was not received by window 1 from window 1')
self._switch_to_window(1)
WebDriverWait(self.driver, 2).until(lambda _:
'hello' in self._chat_log_value,
'Message was not received by window 2 from window 1')
finally:
self._close_all_new_windows()
def test_when_chat_message_posted_then_not_seen_by_anyone_in_different_room(self):
try:
self._enter_chat_room('room_1')
self._open_new_window()
self._enter_chat_room('room_2')
self._switch_to_window(0)
self._post_message('hello')
WebDriverWait(self.driver, 2).until(lambda _:
'hello' in self._chat_log_value,
'Message was not received by window 1 from window 1')
self._switch_to_window(1)
self._post_message('world')
WebDriverWait(self.driver, 2).until(lambda _:
'world' in self._chat_log_value,
'Message was not received by window 2 from window 2')
self.assertTrue('hello' not in self._chat_log_value,
'Message was improperly received by window 2 from window 1')
finally:
self._close_all_new_windows()
# === Utility ===
def _enter_chat_room(self, room_name):
self.driver.get(self.live_server_url + '/chat/')
ActionChains(self.driver).send_keys(room_name + '\n').perform()
WebDriverWait(self.driver, 2).until(lambda _:
room_name in self.driver.current_url)
def _open_new_window(self):
self.driver.execute_script('window.open("about:blank", "_blank");')
self.driver.switch_to.window(self.driver.window_handles[-1])
def _close_all_new_windows(self):
while len(self.driver.window_handles) > 1:
self.driver.switch_to.window(self.driver.window_handles[-1])
self.driver.execute_script('window.close();')
if len(self.driver.window_handles) == 1:
self.driver.switch_to.window(self.driver.window_handles[0])
def _switch_to_window(self, window_index):
self.driver.switch_to.window(self.driver.window_handles[window_index])
def _post_message(self, message):
ActionChains(self.driver).send_keys(message + '\n').perform()
@property
def _chat_log_value(self):
return self.driver.find_element_by_css_selector('#chat-log').get_property('value')
我正在使用 Python 3.5 和 Django 2.0。
reduction.py 无法序列化包含 lambda 的对象。经过一些研究，这似乎与 Windows 环境中的多处理问题有关（并且不限于此示例。）
解决此问题的一种方法是 reduction.py
替换：import pickle 为 import dill as pickle
dill 包可以序列化 pickle 失败的这些对象。但是，如果不深入研究以确保此更改不会破坏其他任何内容，我不会建议将其用于生产环境。
我正在尝试学习 Django 频道教程。我能够实现所描述的聊天功能 here. But unittests completly copy pasted from this page 失败，出现以下错误 AttributeError: Can't pickle local object 'DaphneProcess.__init__.<locals>.<lambda>'。
完整追溯：
Traceback (most recent call last):
File "C:\Users\user\PycharmProjects\django_channels_test\venv35\lib\site-packages\django\test\testcases.py", line 202, in __call__
self._pre_setup()
File "C:\Users\user\PycharmProjects\django_channels_test\venv35\lib\site-packages\channels\testing\live.py", line 42, in _pre_setup
self._server_process.start()
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\process.py", line 105, in start
self._popen = self._Popen(self)
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\context.py", line 212, in _Popen
return _default_context.get_context().Process._Popen(process_obj)
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\context.py", line 313, in _Popen
return Popen(process_obj)
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\popen_spawn_win32.py", line 66, in __init__
reduction.dump(process_obj, to_child)
File "C:\Users\user\AppData\Local\Programs\Python\Python35\Lib\multiprocessing\reduction.py", line 59, in dump
ForkingPickler(file, protocol).dump(obj)
AttributeError: Can't pickle local object 'DaphneProcess.__init__.<locals>.<lambda>'
我的消费者class：
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.room_name = self.scope['url_route']['kwargs']['room_name']
self.room_group_name = 'chat_%s' % self.room_name
# Join room group
await self.channel_layer.group_add(
self.room_group_name,
self.channel_name
)
await self.accept()
async def disconnect(self, close_code):
# Leave room group
await self.channel_layer.group_discard(
self.room_group_name,
self.channel_name
)
# Receive message from WebSocket
async def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']
# Send message to room group
await self.channel_layer.group_send(
self.room_group_name,
{
'type': 'chat_message',
'message': message
}
)
# Receive message from room group
async def chat_message(self, event):
message = event['message']
# Send message to WebSocket
await self.send(text_data=json.dumps({
'message': message
}))
我的测试模块：
from channels.testing import ChannelsLiveServerTestCase
from selenium import webdriver
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.support.wait import WebDriverWait
class ChatTests(ChannelsLiveServerTestCase):
serve_static = True # emulate StaticLiveServerTestCase
@classmethod
def setUpClass(cls):
super().setUpClass()
try:
# NOTE: Requires "chromedriver" binary to be installed in $PATH
cls.driver = webdriver.Chrome()
except:
super().tearDownClass()
raise
@classmethod
def tearDownClass(cls):
cls.driver.quit()
super().tearDownClass()
def test_when_chat_message_posted_then_seen_by_everyone_in_same_room(self):
try:
self._enter_chat_room('room_1')
self._open_new_window()
self._enter_chat_room('room_1')
self._switch_to_window(0)
self._post_message('hello')
WebDriverWait(self.driver, 2).until(lambda _:
'hello' in self._chat_log_value,
'Message was not received by window 1 from window 1')
self._switch_to_window(1)
WebDriverWait(self.driver, 2).until(lambda _:
'hello' in self._chat_log_value,
'Message was not received by window 2 from window 1')
finally:
self._close_all_new_windows()
def test_when_chat_message_posted_then_not_seen_by_anyone_in_different_room(self):
try:
self._enter_chat_room('room_1')
self._open_new_window()
self._enter_chat_room('room_2')
self._switch_to_window(0)
self._post_message('hello')
WebDriverWait(self.driver, 2).until(lambda _:
'hello' in self._chat_log_value,
'Message was not received by window 1 from window 1')
self._switch_to_window(1)
self._post_message('world')
WebDriverWait(self.driver, 2).until(lambda _:
'world' in self._chat_log_value,
'Message was not received by window 2 from window 2')
self.assertTrue('hello' not in self._chat_log_value,
'Message was improperly received by window 2 from window 1')
finally:
self._close_all_new_windows()
# === Utility ===
def _enter_chat_room(self, room_name):
self.driver.get(self.live_server_url + '/chat/')
ActionChains(self.driver).send_keys(room_name + '\n').perform()
WebDriverWait(self.driver, 2).until(lambda _:
room_name in self.driver.current_url)
def _open_new_window(self):
self.driver.execute_script('window.open("about:blank", "_blank");')
self.driver.switch_to.window(self.driver.window_handles[-1])
def _close_all_new_windows(self):
while len(self.driver.window_handles) > 1:
self.driver.switch_to.window(self.driver.window_handles[-1])
self.driver.execute_script('window.close();')
if len(self.driver.window_handles) == 1:
self.driver.switch_to.window(self.driver.window_handles[0])
def _switch_to_window(self, window_index):
self.driver.switch_to.window(self.driver.window_handles[window_index])
def _post_message(self, message):
ActionChains(self.driver).send_keys(message + '\n').perform()
@property
def _chat_log_value(self):
return self.driver.find_element_by_css_selector('#chat-log').get_property('value')
我正在使用 Python 3.5 和 Django 2.0。
reduction.py 无法序列化包含 lambda 的对象。经过一些研究，这似乎与 Windows 环境中的多处理问题有关（并且不限于此示例。）
解决此问题的一种方法是 reduction.py
替换：import pickle 为 import dill as pickle
dill 包可以序列化 pickle 失败的这些对象。但是，如果不深入研究以确保此更改不会破坏其他任何内容，我不会建议将其用于生产环境。