Copilot commented on code in PR #65566:
URL: https://github.com/apache/airflow/pull/65566#discussion_r3113197078
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py:
##########
@@ -1935,3 +1952,96 @@ async def test_wait_until_container_started(
]
)
mock_sleep.assert_awaited_once_with(10)
+
+ # -------------------------------------------------------------------------
+ # Tests for the per-instance Configuration isolation fix
+ # -------------------------------------------------------------------------
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_load_config_initialises_client_configuration(self,
mock_load):
+ """_load_config allocates a fresh per-instance Configuration when none
was supplied."""
+ mock_load.return_value = None
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False,
config_dict={"a": "b"})
+ assert hook.client_configuration is None
+
+ await hook._load_config()
+
+ assert isinstance(hook.client_configuration,
async_client.Configuration)
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_incluster_config")
+ async def test_load_config_incluster_passes_client_configuration(self,
mock_incluster):
+ """load_incluster_config is called with the per-instance
client_configuration."""
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=True)
+
+ await hook._load_config()
+
+
mock_incluster.assert_called_once_with(client_configuration=hook.client_configuration)
+ assert isinstance(hook.client_configuration,
async_client.Configuration)
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_load_config_config_dict_passes_client_configuration(self,
mock_load):
+ """load_kube_config_from_dict is called with the per-instance
client_configuration."""
+ mock_load.return_value = None
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False,
config_dict={"a": "b"})
+
+ await hook._load_config()
+
+ mock_load.assert_called_once()
+ _, kwargs = mock_load.call_args
+ assert "client_configuration" in kwargs
+ assert kwargs["client_configuration"] is hook.client_configuration
+ assert isinstance(hook.client_configuration,
async_client.Configuration)
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_get_conn_passes_client_configuration_to_async_client(self,
mock_load):
+ """get_conn passes the per-instance configuration= to
_TimeoutAsyncK8sApiClient."""
+ mock_load.return_value = None
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False,
config_dict={"a": "b"})
+
+ with mock.patch(HOOK_MODULE + "._TimeoutAsyncK8sApiClient") as
mock_client_cls:
+ mock_client_cls.return_value.close = mock.AsyncMock()
+ async with hook.get_conn():
+ pass
+
+
mock_client_cls.assert_called_once_with(configuration=hook.client_configuration)
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_concurrent_hooks_have_isolated_configurations(self,
mock_load):
+ """Two concurrent hooks for different clusters each use an independent
Configuration.
+
+ This is the core regression test for the race condition: before the
fix,
+ load_kube_config_from_dict was called with client_configuration=None,
causing it
+ to overwrite the global kubernetes_asyncio Configuration via
Configuration.set_default().
+ Concurrent triggerer coroutines for different clusters would
interleave at the await
+ inside load_kube_config_from_dict and produce a mixed global state
(e.g. cluster-a's
+ server URL paired with cluster-b's bearer token).
+ """
+
+ async def populate_config(config_dict, context, client_configuration):
Review Comment:
The “concurrent hooks” regression test doesn’t currently force any
interleaving: the `populate_config` side effect has no `await` points, so each
`_load_config()` call can run to completion without yielding to the event loop.
Adding at least one scheduling yield inside the side effect (or otherwise
synchronizing both coroutines mid-call) would make this test actually exercise
the race condition scenario it describes.
```suggestion
async def populate_config(config_dict, context,
client_configuration):
# Simulate the library yielding while loading config so
concurrent calls interleave.
await asyncio.sleep(0)
```
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -894,6 +900,9 @@ async def _load_config(self):
and cached. For exec-based auth (EKS, GKE), the config is reloaded on
every call so
that short-lived tokens are always refreshed.
"""
+ if self.client_configuration is None:
+ self.client_configuration = async_client.Configuration()
+
Review Comment:
`AsyncKubernetesHook` assigns `self.client_configuration` to an
`async_client.Configuration`, but the inherited
`KubernetesHook.client_configuration` parameter/attribute is typed as
`kubernetes.client.Configuration`. Now that `get_conn()` forwards this object
into `kubernetes_asyncio.client.ApiClient`, callers passing a sync
`Configuration` via `client_configuration=` could hit runtime
incompatibilities. Consider explicitly typing/overriding `client_configuration`
on `AsyncKubernetesHook` (or using a separate attribute) and
validating/converting the provided value to an `async_client.Configuration` to
make the public API unambiguous.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
