Files
test/keywords/k8s/deployments/kubectl_get_deployments_keywords.py
Vitor Vidal 0ea45495ca Adding file resources used by system test TCs.
Adding first system test TC (deploy, scale and remove deployments).
Adding new object for k8s get deployments.
Updating existing k8s objects to support "namespace" parameter.

Change-Id: Ifafa33638ab2949cac9d985b56ed90bebe2495a8

---
Signed-off-by: Vitor Vidal <vitor.vidaldenegreiros@windriver.com>
---
Change-Id: I6c0cd323826f0c38db2289fb420d49bf432e1930
2025-08-28 14:07:49 -03:00

53 lines
1.7 KiB
Python

from keywords.base_keyword import BaseKeyword
from keywords.k8s.k8s_command_wrapper import export_k8s_config
from keywords.k8s.deployments.object.kubectl_get_deployments_output import KubectlGetDeploymentOutput
class KubectlGetDeploymentsKeywords(BaseKeyword):
"""
Class for Expose Deployment Keywords
"""
def __init__(self, ssh_connection):
"""
Constructor
Args:
ssh_connection:
"""
self.ssh_connection = ssh_connection
def get_deployment(self, deployment_name: str):
"""
Get the deployments from the specified name
Args:
deployment_name (str): the deployment name
Returns:
A list
"""
kubectl_get_deployments_output = self.ssh_connection.send(export_k8s_config(f"kubectl get deployment {deployment_name}"))
self.validate_success_return_code(self.ssh_connection)
deployments_list_output = KubectlGetDeploymentOutput(kubectl_get_deployments_output)
return deployments_list_output
def get_deployments(self, namespace: str = None):
"""
Get the deployments from the specified namespace, or all namespaces if not specified.
Args:
deployment_name (str): the deployment name
namespace (str): the namespace
Returns:
A list
"""
cmd = f"kubectl get deployment"
if namespace:
cmd = f"{cmd} -n {namespace}"
kubectl_get_deployments_output = self.ssh_connection.send(export_k8s_config(cmd))
self.validate_success_return_code(self.ssh_connection)
deployments_list_output = KubectlGetDeploymentOutput(kubectl_get_deployments_output)
return deployments_list_output