在gcdw日常测试中出现过一个这样的问题,在创建warehouse时出现过warehouse无法resume唤醒的问题,这里使用selenium对gcdw集群warehouse部分进行自动化测试,相关片段如下:
# -*- coding: utf-8 -*-
'''
author : anonymity
case description: 创建warehouse、勾选自启动、启动warehouse、验证warehouse是否可以启动
'''
from time import sleep
import config
import pytest
from selenium.webdriver.common.by import By
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import paramiko
from toolbox.sql_client import conn_sql
import re
def start():
"""
用例执行前需要做的处理
检查warehouse(warehouseresume1)是否存在,如果存在则删除
"""
db = conn_sql()
cursor = db.cursor()
try:
cursor.execute("show warehouses")
warehouses = cursor.fetchall()
for warehouse in warehouses:
if warehouse[0] == 'warehouseresume1':
cursor.execute("drop warehouse warehouseresume1")
db.commit()
except Exception as e:
print(e)
db.rollback()
cursor.close()
db.close()
print("setup方法")
def end():
"""
用例执行后需要做的处理
"""
print("teardown方法")
@pytest.mark.mark_11
@pytest.mark.skipif(condition=config.is_run("test_11_warehouse_auto_resume"), reason='用例下线')
def test_11_warehouse_auto_resume():
# 获取驱动
driver = config.DRIVER
WebDriverWait(driver, 30).until(EC.element_to_be_clickable((By.LINK_TEXT, "Warehouses"))).click()
WebDriverWait(driver, 30).until(EC.element_to_be_clickable((By.XPATH, "//button/span[contains(text(), 'Create')]"))).click()
driver.find_element(By.XPATH, "//div[@aria-label='Create warehouse']//form/div[1]//input").send_keys('warehouseresume1')
driver.find_element(By.XPATH, "//div[@aria-label='Create warehouse']//form/div[2]//input").click()
WebDriverWait(driver, 30).until(EC.visibility_of_element_located((By.XPATH, "/html/body/div[4]/div[1]/div[1]/ul/li[1]"))).click()
driver.find_element(By.XPATH, "//div[@aria-label='Create warehouse']//form/div[4]//input").click()
WebDriverWait(driver, 30).until(EC.visibility_of_element_located((By.XPATH, "/html/body/div[5]/div[1]/div[1]/ul/li[2]"))).click()
driver.find_element(By.XPATH, "//div[@aria-label='Create warehouse']//form/div[10]//input").click()
button_5_min = WebDriverWait(driver, 30).until(EC.visibility_of_element_located((By.XPATH, "//li/span[text()='5 minutes']")))
driver.execute_script("arguments[0].click();", button_5_min)
# checkbox = driver.find_element(By.XPATH, "//*[@id='app']/section/main/section/div[1]/div/div[2]/form/div[12]/div/label/span[1]")
# is_checked = checkbox.get_attribute('class')
# if 'is-checked' in is_checked:
# checkbox.click()
driver.find_element(By.XPATH, "//div[@aria-label='Create warehouse']//div//button/span[text()='Finish']").click()
# 创建warehouse完毕
sleep(1)
table = driver.find_element(By.XPATH, "//*[@id='app']/section/main/section/main/div/div[1]/div[2]/div/div[3]/table")
rows = table.find_elements(By.TAG_NAME, "tr")
table_header = driver.find_element(By.XPATH, "//*[@id='app']/section/main/section/main/div/div[1]/div[2]/div/div[2]/table")
cols = table_header.find_elements(By.TAG_NAME, "th")
cols_texts = []
for col in cols:
cols_texts.append(col.text)
name_index = cols_texts.index('Name')
status_index = cols_texts.index('Status')
warehouse_list = []
for row in rows:
cells = row.find_elements(By.TAG_NAME, 'td')
warehouse_list.append(cells[name_index].text)
print("warehouse_list:", warehouse_list)
warehouse_exist = 0
for warehouse in warehouse_list:
if warehouse == 'warehouseresume1':
warehouse_exist = 1
# 断言界面中是否存在warehouse
assert warehouse_exist, f"warehouse没有创建成功,web界面中未找到【warehouseresume1】"
print("web界面中找到warehouse【warehouseresume1】")
for row in rows:
cells = row.find_elements(By.TAG_NAME, 'td')
if cells[name_index].text == 'warehouseresume1':
cells[name_index].click()
# 点击resume button启动warehouse
driver.find_element(By.XPATH, "//button/span[contains(text(), 'Resume')]").click()
driver.find_element(By.XPATH, "//div[@aria-label='Resume Warehouse']//button/span[contains(text(), 'OK')]").click()
WebDriverWait(driver, 120).until(EC.invisibility_of_element_located((By.XPATH, "//div[@aria-label='Resume Warehouse']")))
sleep(2)
# 断言界面状态是否是STARTED
for row in rows:
cells = row.find_elements(By.TAG_NAME, 'td')
if cells[name_index].text == 'warehouseresume1':
assert cells[status_index].text == 'STARTED', "界面中warehouse:[warehouseresume1]状态未显示STARTED"
print("界面中warehouse:[warehouseresume1]状态显示STARTED")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect('10.10.55.74', username='root', password='1q2w3e4r')
command = 'kubectl get all -n zhaozhetest1'
stdin, stdout, stderr = ssh.exec_command(command)
output = stdout.read().decode('utf-8')
error = stderr.read().decode('utf-8')
if error:
print(f"Error: {error}")
return None
pattern = r'^pod/warehouseresume1.*'
matches = re.findall(pattern, output, re.MULTILINE)
print("matches:", matches)
assert 'Running' in matches[0], "kubernetes 客户端为查询warehouseresume1相关warehouse正在运行"
print('warehouseresume1 正在运行')
finally:
ssh.close()
最后修改时间:2024-10-31 17:27:04
「喜欢这篇文章,您的关注和赞赏是给作者最好的鼓励」
关注作者
【版权声明】本文为墨天轮用户原创内容,转载时必须标注文章的来源(墨天轮),文章链接,文章作者等基本信息,否则作者和墨天轮有权追究责任。如果您发现墨天轮中有涉嫌抄袭或者侵权的内容,欢迎发送邮件至:contact@modb.pro进行举报,并提供相关证据,一经查实,墨天轮将立刻删除相关内容。




