login_with_kubeconfig - Support for the file picker
- WebActionSendKeys - WebConditionAttributeEquals - Fixed issues with test_k8s_dashboard_access - login_with_kubeconfig now works with the input Change-Id: Ia98370730713ccf2841309b3631bc82ecb39cc67 Signed-off-by: croy <Christian.Roy@windriver.com>
This commit is contained in:
croy
parent
7dcca5b7a7
commit
4207ece6f5
32
framework/web/action/web_action_send_keys.py
Normal file
32
framework/web/action/web_action_send_keys.py
Normal file
@ -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"
|
||||
52
framework/web/condition/web_condition_attribute_equals.py
Normal file
52
framework/web/condition/web_condition_attribute_equals.py
Normal file
@ -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()}"
|
||||
@ -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()
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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()
|
||||
|
||||
|
||||
@ -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.
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user