Merge pull request #194 from cloudblue/LITE-31014-fix-memleak

LITE-31014: Fix memory leak in runner, support new connect-eaas-core

Author: r-s11v

.github/workflows/build.yml

@@ -15,7 +15,7 @@ jobs:
     timeout-minutes: 10
     strategy:
       matrix:
-        python-version: ["3.9", "3.10"]
+        python-version: ["3.9", "3.10", "3.11"]
     steps:
     - uses: actions/checkout@v3
       with:

connect/eaas/runner/managers/background.py

@@ -49,11 +49,14 @@ async def get_argument(self, task_data):
         """
         client = self.client
         if task_data.options.api_key:
-            client = AsyncConnectClient(
+            client = self._task_api_key_clients.setdefault(
                 task_data.options.api_key,
-                endpoint=self.config.get_api_url(),
-                use_specs=False,
-                default_headers=self.config.get_user_agent(),
+                AsyncConnectClient(
+                    task_data.options.api_key,
+                    endpoint=self.config.get_api_url(),
+                    use_specs=False,
+                    default_headers=self.config.get_user_agent(),
+                ),
             )
         object_exists = await self.filter_collection_by_event_definition(
             client,

connect/eaas/runner/managers/base.py

@@ -63,6 +63,7 @@ def __init__(self, config: ConfigHelper, handler: Union[EventsApp, TfnApp], enqu
             use_specs=False,
             default_headers=self.config.get_user_agent(),
         )
+        self._task_api_key_clients = {}
         self.running_tasks = 0
 
     async def filter_collection_by_event_definition(self, client, task_data):

connect/eaas/runner/managers/transformation.py

@@ -73,12 +73,15 @@ class TransformationTasksManager(TasksManagerBase):
 
     def get_client(self, task_data):
         if task_data.options.api_key:
-            return AsyncConnectClient(
+            return self._task_api_key_clients.setdefault(
                 task_data.options.api_key,
-                endpoint=self.config.get_api_url(),
-                use_specs=False,
-                default_headers=self.config.get_user_agent(),
-                logger=RequestLogger(logger),
+                AsyncConnectClient(
+                    task_data.options.api_key,
+                    endpoint=self.config.get_api_url(),
+                    use_specs=False,
+                    default_headers=self.config.get_user_agent(),
+                    logger=RequestLogger(logger),
+                ),
             )
 
         return self.client

poetry.lock

@@ -17,6 +17,7 @@ classifiers = [
     "Intended Audience :: Developers",
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
     "Topic :: Utilities",
     "Topic :: Software Development :: Libraries",
 ]
@@ -28,13 +29,13 @@ cextrun = 'connect.eaas.runner.main:main'
 python = ">=3.9,<4"
 websockets = "13.*"
 connect-openapi-client = ">=29.0,<30"
-logzio-python-handler = "^3.1.1"
+logzio-python-handler = "^4.1.4"
 backoff = "^2.2.1"
-connect-eaas-core = ">=30.3,<31"
+connect-eaas-core = ">=33.0,<34"
 httpx = ">=0.23,<1"
 rich = ">=12"
 pyfiglet = "^1.0.2"
-devtools = "^0.10.0"
+devtools = "^0.12.2"
 watchfiles = "^0.24"
 openpyxl = ">=3.0.0,<4"
 lxml = "^4.9.2"
@@ -57,7 +58,7 @@ flake8-pyproject = "^1.2"
 pytest-asyncio = "0.20.*"
 pytest-httpx = ">=0.20"
 responses = "^0.23.0"
-freezegun = "^1.2.1"
+freezegun = "^1.5.1"
 flaky = "^3.8"
 flake8-isort = "^6.0.0"
 

pyproject.toml

@@ -10,6 +10,9 @@
 import responses as sentry_responses
 import websockets
 
+from connect.eaas.core.decorators import (
+    router as router_decorator,
+)
 from connect.eaas.core.extension import (
     Extension,
 )
@@ -185,3 +188,9 @@ def default_env(mocker, unused_port):
             'row_transformation_task_max_execution_time': 60,
         },
     )
+
+
+@pytest.fixture(scope='function')
+def router():
+    router_decorator.routes = []
+    return router_decorator

tests/conftest.py

@@ -1,3 +1,4 @@
+import sys
 from importlib.metadata import (
     EntryPoint,
 )
@@ -195,7 +196,10 @@ def get_features(self):
 
     mocker.patch(
         'connect.eaas.runner.handlers.base.sys.modules',
-        {'my_module': mocked_module},
+        {
+            **sys.modules,
+            'my_module': mocked_module,
+        },
     )
 
     mocker.patch(

tests/handlers/test_base.py

@@ -362,19 +362,19 @@ def example_no_auth(self):
                 {
                     'method': 'GET',
                     'path': '/auth',
-                    'summary': 'Example Auth',
+                    'summary': 'Myextension.Example Auth',
                 },
             ],
             'no_auth': [
                 {
                     'method': 'GET',
                     'path': '/no_auth_deprecated',
-                    'summary': 'Example No Auth Deprecated',
+                    'summary': 'Myextension.Example No Auth Deprecated',
                 },
                 {
                     'method': 'GET',
                     'path': '/no_auth',
-                    'summary': 'Example No Auth',
+                    'summary': 'Myextension.Example No Auth',
                 },
             ],
         },

tests/handlers/test_web.py

@@ -10,7 +10,6 @@
 from connect.eaas.core.decorators import (
     account_settings_page,
     customer_pages,
-    router,
     web_app,
 )
 from connect.eaas.core.extension import (
@@ -169,7 +168,7 @@ def get_proxied_connect_api(cls):
     ),
 )
 @pytest.mark.asyncio
-async def test_http_call(mocker, ws_server, unused_port, settings_payload, task_options):
+async def test_http_call(mocker, ws_server, unused_port, settings_payload, task_options, router):
     setup_response = copy.deepcopy(settings_payload)
     setup_response['logging']['logging_api_key'] = 'logging_api_key'
     setup_response['logging']['log_level'] = None
@@ -340,7 +339,9 @@ def test_url(self):
     ),
 )
 @pytest.mark.asyncio
-async def test_http_call_redirect(mocker, ws_server, unused_port, settings_payload, task_options):
+async def test_http_call_redirect(
+    mocker, ws_server, unused_port, settings_payload, task_options, router,
+):
     setup_response = copy.deepcopy(settings_payload)
     setup_response['logging']['logging_api_key'] = 'logging_api_key'
     setup_response['logging']['log_level'] = None
@@ -477,7 +478,7 @@ def test_url(self):
 
 
 @pytest.mark.asyncio
-async def test_http_call_exception(mocker, ws_server, unused_port, settings_payload):
+async def test_http_call_exception(mocker, ws_server, unused_port, settings_payload, router):
     setup_response = copy.deepcopy(settings_payload)
     setup_response['logging']['logging_api_key'] = 'logging_api_key'
     mocker.patch(
@@ -1017,7 +1018,7 @@ def get_descriptor(cls):
 
 
 @pytest.mark.asyncio
-async def test_proper_internal_headers(mocker, ws_server, unused_port, settings_payload):
+async def test_proper_internal_headers(mocker, ws_server, unused_port, settings_payload, router):
 
     task_options = WebTaskOptions(
         correlation_id='correlation_id',
@@ -1148,7 +1149,7 @@ def test_url(self):
 
 
 @pytest.mark.asyncio
-async def test_optional_internal_headers(mocker, ws_server, unused_port, settings_payload):
+async def test_optional_internal_headers(mocker, ws_server, unused_port, settings_payload, router):
 
     task_options = WebTaskOptions(
         correlation_id='correlation_id',