Add workaround for Pytest 7.0.0 stacktrace normalization bug

tests/http1_tests.py

@@ -148,8 +148,10 @@ def get_request(self) -> Tuple[socket.socket, Any]:
         self._client_sockets.append(sock)
         return sock, addr
 
-    def shutdown_request(self, request: socket.socket) -> None:
+    def shutdown_request(self, request) -> None:
         """Forget the client socket"""
+        assert isinstance(request, socket.socket)
+
         self._client_sockets.remove(request)
         super().shutdown_request(request)  # type: ignore[misc]  # error: "shutdown_request" undefined in superclass
 

tests/router_engine_test.py

@@ -146,6 +146,7 @@ def neighbor_refresh(self, node_id, ProtocolVersion, instance, link_id, cost, no
         self.neighbors[node_id] = (instance, link_id, cost, now)
 
     def setUp(self):
+        super().setUp()
         self.sent = []
         self.neighbors = {}
         self.id = "R1"
@@ -210,6 +211,7 @@ def test_establish_multiple_peers(self):
 
 class PathTest(unittest.TestCase):
     def setUp(self):
+        super().setUp()
         self.id = 'R1'
         self.engine = PathEngine(self)
 

tests/system_test.py

@@ -877,12 +877,13 @@ class TestCase(unittest.TestCase, Tester):  # pylint: disable=too-many-public-me
 
     tester: Tester
 
-    def __init__(self, test_method):
-        unittest.TestCase.__init__(self, test_method)
+    def __init__(self, methodName='runTest'):
+        unittest.TestCase.__init__(self, methodName)
         Tester.__init__(self, self.id())
 
     @classmethod
     def setUpClass(cls):
+        super().setUpClass()
         cls.maxDiff = None
         cls.tester = Tester('.'.join([cls.__module__, cls.__name__, 'setUpClass']))
         cls.tester.rmtree()
@@ -893,12 +894,15 @@ def tearDownClass(cls):
         if hasattr(cls, 'tester'):
             cls.tester.teardown()
             del cls.tester
+        super().tearDownClass()
 
     def setUp(self):
+        super().setUp()
         Tester.setup(self)
 
     def tearDown(self):
         Tester.teardown(self)
+        super().tearDown()
 
     def assert_fair(self, seq):
         avg = sum(seq) / len(seq)

tests/system_tests_autolinks.py

@@ -205,8 +205,8 @@ def setUpClass(cls):
                           'address': 'examples', 'direction': 'out'}),
         ])
 
-    def __init__(self, test_method):
-        TestCase.__init__(self, test_method)
+    def setUp(self):
+        super().setUp()
         self.success = False
         self.timer_delay = 6
         self.max_attempts = 2

tests/system_tests_bad_configuration.py

@@ -26,6 +26,8 @@
 import os
 from threading import Timer
 from subprocess import PIPE, STDOUT
+from typing import ClassVar
+
 from system_test import TestCase, Qdrouterd, TIMEOUT, Process
 
 
@@ -36,12 +38,15 @@ class RouterTestBadConfiguration(TestCase):
     well defined, but are not supposed to cause a crash to the router
     process.
     """
+
+    config: ClassVar[Qdrouterd.Config]
+    name: ClassVar[str]
+    unresolvable_host_name: ClassVar[str]
+    router: ClassVar[Qdrouterd]
+
     @classmethod
-    def setUpClass(cls):
-        """
-        Set up router instance configuration to be used for testing.
-        :return:
-        """
+    def setUpClass(cls) -> None:
+        """Set up router instance configuration to be used for testing."""
         super(RouterTestBadConfiguration, cls).setUpClass()
         cls.name = "test-router"
         cls.unresolvable_host_name = 'unresolvable.host.name'
@@ -61,19 +66,23 @@ def setUpClass(cls):
         except OSError:
             pass
 
-    def __init__(self, test_method):
-        TestCase.__init__(self, test_method)
+    def setUp(self):
+        super().setUp()
         self.error_caught = False
         self.timer_delay = 0.2
         self.max_attempts = 100
         self.attempts_made = 0
+
         self.schedule_timer()
 
-    def schedule_timer(self):
+        # Wait till error is found or timed out waiting for it.
+        while self.waiting_for_error():
+            pass
+
+    def schedule_timer(self) -> None:
         """
         Schedules a timer triggers wait_for_unresolvable_host after
         timer_delay has been elapsed.
-        :return:
         """
         Timer(self.timer_delay, self.wait_for_unresolvable_host).start()
 
@@ -85,23 +94,20 @@ def address(self):
         """
         Returns the address that can be used along with qdmanage
         to query the running instance of the dispatch router.
