diff --git a/framework/web/action/web_action_send_keys.py b/framework/web/action/web_action_send_keys.py new file mode 100644 index 00000000..c2c7c025 --- /dev/null +++ b/framework/web/action/web_action_send_keys.py @@ -0,0 +1,32 @@ +from selenium.webdriver.remote.webelement import WebElement + +from framework.web.action.web_action import WebAction + + +class WebActionSendKeys(WebAction): + """ + Class representing a Web action of Sending Keys to an element. SetText should be used whenever possible, but for some cases, we need this one. + """ + + def perform_action(self, web_element: WebElement, *args: str) -> None: + """ + Override the parent's perform action - Sends Keys to a WebElement + + Args: + web_element (WebElement): Element to send keys to + *args (str): One 'str' argument; Keys to send + + Returns: None + + """ + text_to_set = args[0] + web_element.send_keys(text_to_set) + + def __str__(self) -> str: + """ + String representation of this action. + + Returns: + str: + """ + return "SendKeys" diff --git a/framework/web/condition/web_condition_attribute_equals.py b/framework/web/condition/web_condition_attribute_equals.py new file mode 100644 index 00000000..ead7d71b --- /dev/null +++ b/framework/web/condition/web_condition_attribute_equals.py @@ -0,0 +1,52 @@ +from selenium import webdriver + +from framework.logging.automation_logger import get_logger +from framework.web.condition.web_condition import WebCondition +from framework.web.web_locator import WebLocator + + +class WebConditionAttributeEquals(WebCondition): + """ + This Web Condition will check if the Attribute of the element has the expected value. + """ + + def __init__(self, web_locator: WebLocator, attribute_name: str, expected_value: str): + """ + Constructor which will instantiate the driver object. + + Args: + web_locator (WebLocator): Locator for the WebElement of interest + attribute_name (str): Name of the attribute to expect. + expected_value (str): The expected value of the attribute. + """ + self.web_locator = web_locator + self.attribute_name = attribute_name + self.expected_value = expected_value + + def is_condition_satisfied(self, webdriver: webdriver) -> bool: + """ + This function will evaluate the web_condition and return True if the text of the element is as expected. + + Args: + webdriver (webdriver): The Selenium webdriver instance + + Returns: + bool: True if the value of the attribute matches the expected_value + + """ + web_element = webdriver.find_element(self.get_web_locator().get_by(), self.get_web_locator().get_locator()) + web_element_actual_attribute_value = web_element.get_attribute(self.attribute_name) + + get_logger().log_debug(f"Attribute {self.attribute_name} Expected: '{self.expected_value}' | Actual: '{web_element_actual_attribute_value}'") + + return web_element_actual_attribute_value == self.expected_value + + def __str__(self) -> str: + """ + Nice String representation for this condition. + + Returns: + str: + + """ + return f"ElementAttributeEquals {self.attribute_name} = {self.expected_value} - {self.get_web_locator()}" diff --git a/framework/web/webdriver_core.py b/framework/web/webdriver_core.py index cc59afef..e2a0d2d7 100644 --- a/framework/web/webdriver_core.py +++ b/framework/web/webdriver_core.py @@ -8,6 +8,7 @@ from config.configuration_manager import ConfigurationManager from framework.logging.automation_logger import get_logger from framework.web.action.web_action_click import WebActionClick from framework.web.action.web_action_get_text import WebActionGetText +from framework.web.action.web_action_send_keys import WebActionSendKeys from framework.web.action.web_action_set_text import WebActionSetText from framework.web.condition.web_condition import WebCondition from framework.web.condition.web_condition_text_equals import WebConditionTextEquals @@ -23,8 +24,8 @@ class WebDriverCore: def __init__(self): """ Constructor which will instantiate the driver object. - """ + """ chrome_options = selenium.webdriver.chrome.options.Options() chrome_options.add_argument("--ignore-certificate-errors") if ConfigurationManager.get_web_config().get_run_headless(): @@ -34,6 +35,7 @@ class WebDriverCore: def close(self) -> None: """ Close the WebDriver and browser window. + Returns: None """ @@ -42,11 +44,12 @@ class WebDriverCore: def navigate_to_url(self, url: str, conditions: List[WebCondition] = []) -> None: """ This function will navigate to the specified url. + The navigation will get retried if until one of the conditions is met (or we time out). Args: - url: URL to navigate to. - conditions: Conditions for successful navigation to this URL. + url (str): URL to navigate to. + conditions (List[WebCondition]): Conditions for successful navigation to this URL. Returns: None @@ -81,30 +84,46 @@ class WebDriverCore: def click(self, locator: WebLocator, conditions: List[WebCondition] = []) -> None: """ Click on the target element + Args: - locator: The locator of the element that we want to click on. - conditions: Conditions that must be satisfied for the Action to be declared successful. + locator (WebLocator): The locator of the element that we want to click on. + conditions (List[WebCondition]): Conditions that must be satisfied for the Action to be declared successful. Returns: None """ - action = WebActionClick(self.driver, locator, conditions) action_executor = WebActionExecutor(action) action_executor.execute_action() - def set_text(self, locator: WebLocator, text: str, conditions: List[WebCondition] = []) -> None: + def send_keys(self, locator: WebLocator, keys: str, conditions: List[WebCondition] = []) -> None: """ - Clears the text content of the element, then sets the text of the element. + Sends Keys directly to a WebElement. SendText should be favored wherever it can be used. + Args: - locator: The locator of the element that we want to set the text of. - text: The text that we want to set. - conditions: Conditions that must be satisfied for the Action to be declared successful. + locator (WebLocator): The locator of the element that we want to set the text of. + keys (str): The keys that we want to send + conditions (List[WebCondition]): Conditions that must be satisfied for the Action to be declared successful. Returns: None """ + action = WebActionSendKeys(self.driver, locator, conditions) + action_executor = WebActionExecutor(action) + action_executor.execute_action(keys) + def set_text(self, locator: WebLocator, text: str, conditions: List[WebCondition] = []) -> None: + """ + Clears the text content of the element, then sets the text of the element. + + Args: + locator (WebLocator): The locator of the element that we want to set the text of. + text (str): The text that we want to set. + conditions (List[WebCondition]): Conditions that must be satisfied for the Action to be declared successful. + + Returns: None + + """ conditions_clone = [condition for condition in conditions] conditions_clone.append(WebConditionTextEquals(locator, text)) action = WebActionSetText(self.driver, locator, conditions_clone) @@ -114,14 +133,15 @@ class WebDriverCore: def get_text(self, locator: WebLocator, conditions: List[WebCondition] = []) -> str: """ Gets the Text content of the element - Args: - locator: The locator of the element from which we want to get the text contents. - conditions: Conditions that must be satisfied for the Action to be declared successful. - Returns: None + Args: + locator (WebLocator): The locator of the element from which we want to get the text contents. + conditions (List[WebCondition]): Conditions that must be satisfied for the Action to be declared successful. + + Returns: + str: """ - action = WebActionGetText(self.driver, locator, conditions) action_executor = WebActionExecutor(action) return action_executor.execute_action() @@ -129,14 +149,15 @@ class WebDriverCore: def get_all_elements_text(self, locator: WebLocator, conditions: List[WebCondition] = []) -> List[str]: """ Gets the text content of all the elements that are matching the locator. - Args: - locator: A locator that matches multiple elements from which we want to get the text. - conditions: Conditions that must be satisfied for the Action to be declared successful. - Returns: None + Args: + locator (WebLocator): A locator that matches multiple elements from which we want to get the text. + conditions (List[WebCondition]): Conditions that must be satisfied for the Action to be declared successful. + + Returns: + List[str]: """ - action = WebActionGetText(self.driver, locator, conditions) action_executor = WebActionExecutor(action) return action_executor.execute_mass_action() diff --git a/keywords/k8s/pods/kubectl_get_pods_keywords.py b/keywords/k8s/pods/kubectl_get_pods_keywords.py index 3b85cccb..b7b14a40 100644 --- a/keywords/k8s/pods/kubectl_get_pods_keywords.py +++ b/keywords/k8s/pods/kubectl_get_pods_keywords.py @@ -1,5 +1,6 @@ import time +from framework.exceptions.keyword_exception import KeywordException from framework.ssh.ssh_connection import SSHConnection from framework.logging.automation_logger import get_logger from framework.validation.validation import validate_equals_with_retry @@ -42,7 +43,7 @@ class KubectlGetPodsKeywords(BaseKeyword): pods_list_output = KubectlGetPodsOutput(kubectl_get_pods_output) return pods_list_output - + def get_pods_no_validation(self, namespace: str = None) -> KubectlGetPodsOutput: """ Gets the k8s pods that are available using '-o wide'. @@ -66,7 +67,6 @@ class KubectlGetPodsKeywords(BaseKeyword): return pods_list_output - def get_pods_all_namespaces(self) -> KubectlGetPodsOutput: """ Gets the k8s pods that are available using '-o wide' for all namespaces. @@ -128,33 +128,40 @@ 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 + def wait_for_pods_to_reach_status(self, expected_status: str, pod_names: list = None, 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 to look for. If left as None, we will check for all the pods. + namespace (str): the namespace + poll_interval (int): the interval in secs to poll for status + timeout (int): the timeout in secs - pod_status_timeout = time.time() + timeout + Returns: + bool: True if pod is in expected status else False - 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) + """ + pod_status_timeout = time.time() + timeout + + while time.time() < pod_status_timeout: + + pods = self.get_pods(namespace).get_pods() + + # We need to filter the list for only the pods matching the pod names if specified + if pod_names: + pods = [pod for pod in pods if pod.get_name() in pod_names] + + pods_in_incorrect_status = [pod for pod in pods if pod.get_status() != expected_status] + + if len(pods_in_incorrect_status) == 0: + return True + time.sleep(poll_interval) + + raise KeywordException(f"Pods {pods_in_incorrect_status} in namespace {namespace} did not reach status {expected_status} within {timeout} seconds") - raise KeywordException(f"Pods {pod_names} in namespace {namespace} did not reach status {expected_status} within {timeout} seconds") - def wait_for_kubernetes_to_restart(self, timeout: int = 600, check_interval: int = 20) -> bool: """ Wait for the Kubernetes API to go down, then wait for the kube-apiserver pod to be Running. diff --git a/testcases/cloud_platform/regression/containers/test_k8s_dashboard.py b/testcases/cloud_platform/regression/containers/test_k8s_dashboard.py index 3e36f75e..68d812b0 100644 --- a/testcases/cloud_platform/regression/containers/test_k8s_dashboard.py +++ b/testcases/cloud_platform/regression/containers/test_k8s_dashboard.py @@ -1,5 +1,3 @@ -import os - from pytest import fixture, mark from config.configuration_manager import ConfigurationManager @@ -72,30 +70,29 @@ def copy_k8s_files(request: fixture, ssh_connection: SSHConnection): FileKeywords(ssh_connection).upload_file(local_path, f"/home/sysadmin/{k8s_dashboard_dir}/{dashboard_file_name}") -def create_k8s_dashboard(request: fixture, namespace: str, con_ssh: SSHConnection): +def create_k8s_dashboard(namespace: str, con_ssh: SSHConnection): """ Create all necessary resources for the k8s dashboard + Args: - request (fixture): pytest fixture namespace (str): kubernetes_dashboard namespace name con_ssh (SSHConnection): the SSH connection Raises: KeywordException: if the k8s dashboard is not accessible """ - - k8s_dashboard_file_path = os.path.join(HOME_K8S_DIR, K8S_DASHBOARD_FILE) + k8s_dashboard_file_path = f"{HOME_K8S_DIR}/{K8S_DASHBOARD_FILE}" sys_domain_name = ConfigurationManager.get_lab_config().get_floating_ip() - path_cert = os.path.join(HOME_K8S_DIR, K8S_CERT_DIR) + path_cert = f"{HOME_K8S_DIR}/{K8S_CERT_DIR}" get_logger().log_info(f"Creating {path_cert} directory") 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_DIR, dashboard_key) - crt = os.path.join(HOME_K8S_DIR, dashboard_cert) + key = f"{HOME_K8S_DIR}/{dashboard_key}" + crt = f"{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=K8S_DASHBOARD_SECRETS_NAME, tls_crt=crt, tls_key=key, namespace=namespace) @@ -104,11 +101,9 @@ def create_k8s_dashboard(request: fixture, namespace: str, con_ssh: SSHConnectio 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, - ) + is_dashboard_pods_running = kubectl_get_pods_keywords.wait_for_pods_to_reach_status("Running", namespace=namespace) assert is_dashboard_pods_running, f"Not all pods in {namespace} namespace reached 'Running' status" get_logger().log_info(f"Updating {K8S_DASHBOARD_NAME} service to be exposed on port {K8S_DASHBOARD_PORT}") @@ -141,7 +136,7 @@ def get_k8s_token(request: fixture, con_ssh: SSHConnection) -> str: adminuserfile = "admin-user.yaml" serviceaccount = "admin-user" - admin_user_file_path = os.path.join(HOME_K8S_DIR, adminuserfile) + admin_user_file_path = f"{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) @@ -266,7 +261,7 @@ def test_k8s_dashboard_access(request): request.addfinalizer(teardown_secret) - create_k8s_dashboard(request, namespace=test_namespace, con_ssh=ssh_connection) + create_k8s_dashboard(namespace=test_namespace, con_ssh=ssh_connection) # Step 4: Create the token for the dashboard def teardown_svc_account(): @@ -291,5 +286,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) diff --git a/web_pages/k8s_dashboard/login/k8s_login_page.py b/web_pages/k8s_dashboard/login/k8s_login_page.py index a1b94fd6..ac6e1b95 100644 --- a/web_pages/k8s_dashboard/login/k8s_login_page.py +++ b/web_pages/k8s_dashboard/login/k8s_login_page.py @@ -1,4 +1,7 @@ +import os + from framework.web.condition.web_condition_element_visible import WebConditionElementVisible +from framework.web.condition.web_condition_text_equals import WebConditionTextEquals from framework.web.webdriver_core import WebDriverCore from web_pages.base_page import BasePage from web_pages.k8s_dashboard.login.k8s_login_page_locators import K8sLoginPageLocators @@ -41,13 +44,17 @@ class K8sLoginPage(BasePage): Args: kubeconfig_path (str): The file path to the kubeconfig file. """ - condition = WebConditionElementVisible(self.locators.get_locator_input_kubeconfig_file()) + visible_input_locator = self.locators.get_locator_input_kubeconfig_file() + + condition = WebConditionElementVisible(visible_input_locator) kubeconfig_option = self.locators.get_locator_kubeconfig_option() self.driver.click(kubeconfig_option, conditions=[condition]) # this actually needs to be changed to send_keys - kubeconfig_input = self.locators.get_locator_input_kubeconfig_file() - self.driver.set_text(kubeconfig_input, kubeconfig_path) + file_name = os.path.basename(kubeconfig_path) + file_is_displayed_in_visible_input = WebConditionTextEquals(visible_input_locator, file_name) + kubeconfig_input = self.locators.get_locator_hidden_input_kubeconfig_file() + self.driver.send_keys(kubeconfig_input, kubeconfig_path, [file_is_displayed_in_visible_input]) self.click_signin() diff --git a/web_pages/k8s_dashboard/login/k8s_login_page_locators.py b/web_pages/k8s_dashboard/login/k8s_login_page_locators.py index e4e6fdd6..88f5ace7 100644 --- a/web_pages/k8s_dashboard/login/k8s_login_page_locators.py +++ b/web_pages/k8s_dashboard/login/k8s_login_page_locators.py @@ -44,10 +44,23 @@ class K8sLoginPageLocators: """ Locator for the Kubeconfig File Input. + This is the visible input, which will show the name of the file uploaded. + To upload a file, you need to use the hidden_input instead. + Returns: WebLocator """ return WebLocator("""[title="fileInput"]""", By.CSS_SELECTOR) + def get_locator_hidden_input_kubeconfig_file(self) -> WebLocator: + """ + Locator for the Hidden Kubeconfig File Input. + + This is the input that needs to be used to upload the file using set_text. + + Returns: WebLocator + """ + return WebLocator("""input[type="file"]""", By.CSS_SELECTOR) + def get_locator_user_button(self) -> WebLocator: """ Locator for the User Button.