Mekhanik evgenii/fix 1346 1

framework/deproxy/deproxy_client.py

@@ -608,6 +608,11 @@ def make_request(
             self.stream_id += 2
             self._valid_req_num += 1
 
+    def send_ping(self, data: bytes = b"\x00\x01\x02\x03\x04\x05\x06\x07") -> None:
+        self.h2_connection.ping(opaque_data=data)
+        self.send_bytes(self.h2_connection.data_to_send())
+        self.h2_connection.clear_outbound_data_buffer()
+
     @staticmethod
     def create_request(
         method,

framework/test_suite/tester.py

@@ -7,6 +7,7 @@
 import re
 import signal
 import subprocess
+import threading
 import typing
 import unittest
 from unittest.util import strclass
@@ -375,6 +376,23 @@ def start_all_clients(self):
             if not client.is_running():
                 raise Exception("Can not start client %s" % cid)
 
+    def create_task(self, func):
+        stop_event = threading.Event()
+
+        task = threading.Thread(target=func, args=(stop_event,))
+        self.__tasks.append((task, stop_event))
+
+        return task
+
+    async def __cleanup_tasks(self):
+        for task, stop_event in self.__tasks:
+            stop_event.set()
+
+        await asyncio.gather(
+            *[asyncio.to_thread(task.join) for task, _ in self.__tasks if task.ident is not None]
+        )
+        self.__tasks.clear()
+
     async def asyncSetUp(self):
         # `unittest.TestLoader.discover` returns initialized objects, we can't
         # raise `SkipTest` inside of `TempestaTest.__init__` because we are unable
@@ -386,6 +404,7 @@ async def asyncSetUp(self):
         test_logger.info(f"setUp '{self.id()}'")
         if not await remote.wait_available():
             raise Exception("Tempesta node is unavailable")
+        self.__tasks = []
         self.__exceptions = dict()
         self.__servers = {}
         self.__clients = {}
@@ -412,6 +431,7 @@ async def asyncSetUp(self):
         self.addAsyncCleanup(self.cleanup_interfaces)
         self.addAsyncCleanup(self.cleanup_deproxy)
         self.addAsyncCleanup(self.cleanup_services)
+        self.addAsyncCleanup(self.__cleanup_tasks)
         test_logger.info(f"setUp completed '{self.id()}'")
 
     async def cleanup_services(self):

tests/client_mem/__init__.py

@@ -0,0 +1,3 @@
+__all__ = ["test_client_mem"]
+
+# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4