-        :return:
         """
         return self.router.addresses[0]
 
-    def waiting_for_error(self):
+    def waiting_for_error(self) -> bool:
         """
         Returns True if max_attempts not yet reached and error is still not found.
-        :return: bool
         """
         return not self.error_caught and self.attempts_made < self.max_attempts
 
-    def wait_for_unresolvable_host(self):
+    def wait_for_unresolvable_host(self) -> None:
         """
         Wait for error to show up in the logs based on pre-defined max_attempts
         and timer_delay. If error is not caught within max_attempts * timer_delay
         then it stops scheduling new attempts.
-        :return:
         """
         try:
             # mode 'r' and 't' are defaults
@@ -125,22 +131,11 @@ def wait_for_unresolvable_host(self):
                 self.attempts_made += 1
                 self.schedule_timer()
 
-    def setUp(self):
-        """
-        Causes tests to wait till timer has found the expected error or
-        after it times out.
-        :return:
-        """
-        # Wait till error is found or timed out waiting for it.
-        while self.waiting_for_error():
-            pass
-
     def test_unresolvable_host_caught(self):
         """
         Validate if the error message stating host is unresolvable is printed
         to the router log.
         It expects that the error can be caught in the logs.
-        :return:
         """
         self.assertTrue(self.error_caught)
 
@@ -172,9 +167,6 @@ def setUpClass(cls):
         super(RouterTestIdFailCtrlChar, cls).setUpClass()
         cls.name = "test-router-ctrl-char"
 
-    def __init__(self, test_method):
-        TestCase.__init__(self, test_method)
-
     @classmethod
     def tearDownClass(cls):
         super(RouterTestIdFailCtrlChar, cls).tearDownClass()
@@ -214,9 +206,6 @@ def setUpClass(cls):
         super(RouterTestIdFailWhiteSpace, cls).setUpClass()
         cls.name = "test-router-ctrl-char"
 
-    def __init__(self, test_method):
-        TestCase.__init__(self, test_method)
-
     @classmethod
     def tearDownClass(cls):
         super(RouterTestIdFailWhiteSpace, cls).tearDownClass()

tests/system_tests_connector_status.py

@@ -62,8 +62,8 @@ def router(name, config):
         cls.routers[0].wait_ports()
         cls.routers[1].wait_ports()
 
-    def __init__(self, test_method):
-        TestCase.__init__(self, test_method)
+    def setUp(self):
+        super().setUp()
         self.success = False
         self.timer_delay = 2
         self.max_attempts = 5

tests/system_tests_distribution.py

@@ -118,7 +118,7 @@ class DistributionSkipMapper:
 #     Setup
 # ================================================================
 
-class DistributionTests (TestCase):
+class DistributionTests(TestCase):
 
     @classmethod
     def setUpClass(cls):

tests/system_tests_edge_router.py

@@ -97,8 +97,8 @@ def router(name, mode, connection, extra=None):
         cls.skip = {'test_01' : 0
                     }
 
-    def __init__(self, test_method):
-        TestCase.__init__(self, test_method)
+    def setUp(self):
+        super().setUp()
         self.success = False
         self.timer_delay = 2
         self.max_attempts = 3

tests/system_tests_handle_failover.py

@@ -90,8 +90,8 @@ def setUpClass(cls):
 
         cls.routers[1].wait_router_connected('B')
 
-    def __init__(self, test_method):
-        TestCase.__init__(self, test_method)
+    def setUp(self):
+        super().setUp()
         self.success = False
         self.timer_delay = 2
         self.max_attempts = 10

tests/system_tests_two_routers.py

@@ -1774,8 +1774,8 @@ def run(self):
 
 
 class TwoRouterConnection(TestCase):
-    def __init__(self, test_method):
-        TestCase.__init__(self, test_method)
+    def setUp(self):
+        super().setUp()
         self.success = False
         self.timer_delay = 4
         self.max_attempts = 2

tests/test_command.py

@@ -62,6 +62,7 @@ def show_all(self): pass
 
 class TestParseArgsQdstat(unittest.TestCase):
     def setUp(self):
+        super().setUp()
         self.parser = _qdstat_parser(BusManager=FBM)
 
     def test_parse_args_qdstat_print_help(self):
@@ -111,6 +112,7 @@ def test_parse_args_qdstat_limit(self):
 
 class TestParseArgsQdmanage(unittest.TestCase):
     def setUp(self):
+        super().setUp()
         self.operations = ["HERE", "SOME", "OPERATIONS"]
         self.parser = _qdmanage_parser(operations=self.operations)