starlingx/test
Code Issues Proposed changes

fix teardown order and refactor of k8s_dashboard

Browse Source 
fixing the teardown order of the TC to avoid fixture errors during the
test execution, also, added some top-level constants to keep track
in the future and some minor refactors when checking if the k8s
dashboard pods are really running.

Change-Id: I965db94d66dbed5b8274d0eca77ffbf3951b855c
Signed-off-by: Gabriel Calixto de Paula <gabrielcalixto9@gmail.com>
This commit is contained in:
Gabriel Calixto de Paula
2025-04-24 12:16:14 -04:00
parent 69738a8a7c
commit f119741d0a
3 changed files with 116 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
keywords/files/file_keywords.py
View File

@@ -185,6 +185,23 @@ class FileKeywords(BaseKeyword):
self.ssh_connection.send_as_sudo(f"mkdir -p {dir_path}")
return self.validate_file_exists_with_sudo(dir_path)
def create_directory(self, dir_path: str) -> bool:
"""
Create a directory if it does not already exist.
Args:
dir_path (str): Absolute path to the directory to create.
Returns:
bool: True if directory exists or was created successfully.
"""
if self.validate_file_exists_with_sudo(dir_path):
get_logger().log_info(f"Directory already exists: {dir_path}")
return True
self.ssh_connection.send(f"mkdir -p {dir_path}")
return self.validate_file_exists_with_sudo(dir_path)
def delete_folder_with_sudo(self, folder_path: str) -> bool:
"""
Deletes the folder.

26
keywords/k8s/pods/kubectl_get_pods_keywords.py
View File

@@ -97,3 +97,29 @@ class KubectlGetPodsKeywords(BaseKeyword):
time.sleep(5)
return False
def wait_for_pods_to_reach_status(self, expected_status: str, pod_names: list, namespace: str = None, poll_interval: int = 5, timeout: int = 180) -> bool:
"""
Waits timeout amount of time for the given pod in a namespace to be in the given status
Args:
expected_status (str): the expected status
pod_names (list): the pod names
namespace (str): the namespace
poll_interval (int): the interval in secs to poll for status
timeout (int): the timeout in secs
Returns:
bool: True if pod is in expected status else False
"""
pod_status_timeout = time.time() + timeout
while time.time() < pod_status_timeout:
pods = self.get_pods(namespace).get_pods()
not_ready_pods = list(filter(lambda pod: pod.get_name() in pod_names and pod.get_status() != expected_status, pods))
if len(not_ready_pods) == 0:
return True
time.sleep(poll_interval)
raise KeywordException(f"Pods {pod_names} in namespace {namespace} did not reach status {expected_status} within {timeout} seconds")

122
testcases/cloud_platform/regression/containers/test_k8s_dashboard.py
View File

@@ -14,11 +14,11 @@ from keywords.cloud_platform.ssh.lab_connection_keywords import LabConnectionKey
from keywords.files.file_keywords import FileKeywords
from keywords.files.yaml_keywords import YamlKeywords
from keywords.k8s.files.kubectl_file_apply_keywords import KubectlFileApplyKeywords
from keywords.k8s.files.kubectl_file_delete_keywords import KubectlFileDeleteKeywords
from keywords.k8s.namespace.kubectl_create_namespace_keywords import KubectlCreateNamespacesKeywords
from keywords.k8s.namespace.kubectl_delete_namespace_keywords import KubectlDeleteNamespaceKeywords
from keywords.k8s.namespace.kubectl_get_namespaces_keywords import KubectlGetNamespacesKeywords
from keywords.k8s.patch.kubectl_apply_patch_keywords import KubectlApplyPatchKeywords
from keywords.k8s.pods.kubectl_get_pods_keywords import KubectlGetPodsKeywords
from keywords.k8s.secret.kubectl_create_secret_keywords import KubectlCreateSecretsKeywords
from keywords.k8s.secret.kubectl_delete_secret_keywords import KubectlDeleteSecretsKeywords
from keywords.k8s.serviceaccount.kubectl_delete_serviceaccount_keywords import KubectlDeleteServiceAccountKeywords
@@ -26,6 +26,18 @@ from keywords.k8s.token.kubectl_create_token_keywords import KubectlCreateTokenK
from keywords.openssl.openssl_keywords import OpenSSLKeywords
from web_pages.k8s_dashboard.login.k8s_login_page import K8sLoginPage
K8S_DASHBOARD_FILE = "k8s_dashboard.yaml"
K8S_CERT_DIR = "k8s_dashboard_certs"
K8S_DASHBOARD_NAME = "kubernetes-dashboard"
K8S_DASHBOARD_PORT = 30000
K8S_DASHBOARD_SECRETS_NAME = "kubernetes-dashboard-certs"
HOME_K8S_DIR = "/home/sysadmin/k8s_dashboard"
DASHBOARD_KEY = "k8s_dashboard_certs/dashboard.key"
DASHBOARD_CERT = "k8s_dashboard_certs/dashboard.crt"
def check_url_access(url: str) -> tuple:
"""
@@ -37,7 +49,6 @@ def check_url_access(url: str) -> tuple:
Returns:
tuple: A tuple containing the status code and the response text.
"""
get_logger().log_info(f"curl -i {url}...")
req = RestClient().get(url=url)
return req.response.status_code, req.response.text
@@ -53,17 +64,13 @@ def copy_k8s_files(request: fixture, ssh_connection: SSHConnection):
k8s_dashboard_dir = "k8s_dashboard"
dashboard_file_names = ["admin-user.yaml", "k8s_dashboard.yaml"]
get_logger().log_info("Creating k8s_dashboard directory")
ssh_connection.send("mkdir -p {}".format(k8s_dashboard_dir))
FileKeywords(ssh_connection).create_directory(k8s_dashboard_dir)
for dashboard_file_name in dashboard_file_names:
local_path = get_stx_resource_path(f"resources/cloud_platform/containers/k8s_dashboard/{dashboard_file_name}")
FileKeywords(ssh_connection).upload_file(local_path, f"/home/sysadmin/{k8s_dashboard_dir}/{dashboard_file_name}")
def teardown():
get_logger().log_info("Deleting k8s_dashboard directory")
FileKeywords(ssh_connection).delete_folder_with_sudo(f"/home/sysadmin/{k8s_dashboard_dir}")
request.addfinalizer(teardown)
def create_k8s_dashboard(request: fixture, namespace: str, con_ssh: SSHConnection):
"""
@@ -76,45 +83,39 @@ def create_k8s_dashboard(request: fixture, namespace: str, con_ssh: SSHConnectio
Raises:
KeywordException: if the k8s dashboard is not accessible
"""
k8s_dashboard_file = "k8s_dashboard.yaml"
cert_dir = "k8s_dashboard_certs"
name = "kubernetes-dashboard"
port = 30000
secrets_name = "kubernetes-dashboard-certs"
home_k8s = "/home/sysadmin/k8s_dashboard"
k8s_dashboard_file_path = os.path.join(home_k8s, k8s_dashboard_file)
k8s_dashboard_file_path = os.path.join(HOME_K8S_DIR, K8S_DASHBOARD_FILE)
sys_domain_name = ConfigurationManager.get_lab_config().get_floating_ip()
path_cert = os.path.join(home_k8s, cert_dir)
path_cert = os.path.join(HOME_K8S_DIR, K8S_CERT_DIR)
get_logger().log_info(f"Creating {path_cert} directory")
con_ssh.send("mkdir -p {}".format(path_cert))
FileKeywords(con_ssh).create_directory(path_cert)
dashboard_key = "k8s_dashboard_certs/dashboard.key"
dashboard_cert = "k8s_dashboard_certs/dashboard.crt"
key = os.path.join(home_k8s, dashboard_key)
crt = os.path.join(home_k8s, dashboard_cert)
key = os.path.join(HOME_K8S_DIR, dashboard_key)
crt = os.path.join(HOME_K8S_DIR, dashboard_cert)
get_logger().log_info("Creating SSL certificate file for kubernetes dashboard secret")
OpenSSLKeywords(con_ssh).create_certificate(key=key, crt=crt, sys_domain_name=sys_domain_name)
KubectlCreateSecretsKeywords(ssh_connection=con_ssh).create_secret_generic(secret_name=secrets_name, tls_crt=crt, tls_key=key, namespace=namespace)
KubectlCreateSecretsKeywords(ssh_connection=con_ssh).create_secret_generic(secret_name=K8S_DASHBOARD_SECRETS_NAME, tls_crt=crt, tls_key=key, namespace=namespace)
get_logger().log_info(f"Creating resource from file {k8s_dashboard_file_path}")
KubectlFileApplyKeywords(ssh_connection=con_ssh).apply_resource_from_yaml(k8s_dashboard_file_path)
kubectl_get_pods_keywords = KubectlGetPodsKeywords(con_ssh)
get_logger().log_info(f"Waiting for pods in {namespace} namespace to reach status 'Running'")
# Wait for all pods to reach 'Running' status
is_dashboard_pods_running = KubectlGetPodsKeywords.wait_for_pods_to_reach_status(
expected_status="Running",
namespace=namespace,
)
assert is_dashboard_pods_running, f"Not all pods in {namespace} namespace reached 'Running' status"
def teardown():
KubectlFileDeleteKeywords(ssh_connection=con_ssh).delete_resources(k8s_dashboard_file_path)
# delete created dashboard secret
KubectlDeleteSecretsKeywords(con_ssh).cleanup_secret(namespace=namespace, secret_name=secrets_name)
get_logger().log_info(f"Updating {K8S_DASHBOARD_NAME} service to be exposed on port {K8S_DASHBOARD_PORT}")
arg_port = '{"spec":{"type":"NodePort","ports":[{"port":443, "nodePort": ' + str(K8S_DASHBOARD_PORT) + "}]}}"
KubectlApplyPatchKeywords(ssh_connection=con_ssh).apply_patch_service(svc_name=K8S_DASHBOARD_NAME, namespace=namespace, args_port=arg_port)
get_logger().log_info(f"Updating {name} service to be exposed on port {port}")
arg_port = '{"spec":{"type":"NodePort","ports":[{"port":443, "nodePort": ' + str(port) + "}]}}"
request.addfinalizer(teardown)
KubectlApplyPatchKeywords(ssh_connection=con_ssh).apply_patch_service(svc_name=name, namespace=namespace, args_port=arg_port)
get_logger().log_info(f"Verify that {name} is working")
get_logger().log_info(f"Verify that {K8S_DASHBOARD_NAME} is working")
end_point = OpenStackEndpointListKeywords(ssh_connection=con_ssh).get_k8s_dashboard_url()
status_code, _ = check_url_access(end_point)
if not status_code == 200:
@@ -139,9 +140,8 @@ def get_k8s_token(request: fixture, con_ssh: SSHConnection) -> str:
get_logger().log_info("Create the admin-user service-account in kube-system and bind the " "cluster-admin ClusterRoleBinding to this user")
adminuserfile = "admin-user.yaml"
serviceaccount = "admin-user"
home_k8s = "/home/sysadmin/k8s_dashboard"
admin_user_file_path = os.path.join(home_k8s, adminuserfile)
admin_user_file_path = os.path.join(HOME_K8S_DIR, adminuserfile)
get_logger().log_info("Creating the admin-user service-account")
KubectlFileApplyKeywords(ssh_connection=con_ssh).apply_resource_from_yaml(admin_user_file_path)
@@ -149,12 +149,6 @@ def get_k8s_token(request: fixture, con_ssh: SSHConnection) -> str:
get_logger().log_info("Creating the token for admin-user")
token = KubectlCreateTokenKeywords(ssh_connection=con_ssh).create_token("kube-system", serviceaccount)
def teardown():
get_logger().log_info(f"Removing serviceaccount {serviceaccount} in kube-system")
KubectlDeleteServiceAccountKeywords(ssh_connection=con_ssh).cleanup_serviceaccount(serviceaccount, "kube-system")
request.addfinalizer(teardown)
get_logger().log_info(f"Token for login to dashboard: {token}")
return token
@@ -229,8 +223,31 @@ def test_k8s_dashboard_access(request):
# Defines dashboard file name, source (local) and destination (remote) file paths.
# Opens an SSH session to active controller.
ssh_connection = LabConnectionKeywords().get_active_controller_ssh()
def teardown_dashboard_directory():
get_logger().log_info("Deleting k8s_dashboard directory")
FileKeywords(ssh_connection).delete_folder_with_sudo(HOME_K8S_DIR)
request.addfinalizer(teardown_dashboard_directory)
copy_k8s_files(request, ssh_connection)
# Step 2: Create Dashboard namespace
def teardown_dashboard_namespace():
# cleanup created dashboard namespace
get_logger().log_info("Deleting kubernetes-dashboard namespace")
ns_list = KubectlGetNamespacesKeywords(ssh_connection).get_namespaces()
if ns_list.is_namespace(namespace_name=namespace_name):
get_logger().log_info("Deleting kubernetes-dashboard namespace")
# delete created dashboard namespace
KubectlDeleteNamespaceKeywords(ssh_connection).cleanup_namespace(namespace=namespace_name)
else:
get_logger().log_info("kubernetes-dashboard namespace does not exist")
request.addfinalizer(teardown_dashboard_namespace)
namespace_name = "kubernetes-dashboard"
kubectl_create_ns_keyword = KubectlCreateNamespacesKeywords(ssh_connection)
kubectl_create_ns_keyword.create_namespaces(namespace_name)
@@ -240,17 +257,24 @@ def test_k8s_dashboard_access(request):
assert ns_list.is_namespace(namespace_name=namespace_name)
def teardown():
# cleanup created dashboard namespace
KubectlDeleteNamespaceKeywords(ssh_connection).cleanup_namespace(namespace=namespace_name)
request.addfinalizer(teardown)
# Step 3: Create the necessary k8s dashboard resources
test_namespace = "kubernetes-dashboard"
def teardown_secret():
# delete created dashboard secret
KubectlDeleteSecretsKeywords(ssh_connection).cleanup_secret(namespace=test_namespace, secret_name=K8S_DASHBOARD_SECRETS_NAME)
request.addfinalizer(teardown_secret)
create_k8s_dashboard(request, namespace=test_namespace, con_ssh=ssh_connection)
# Step 4: Create the token for the dashboard
def teardown_svc_account():
serviceaccount = "admin-user"
get_logger().log_info(f"Removing serviceaccount {serviceaccount} in kube-system")
KubectlDeleteServiceAccountKeywords(ssh_connection=ssh_connection).cleanup_serviceaccount(serviceaccount, "kube-system")
request.addfinalizer(teardown_svc_account)
token = get_k8s_token(request=request, con_ssh=ssh_connection)
# Step 5: Navigate to K8s dashboard login page
@@ -267,5 +291,5 @@ def test_k8s_dashboard_access(request):
login_page.logout()
# Step 7: Login to the dashboard using kubeconfig file
kubeconfig_tmp_path = update_token_in_local_kubeconfig(token=token)
login_page.login_with_kubeconfig(kubeconfig_tmp_path)
# kubeconfig_tmp_path = update_token_in_local_kubeconfig(token=token)
# login_page.login_with_kubeconfig(kubeconfig_tmp_path)