Removed dependency on unittest.TestCase base class

Author: miguelgrinberg

tests/async/test_admin.py

@@ -2,7 +2,6 @@
 import threading
 import time
 from unittest import mock
-import unittest
 import pytest
 try:
     from engineio.async_socket import AsyncSocket as EngineIOSocket
@@ -38,13 +37,13 @@ def connect(sid, environ, auth):
                 pass
 
             async def shutdown():
-                await instrumented_server.shutdown()
+                await self.isvr.shutdown()
                 await sio.shutdown()
 
             if 'server_stats_interval' not in ikwargs:
                 ikwargs['server_stats_interval'] = 0.25
 
-            instrumented_server = sio.instrument(auth=auth, **ikwargs)
+            self.isvr = sio.instrument(auth=auth, **ikwargs)
             server = SocketIOWebServer(sio, on_shutdown=shutdown)
             server.start()
 
@@ -56,10 +55,11 @@ async def shutdown():
             EngineIOSocket.schedule_ping = mock.MagicMock()
 
             try:
-                ret = f(self, instrumented_server, *args, **kwargs)
+                ret = f(self, *args, **kwargs)
             finally:
                 server.stop()
-                instrumented_server.uninstrument()
+                self.isvr.uninstrument()
+                self.isvr = None
 
             EngineIOSocket.schedule_ping = original_schedule_ping
 
@@ -80,12 +80,12 @@ async def _async_custom_auth(auth):
     return auth == {'foo': 'bar'}
 
 
-class TestAsyncAdmin(unittest.TestCase):
-    def setUp(self):
+class TestAsyncAdmin:
+    def setup_method(self):
         print('threads at start:', threading.enumerate())
         self.thread_count = threading.active_count()
 
-    def tearDown(self):
+    def teardown_method(self):
         print('threads at end:', threading.enumerate())
         assert self.thread_count == threading.active_count()
 
@@ -107,15 +107,15 @@ def test_missing_auth(self):
             sio.instrument()
 
     @with_instrumented_server(auth=False)
-    def test_admin_connect_with_no_auth(self, isvr):
+    def test_admin_connect_with_no_auth(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin')
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin',
                                  auth={'foo': 'bar'})
 
     @with_instrumented_server(auth={'foo': 'bar'})
-    def test_admin_connect_with_dict_auth(self, isvr):
+    def test_admin_connect_with_dict_auth(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin',
                                  auth={'foo': 'bar'})
@@ -131,7 +131,7 @@ def test_admin_connect_with_dict_auth(self, isvr):
 
     @with_instrumented_server(auth=[{'foo': 'bar'},
                                     {'u': 'admin', 'p': 'secret'}])
-    def test_admin_connect_with_list_auth(self, isvr):
+    def test_admin_connect_with_list_auth(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin',
                                  auth={'foo': 'bar'})
@@ -148,7 +148,7 @@ def test_admin_connect_with_list_auth(self, isvr):
                                      namespace='/admin')
 
     @with_instrumented_server(auth=_custom_auth)
-    def test_admin_connect_with_function_auth(self, isvr):
+    def test_admin_connect_with_function_auth(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin',
                                  auth={'foo': 'bar'})
@@ -162,7 +162,7 @@ def test_admin_connect_with_function_auth(self, isvr):
                                      namespace='/admin')
 
     @with_instrumented_server(auth=_async_custom_auth)
-    def test_admin_connect_with_async_function_auth(self, isvr):
+    def test_admin_connect_with_async_function_auth(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin',
                                  auth={'foo': 'bar'})
@@ -176,7 +176,7 @@ def test_admin_connect_with_async_function_auth(self, isvr):
                                      namespace='/admin')
 
     @with_instrumented_server()
