Adding the tests for Pod Security Admission.

JIRA: CGTS-74433

Change-Id: I2177683f45e46a9b4da45a7dfb1276a10d99afdd
Signed-off-by: vshivara <vanamala.shivaramakrishna@windriver.com>
This commit is contained in:
vshivara
2025-03-11 07:56:41 -04:00
committed by Vanamala Shivaramakrishna
parent 78631f96a6
commit 756f667163
12 changed files with 550 additions and 20 deletions

View File

@@ -29,3 +29,17 @@ class KubectlGetNamespacesKeywords(BaseKeyword):
namespaces_list_output = KubectlGetNamespacesOutput(kubectl_get_namespaces_output)
return namespaces_list_output
def get_namespaces_by_label(self, label) -> KubectlGetNamespacesOutput:
"""
Gets the k8s namespaces available for a given label.
Args:
Returns: KubectlGetNamespacesOutput
"""
kubectl_get_namespaces_output = self.ssh_connection.send(export_k8s_config(f"kubectl get ns -l={label}"))
self.validate_success_return_code(self.ssh_connection)
namespaces_list_output = KubectlGetNamespacesOutput(kubectl_get_namespaces_output)
return namespaces_list_output

View File

@@ -5,32 +5,34 @@ from keywords.k8s.namespace.object.kubectl_namespace_object import KubectlNamesp
class KubectlGetNamespacesOutput:
"""
Class for 'kubectl get ns output' keywords
"""
def __init__(self, kubectl_get_namespaces_output: str):
"""
Constructor
Args:
kubectl_get_namespaces_output: Raw string output from running a "kubectl get ns" command.
kubectl_get_namespaces_output (str): Raw string output from running a "kubectl get ns" command.
"""
self.kubectl_namespaces: [KubectlNamespaceObject] = []
kubectl_get_namespaces_table_parser = KubectlGetNamespacesTableParser(kubectl_get_namespaces_output)
output_values_list = kubectl_get_namespaces_table_parser.get_output_values_list()
for namespace_dict in output_values_list:
if 'NAME' not in namespace_dict:
if "NAME" not in namespace_dict:
raise ValueError(f"There is no NAME associated with the namespace: {namespace_dict}")
namespace = KubectlNamespaceObject(namespace_dict['NAME'])
namespace = KubectlNamespaceObject(namespace_dict["NAME"])
if 'STATUS' in namespace_dict:
namespace.set_status(namespace_dict['STATUS'])
if "STATUS" in namespace_dict:
namespace.set_status(namespace_dict["STATUS"])
if 'AGE' in namespace_dict:
namespace.set_age(namespace_dict['AGE'])
if "AGE" in namespace_dict:
namespace.set_age(namespace_dict["AGE"])
self.kubectl_namespaces.append(namespace)
@@ -43,13 +45,15 @@ class KubectlGetNamespacesOutput:
"""
return self.kubectl_namespaces
def is_namespace(self, namespace_name) -> bool:
def is_namespace(self, namespace_name: str) -> bool:
"""
This function will get the namespace with the name specified from this get_namespace_output.
Args:
namespace_name: The name of the namespace of interest.
Returns: bool
Args:
namespace_name (str): The name of the namespace of interest.
Returns:
bool: This function return a bool value.
"""
for ns in self.kubectl_namespaces:

View File

@@ -1,28 +1,49 @@
from keywords.base_keyword import BaseKeyword
from keywords.k8s.k8s_command_wrapper import export_k8s_config
from starlingx.framework.ssh.ssh_connection import SSHConnection
class KubectlApplyPodsKeywords(BaseKeyword):
"""
Class for Kubectl apply pod keywords
"""
def __init__(self, ssh_connection):
def __init__(self, ssh_connection: SSHConnection):
"""
Constructor
Args:
ssh_connection:
ssh_connection (SSHConnection): ssh connection
"""
self.ssh_connection = ssh_connection
def apply_from_yaml(self, yaml_file: str):
def apply_from_yaml(self, yaml_file: str) -> None:
"""
Applies a pod yaml config
Args:
yaml_file (): the yaml file
Returns:
Args:
yaml_file (str): the yaml file
Returns: None
"""
self.ssh_connection.send(export_k8s_config(f"kubectl apply -f {yaml_file}"))
self.validate_success_return_code(self.ssh_connection)
def fail_apply_from_yaml(self, yaml_file: str) -> None:
"""
Checks if applying a pod yaml config fails
Args:
yaml_file (str): the yaml file
Returns:
None: This function does not return a value.
"""
self.ssh_connection.send(export_k8s_config(f"kubectl apply -f {yaml_file}"))
rc = self.ssh_connection.get_return_code()
if 1 != rc:
raise Exception(f"Expected deployment of {yaml_file} to fail, instead it passed, investigate")

View File

@@ -16,12 +16,13 @@ class KubectlCreateSecretsKeywords(BaseKeyword):
"""
self.ssh_connection = ssh_connection
def create_secret_for_registry(self, registry: Registry, secret_name: str):
def create_secret_for_registry(self, registry: Registry, secret_name: str, namespace: str = "default"):
"""
Create a secret for the registry
Args:
registry (): the registry
secret_name (): the secret name
namespace (): the namespace
Returns:
@@ -30,7 +31,7 @@ class KubectlCreateSecretsKeywords(BaseKeyword):
password = registry.get_password()
docker_server = registry.get_registry_url()
self.ssh_connection.send(
export_k8s_config(f"kubectl create secret docker-registry {secret_name} --docker-server={docker_server} " f"--docker-username={user_name} --docker-password={password}")
export_k8s_config(f"kubectl create secret -n {namespace} docker-registry {secret_name} --docker-server={docker_server} " f"--docker-username={user_name} --docker-password={password}")
)
def create_secret_generic(self, secret_name: str, tls_crt: str, tls_key: str, namespace: str):

View File

@@ -0,0 +1,13 @@
apiVersion: v1
kind: Pod
metadata:
name: baseline-pod-allow
namespace: baseline-ns
spec:
containers:
- name: pause
image: registry.local:9001/pause
securityContext:
privileged: false
imagePullSecrets:
- name: local-secret

View File

@@ -0,0 +1,13 @@
apiVersion: v1
kind: Pod
metadata:
name: baseline-pod-deny
namespace: baseline-ns
spec:
containers:
- name: pause
image: registry.local:9001/pause
securityContext:
privileged: true
imagePullSecrets:
- name: local-secret

View File

@@ -0,0 +1,70 @@
---
apiVersion: v1
kind: Pod
metadata:
name: privileged-1
namespace: privileged-ns
spec:
containers:
- name: pause
image: registry.local:9001/pause
securityContext:
privileged: false
imagePullSecrets:
- name: local-secret
---
apiVersion: v1
kind: Pod
metadata:
name: privileged-2
namespace: privileged-ns
spec:
containers:
- name: pause
image: registry.local:9001/pause
securityContext:
privileged: true
imagePullSecrets:
- name: local-secret
---
apiVersion: v1
kind: Pod
metadata:
name: privileged-3
namespace: privileged-ns
spec:
containers:
- name: pause
image: registry.local:9001/pause
securityContext:
runAsUser: 1000
allowPrivilegeEscalation: false
runAsNonRoot: true
seccompProfile:
type: RuntimeDefault
capabilities:
drop:
- ALL
imagePullSecrets:
- name: local-secret
---
apiVersion: v1
kind: Pod
metadata:
name: privileged-4
namespace: privileged-ns
spec:
containers:
- name: pause
image: registry.local:9001/pause
securityContext:
runAsUser: 1000
allowPrivilegeEscalation: false
runAsNonRoot: false
seccompProfile:
type: RuntimeDefault
capabilities:
drop:
- ALL
imagePullSecrets:
- name: local-secret

View File

@@ -0,0 +1,20 @@
apiVersion: v1
kind: Pod
metadata:
name: restricted-pod-allow
namespace: restricted-ns
spec:
containers:
- name: pause
image: registry.local:9001/pause
securityContext:
runAsUser: 1000
allowPrivilegeEscalation: false
runAsNonRoot: true
seccompProfile:
type: RuntimeDefault
capabilities:
drop:
- ALL
imagePullSecrets:
- name: local-secret

View File

@@ -0,0 +1,20 @@
apiVersion: v1
kind: Pod
metadata:
name: restricted-pod-deny
namespace: restricted-ns
spec:
containers:
- name: pause
image: registry.local:9001/pause
securityContext:
runAsUser: 1000
allowPrivilegeEscalation: false
runAsNonRoot: false
seccompProfile:
type: RuntimeDefault
capabilities:
drop:
- ALL
imagePullSecrets:
- name: local-secret

View File

@@ -0,0 +1,37 @@
---
apiVersion: v1
kind: Namespace
metadata:
name: baseline-ns
labels:
pod-security.kubernetes.io/enforce: baseline
pod-security.kubernetes.io/enforce-version: latest
pod-security.kubernetes.io/warn: baseline
pod-security.kubernetes.io/warn-version: latest
pod-security.kubernetes.io/audit: baseline
pod-security.kubernetes.io/audit-version: latest
---
apiVersion: v1
kind: Namespace
metadata:
name: privileged-ns
labels:
pod-security.kubernetes.io/enforce: privileged
pod-security.kubernetes.io/enforce-version: latest
pod-security.kubernetes.io/warn: privileged
pod-security.kubernetes.io/warn-version: latest
pod-security.kubernetes.io/audit: privileged
pod-security.kubernetes.io/audit-version: latest
---
apiVersion: v1
kind: Namespace
metadata:
name: restricted-ns
labels:
pod-security.kubernetes.io/enforce: restricted
pod-security.kubernetes.io/enforce-version: latest
pod-security.kubernetes.io/warn: restricted
pod-security.kubernetes.io/warn-version: latest
pod-security.kubernetes.io/audit: restricted
pod-security.kubernetes.io/audit-version: latest
---

BIN
resources/images/pause.tar Executable file

Binary file not shown.

View File

@@ -0,0 +1,317 @@
from pytest import FixtureRequest, mark
from starlingx.config.configuration_file_locations_manager import ConfigurationFileLocationsManager
from starlingx.config.configuration_manager import ConfigurationManager
from starlingx.framework.logging.automation_logger import get_logger
from starlingx.framework.resources.resource_finder import get_stx_resource_path
from starlingx.framework.ssh.ssh_connection import SSHConnection
from starlingx.framework.validation.validation import validate_equals
from starlingx.keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKeywords
from starlingx.keywords.docker.images.docker_load_image_keywords import DockerLoadImageKeywords
from starlingx.keywords.files.file_keywords import FileKeywords
from starlingx.keywords.k8s.files.kubectl_file_delete_keywords import KubectlFileDeleteKeywords
from starlingx.keywords.k8s.namespace.kubectl_get_namespaces_keywords import KubectlGetNamespacesKeywords
from starlingx.keywords.k8s.pods.kubectl_apply_pods_keywords import KubectlApplyPodsKeywords
from starlingx.keywords.k8s.pods.kubectl_get_pods_keywords import KubectlGetPodsKeywords
from starlingx.keywords.k8s.secret.kubectl_create_secret_keywords import KubectlCreateSecretsKeywords
@mark.p0
def test_policy_labels_on_platform_namespaces():
"""
Verify that cert-manager, armada, kube-system, deployment namespaces contains the
privileged policy labels
Steps:
- Get the platform namespaces with the privileged policy labels
- Assert the expected namespace is ["cert-manager", "kube-system",
"deployment", "flux-helm"]
"""
label = "pod-security.kubernetes.io/audit=privileged,pod-security.kubernetes.io/audit-version=latest," "pod-security.kubernetes.io/enforce=privileged,pod-security.kubernetes.io/enforce-version=latest," "pod-security.kubernetes.io/warn=privileged,pod-security.kubernetes.io/warn-version=latest "
active_ssh = LabConnectionKeywords().get_active_controller_ssh()
get_logger().log_info("Get the privileged namespace list based on release")
expected_ns = ["cert-manager", "deployment", "flux-helm", "kube-system"]
get_logger().log_info("Get the platform namespaces with labels")
actual_ns = KubectlGetNamespacesKeywords(active_ssh).get_namespaces_by_label(label=label).get_namespaces()
actual_ns_names = [ns.get_name() for ns in actual_ns]
validate_equals(actual_ns_names, expected_ns, "Verifying that the retrieved namespaces match the expected values.")
def deploy_policy_ns(request: FixtureRequest, ssh_connection: SSHConnection) -> None:
"""
Function to deploy the privileged-ns, baseline-ns and restricted-ns namespaces
in setup and delete the namespaces in teardown
Args:
request (FixtureRequest): request needed for adding teardown
ssh_connection (SSHConnection): the ssh connection
Returns:
None: This function does not return a value.
"""
# Transfer the pod yaml file to the active controller
# Defines pod definition file name, source (local) and destination (remote) file paths.
psa_file_name = "psa_ns.yaml"
config_file_locations = ConfigurationFileLocationsManager()
ConfigurationManager.load_configs(config_file_locations)
local_path = get_stx_resource_path(f"resources/cloud_platform/nightly_regression/{psa_file_name}")
remote_path = f"/home/{ConfigurationManager.get_lab_config().get_admin_credentials().get_user_name()}/{psa_file_name}"
file_keywords = FileKeywords(ssh_connection)
file_keywords.upload_file(get_stx_resource_path(local_path), remote_path, overwrite=False)
get_logger().log_info(f"Creating resource from file {remote_path}")
kubectl_apply_pods_keywords = KubectlApplyPodsKeywords(ssh_connection)
kubectl_apply_pods_keywords.apply_from_yaml(yaml_file=remote_path)
def teardown():
KubectlFileDeleteKeywords(ssh_connection).delete_resources(remote_path)
request.addfinalizer(teardown)
def deploy_docker_image_to_local_registry(ssh_connection: SSHConnection, secret_namespace: str) -> None:
"""
Deploy images to the local registry
Args:
ssh_connection (SSHConnection): the ssh connection
secret_namespace (str): the namespace
Returns:
None: This function does not return a value.
"""
local_registry = ConfigurationManager.get_docker_config().get_registry("local_registry")
get_logger().log_info(f"Deploy docker images to local registry to {local_registry}")
FileKeywords(ssh_connection).upload_file(get_stx_resource_path("resources/images/pause.tar"), "/home/sysadmin/pause.tar")
KubectlCreateSecretsKeywords(ssh_connection).create_secret_for_registry(local_registry, "local-secret", secret_namespace)
docker_load_image_keywords = DockerLoadImageKeywords(ssh_connection)
docker_load_image_keywords.load_docker_image_to_host("pause.tar")
docker_load_image_keywords.tag_docker_image_for_registry("k8s.gcr.io/pause:latest", "pause", local_registry)
docker_load_image_keywords.push_docker_image_to_registry("pause", local_registry)
@mark.p1
def test_deny_pod_in_baseline_ns(request: FixtureRequest):
"""
Test to verify that baseline-ns rejects the pod with "privileged: true"
Args:
request (FixtureRequest): request needed for adding teardown
Steps:
- Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
- Upload the psa pod yaml to the lab
- Deploy images to the local registry for this testcase
- Deploy the pod with "privileged: true" and expect it to reject
"""
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
psa_baseline_pod_deny_file_name = "psa-baseline-pod-deny.yaml"
namespace = "baseline-ns"
# Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
deploy_policy_ns(request, ssh_connection)
# Upload the psa pod yaml to the lab
local_path = get_stx_resource_path(f"resources/cloud_platform/nightly_regression/{psa_baseline_pod_deny_file_name}")
config_file_locations = ConfigurationFileLocationsManager()
ConfigurationManager.load_configs(config_file_locations)
remote_path = f"/home/{ConfigurationManager.get_lab_config().get_admin_credentials().get_user_name()}/{psa_baseline_pod_deny_file_name}"
get_logger().log_info(f"Upload yaml on local {local_path} to {remote_path} on active controller")
file_keywords = FileKeywords(ssh_connection)
file_keywords.upload_file(get_stx_resource_path(local_path), remote_path, overwrite=True)
# Deploy images to the local registry for this testcase
deploy_docker_image_to_local_registry(ssh_connection, namespace)
# Deploy the pod with "privileged: true" and expect it to reject
get_logger().log_info(f"Deploy {remote_path} with 'privileged: true' on active controller and expect it to reject")
kubectl_apply_pods_keywords = KubectlApplyPodsKeywords(ssh_connection)
kubectl_apply_pods_keywords.fail_apply_from_yaml(remote_path)
@mark.p1
def test_allow_pod_in_baseline_ns(request: FixtureRequest):
"""
Test to verify that baseline-ns accepts the pod with "privileged: false"
Args:
request (FixtureRequest): request needed for adding teardown
Steps:
- Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
- Upload the psa pod yaml to the lab
- Deploy images to the local registry for this testcase
- Deploy the pod with "privileged: false" and expect it to accept
Raises:
AssertionError: If the pod does not reach 'running' status within the
specified timeout, an AssertionError will be raised
with a descriptive error message.
"""
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
psa_baseline_pod_deny_file_name = "psa-baseline-pod-allow.yaml"
namespace = "baseline-ns"
# Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
deploy_policy_ns(request, ssh_connection)
# Upload the psa pod yaml to the lab
local_path = get_stx_resource_path(f"resources/cloud_platform/nightly_regression/{psa_baseline_pod_deny_file_name}")
config_file_locations = ConfigurationFileLocationsManager()
ConfigurationManager.load_configs(config_file_locations)
remote_path = f"/home/{ConfigurationManager.get_lab_config().get_admin_credentials().get_user_name()}/{psa_baseline_pod_deny_file_name}"
get_logger().log_info(f"Upload yaml on local {local_path} to {remote_path} on active controller")
file_keywords = FileKeywords(ssh_connection)
file_keywords.upload_file(get_stx_resource_path(local_path), remote_path, overwrite=True)
# Deploy images to the local registry for this testcase
deploy_docker_image_to_local_registry(ssh_connection, namespace)
# Deploy the pod with "privileged: false" and expect it to accept
get_logger().log_info(f"Deploy {remote_path} with 'privileged: false' on active controller and expect it to accept")
kubectl_apply_pods_keywords = KubectlApplyPodsKeywords(ssh_connection)
kubectl_apply_pods_keywords.apply_from_yaml(remote_path)
get_pod_obj = KubectlGetPodsKeywords(ssh_connection)
pod_obj = get_pod_obj.get_pods(namespace).get_pods()
pod_name = [pod.get_name() for pod in pod_obj][0]
pod_status = get_pod_obj.wait_for_pod_status(pod_name, "Running", namespace)
validate_equals(pod_status, True, "Veryfing that the baseline-ns namespace accepts the pod with the 'privileged: false' configuration.")
@mark.p1
def test_deny_pod_in_restricted_ns(request: FixtureRequest):
"""
Test to verify that restricted-ns rejects the pod with restricted values
Args:
request (FixtureRequest): request needed for adding teardown
Steps:
- Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
- Upload the psa pod yaml to the lab
- Deploy images to the local registry for this testcase
- Deploy the pod with restricted values and expect it to reject in restricted-ns
"""
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
psa_restricted_pod_deny_file_name = "psa-restricted-pod-deny.yaml"
namespace = "restricted-ns"
# Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
deploy_policy_ns(request, ssh_connection)
# Upload the psa pod yaml to the lab
local_path = get_stx_resource_path(f"resources/cloud_platform/nightly_regression/{psa_restricted_pod_deny_file_name}")
config_file_locations = ConfigurationFileLocationsManager()
ConfigurationManager.load_configs(config_file_locations)
remote_path = f"/home/{ConfigurationManager.get_lab_config().get_admin_credentials().get_user_name()}/{psa_restricted_pod_deny_file_name}"
get_logger().log_info(f"Upload yaml on local {local_path} to {remote_path} on active controller")
file_keywords = FileKeywords(ssh_connection)
file_keywords.upload_file(get_stx_resource_path(local_path), remote_path, overwrite=True)
# Deploy images to the local registry for this testcase
deploy_docker_image_to_local_registry(ssh_connection, namespace)
# Deploy the pod with restricted values and expect it to reject in restricted-ns
get_logger().log_info(f"Deploy {remote_path} with restricted values and expect it to reject in restricted-ns")
kubectl_apply_pods_keywords = KubectlApplyPodsKeywords(ssh_connection)
kubectl_apply_pods_keywords.fail_apply_from_yaml(remote_path)
@mark.p1
def test_allow_pod_in_restricted_ns(request: FixtureRequest):
"""
Test to verify that restricted-ns accepts the pod
Args:
request (FixtureRequest): request needed for adding teardown
Steps:
- Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
- Upload the psa pod yaml to the lab
- Deploy images to the local registry for this testcase
- Deploy the pod and expect it to accept in restricted-ns
Raises:
AssertionError: If the pod does not reach 'running' status within the
specified timeout, an AssertionError will be raised
with a descriptive error message.
"""
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
psa_restricted_pod_allow_file_name = "psa-restricted-pod-allow.yaml"
namespace = "restricted-ns"
# Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
deploy_policy_ns(request, ssh_connection)
# Upload the psa pod yaml to the lab
local_path = get_stx_resource_path(f"resources/cloud_platform/nightly_regression/{psa_restricted_pod_allow_file_name}")
config_file_locations = ConfigurationFileLocationsManager()
ConfigurationManager.load_configs(config_file_locations)
remote_path = f"/home/{ConfigurationManager.get_lab_config().get_admin_credentials().get_user_name()}/{psa_restricted_pod_allow_file_name}"
get_logger().log_info(f"Upload yaml on local {local_path} to {remote_path} on active controller")
file_keywords = FileKeywords(ssh_connection)
file_keywords.upload_file(get_stx_resource_path(local_path), remote_path, overwrite=True)
# Deploy images to the local registry for this testcase
deploy_docker_image_to_local_registry(ssh_connection, namespace)
# Deploy the pod and expect it to accept in restricted-ns
get_logger().log_info(f"Deploy {remote_path} in restricted-ns on active controller and expect it to accept")
kubectl_apply_pods_keywords = KubectlApplyPodsKeywords(ssh_connection)
kubectl_apply_pods_keywords.apply_from_yaml(remote_path)
get_pod_obj = KubectlGetPodsKeywords(ssh_connection)
pod_obj = get_pod_obj.get_pods(namespace).get_pods()
pod_name = [pod.get_name() for pod in pod_obj][0]
pod_status = get_pod_obj.wait_for_pod_status(pod_name, "Running", namespace)
validate_equals(pod_status, True, "Veryfing that the restricted-ns namespace accepts the pod.")
@mark.p1
def test_allow_any_in_privileged_ns(request: FixtureRequest):
"""
Test to verify that privileged-ns accepts any pod
Args:
request (FixtureRequest): request needed for adding teardown
Steps:
- Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
- Upload the psa pod yaml to the lab
- Deploy images to the local registry for this testcase
- Deploy the pods in privileged-ns and expect it to accept
Raises:
AssertionError: If the pod does not reach 'running' status within the
specified timeout, an AssertionError will be raised
with a descriptive error message.
"""
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
psa_privileged_pod_allow_file_name = "psa-privileged-allow-any.yaml"
namespace = "privileged-ns"
# Deploy the privileged-ns, baseline-ns and restricted-ns namespaces
deploy_policy_ns(request, ssh_connection)
# Upload the psa pod yaml to the lab
local_path = get_stx_resource_path(f"resources/cloud_platform/nightly_regression/{psa_privileged_pod_allow_file_name}")
config_file_locations = ConfigurationFileLocationsManager()
ConfigurationManager.load_configs(config_file_locations)
remote_path = f"/home/{ConfigurationManager.get_lab_config().get_admin_credentials().get_user_name()}/{psa_privileged_pod_allow_file_name}"
get_logger().log_info(f"Upload yaml on local {local_path} to {remote_path} on active controller")
file_keywords = FileKeywords(ssh_connection)
file_keywords.upload_file(get_stx_resource_path(local_path), remote_path, overwrite=True)
# Deploy images to the local registry for this testcase
deploy_docker_image_to_local_registry(ssh_connection, namespace)
# Deploy the pods in privileged-ns and expect it to accept
get_logger().log_info(f"Deploy {remote_path} in privileged-ns on active controller and expect it to accept")
kubectl_apply_pods_keywords = KubectlApplyPodsKeywords(ssh_connection)
kubectl_apply_pods_keywords.apply_from_yaml(remote_path)
get_pod_obj = KubectlGetPodsKeywords(ssh_connection)
pod_obj = get_pod_obj.get_pods(namespace).get_pods()
pod_name = [pod.get_name() for pod in pod_obj][0]
pod_status = get_pod_obj.wait_for_pod_status(pod_name, "Running", namespace)
validate_equals(pod_status, True, "Veryfing that the privileged-ns namespace accepts the pod.")