Factor out backport of assertNoLogs.

Fix previous commit on Python 3.9.

Author: aaugustin

tests/asyncio/test_connection.py

@@ -19,7 +19,7 @@
 from websockets.protocol import CLIENT, SERVER, Protocol, State
 
 from ..protocol import RecordingProtocol
-from ..utils import MS
+from ..utils import MS, AssertNoLogsMixin
 from .connection import InterceptingConnection
 from .utils import alist
 
@@ -28,7 +28,7 @@
 # All tests run on the client side and the server side to validate this.
 
 
-class ClientConnectionTests(unittest.IsolatedAsyncioTestCase):
+class ClientConnectionTests(AssertNoLogsMixin, unittest.IsolatedAsyncioTestCase):
     LOCAL = CLIENT
     REMOTE = SERVER
 
@@ -48,23 +48,6 @@ async def asyncTearDown(self):
         await self.remote_connection.close()
         await self.connection.close()
 
-    if sys.version_info[:2] < (3, 10):  # pragma: no cover
-
-        @contextlib.contextmanager
-        def assertNoLogs(self, logger=None, level=None):
-            """
-            No message is logged on the given logger with at least the given level.
-
-            """
-            with self.assertLogs(logger, level) as logs:
-                # We want to test that no log message is emitted
-                # but assertLogs expects at least one log message.
-                logging.getLogger(logger).log(level, "dummy")
-                yield
-
-            level_name = logging.getLevelName(level)
-            self.assertEqual(logs.output, [f"{level_name}:{logger}:dummy"])
-
     # Test helpers built upon RecordingProtocol and InterceptingConnection.
 
     async def assertFrameSent(self, frame):
@@ -1277,7 +1260,7 @@ async def test_broadcast_skips_closed_connection(self):
         await self.connection.close()
         await self.assertFrameSent(Frame(Opcode.CLOSE, b"\x03\xe8"))
 
-        with self.assertNoLogs():
+        with self.assertNoLogs("websockets", logging.WARNING):
             broadcast([self.connection], "😀")
         await self.assertNoFrameSent()
 
@@ -1288,7 +1271,7 @@ async def test_broadcast_skips_closing_connection(self):
             await asyncio.sleep(0)
             await self.assertFrameSent(Frame(Opcode.CLOSE, b"\x03\xe8"))
 
-            with self.assertNoLogs():
+            with self.assertNoLogs("websockets", logging.WARNING):
                 broadcast([self.connection], "😀")
             await self.assertNoFrameSent()
 

tests/asyncio/test_server.py

@@ -21,6 +21,7 @@
     CLIENT_CONTEXT,
     MS,
     SERVER_CONTEXT,
+    AssertNoLogsMixin,
     temp_unix_socket_path,
 )
 from .server import (
@@ -32,7 +33,7 @@
 )
 
 
-class ServerTests(EvalShellMixin, unittest.IsolatedAsyncioTestCase):
+class ServerTests(EvalShellMixin, AssertNoLogsMixin, unittest.IsolatedAsyncioTestCase):
     async def test_connection(self):
         """Server receives connection from client and the handshake succeeds."""
         async with serve(*args) as server:

tests/legacy/test_protocol.py

@@ -938,7 +938,7 @@ def test_answer_ping_does_not_crash_if_connection_closing(self):
         self.receive_frame(Frame(True, OP_PING, b"test"))
         self.run_loop_once()
 
-        with self.assertNoLogs():
+        with self.assertNoLogs("websockets", logging.ERROR):
             self.loop.run_until_complete(self.protocol.close())
 
         self.loop.run_until_complete(close_task)  # cleanup
@@ -951,7 +951,7 @@ def test_answer_ping_does_not_crash_if_connection_closed(self):
         self.receive_eof()
         self.run_loop_once()
 
-        with self.assertNoLogs():
+        with self.assertNoLogs("websockets", logging.ERROR):
             self.loop.run_until_complete(self.protocol.close())
 
     def test_ignore_pong(self):