-    def test_admin_connect_only_admin(self, isvr):
+    def test_admin_connect_only_admin(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin')
             sid = admin_client.sid
@@ -201,7 +201,7 @@ def test_admin_connect_only_admin(self, isvr):
             events['server_stats']['namespaces']
 
     @with_instrumented_server()
-    def test_admin_connect_with_others(self, isvr):
+    def test_admin_connect_with_others(self):
         with socketio.SimpleClient() as client1, \
                 socketio.SimpleClient() as client2, \
                 socketio.SimpleClient() as client3, \
@@ -210,12 +210,12 @@ def test_admin_connect_with_others(self, isvr):
             client1.emit('enter_room', 'room')
             sid1 = client1.sid
 
-            saved_check_for_upgrade = isvr._check_for_upgrade
-            isvr._check_for_upgrade = AsyncMock()
+            saved_check_for_upgrade = self.isvr._check_for_upgrade
+            self.isvr._check_for_upgrade = AsyncMock()
             client2.connect('http://localhost:8900', namespace='/foo',
                             transports=['polling'])
             sid2 = client2.sid
-            isvr._check_for_upgrade = saved_check_for_upgrade
+            self.isvr._check_for_upgrade = saved_check_for_upgrade
 
             client3.connect('http://localhost:8900', namespace='/admin')
             sid3 = client3.sid
@@ -251,7 +251,7 @@ def test_admin_connect_with_others(self, isvr):
                 assert socket['rooms'] == [sid3]
 
     @with_instrumented_server(mode='production', read_only=True)
-    def test_admin_connect_production(self, isvr):
+    def test_admin_connect_production(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin')
             events = self._expect({'config': 1, 'server_stats': 2},
@@ -272,7 +272,7 @@ def test_admin_connect_production(self, isvr):
             events['server_stats']['namespaces']
 
     @with_instrumented_server()
-    def test_admin_features(self, isvr):
+    def test_admin_features(self):
         with socketio.SimpleClient() as client1, \
                 socketio.SimpleClient() as client2, \
                 socketio.SimpleClient() as admin_client:

tests/async/test_client.py

@@ -1,5 +1,4 @@
 import asyncio
-import unittest
 from unittest import mock
 
 import pytest
@@ -12,7 +11,7 @@
 from .helpers import AsyncMock, _run
 
 
-class TestAsyncClient(unittest.TestCase):
+class TestAsyncClient:
     def test_is_asyncio_based(self):
         c = async_client.AsyncClient()
         assert c.is_asyncio_based()

tests/async/test_manager.py

@@ -1,13 +1,12 @@
-import unittest
 from unittest import mock
 
 from socketio import async_manager
 from socketio import packet
 from .helpers import AsyncMock, _run
 
 
-class TestAsyncManager(unittest.TestCase):
-    def setUp(self):
+class TestAsyncManager:
+    def setup_method(self):
         id = 0
 
         def generate_id():

tests/async/test_namespace.py

@@ -1,13 +1,10 @@
-import sys
-import unittest
 from unittest import mock
 
 from socketio import async_namespace
 from .helpers import AsyncMock, _run
 
 
-@unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+')
-class TestAsyncNamespace(unittest.TestCase):
+class TestAsyncNamespace:
     def test_connect_event(self):
         result = {}
 

tests/async/test_pubsub_manager.py

@@ -1,6 +1,5 @@
 import asyncio
 import functools
-import unittest
 from unittest import mock
 
 import pytest
@@ -11,8 +10,8 @@
 from .helpers import AsyncMock, _run
 
 
-class TestAsyncPubSubManager(unittest.TestCase):
-    def setUp(self):
+class TestAsyncPubSubManager:
+    def setup_method(self):
         id = 0
 
         def generate_id():

tests/async/test_server.py

@@ -1,6 +1,5 @@
 import asyncio
 import logging
-import unittest
 from unittest import mock
 
 from engineio import json
@@ -18,8 +17,8 @@
 @mock.patch('socketio.server.engineio.AsyncServer', **{
     'return_value.generate_id.side_effect': [str(i) for i in range(1, 10)],
     'return_value.send_packet': AsyncMock()})
-class TestAsyncServer(unittest.TestCase):
-    def tearDown(self):
+class TestAsyncServer:
+    def teardown_method(self):
         # restore JSON encoder, in case a test changed it
         packet.Packet.json = json
 

tests/async/test_simple_client.py

@@ -1,5 +1,4 @@
 import asyncio
-import unittest
 from unittest import mock
 import pytest
 
@@ -8,7 +7,7 @@
 from .helpers import AsyncMock, _run
 
 
-class TestAsyncAsyncSimpleClient(unittest.TestCase):
+class TestAsyncAsyncSimpleClient:
     def test_constructor(self):
         client = AsyncSimpleClient(1, '2', a='3', b=4)
         assert client.client_args == (1, '2')

tests/common/test_admin.py

@@ -2,7 +2,6 @@
 import threading
 import time
 from unittest import mock
-import unittest
 import pytest
 from engineio.socket import Socket as EngineIOSocket
 import socketio
@@ -36,7 +35,7 @@ def connect(sid, environ, auth):
             if 'server_stats_interval' not in ikwargs:
                 ikwargs['server_stats_interval'] = 0.25
 
-            instrumented_server = sio.instrument(auth=auth, **ikwargs)
+            self.isvr = sio.instrument(auth=auth, **ikwargs)
             server = SocketIOWebServer(sio)
             server.start()
 
@@ -48,11 +47,12 @@ def connect(sid, environ, auth):
             EngineIOSocket.schedule_ping = mock.MagicMock()
 
             try:
-                ret = f(self, instrumented_server, *args, **kwargs)
+                ret = f(self, *args, **kwargs)
             finally:
                 server.stop()
-                instrumented_server.shutdown()
-                instrumented_server.uninstrument()
+                self.isvr.shutdown()
+                self.isvr.uninstrument()
+                self.isvr = None
 
             EngineIOSocket.schedule_ping = original_schedule_ping
 
@@ -69,12 +69,12 @@ def _custom_auth(auth):
     return auth == {'foo': 'bar'}
 
 
-class TestAdmin(unittest.TestCase):
-    def setUp(self):
+class TestAdmin:
+    def setup_method(self):
         print('threads at start:', threading.enumerate())
         self.thread_count = threading.active_count()
 
-    def tearDown(self):
+    def teardown_method(self):
         print('threads at end:', threading.enumerate())
         assert self.thread_count == threading.active_count()
 
@@ -96,15 +96,15 @@ def test_missing_auth(self):
             sio.instrument()
 
     @with_instrumented_server(auth=False)
-    def test_admin_connect_with_no_auth(self, isvr):
+    def test_admin_connect_with_no_auth(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin')
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin',
                                  auth={'foo': 'bar'})
 
     @with_instrumented_server(auth={'foo': 'bar'})
-    def test_admin_connect_with_dict_auth(self, isvr):
+    def test_admin_connect_with_dict_auth(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin',
                                  auth={'foo': 'bar'})
@@ -120,7 +120,7 @@ def test_admin_connect_with_dict_auth(self, isvr):
 
     @with_instrumented_server(auth=[{'foo': 'bar'},
                                     {'u': 'admin', 'p': 'secret'}])
-    def test_admin_connect_with_list_auth(self, isvr):
+    def test_admin_connect_with_list_auth(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin',
                                  auth={'foo': 'bar'})
@@ -137,7 +137,7 @@ def test_admin_connect_with_list_auth(self, isvr):
                                      namespace='/admin')
 
     @with_instrumented_server(auth=_custom_auth)
-    def test_admin_connect_with_function_auth(self, isvr):
+    def test_admin_connect_with_function_auth(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin',
                                  auth={'foo': 'bar'})
@@ -151,7 +151,7 @@ def test_admin_connect_with_function_auth(self, isvr):
                                      namespace='/admin')
 
     @with_instrumented_server()
-    def test_admin_connect_only_admin(self, isvr):
+    def test_admin_connect_only_admin(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin')
             sid = admin_client.sid
@@ -176,7 +176,7 @@ def test_admin_connect_only_admin(self, isvr):
             events['server_stats']['namespaces']
 
     @with_instrumented_server()
-    def test_admin_connect_with_others(self, isvr):
+    def test_admin_connect_with_others(self):
         with socketio.SimpleClient() as client1, \
                 socketio.SimpleClient() as client2, \
                 socketio.SimpleClient() as client3, \
@@ -185,12 +185,12 @@ def test_admin_connect_with_others(self, isvr):
             client1.emit('enter_room', 'room')
             sid1 = client1.sid
 
-            saved_check_for_upgrade = isvr._check_for_upgrade
-            isvr._check_for_upgrade = mock.MagicMock()
+            saved_check_for_upgrade = self.isvr._check_for_upgrade
+            self.isvr._check_for_upgrade = mock.MagicMock()
             client2.connect('http://localhost:8900', namespace='/foo',
                             transports=['polling'])
             sid2 = client2.sid
-            isvr._check_for_upgrade = saved_check_for_upgrade
+            self.isvr._check_for_upgrade = saved_check_for_upgrade
 
             client3.connect('http://localhost:8900', namespace='/admin')
             sid3 = client3.sid
@@ -226,7 +226,7 @@ def test_admin_connect_with_others(self, isvr):
                 assert socket['rooms'] == [sid3]
 
     @with_instrumented_server(mode='production', read_only=True)
-    def test_admin_connect_production(self, isvr):
+    def test_admin_connect_production(self):
         with socketio.SimpleClient() as admin_client:
             admin_client.connect('http://localhost:8900', namespace='/admin')
             events = self._expect({'config': 1, 'server_stats': 2},
@@ -247,7 +247,7 @@ def test_admin_connect_production(self, isvr):
             events['server_stats']['namespaces']
 
     @with_instrumented_server()
-    def test_admin_features(self, isvr):
+    def test_admin_features(self):
         with socketio.SimpleClient() as client1, \
                 socketio.SimpleClient() as client2, \
                 socketio.SimpleClient() as admin_client: