Switch thread polling to a denylist with OCI version fallback

Author: Stefan9283

awslambdaric/lambda_config.py

@@ -3,15 +3,16 @@
 """
 
 import os
+import sys
 
 
 class LambdaConfigProvider:
-    SUPPORTED_THREADPOLLING_ENVS = {
-        "AWS_Lambda_python3.12",
-        "AWS_Lambda_python3.13",
-        "AWS_Lambda_python3.14",
-        "AWS_Lambda_python3.15",
+    UNSUPPORTED_THREADPOLLING_ENVS = {
+        "AWS_Lambda_python3.9",
+        "AWS_Lambda_python3.10",
+        "AWS_Lambda_python3.11",
     }
+
     SOCKET_PATH_ENV = "_LAMBDA_TELEMETRY_LOG_FD_PROVIDER_SOCKET"
     AWS_LAMBDA_RUNTIME_API = "AWS_LAMBDA_RUNTIME_API"
     AWS_LAMBDA_MAX_CONCURRENCY = "AWS_LAMBDA_MAX_CONCURRENCY"
@@ -38,10 +39,9 @@ def _parse_concurrency(self):
         return self._environ.get(self.AWS_LAMBDA_MAX_CONCURRENCY)
 
     def _parse_thread_polling(self):
-        return (
-            self._environ.get(self.AWS_EXECUTION_ENV)
-            in self.SUPPORTED_THREADPOLLING_ENVS
-        )
+        if self._environ.get(self.AWS_EXECUTION_ENV) in self.UNSUPPORTED_THREADPOLLING_ENVS:
+            return False
+        return sys.version_info >= (3, 4)
 
     def _parse_lmi_socket_path(self):
         return self._environ.get(self.SOCKET_PATH_ENV)

tests/test_lambda_config.py

@@ -3,6 +3,7 @@
 """
 
 import os
+import sys
 import unittest
 from awslambdaric.lambda_config import LambdaConfigProvider
 
@@ -41,16 +42,45 @@ def test_concurrency_and_is_multi_concurrent(self):
         self.assertIsNone(cfg2.max_concurrency)
         self.assertFalse(cfg2.is_multi_concurrent)
 
-    def test_use_thread_polling_flag(self):
+    def test_use_thread_polling_disabled_for_unsupported_managed_envs(self):
+        # Managed runtimes on the denylist never use thread polling,
+        # regardless of the Python version the code happens to run on.
+        for exec_env in LambdaConfigProvider.UNSUPPORTED_THREADPOLLING_ENVS:
+            env = {
+                "AWS_LAMBDA_RUNTIME_API": "a",
+                "AWS_EXECUTION_ENV": exec_env,
+            }
+            cfg = LambdaConfigProvider(["p", "h.fn"], environ=env)
+            self.assertFalse(
+                cfg.use_thread_polling,
+                msg=f"expected thread polling disabled for {exec_env}",
+            )
+
+    def test_use_thread_polling_enabled_for_custom_oci_image(self):
+        # Custom OCI images (AWS_Lambda_Image) are not on the denylist and
+        # fall back to the minimum-supported Python version check.
+        env = {
+            "AWS_LAMBDA_RUNTIME_API": "a",
+            "AWS_EXECUTION_ENV": "AWS_Lambda_Image",
+        }
+        cfg = LambdaConfigProvider(["p", "h.fn"], environ=env)
+        self.assertEqual(cfg.use_thread_polling, sys.version_info >= (3, 4))
+
+    def test_use_thread_polling_enabled_for_supported_managed_env(self):
+        # Managed runtimes not on the denylist (e.g. newer versions) fall
+        # back to the Python version check.
         env = {
             "AWS_LAMBDA_RUNTIME_API": "a",
             "AWS_EXECUTION_ENV": "AWS_Lambda_python3.12",
         }
         cfg = LambdaConfigProvider(["p", "h.fn"], environ=env)
-        self.assertTrue(cfg.use_thread_polling)
-        env2 = {"AWS_LAMBDA_RUNTIME_API": "a", "AWS_EXECUTION_ENV": "OTHER"}
-        cfg2 = LambdaConfigProvider(["p", "h.fn"], environ=env2)
-        self.assertFalse(cfg2.use_thread_polling)
+        self.assertEqual(cfg.use_thread_polling, sys.version_info >= (3, 4))
+
+    def test_use_thread_polling_without_execution_env(self):
+        # With no AWS_EXECUTION_ENV set, fall back to the version check.
+        env = {"AWS_LAMBDA_RUNTIME_API": "a"}
+        cfg = LambdaConfigProvider(["p", "h.fn"], environ=env)
+        self.assertEqual(cfg.use_thread_polling, sys.version_info >= (3, 4))
 
     def test_lmi_socket_path_property(self):
         env = {