@@ -1028,7 +1028,7 @@ def test_acknowledge_aborted_ping(self):
             pong_waiter.result()
 
         # transfer_data doesn't crash, which would be logged.
-        with self.assertNoLogs():
+        with self.assertNoLogs("websockets", logging.ERROR):
             # Unclog incoming queue.
             self.loop.run_until_complete(self.protocol.recv())
             self.loop.run_until_complete(self.protocol.recv())
@@ -1375,7 +1375,7 @@ def test_remote_close_and_connection_lost(self):
         self.receive_eof()
         self.run_loop_once()
 
-        with self.assertNoLogs():
+        with self.assertNoLogs("websockets", logging.ERROR):
             self.loop.run_until_complete(self.protocol.close(reason="oh noes!"))
 
         self.assertConnectionClosed(CloseCode.NORMAL_CLOSURE, "close")
@@ -1500,14 +1500,14 @@ def test_broadcast_two_clients(self):
     def test_broadcast_skips_closed_connection(self):
         self.close_connection()
 
-        with self.assertNoLogs():
+        with self.assertNoLogs("websockets", logging.ERROR):
             broadcast([self.protocol], "café")
         self.assertNoFrameSent()
 
     def test_broadcast_skips_closing_connection(self):
         close_task = self.half_close_connection_local()
 
-        with self.assertNoLogs():
+        with self.assertNoLogs("websockets", logging.ERROR):
             broadcast([self.protocol], "café")
         self.assertNoFrameSent()
 

tests/legacy/utils.py

@@ -1,12 +1,12 @@
 import asyncio
-import contextlib
 import functools
-import logging
 import sys
 import unittest
 
+from ..utils import AssertNoLogsMixin
 
-class AsyncioTestCase(unittest.TestCase):
+
+class AsyncioTestCase(AssertNoLogsMixin, unittest.TestCase):
     """
     Base class for tests that sets up an isolated event loop for each test.
 
@@ -56,23 +56,6 @@ def run_loop_once(self):
         self.loop.call_soon(self.loop.stop)
         self.loop.run_forever()
 
-    if sys.version_info[:2] < (3, 10):  # pragma: no cover
-
-        @contextlib.contextmanager
-        def assertNoLogs(self, logger="websockets", level=logging.ERROR):
-            """
-            No message is logged on the given logger with at least the given level.
-
-            """
-            with self.assertLogs(logger, level) as logs:
-                # We want to test that no log message is emitted
-                # but assertLogs expects at least one log message.
-                logging.getLogger(logger).log(level, "dummy")
-                yield
-
-            level_name = logging.getLevelName(level)
-            self.assertEqual(logs.output, [f"{level_name}:{logger}:dummy"])
-
     def assertDeprecationWarnings(self, recorded_warnings, expected_warnings):
         """
         Check recorded deprecation warnings match a list of expected messages.

tests/utils.py

@@ -1,9 +1,11 @@
 import contextlib
 import email.utils
+import logging
 import os
 import pathlib
 import platform
 import ssl
+import sys
 import tempfile
 import time
 import unittest
@@ -113,6 +115,30 @@ def assertDeprecationWarning(self, message):
         self.assertEqual(str(warning.message), message)
 
 
+class AssertNoLogsMixin:
+    """
+    Backport of assertNoLogs for Python 3.9.
+
+    """
+
+    if sys.version_info[:2] < (3, 10):  # pragma: no cover
+
+        @contextlib.contextmanager
+        def assertNoLogs(self, logger=None, level=None):
+            """
+            No message is logged on the given logger with at least the given level.
+
+            """
+            with self.assertLogs(logger, level) as logs:
+                # We want to test that no log message is emitted
+                # but assertLogs expects at least one log message.
+                logging.getLogger(logger).log(level, "dummy")
+                yield
+
+            level_name = logging.getLevelName(level)
+            self.assertEqual(logs.output, [f"{level_name}:{logger}:dummy"])
+
+
 @contextlib.contextmanager
 def temp_unix_socket_path():
     with tempfile.TemporaryDirectory() as temp_dir: