jedcunningham commented on a change in pull request #19695:
URL: https://github.com/apache/airflow/pull/19695#discussion_r760582442
##########
File path: tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
##########
@@ -37,107 +35,181 @@
KUBE_CONFIG_PATH = os.getenv('KUBECONFIG', '~/.kube/config')
-class TestKubernetesHook(unittest.TestCase):
+class TestKubernetesHook:
@classmethod
- def setUpClass(cls) -> None:
+ def setup_class(cls) -> None:
for conn_id, extra in [
- ('kubernetes_in_cluster', {'extra__kubernetes__in_cluster': True}),
- ('kubernetes_kube_config', {'extra__kubernetes__kube_config':
'{"test": "kube"}'}),
- ('kubernetes_kube_config_path',
{'extra__kubernetes__kube_config_path': 'path/to/file'}),
- ('kubernetes_in_cluster_empty', {'extra__kubernetes__in_cluster':
''}),
- ('kubernetes_kube_config_empty',
{'extra__kubernetes__kube_config': ''}),
- ('kubernetes_kube_config_path_empty',
{'extra__kubernetes__kube_config_path': ''}),
- ('kubernetes_with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
- ('kubernetes_default_kube_config', {}),
+ ('in_cluster', {'extra__kubernetes__in_cluster': True}),
+ ('kube_config', {'extra__kubernetes__kube_config': '{"test":
"kube"}'}),
+ ('kube_config_path', {'extra__kubernetes__kube_config_path':
'path/to/file'}),
+ ('in_cluster_empty', {'extra__kubernetes__in_cluster': ''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('context_empty', {'extra__kubernetes__cluster_context': ''}),
+ ('context', {'extra__kubernetes__cluster_context': 'my-context'}),
+ ('with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
+ ('default_kube_config', {}),
]:
db.merge_conn(Connection(conn_type='kubernetes', conn_id=conn_id,
extra=json.dumps(extra)))
@classmethod
- def tearDownClass(cls) -> None:
+ def teardown_class(cls) -> None:
clear_db_connections()
- @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
- def test_in_cluster_connection(self, mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
- @patch("kubernetes.config.kube_config.KubeConfigMerger")
- @patch("kubernetes.config.kube_config.KubeConfigLoader")
- def test_in_cluster_connection_empty(self, mock_kube_config_merger,
mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster_empty')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
+ @pytest.mark.parametrize(
+ 'in_cluster_param, conn_id, in_cluster_called',
+ (
+ pytest.param(True, None, True),
+ pytest.param(None, None, False),
+ pytest.param(None, 'in_cluster', True),
+ pytest.param(True, 'in_cluster', True),
+ pytest.param(False, 'in_cluster', False),
+ pytest.param(None, 'in_cluster_empty', False),
+ pytest.param(True, 'in_cluster_empty', True),
+ pytest.param(False, 'in_cluster_empty', False),
Review comment:
```suggestion
(True, None, True),
(None, None, False),
(None, 'in_cluster', True),
(True, 'in_cluster', True),
(False, 'in_cluster', False),
(None, 'in_cluster_empty', False),
(True, 'in_cluster_empty', True),
(False, 'in_cluster_empty', False),
```
Nit: these can just be normal tuples.
##########
File path: tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
##########
@@ -37,107 +35,181 @@
KUBE_CONFIG_PATH = os.getenv('KUBECONFIG', '~/.kube/config')
-class TestKubernetesHook(unittest.TestCase):
+class TestKubernetesHook:
@classmethod
- def setUpClass(cls) -> None:
+ def setup_class(cls) -> None:
for conn_id, extra in [
- ('kubernetes_in_cluster', {'extra__kubernetes__in_cluster': True}),
- ('kubernetes_kube_config', {'extra__kubernetes__kube_config':
'{"test": "kube"}'}),
- ('kubernetes_kube_config_path',
{'extra__kubernetes__kube_config_path': 'path/to/file'}),
- ('kubernetes_in_cluster_empty', {'extra__kubernetes__in_cluster':
''}),
- ('kubernetes_kube_config_empty',
{'extra__kubernetes__kube_config': ''}),
- ('kubernetes_kube_config_path_empty',
{'extra__kubernetes__kube_config_path': ''}),
- ('kubernetes_with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
- ('kubernetes_default_kube_config', {}),
+ ('in_cluster', {'extra__kubernetes__in_cluster': True}),
+ ('kube_config', {'extra__kubernetes__kube_config': '{"test":
"kube"}'}),
+ ('kube_config_path', {'extra__kubernetes__kube_config_path':
'path/to/file'}),
+ ('in_cluster_empty', {'extra__kubernetes__in_cluster': ''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('context_empty', {'extra__kubernetes__cluster_context': ''}),
+ ('context', {'extra__kubernetes__cluster_context': 'my-context'}),
+ ('with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
+ ('default_kube_config', {}),
]:
db.merge_conn(Connection(conn_type='kubernetes', conn_id=conn_id,
extra=json.dumps(extra)))
@classmethod
- def tearDownClass(cls) -> None:
+ def teardown_class(cls) -> None:
clear_db_connections()
- @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
- def test_in_cluster_connection(self, mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
- @patch("kubernetes.config.kube_config.KubeConfigMerger")
- @patch("kubernetes.config.kube_config.KubeConfigLoader")
- def test_in_cluster_connection_empty(self, mock_kube_config_merger,
mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster_empty')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
+ @pytest.mark.parametrize(
+ 'in_cluster_param, conn_id, in_cluster_called',
+ (
+ pytest.param(True, None, True),
+ pytest.param(None, None, False),
+ pytest.param(None, 'in_cluster', True),
+ pytest.param(True, 'in_cluster', True),
+ pytest.param(False, 'in_cluster', False),
+ pytest.param(None, 'in_cluster_empty', False),
+ pytest.param(True, 'in_cluster_empty', True),
+ pytest.param(False, 'in_cluster_empty', False),
+ ),
+ )
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
- def test_kube_config_path(self, mock_kube_config_loader,
mock_kube_config_merger):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_kube_config_path')
+ @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
+ def test_in_cluster_connection(
+ self,
+ mock_in_cluster_loader,
+ mock_merger,
+ mock_loader,
+ in_cluster_param,
+ conn_id,
+ in_cluster_called,
+ ):
+ """
+ Verifies whether in_cluster is called depending on combination of hook
param and connection extra.
+ Hook param should beat extra.
+ """
+ kubernetes_hook = KubernetesHook(conn_id=conn_id,
in_cluster=in_cluster_param)
api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with("path/to/file")
- mock_kube_config_merger.assert_called_once()
+ if in_cluster_called:
+ mock_in_cluster_loader.assert_called_once()
+ mock_merger.assert_not_called()
+ mock_loader.assert_not_called()
+ else:
+ mock_in_cluster_loader.assert_not_called()
+ mock_merger.assert_called_once_with(KUBE_CONFIG_PATH)
+ mock_loader.assert_called_once()
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.parametrize(
+ 'config_path_param, conn_id, call_path',
+ (
+ pytest.param(None, None, KUBE_CONFIG_PATH),
+ pytest.param('/my/path/override', None, '/my/path/override'),
+ pytest.param(None, 'kube_config_path', 'path/to/file'),
+ pytest.param('/my/path/override', 'kube_config_path',
'/my/path/override'),
+ pytest.param(None, 'kube_config_path_empty', KUBE_CONFIG_PATH),
+ pytest.param('/my/path/override', 'kube_config_path_empty',
'/my/path/override'),
+ ),
+ )
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
- def test_kube_config_path_empty(self, mock_kube_config_loader,
mock_kube_config_merger):
- kubernetes_hook =
KubernetesHook(conn_id='kubernetes_kube_config_path_empty')
+ def test_kube_config_path(
+ self, mock_kube_config_merger, mock_kube_config_loader,
config_path_param, conn_id, call_path
+ ):
+ """
+ Verifies kube config path depending on combination of hook param and
connection extra.
+ Hook param should beat extra.
+ """
+ kubernetes_hook = KubernetesHook(conn_id=conn_id,
config_file=config_path_param)
api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
+ mock_kube_config_merger.assert_called_once_with(call_path)
+ mock_kube_config_loader.assert_called_once()
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.parametrize(
+ 'conn_id, has_config',
+ (
+ pytest.param(None, False),
+ pytest.param('kube_config', True),
+ pytest.param('kube_config_empty', False),
Review comment:
```suggestion
(None, False),
('kube_config', True),
('kube_config_empty', False),
```
nit
##########
File path: tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
##########
@@ -37,107 +35,181 @@
KUBE_CONFIG_PATH = os.getenv('KUBECONFIG', '~/.kube/config')
-class TestKubernetesHook(unittest.TestCase):
+class TestKubernetesHook:
@classmethod
- def setUpClass(cls) -> None:
+ def setup_class(cls) -> None:
for conn_id, extra in [
- ('kubernetes_in_cluster', {'extra__kubernetes__in_cluster': True}),
- ('kubernetes_kube_config', {'extra__kubernetes__kube_config':
'{"test": "kube"}'}),
- ('kubernetes_kube_config_path',
{'extra__kubernetes__kube_config_path': 'path/to/file'}),
- ('kubernetes_in_cluster_empty', {'extra__kubernetes__in_cluster':
''}),
- ('kubernetes_kube_config_empty',
{'extra__kubernetes__kube_config': ''}),
- ('kubernetes_kube_config_path_empty',
{'extra__kubernetes__kube_config_path': ''}),
- ('kubernetes_with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
- ('kubernetes_default_kube_config', {}),
+ ('in_cluster', {'extra__kubernetes__in_cluster': True}),
+ ('kube_config', {'extra__kubernetes__kube_config': '{"test":
"kube"}'}),
+ ('kube_config_path', {'extra__kubernetes__kube_config_path':
'path/to/file'}),
+ ('in_cluster_empty', {'extra__kubernetes__in_cluster': ''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('context_empty', {'extra__kubernetes__cluster_context': ''}),
+ ('context', {'extra__kubernetes__cluster_context': 'my-context'}),
+ ('with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
+ ('default_kube_config', {}),
]:
db.merge_conn(Connection(conn_type='kubernetes', conn_id=conn_id,
extra=json.dumps(extra)))
@classmethod
- def tearDownClass(cls) -> None:
+ def teardown_class(cls) -> None:
clear_db_connections()
- @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
- def test_in_cluster_connection(self, mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
- @patch("kubernetes.config.kube_config.KubeConfigMerger")
- @patch("kubernetes.config.kube_config.KubeConfigLoader")
- def test_in_cluster_connection_empty(self, mock_kube_config_merger,
mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster_empty')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
+ @pytest.mark.parametrize(
+ 'in_cluster_param, conn_id, in_cluster_called',
+ (
+ pytest.param(True, None, True),
+ pytest.param(None, None, False),
+ pytest.param(None, 'in_cluster', True),
+ pytest.param(True, 'in_cluster', True),
+ pytest.param(False, 'in_cluster', False),
+ pytest.param(None, 'in_cluster_empty', False),
+ pytest.param(True, 'in_cluster_empty', True),
+ pytest.param(False, 'in_cluster_empty', False),
+ ),
+ )
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
- def test_kube_config_path(self, mock_kube_config_loader,
mock_kube_config_merger):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_kube_config_path')
+ @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
+ def test_in_cluster_connection(
+ self,
+ mock_in_cluster_loader,
+ mock_merger,
+ mock_loader,
+ in_cluster_param,
+ conn_id,
+ in_cluster_called,
+ ):
+ """
+ Verifies whether in_cluster is called depending on combination of hook
param and connection extra.
+ Hook param should beat extra.
+ """
+ kubernetes_hook = KubernetesHook(conn_id=conn_id,
in_cluster=in_cluster_param)
api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with("path/to/file")
- mock_kube_config_merger.assert_called_once()
+ if in_cluster_called:
+ mock_in_cluster_loader.assert_called_once()
+ mock_merger.assert_not_called()
+ mock_loader.assert_not_called()
+ else:
+ mock_in_cluster_loader.assert_not_called()
+ mock_merger.assert_called_once_with(KUBE_CONFIG_PATH)
+ mock_loader.assert_called_once()
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.parametrize(
+ 'config_path_param, conn_id, call_path',
+ (
+ pytest.param(None, None, KUBE_CONFIG_PATH),
+ pytest.param('/my/path/override', None, '/my/path/override'),
+ pytest.param(None, 'kube_config_path', 'path/to/file'),
+ pytest.param('/my/path/override', 'kube_config_path',
'/my/path/override'),
+ pytest.param(None, 'kube_config_path_empty', KUBE_CONFIG_PATH),
+ pytest.param('/my/path/override', 'kube_config_path_empty',
'/my/path/override'),
Review comment:
```suggestion
(None, None, KUBE_CONFIG_PATH),
('/my/path/override', None, '/my/path/override'),
(None, 'kube_config_path', 'path/to/file'),
('/my/path/override', 'kube_config_path', '/my/path/override'),
(None, 'kube_config_path_empty', KUBE_CONFIG_PATH),
('/my/path/override', 'kube_config_path_empty',
'/my/path/override'),
```
nit
##########
File path: tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
##########
@@ -37,107 +35,181 @@
KUBE_CONFIG_PATH = os.getenv('KUBECONFIG', '~/.kube/config')
-class TestKubernetesHook(unittest.TestCase):
+class TestKubernetesHook:
@classmethod
- def setUpClass(cls) -> None:
+ def setup_class(cls) -> None:
for conn_id, extra in [
- ('kubernetes_in_cluster', {'extra__kubernetes__in_cluster': True}),
- ('kubernetes_kube_config', {'extra__kubernetes__kube_config':
'{"test": "kube"}'}),
- ('kubernetes_kube_config_path',
{'extra__kubernetes__kube_config_path': 'path/to/file'}),
- ('kubernetes_in_cluster_empty', {'extra__kubernetes__in_cluster':
''}),
- ('kubernetes_kube_config_empty',
{'extra__kubernetes__kube_config': ''}),
- ('kubernetes_kube_config_path_empty',
{'extra__kubernetes__kube_config_path': ''}),
- ('kubernetes_with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
- ('kubernetes_default_kube_config', {}),
+ ('in_cluster', {'extra__kubernetes__in_cluster': True}),
+ ('kube_config', {'extra__kubernetes__kube_config': '{"test":
"kube"}'}),
+ ('kube_config_path', {'extra__kubernetes__kube_config_path':
'path/to/file'}),
+ ('in_cluster_empty', {'extra__kubernetes__in_cluster': ''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('context_empty', {'extra__kubernetes__cluster_context': ''}),
+ ('context', {'extra__kubernetes__cluster_context': 'my-context'}),
+ ('with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
+ ('default_kube_config', {}),
]:
db.merge_conn(Connection(conn_type='kubernetes', conn_id=conn_id,
extra=json.dumps(extra)))
@classmethod
- def tearDownClass(cls) -> None:
+ def teardown_class(cls) -> None:
clear_db_connections()
- @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
- def test_in_cluster_connection(self, mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
- @patch("kubernetes.config.kube_config.KubeConfigMerger")
- @patch("kubernetes.config.kube_config.KubeConfigLoader")
- def test_in_cluster_connection_empty(self, mock_kube_config_merger,
mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster_empty')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
+ @pytest.mark.parametrize(
+ 'in_cluster_param, conn_id, in_cluster_called',
+ (
+ pytest.param(True, None, True),
Review comment:
```suggestion
pytest.param(True, None, True),
pytest.param(False, None, False),
```
Was this case omitted intentionally?
##########
File path: tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
##########
@@ -37,107 +35,181 @@
KUBE_CONFIG_PATH = os.getenv('KUBECONFIG', '~/.kube/config')
-class TestKubernetesHook(unittest.TestCase):
+class TestKubernetesHook:
@classmethod
- def setUpClass(cls) -> None:
+ def setup_class(cls) -> None:
for conn_id, extra in [
- ('kubernetes_in_cluster', {'extra__kubernetes__in_cluster': True}),
- ('kubernetes_kube_config', {'extra__kubernetes__kube_config':
'{"test": "kube"}'}),
- ('kubernetes_kube_config_path',
{'extra__kubernetes__kube_config_path': 'path/to/file'}),
- ('kubernetes_in_cluster_empty', {'extra__kubernetes__in_cluster':
''}),
- ('kubernetes_kube_config_empty',
{'extra__kubernetes__kube_config': ''}),
- ('kubernetes_kube_config_path_empty',
{'extra__kubernetes__kube_config_path': ''}),
- ('kubernetes_with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
- ('kubernetes_default_kube_config', {}),
+ ('in_cluster', {'extra__kubernetes__in_cluster': True}),
+ ('kube_config', {'extra__kubernetes__kube_config': '{"test":
"kube"}'}),
+ ('kube_config_path', {'extra__kubernetes__kube_config_path':
'path/to/file'}),
+ ('in_cluster_empty', {'extra__kubernetes__in_cluster': ''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('context_empty', {'extra__kubernetes__cluster_context': ''}),
+ ('context', {'extra__kubernetes__cluster_context': 'my-context'}),
+ ('with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
+ ('default_kube_config', {}),
]:
db.merge_conn(Connection(conn_type='kubernetes', conn_id=conn_id,
extra=json.dumps(extra)))
@classmethod
- def tearDownClass(cls) -> None:
+ def teardown_class(cls) -> None:
clear_db_connections()
- @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
- def test_in_cluster_connection(self, mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
- @patch("kubernetes.config.kube_config.KubeConfigMerger")
- @patch("kubernetes.config.kube_config.KubeConfigLoader")
- def test_in_cluster_connection_empty(self, mock_kube_config_merger,
mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster_empty')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
+ @pytest.mark.parametrize(
+ 'in_cluster_param, conn_id, in_cluster_called',
+ (
+ pytest.param(True, None, True),
+ pytest.param(None, None, False),
+ pytest.param(None, 'in_cluster', True),
+ pytest.param(True, 'in_cluster', True),
+ pytest.param(False, 'in_cluster', False),
+ pytest.param(None, 'in_cluster_empty', False),
+ pytest.param(True, 'in_cluster_empty', True),
+ pytest.param(False, 'in_cluster_empty', False),
+ ),
+ )
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
- def test_kube_config_path(self, mock_kube_config_loader,
mock_kube_config_merger):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_kube_config_path')
+ @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
+ def test_in_cluster_connection(
+ self,
+ mock_in_cluster_loader,
+ mock_merger,
+ mock_loader,
+ in_cluster_param,
+ conn_id,
+ in_cluster_called,
+ ):
+ """
+ Verifies whether in_cluster is called depending on combination of hook
param and connection extra.
+ Hook param should beat extra.
+ """
+ kubernetes_hook = KubernetesHook(conn_id=conn_id,
in_cluster=in_cluster_param)
api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with("path/to/file")
- mock_kube_config_merger.assert_called_once()
+ if in_cluster_called:
+ mock_in_cluster_loader.assert_called_once()
+ mock_merger.assert_not_called()
+ mock_loader.assert_not_called()
+ else:
+ mock_in_cluster_loader.assert_not_called()
+ mock_merger.assert_called_once_with(KUBE_CONFIG_PATH)
+ mock_loader.assert_called_once()
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.parametrize(
+ 'config_path_param, conn_id, call_path',
+ (
+ pytest.param(None, None, KUBE_CONFIG_PATH),
+ pytest.param('/my/path/override', None, '/my/path/override'),
+ pytest.param(None, 'kube_config_path', 'path/to/file'),
+ pytest.param('/my/path/override', 'kube_config_path',
'/my/path/override'),
+ pytest.param(None, 'kube_config_path_empty', KUBE_CONFIG_PATH),
+ pytest.param('/my/path/override', 'kube_config_path_empty',
'/my/path/override'),
+ ),
+ )
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
- def test_kube_config_path_empty(self, mock_kube_config_loader,
mock_kube_config_merger):
- kubernetes_hook =
KubernetesHook(conn_id='kubernetes_kube_config_path_empty')
+ def test_kube_config_path(
+ self, mock_kube_config_merger, mock_kube_config_loader,
config_path_param, conn_id, call_path
+ ):
+ """
+ Verifies kube config path depending on combination of hook param and
connection extra.
+ Hook param should beat extra.
+ """
+ kubernetes_hook = KubernetesHook(conn_id=conn_id,
config_file=config_path_param)
api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
+ mock_kube_config_merger.assert_called_once_with(call_path)
+ mock_kube_config_loader.assert_called_once()
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.parametrize(
+ 'conn_id, has_config',
+ (
+ pytest.param(None, False),
+ pytest.param('kube_config', True),
+ pytest.param('kube_config_empty', False),
+ ),
+ )
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
@patch.object(tempfile, 'NamedTemporaryFile')
- def test_kube_config_connection(self, mock_kube_config_loader,
mock_kube_config_merger, mock_tempfile):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_kube_config')
+ def test_kube_config_connection(
+ self, mock_tempfile, mock_kube_config_merger, mock_kube_config_loader,
conn_id, has_config
+ ):
+ """
+ Verifies whether temporary kube config file is created.
+ """
+ mock_tempfile.return_value.__enter__.return_value.name =
"fake-temp-file"
+ mock_kube_config_merger.return_value.config = {"fake_config": "value"}
+ kubernetes_hook = KubernetesHook(conn_id=conn_id)
api_conn = kubernetes_hook.get_conn()
- mock_tempfile.is_called_once()
- mock_kube_config_loader.assert_called_once()
- mock_kube_config_merger.assert_called_once()
+ if has_config:
+ mock_tempfile.is_called_once()
+ mock_kube_config_loader.assert_called_once()
+ mock_kube_config_merger.assert_called_once_with('fake-temp-file')
+ else:
+ mock_tempfile.assert_not_called()
+ mock_kube_config_loader.assert_called_once()
+ mock_kube_config_merger.assert_called_once_with(KUBE_CONFIG_PATH)
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
- @patch("kubernetes.config.kube_config.KubeConfigLoader")
- @patch("kubernetes.config.kube_config.KubeConfigMerger")
- def test_kube_config_connection_empty(self, mock_kube_config_loader,
mock_kube_config_merger):
- kubernetes_hook =
KubernetesHook(conn_id='kubernetes_kube_config_empty')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.parametrize(
+ 'context_param, conn_id, expected_context',
+ (
+ pytest.param('param-context', None, 'param-context'),
+ pytest.param(None, None, None),
+ pytest.param('param-context', 'context', 'param-context'),
+ pytest.param(None, 'context', 'my-context'),
+ pytest.param('param-context', 'context_empty', 'param-context'),
+ pytest.param(None, 'context_empty', None),
Review comment:
```suggestion
('param-context', None, 'param-context'),
(None, None, None),
('param-context', 'context', 'param-context'),
(None, 'context', 'my-context'),
('param-context', 'context_empty', 'param-context'),
(None, 'context_empty', None),
```
nit
##########
File path: tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
##########
@@ -37,107 +35,181 @@
KUBE_CONFIG_PATH = os.getenv('KUBECONFIG', '~/.kube/config')
-class TestKubernetesHook(unittest.TestCase):
+class TestKubernetesHook:
@classmethod
- def setUpClass(cls) -> None:
+ def setup_class(cls) -> None:
for conn_id, extra in [
- ('kubernetes_in_cluster', {'extra__kubernetes__in_cluster': True}),
- ('kubernetes_kube_config', {'extra__kubernetes__kube_config':
'{"test": "kube"}'}),
- ('kubernetes_kube_config_path',
{'extra__kubernetes__kube_config_path': 'path/to/file'}),
- ('kubernetes_in_cluster_empty', {'extra__kubernetes__in_cluster':
''}),
- ('kubernetes_kube_config_empty',
{'extra__kubernetes__kube_config': ''}),
- ('kubernetes_kube_config_path_empty',
{'extra__kubernetes__kube_config_path': ''}),
- ('kubernetes_with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
- ('kubernetes_default_kube_config', {}),
+ ('in_cluster', {'extra__kubernetes__in_cluster': True}),
+ ('kube_config', {'extra__kubernetes__kube_config': '{"test":
"kube"}'}),
+ ('kube_config_path', {'extra__kubernetes__kube_config_path':
'path/to/file'}),
+ ('in_cluster_empty', {'extra__kubernetes__in_cluster': ''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('kube_config_empty', {'extra__kubernetes__kube_config': ''}),
+ ('kube_config_path_empty', {'extra__kubernetes__kube_config_path':
''}),
+ ('context_empty', {'extra__kubernetes__cluster_context': ''}),
+ ('context', {'extra__kubernetes__cluster_context': 'my-context'}),
+ ('with_namespace', {'extra__kubernetes__namespace':
'mock_namespace'}),
+ ('default_kube_config', {}),
]:
db.merge_conn(Connection(conn_type='kubernetes', conn_id=conn_id,
extra=json.dumps(extra)))
@classmethod
- def tearDownClass(cls) -> None:
+ def teardown_class(cls) -> None:
clear_db_connections()
- @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
- def test_in_cluster_connection(self, mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
- @patch("kubernetes.config.kube_config.KubeConfigMerger")
- @patch("kubernetes.config.kube_config.KubeConfigLoader")
- def test_in_cluster_connection_empty(self, mock_kube_config_merger,
mock_kube_config_loader):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_in_cluster_empty')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
-
+ @pytest.mark.parametrize(
+ 'in_cluster_param, conn_id, in_cluster_called',
+ (
+ pytest.param(True, None, True),
+ pytest.param(None, None, False),
+ pytest.param(None, 'in_cluster', True),
+ pytest.param(True, 'in_cluster', True),
+ pytest.param(False, 'in_cluster', False),
+ pytest.param(None, 'in_cluster_empty', False),
+ pytest.param(True, 'in_cluster_empty', True),
+ pytest.param(False, 'in_cluster_empty', False),
+ ),
+ )
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
- def test_kube_config_path(self, mock_kube_config_loader,
mock_kube_config_merger):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_kube_config_path')
+ @patch("kubernetes.config.incluster_config.InClusterConfigLoader")
+ def test_in_cluster_connection(
+ self,
+ mock_in_cluster_loader,
+ mock_merger,
+ mock_loader,
+ in_cluster_param,
+ conn_id,
+ in_cluster_called,
+ ):
+ """
+ Verifies whether in_cluster is called depending on combination of hook
param and connection extra.
+ Hook param should beat extra.
+ """
+ kubernetes_hook = KubernetesHook(conn_id=conn_id,
in_cluster=in_cluster_param)
api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with("path/to/file")
- mock_kube_config_merger.assert_called_once()
+ if in_cluster_called:
+ mock_in_cluster_loader.assert_called_once()
+ mock_merger.assert_not_called()
+ mock_loader.assert_not_called()
+ else:
+ mock_in_cluster_loader.assert_not_called()
+ mock_merger.assert_called_once_with(KUBE_CONFIG_PATH)
+ mock_loader.assert_called_once()
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.parametrize(
+ 'config_path_param, conn_id, call_path',
+ (
+ pytest.param(None, None, KUBE_CONFIG_PATH),
+ pytest.param('/my/path/override', None, '/my/path/override'),
+ pytest.param(None, 'kube_config_path', 'path/to/file'),
+ pytest.param('/my/path/override', 'kube_config_path',
'/my/path/override'),
+ pytest.param(None, 'kube_config_path_empty', KUBE_CONFIG_PATH),
+ pytest.param('/my/path/override', 'kube_config_path_empty',
'/my/path/override'),
+ ),
+ )
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
- def test_kube_config_path_empty(self, mock_kube_config_loader,
mock_kube_config_merger):
- kubernetes_hook =
KubernetesHook(conn_id='kubernetes_kube_config_path_empty')
+ def test_kube_config_path(
+ self, mock_kube_config_merger, mock_kube_config_loader,
config_path_param, conn_id, call_path
+ ):
+ """
+ Verifies kube config path depending on combination of hook param and
connection extra.
+ Hook param should beat extra.
+ """
+ kubernetes_hook = KubernetesHook(conn_id=conn_id,
config_file=config_path_param)
api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
+ mock_kube_config_merger.assert_called_once_with(call_path)
+ mock_kube_config_loader.assert_called_once()
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.parametrize(
+ 'conn_id, has_config',
+ (
+ pytest.param(None, False),
+ pytest.param('kube_config', True),
+ pytest.param('kube_config_empty', False),
+ ),
+ )
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
@patch.object(tempfile, 'NamedTemporaryFile')
- def test_kube_config_connection(self, mock_kube_config_loader,
mock_kube_config_merger, mock_tempfile):
- kubernetes_hook = KubernetesHook(conn_id='kubernetes_kube_config')
+ def test_kube_config_connection(
+ self, mock_tempfile, mock_kube_config_merger, mock_kube_config_loader,
conn_id, has_config
+ ):
+ """
+ Verifies whether temporary kube config file is created.
+ """
+ mock_tempfile.return_value.__enter__.return_value.name =
"fake-temp-file"
+ mock_kube_config_merger.return_value.config = {"fake_config": "value"}
+ kubernetes_hook = KubernetesHook(conn_id=conn_id)
api_conn = kubernetes_hook.get_conn()
- mock_tempfile.is_called_once()
- mock_kube_config_loader.assert_called_once()
- mock_kube_config_merger.assert_called_once()
+ if has_config:
+ mock_tempfile.is_called_once()
+ mock_kube_config_loader.assert_called_once()
+ mock_kube_config_merger.assert_called_once_with('fake-temp-file')
+ else:
+ mock_tempfile.assert_not_called()
+ mock_kube_config_loader.assert_called_once()
+ mock_kube_config_merger.assert_called_once_with(KUBE_CONFIG_PATH)
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
- @patch("kubernetes.config.kube_config.KubeConfigLoader")
- @patch("kubernetes.config.kube_config.KubeConfigMerger")
- def test_kube_config_connection_empty(self, mock_kube_config_loader,
mock_kube_config_merger):
- kubernetes_hook =
KubernetesHook(conn_id='kubernetes_kube_config_empty')
- api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with(KUBE_CONFIG_PATH)
- mock_kube_config_merger.assert_called_once()
- assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.parametrize(
+ 'context_param, conn_id, expected_context',
+ (
+ pytest.param('param-context', None, 'param-context'),
+ pytest.param(None, None, None),
+ pytest.param('param-context', 'context', 'param-context'),
+ pytest.param(None, 'context', 'my-context'),
+ pytest.param('param-context', 'context_empty', 'param-context'),
+ pytest.param(None, 'context_empty', None),
+ ),
+ )
+ @patch("kubernetes.config.load_kube_config")
+ def test_cluster_context(self, mock_load_kube_config, context_param,
conn_id, expected_context):
+ """
+ Verifies cluster context depending on combination of hook param and
connection extra.
+ Hook param should beat extra.
+ """
+ kubernetes_hook = KubernetesHook(conn_id=conn_id,
cluster_context=context_param)
+ kubernetes_hook.get_conn()
+ mock_load_kube_config.assert_called_with(client_configuration=None,
context=expected_context)
@patch("kubernetes.config.kube_config.KubeConfigLoader")
@patch("kubernetes.config.kube_config.KubeConfigMerger")
@patch("kubernetes.config.kube_config.KUBE_CONFIG_DEFAULT_LOCATION",
"/mock/config")
- def test_default_kube_config_connection(
- self,
- mock_kube_config_loader,
- mock_kube_config_merger,
- ):
- kubernetes_hook =
KubernetesHook(conn_id='kubernetes_default_kube_config')
+ def test_default_kube_config_connection(self, mock_kube_config_merger,
mock_kube_config_loader):
+ kubernetes_hook = KubernetesHook(conn_id='default_kube_config')
api_conn = kubernetes_hook.get_conn()
- mock_kube_config_loader.assert_called_once_with("/mock/config")
- mock_kube_config_merger.assert_called_once()
+ mock_kube_config_merger.assert_called_once_with("/mock/config")
+ mock_kube_config_loader.assert_called_once()
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
def test_get_namespace(self):
- kubernetes_hook_with_namespace =
KubernetesHook(conn_id='kubernetes_with_namespace')
- kubernetes_hook_without_namespace =
KubernetesHook(conn_id='kubernetes_default_kube_config')
+ kubernetes_hook_with_namespace =
KubernetesHook(conn_id='with_namespace')
+ kubernetes_hook_without_namespace =
KubernetesHook(conn_id='default_kube_config')
assert kubernetes_hook_with_namespace.get_namespace() ==
'mock_namespace'
assert kubernetes_hook_without_namespace.get_namespace() == 'default'
Review comment:
Missing a case for a hook without a connection.
--
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]
