piotrlinski commented on code in PR #61527: URL: https://github.com/apache/airflow/pull/61527#discussion_r2802622345
########## providers/cncf/kubernetes/tests/unit/cncf/kubernetes/secrets/test_kubernetes_secrets_backend.py: ########## @@ -0,0 +1,403 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import base64 +import json +from unittest import mock + +import pytest +from kubernetes.client.exceptions import ApiException + +from airflow.exceptions import AirflowException +from airflow.providers.cncf.kubernetes.secrets.kubernetes_secrets_backend import ( + KubernetesSecretsBackend, +) + +MODULE_PATH = "airflow.providers.cncf.kubernetes.secrets.kubernetes_secrets_backend.KubernetesSecretsBackend" + + +def _make_secret(data: dict[str, str], name: str = "some-secret"): + """Create a mock V1Secret with base64-encoded data.""" + encoded = {k: base64.b64encode(v.encode("utf-8")).decode("utf-8") for k, v in data.items()} + secret = mock.MagicMock() + secret.data = encoded + secret.metadata.name = name + return secret + + +def _make_secret_list(secrets: list): + """Create a mock V1SecretList with the given items.""" + secret_list = mock.MagicMock() + secret_list.items = secrets + return secret_list + + +class TestKubernetesSecretsBackendConnections: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_conn_value_uri(self, mock_client, mock_namespace): + """Test reading a connection URI from a Kubernetes secret.""" + uri = "postgresql://user:pass@host:5432/db" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": uri})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("my_db") + + assert result == uri + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="airflow.apache.org/connection-id=my_db", + resource_version="0", + ) + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_conn_value_json(self, mock_client, mock_namespace): + """Test reading a JSON-formatted connection from a Kubernetes secret.""" + conn_json = json.dumps( + { + "conn_type": "postgres", + "login": "user", + "password": "pass", + "host": "host", + "port": 5432, + "schema": "db", + } + ) + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": conn_json})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("my_db") + + assert result == conn_json + parsed = json.loads(result) + assert parsed["conn_type"] == "postgres" + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_conn_value_not_found(self, mock_client, mock_namespace): + """Test that a missing secret returns None.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list([]) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("nonexistent") + + assert result is None + + +class TestKubernetesSecretsBackendVariables: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_variable(self, mock_client, mock_namespace): + """Test reading a variable from a Kubernetes secret.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": "my-value"})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_variable("api_key") + + assert result == "my-value" + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="airflow.apache.org/variable-key=api_key", + resource_version="0", + ) + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_variable_not_found(self, mock_client, mock_namespace): + """Test that a missing variable secret returns None.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list([]) + + backend = KubernetesSecretsBackend() + result = backend.get_variable("nonexistent") + + assert result is None + + +class TestKubernetesSecretsBackendConfig: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_config(self, mock_client, mock_namespace): + """Test reading a config value from a Kubernetes secret.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": "sqlite:///airflow.db"})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_config("sql_alchemy_conn") + + assert result == "sqlite:///airflow.db" + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="airflow.apache.org/config-key=sql_alchemy_conn", + resource_version="0", + ) + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_config_not_found(self, mock_client, mock_namespace): + """Test that a missing config secret returns None.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list([]) + + backend = KubernetesSecretsBackend() + result = backend.get_config("nonexistent") + + assert result is None + + +class TestKubernetesSecretsBackendCustomConfig: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_custom_label(self, mock_client, mock_namespace): + """Test using a custom label key.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": "postgresql://localhost/db"})] + ) + + backend = KubernetesSecretsBackend(connections_label="my-org/conn") + result = backend.get_conn_value("my_db") + + assert result == "postgresql://localhost/db" + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="my-org/conn=my_db", + resource_version="0", + ) + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_custom_data_key(self, mock_client, mock_namespace): + """Test using a custom data key for connections.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"conn_uri": "postgresql://localhost/db"})] + ) + + backend = KubernetesSecretsBackend(connections_data_key="conn_uri") + result = backend.get_conn_value("my_db") + + assert result == "postgresql://localhost/db" + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_missing_value_data_key_returns_none(self, mock_client, mock_namespace): + """Test that a secret without the 'value' data key returns None.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"wrong_key": "some-value"})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("my_db") + + assert result is None + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_secret_with_none_data_returns_none(self, mock_client, mock_namespace): + """Test that a secret with None data returns None.""" + secret = mock.MagicMock() + secret.data = None + secret.metadata.name = "some-secret" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list([secret]) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("my_db") + + assert result is None + + +class TestKubernetesSecretsBackendTeamName: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_team_name_does_not_affect_conn_lookup(self, mock_client, mock_namespace): + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": "uri://val"})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("my_db", team_name="my-team") + + assert result == "uri://val" + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="airflow.apache.org/connection-id=my_db", + resource_version="0", + ) + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_team_name_does_not_affect_variable_lookup(self, mock_client, mock_namespace): + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": "val"})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_variable("my_key", team_name="my-team") + + assert result == "val" + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="airflow.apache.org/variable-key=my_key", + resource_version="0", + ) + + +class TestKubernetesSecretsBackendLabelNone: + @mock.patch(f"{MODULE_PATH}._get_secret") + def test_connections_label_none(self, mock_get_secret): + """Test that setting connections_label to None skips connection lookups.""" + backend = KubernetesSecretsBackend(connections_label=None) + result = backend.get_conn_value("my_db") + + assert result is None + mock_get_secret.assert_not_called() + + @mock.patch(f"{MODULE_PATH}._get_secret") + def test_variables_label_none(self, mock_get_secret): + """Test that setting variables_label to None skips variable lookups.""" + backend = KubernetesSecretsBackend(variables_label=None) + result = backend.get_variable("my_var") + + assert result is None + mock_get_secret.assert_not_called() + + @mock.patch(f"{MODULE_PATH}._get_secret") + def test_config_label_none(self, mock_get_secret): + """Test that setting config_label to None skips config lookups.""" + backend = KubernetesSecretsBackend(config_label=None) + result = backend.get_config("my_config") + + assert result is None + mock_get_secret.assert_not_called() + + +class TestKubernetesSecretsBackendMultipleMatches: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_multiple_secrets_uses_first_and_warns(self, mock_client, mock_namespace, caplog): + """Test that multiple matching secrets uses the first and logs a warning.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [ + _make_secret({"value": "first-value"}, name="secret-1"), + _make_secret({"value": "second-value"}, name="secret-2"), + ] + ) + + backend = KubernetesSecretsBackend() + import logging + + with caplog.at_level(logging.WARNING): + result = backend.get_conn_value("my_db") + + assert result == "first-value" + assert "Multiple secrets found" in caplog.text + + +class TestKubernetesSecretsBackendResourceVersion: Review Comment: rmoved ########## providers/cncf/kubernetes/tests/unit/cncf/kubernetes/secrets/test_kubernetes_secrets_backend.py: ########## @@ -0,0 +1,403 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import base64 +import json +from unittest import mock + +import pytest +from kubernetes.client.exceptions import ApiException + +from airflow.exceptions import AirflowException +from airflow.providers.cncf.kubernetes.secrets.kubernetes_secrets_backend import ( + KubernetesSecretsBackend, +) + +MODULE_PATH = "airflow.providers.cncf.kubernetes.secrets.kubernetes_secrets_backend.KubernetesSecretsBackend" + + +def _make_secret(data: dict[str, str], name: str = "some-secret"): + """Create a mock V1Secret with base64-encoded data.""" + encoded = {k: base64.b64encode(v.encode("utf-8")).decode("utf-8") for k, v in data.items()} + secret = mock.MagicMock() + secret.data = encoded + secret.metadata.name = name + return secret + + +def _make_secret_list(secrets: list): + """Create a mock V1SecretList with the given items.""" + secret_list = mock.MagicMock() + secret_list.items = secrets + return secret_list + + +class TestKubernetesSecretsBackendConnections: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_conn_value_uri(self, mock_client, mock_namespace): + """Test reading a connection URI from a Kubernetes secret.""" + uri = "postgresql://user:pass@host:5432/db" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": uri})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("my_db") + + assert result == uri + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="airflow.apache.org/connection-id=my_db", + resource_version="0", + ) + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_conn_value_json(self, mock_client, mock_namespace): + """Test reading a JSON-formatted connection from a Kubernetes secret.""" + conn_json = json.dumps( + { + "conn_type": "postgres", + "login": "user", + "password": "pass", + "host": "host", + "port": 5432, + "schema": "db", + } + ) + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": conn_json})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("my_db") + + assert result == conn_json + parsed = json.loads(result) + assert parsed["conn_type"] == "postgres" + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_conn_value_not_found(self, mock_client, mock_namespace): + """Test that a missing secret returns None.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list([]) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("nonexistent") + + assert result is None + + +class TestKubernetesSecretsBackendVariables: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_variable(self, mock_client, mock_namespace): + """Test reading a variable from a Kubernetes secret.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": "my-value"})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_variable("api_key") + + assert result == "my-value" + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="airflow.apache.org/variable-key=api_key", + resource_version="0", + ) + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_variable_not_found(self, mock_client, mock_namespace): + """Test that a missing variable secret returns None.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list([]) + + backend = KubernetesSecretsBackend() + result = backend.get_variable("nonexistent") + + assert result is None + + +class TestKubernetesSecretsBackendConfig: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_config(self, mock_client, mock_namespace): + """Test reading a config value from a Kubernetes secret.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": "sqlite:///airflow.db"})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_config("sql_alchemy_conn") + + assert result == "sqlite:///airflow.db" + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="airflow.apache.org/config-key=sql_alchemy_conn", + resource_version="0", + ) + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_get_config_not_found(self, mock_client, mock_namespace): + """Test that a missing config secret returns None.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list([]) + + backend = KubernetesSecretsBackend() + result = backend.get_config("nonexistent") + + assert result is None + + +class TestKubernetesSecretsBackendCustomConfig: + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_custom_label(self, mock_client, mock_namespace): + """Test using a custom label key.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"value": "postgresql://localhost/db"})] + ) + + backend = KubernetesSecretsBackend(connections_label="my-org/conn") + result = backend.get_conn_value("my_db") + + assert result == "postgresql://localhost/db" + mock_client.return_value.list_namespaced_secret.assert_called_once_with( + "default", + label_selector="my-org/conn=my_db", + resource_version="0", + ) + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_custom_data_key(self, mock_client, mock_namespace): + """Test using a custom data key for connections.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"conn_uri": "postgresql://localhost/db"})] + ) + + backend = KubernetesSecretsBackend(connections_data_key="conn_uri") + result = backend.get_conn_value("my_db") + + assert result == "postgresql://localhost/db" + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_missing_value_data_key_returns_none(self, mock_client, mock_namespace): + """Test that a secret without the 'value' data key returns None.""" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list( + [_make_secret({"wrong_key": "some-value"})] + ) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("my_db") + + assert result is None + + @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default") + @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock) + def test_secret_with_none_data_returns_none(self, mock_client, mock_namespace): + """Test that a secret with None data returns None.""" + secret = mock.MagicMock() + secret.data = None + secret.metadata.name = "some-secret" + mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list([secret]) + + backend = KubernetesSecretsBackend() + result = backend.get_conn_value("my_db") + + assert result is None + + +class TestKubernetesSecretsBackendTeamName: Review Comment: removed -- 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]
