mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
Pepe Fagoaga
676 lines
26 KiB
Python
676 lines
26 KiB
Python
from argparse import Namespace
|
|
from unittest.mock import patch
|
|
|
|
from kubernetes.config.config_exception import ConfigException
|
|
|
|
from kubernetes import client
|
|
from prowler.config.config import (
|
|
default_config_file_path,
|
|
default_fixer_config_file_path,
|
|
load_and_validate_config_file,
|
|
)
|
|
from prowler.providers.kubernetes.exceptions.exceptions import (
|
|
KubernetesSetUpSessionError,
|
|
)
|
|
from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
|
|
from prowler.providers.kubernetes.models import (
|
|
KubernetesIdentityInfo,
|
|
KubernetesSession,
|
|
)
|
|
from tests.providers.kubernetes.kubernetes_fixtures import KUBERNETES_CONFIG
|
|
|
|
|
|
def mock_set_kubernetes_credentials(*_):
|
|
return ("apiclient", "context")
|
|
|
|
|
|
def mock_get_context_user_roles(*_):
|
|
return []
|
|
|
|
|
|
class TestKubernetesProvider:
|
|
def test_kubernetes_provider_no_namespaces(
|
|
self,
|
|
):
|
|
context = {
|
|
"name": "test-context",
|
|
"context": {
|
|
"user": "test-user",
|
|
"cluster": "test-cluster",
|
|
},
|
|
}
|
|
with (
|
|
patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.setup_session",
|
|
return_value=KubernetesSession(
|
|
api_client=client.ApiClient,
|
|
context=context,
|
|
),
|
|
),
|
|
patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.get_all_namespaces",
|
|
return_value=["namespace-1"],
|
|
),
|
|
):
|
|
arguments = Namespace()
|
|
arguments.kubeconfig_file = "dummy_path"
|
|
arguments.context = None
|
|
arguments.only_logs = False
|
|
arguments.namespace = None
|
|
fixer_config = load_and_validate_config_file(
|
|
"kubernetes", default_fixer_config_file_path
|
|
)
|
|
|
|
# Instantiate the KubernetesProvider with mocked arguments
|
|
kubernetes_provider = KubernetesProvider(
|
|
arguments.kubeconfig_file,
|
|
arguments.context,
|
|
arguments.namespace,
|
|
config_path=default_config_file_path,
|
|
fixer_config=fixer_config,
|
|
)
|
|
assert isinstance(kubernetes_provider.session, KubernetesSession)
|
|
assert kubernetes_provider.session.api_client is not None
|
|
assert kubernetes_provider.session.context == {
|
|
"name": "test-context",
|
|
"context": {"cluster": "test-cluster", "user": "test-user"},
|
|
}
|
|
assert kubernetes_provider.namespaces == ["namespace-1"]
|
|
assert isinstance(kubernetes_provider.identity, KubernetesIdentityInfo)
|
|
assert kubernetes_provider.identity.context == "test-context"
|
|
assert kubernetes_provider.identity.cluster == "test-cluster"
|
|
assert kubernetes_provider.identity.user == "test-user"
|
|
|
|
assert kubernetes_provider.audit_config == KUBERNETES_CONFIG
|
|
|
|
@patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
|
)
|
|
@patch("kubernetes.config.list_kube_config_contexts")
|
|
@patch("kubernetes.config.load_kube_config_from_dict")
|
|
def test_kubernetes_test_connection_with_kubeconfig_content(
|
|
self,
|
|
mock_load_kube_config_from_dict,
|
|
mock_list_kube_config_contexts,
|
|
mock_list_namespace,
|
|
):
|
|
mock_load_kube_config_from_dict.return_value = None
|
|
mock_list_kube_config_contexts.return_value = (
|
|
[
|
|
{
|
|
"name": "example-context",
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
}
|
|
],
|
|
None,
|
|
)
|
|
mock_list_namespace.return_value.items = [
|
|
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
|
]
|
|
|
|
kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'
|
|
|
|
connection = KubernetesProvider.test_connection(
|
|
kubeconfig_file=None,
|
|
kubeconfig_content=kubeconfig_content,
|
|
provider_id="example-context",
|
|
raise_on_exception=False,
|
|
)
|
|
|
|
assert connection.is_connected
|
|
assert connection.error is None
|
|
|
|
@patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
|
)
|
|
@patch("kubernetes.config.list_kube_config_contexts")
|
|
@patch("kubernetes.config.load_kube_config")
|
|
def test_kubernetes_test_connection_with_kubeconfig_file(
|
|
self, mock_load_kube_config, mock_list_kube_config_contexts, mock_list_namespace
|
|
):
|
|
mock_load_kube_config.return_value = None
|
|
mock_list_kube_config_contexts.return_value = (
|
|
[
|
|
{
|
|
"name": "test-context",
|
|
"context": {
|
|
"cluster": "test-cluster",
|
|
"user": "test-user",
|
|
},
|
|
}
|
|
],
|
|
None,
|
|
)
|
|
mock_list_namespace.return_value.items = [
|
|
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
|
]
|
|
|
|
connection = KubernetesProvider.test_connection(
|
|
kubeconfig_file="dummy_kubeconfig_path",
|
|
kubeconfig_content="",
|
|
provider_id="test-context",
|
|
raise_on_exception=False,
|
|
)
|
|
|
|
assert connection.is_connected
|
|
assert connection.error is None
|
|
|
|
@patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespaced_pod"
|
|
)
|
|
@patch("kubernetes.config.list_kube_config_contexts")
|
|
@patch("kubernetes.config.load_kube_config")
|
|
def test_kubernetes_test_connection_with_namespace_input(
|
|
self,
|
|
mock_load_kube_config,
|
|
mock_list_kube_config_contexts,
|
|
mock_list_namespaced_pod,
|
|
):
|
|
mock_load_kube_config.return_value = None
|
|
mock_list_kube_config_contexts.return_value = (
|
|
[
|
|
{
|
|
"name": "test-context",
|
|
"context": {
|
|
"cluster": "test-cluster",
|
|
"user": "test-user",
|
|
},
|
|
}
|
|
],
|
|
None,
|
|
)
|
|
mock_list_namespaced_pod.return_value.items = [
|
|
client.V1Pod(metadata=client.V1ObjectMeta(name="pod-1")),
|
|
]
|
|
|
|
connection = KubernetesProvider.test_connection(
|
|
kubeconfig_file="",
|
|
kubeconfig_content="",
|
|
namespace="test-namespace",
|
|
provider_id="test-context",
|
|
raise_on_exception=False,
|
|
)
|
|
|
|
assert connection.is_connected
|
|
assert connection.error is None
|
|
|
|
@patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
|
)
|
|
@patch("kubernetes.config.list_kube_config_contexts")
|
|
@patch("kubernetes.config.load_kube_config_from_dict")
|
|
def test_kubernetes_test_connection_with_kubeconfig_content_invalid_provider_id(
|
|
self,
|
|
mock_load_kube_config_from_dict,
|
|
mock_list_kube_config_contexts,
|
|
mock_list_namespace,
|
|
):
|
|
mock_load_kube_config_from_dict.return_value = None
|
|
mock_list_kube_config_contexts.return_value = (
|
|
[
|
|
{
|
|
"name": "example-context",
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
}
|
|
],
|
|
None,
|
|
)
|
|
mock_list_namespace.return_value.items = [
|
|
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
|
]
|
|
|
|
kubeconfig_content = {
|
|
"apiVersion": "v1",
|
|
"clusters": [
|
|
{
|
|
"cluster": {
|
|
"server": "https://kubernetes.example.com",
|
|
},
|
|
"name": "example-cluster",
|
|
}
|
|
],
|
|
"contexts": [
|
|
{
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
"name": "example-context",
|
|
}
|
|
],
|
|
"current-context": "example-context",
|
|
"kind": "Config",
|
|
"preferences": {},
|
|
"users": [
|
|
{
|
|
"name": "example-user",
|
|
"user": {
|
|
"token": "EXAMPLE_TOKEN",
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
connection = KubernetesProvider.test_connection(
|
|
kubeconfig_file=None,
|
|
kubeconfig_content=kubeconfig_content,
|
|
provider_id="example-context-invalid",
|
|
raise_on_exception=False,
|
|
)
|
|
|
|
assert not connection.is_connected
|
|
assert connection.error is not None
|
|
assert isinstance(connection.error, KubernetesSetUpSessionError)
|
|
|
|
@patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
|
)
|
|
@patch("kubernetes.config.list_kube_config_contexts")
|
|
@patch("kubernetes.config.load_kube_config_from_dict")
|
|
def test_kubernetes_test_connection_with_kubeconfig_content_valid_provider_id(
|
|
self,
|
|
mock_load_kube_config_from_dict,
|
|
mock_list_kube_config_contexts,
|
|
mock_list_namespace,
|
|
):
|
|
mock_load_kube_config_from_dict.return_value = None
|
|
mock_list_kube_config_contexts.return_value = (
|
|
[
|
|
{
|
|
"name": "example-context",
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
}
|
|
],
|
|
None,
|
|
)
|
|
mock_list_namespace.return_value.items = [
|
|
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
|
]
|
|
|
|
kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'
|
|
|
|
connection = KubernetesProvider.test_connection(
|
|
kubeconfig_file=None,
|
|
kubeconfig_content=kubeconfig_content,
|
|
provider_id="example-context",
|
|
raise_on_exception=False,
|
|
)
|
|
|
|
assert connection.is_connected
|
|
assert connection.error is None
|
|
|
|
def test_kubernetes_provider_incluster_with_env_var(self, monkeypatch):
|
|
monkeypatch.setenv("CLUSTER_NAME", "env-cluster-name")
|
|
|
|
with (
|
|
patch(
|
|
"kubernetes.config.load_kube_config",
|
|
side_effect=ConfigException("No kubeconfig"),
|
|
),
|
|
patch("kubernetes.config.load_incluster_config", return_value=None),
|
|
patch("prowler.providers.kubernetes.kubernetes_provider.client.ApiClient"),
|
|
patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.get_all_namespaces",
|
|
return_value=["default"],
|
|
),
|
|
):
|
|
session = KubernetesProvider.setup_session(
|
|
kubeconfig_file=None,
|
|
kubeconfig_content=None,
|
|
context=None,
|
|
cluster_name=None,
|
|
)
|
|
assert isinstance(session, KubernetesSession)
|
|
assert session.context["context"]["cluster"] == "env-cluster-name"
|
|
|
|
def test_kubernetes_provider_incluster_with_cli_flag(self):
|
|
with (
|
|
patch(
|
|
"kubernetes.config.load_kube_config",
|
|
side_effect=ConfigException("No kubeconfig"),
|
|
),
|
|
patch("kubernetes.config.load_incluster_config", return_value=None),
|
|
patch("prowler.providers.kubernetes.kubernetes_provider.client.ApiClient"),
|
|
patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.get_all_namespaces",
|
|
return_value=["default"],
|
|
),
|
|
):
|
|
session = KubernetesProvider.setup_session(
|
|
kubeconfig_file=None,
|
|
kubeconfig_content=None,
|
|
context=None,
|
|
cluster_name="cli-cluster-name",
|
|
)
|
|
assert isinstance(session, KubernetesSession)
|
|
assert session.context["context"]["cluster"] == "cli-cluster-name"
|
|
|
|
@patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
|
)
|
|
@patch("kubernetes.config.list_kube_config_contexts")
|
|
@patch("kubernetes.config.load_kube_config_from_dict")
|
|
def test_kubernetes_provider_proxy_from_env(
|
|
self,
|
|
mock_load_kube_config_from_dict,
|
|
mock_list_kube_config_contexts,
|
|
mock_list_namespace,
|
|
monkeypatch,
|
|
):
|
|
|
|
monkeypatch.setenv("HTTPS_PROXY", "http://my.internal.proxy:8888")
|
|
|
|
mock_load_kube_config_from_dict.return_value = None
|
|
mock_list_kube_config_contexts.return_value = (
|
|
[
|
|
{
|
|
"name": "example-context",
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
}
|
|
],
|
|
None,
|
|
)
|
|
mock_list_namespace.return_value.items = [
|
|
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
|
]
|
|
|
|
kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'
|
|
|
|
session = KubernetesProvider.setup_session(
|
|
kubeconfig_content=kubeconfig_content,
|
|
context="example-context",
|
|
)
|
|
|
|
assert isinstance(session, KubernetesSession)
|
|
assert isinstance(session.api_client, client.ApiClient)
|
|
assert isinstance(session.api_client.configuration, client.Configuration)
|
|
assert session.api_client.configuration.verify_ssl
|
|
assert session.api_client.configuration.proxy == "http://my.internal.proxy:8888"
|
|
|
|
@patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
|
)
|
|
@patch("kubernetes.config.list_kube_config_contexts")
|
|
@patch("kubernetes.config.load_kube_config_from_dict")
|
|
def test_kubernetes_provider_disable_tls_verification(
|
|
self,
|
|
mock_load_kube_config_from_dict,
|
|
mock_list_kube_config_contexts,
|
|
mock_list_namespace,
|
|
monkeypatch,
|
|
):
|
|
monkeypatch.setenv("K8S_SKIP_TLS_VERIFY", "true")
|
|
|
|
mock_load_kube_config_from_dict.return_value = None
|
|
mock_list_kube_config_contexts.return_value = (
|
|
[
|
|
{
|
|
"name": "example-context",
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
}
|
|
],
|
|
None,
|
|
)
|
|
mock_list_namespace.return_value.items = [
|
|
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
|
]
|
|
|
|
kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'
|
|
|
|
session = KubernetesProvider.setup_session(
|
|
kubeconfig_content=kubeconfig_content,
|
|
context="example-context",
|
|
)
|
|
|
|
assert isinstance(session, KubernetesSession)
|
|
assert isinstance(session.api_client, client.ApiClient)
|
|
assert isinstance(session.api_client.configuration, client.Configuration)
|
|
assert session.api_client.configuration.verify_ssl is False
|
|
assert session.api_client.configuration.proxy is None
|
|
|
|
@patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
|
)
|
|
@patch("kubernetes.config.list_kube_config_contexts")
|
|
@patch("kubernetes.config.load_kube_config_from_dict")
|
|
def test_kubernetes_provider_kubeconfig_content(
|
|
self,
|
|
mock_load_kube_config_from_dict,
|
|
mock_list_kube_config_contexts,
|
|
mock_list_namespace,
|
|
):
|
|
mock_load_kube_config_from_dict.return_value = None
|
|
mock_list_kube_config_contexts.return_value = (
|
|
[
|
|
{
|
|
"name": "example-context",
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
}
|
|
],
|
|
None,
|
|
)
|
|
mock_list_namespace.return_value.items = [
|
|
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
|
]
|
|
|
|
kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'
|
|
|
|
session = KubernetesProvider.setup_session(
|
|
kubeconfig_content=kubeconfig_content,
|
|
context="example-context",
|
|
)
|
|
|
|
assert isinstance(session, KubernetesSession)
|
|
assert isinstance(session.api_client, client.ApiClient)
|
|
|
|
assert session.context == {
|
|
"name": "example-context",
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
}
|
|
|
|
@patch(
|
|
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
|
)
|
|
@patch("kubernetes.config.list_kube_config_contexts")
|
|
@patch("kubernetes.config.load_kube_config_from_dict")
|
|
def test_kubernetes_provider_kubeconfig_content_proxy_settings(
|
|
self,
|
|
mock_load_kube_config_from_dict,
|
|
mock_list_kube_config_contexts,
|
|
mock_list_namespace,
|
|
monkeypatch,
|
|
):
|
|
monkeypatch.setenv("HTTPS_PROXY", "http://my.internal.proxy:8888")
|
|
monkeypatch.setenv("K8S_SKIP_TLS_VERIFY", "true")
|
|
|
|
mock_load_kube_config_from_dict.return_value = None
|
|
mock_list_kube_config_contexts.return_value = (
|
|
[
|
|
{
|
|
"name": "example-context",
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
}
|
|
],
|
|
None,
|
|
)
|
|
mock_list_namespace.return_value.items = [
|
|
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
|
]
|
|
|
|
kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'
|
|
|
|
session = KubernetesProvider.setup_session(
|
|
kubeconfig_content=kubeconfig_content,
|
|
context="example-context",
|
|
)
|
|
|
|
assert isinstance(session, KubernetesSession)
|
|
assert isinstance(session.api_client, client.ApiClient)
|
|
|
|
assert session.context == {
|
|
"name": "example-context",
|
|
"context": {
|
|
"cluster": "example-cluster",
|
|
"user": "example-user",
|
|
},
|
|
}
|
|
|
|
assert session.api_client.configuration.proxy == "http://my.internal.proxy:8888"
|
|
assert session.api_client.configuration.verify_ssl is False
|
|
|
|
def test_set_proxy_settings_no_proxy_no_tls_skip(self):
|
|
"""Test set_proxy_settings with no environment variables set."""
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
config = KubernetesProvider.set_proxy_settings()
|
|
|
|
# Verify it's a Configuration instance from kubernetes.client
|
|
from kubernetes.client import Configuration
|
|
|
|
assert isinstance(config, Configuration)
|
|
|
|
assert hasattr(config, "proxy")
|
|
assert config.proxy is None
|
|
assert hasattr(config, "verify_ssl")
|
|
assert config.verify_ssl is True
|
|
|
|
def test_set_proxy_settings_with_https_proxy_uppercase(self):
|
|
"""Test set_proxy_settings with HTTPS_PROXY environment variable."""
|
|
proxy_url = "http://proxy.example.com:8080"
|
|
with patch.dict("os.environ", {"HTTPS_PROXY": proxy_url}, clear=True):
|
|
config = KubernetesProvider.set_proxy_settings()
|
|
|
|
# Verify it's a Configuration instance from kubernetes.client
|
|
from kubernetes.client import Configuration
|
|
|
|
assert isinstance(config, Configuration)
|
|
|
|
assert config.proxy == proxy_url
|
|
assert config.verify_ssl is True
|
|
|
|
def test_set_proxy_settings_with_https_proxy_lowercase(self):
|
|
"""Test set_proxy_settings with https_proxy environment variable."""
|
|
proxy_url = "http://proxy.example.com:3128"
|
|
with patch.dict("os.environ", {"https_proxy": proxy_url}, clear=True):
|
|
config = KubernetesProvider.set_proxy_settings()
|
|
|
|
# Verify it's a Configuration instance from kubernetes.client
|
|
from kubernetes.client import Configuration
|
|
|
|
assert isinstance(config, Configuration)
|
|
|
|
assert config.proxy == proxy_url
|
|
assert config.verify_ssl is True
|
|
|
|
def test_set_proxy_settings_uppercase_proxy_takes_precedence(self):
|
|
"""Test that HTTPS_PROXY takes precedence over https_proxy."""
|
|
uppercase_proxy = "http://uppercase.proxy.com:8080"
|
|
lowercase_proxy = "http://lowercase.proxy.com:3128"
|
|
with patch.dict(
|
|
"os.environ",
|
|
{"HTTPS_PROXY": uppercase_proxy, "https_proxy": lowercase_proxy},
|
|
clear=True,
|
|
):
|
|
config = KubernetesProvider.set_proxy_settings()
|
|
|
|
# Verify it's a Configuration instance from kubernetes.client
|
|
from kubernetes.client import Configuration
|
|
|
|
assert isinstance(config, Configuration)
|
|
|
|
assert config.proxy == uppercase_proxy
|
|
assert config.verify_ssl is True
|
|
|
|
def test_set_proxy_settings_with_tls_skip_true(self):
|
|
"""Test set_proxy_settings with K8S_SKIP_TLS_VERIFY set to true."""
|
|
with patch.dict("os.environ", {"K8S_SKIP_TLS_VERIFY": "true"}, clear=True):
|
|
config = KubernetesProvider.set_proxy_settings()
|
|
|
|
# Verify it's a Configuration instance from kubernetes.client
|
|
from kubernetes.client import Configuration
|
|
|
|
assert isinstance(config, Configuration)
|
|
|
|
assert config.proxy is None
|
|
assert config.verify_ssl is False
|
|
|
|
def test_set_proxy_settings_with_tls_skip_true_uppercase(self):
|
|
"""Test set_proxy_settings with K8S_SKIP_TLS_VERIFY set to TRUE."""
|
|
with patch.dict("os.environ", {"K8S_SKIP_TLS_VERIFY": "TRUE"}, clear=True):
|
|
config = KubernetesProvider.set_proxy_settings()
|
|
|
|
# Verify it's a Configuration instance from kubernetes.client
|
|
from kubernetes.client import Configuration
|
|
|
|
assert isinstance(config, Configuration)
|
|
|
|
assert config.proxy is None
|
|
assert config.verify_ssl is False
|
|
|
|
def test_set_proxy_settings_with_tls_skip_false(self):
|
|
"""Test set_proxy_settings with K8S_SKIP_TLS_VERIFY set to false."""
|
|
with patch.dict("os.environ", {"K8S_SKIP_TLS_VERIFY": "false"}, clear=True):
|
|
config = KubernetesProvider.set_proxy_settings()
|
|
|
|
# Verify it's a Configuration instance from kubernetes.client
|
|
from kubernetes.client import Configuration
|
|
|
|
assert isinstance(config, Configuration)
|
|
|
|
assert config.proxy is None
|
|
assert config.verify_ssl is True
|
|
|
|
def test_set_proxy_settings_with_tls_skip_invalid_value(self):
|
|
"""Test set_proxy_settings with K8S_SKIP_TLS_VERIFY set to invalid value."""
|
|
with patch.dict("os.environ", {"K8S_SKIP_TLS_VERIFY": "invalid"}, clear=True):
|
|
config = KubernetesProvider.set_proxy_settings()
|
|
|
|
# Verify it's a Configuration instance from kubernetes.client
|
|
from kubernetes.client import Configuration
|
|
|
|
assert isinstance(config, Configuration)
|
|
|
|
assert config.proxy is None
|
|
assert config.verify_ssl is True
|
|
|
|
def test_set_proxy_settings_with_both_proxy_and_tls_skip(self):
|
|
"""Test set_proxy_settings with both proxy and TLS skip settings."""
|
|
proxy_url = "http://secure.proxy.com:8080"
|
|
with patch.dict(
|
|
"os.environ",
|
|
{"HTTPS_PROXY": proxy_url, "K8S_SKIP_TLS_VERIFY": "true"},
|
|
clear=True,
|
|
):
|
|
config = KubernetesProvider.set_proxy_settings()
|
|
|
|
# Verify it's a Configuration instance from kubernetes.client
|
|
from kubernetes.client import Configuration
|
|
|
|
assert isinstance(config, Configuration)
|
|
|
|
assert config.proxy == proxy_url
|
|
assert config.verify_ssl is False
|