GBase8a GCDW 审计模块 Web 端自动化测试样例：Selenium + Pytest

在对 GBase8a GCDW (GBase 8a 的数据仓库解决方案) 的审计模块进行自动化测试时，我们采用了 Selenium 和 Pytest 这两种技术的组合。这一方法不仅提高了测试的效率，还确保了审计模块的功能测试覆盖全面，能够及时识别潜在的问题。

在开始测试之前，我们首先设置了测试环境，包括安装必要的库和配置浏览器驱动。接着，我们编写了一系列测试用例，以验证审计模块的关键功能，比如生成审计报告、查看审计记录及导出数据等。

测试用例示例：

# -*- coding: utf-8 -*-
'''
    case description: 点击queryid、检查History SQL Detail
'''

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
def start():
    """
    用例执行前需要做的处理
    """
    sql1 = "show databases"
    sql2 = "CREATE DATABASE zztestaudit"
    sql3 = "show tables"
    sql4 = "CREATE TABLE `zztestaudit`. `zztest_t1` (`name` VARCHAR(50),`age` int,`sex` CHAR(1))"
    sql5 = "show warehouses"
    sql6 = "CREATE WAREHOUSE 'zztestaudit' WITH WAREHOUSE_SIZE = 'XSMALL' AUTO_SUSPEND =600 AUTO_RESUME = 'true' MIN_CLUSTER_COUNT = 2 MAX_CLUSTER_COUNT = 2 SCALING_POLICY = 'standard'"
    sql7 = "use warehouse zztestaudit"
    sql8 = "alter user root default_warehouse = 'zztestaudit'"
    sql9 = "insert into zztestaudit.zztest_t1(name, age, sex) values('James', 24, 'M'),('David', 28, 'M'),('Emily', 34, 'F')"
    db = conn_sql()
    cursor = db.cursor()
    try:
        # 为root用户设置默认warehouse
        cursor.execute(sql5)
        warehouses = cursor.fetchall()
        print(warehouses)
        if not any(warehouse[0] == 'zztestaudit' for warehouse in warehouses):
            cursor.execute(sql6)
        cursor.execute(sql8)
        print("设置默认warehouse成功")
        # 开启审计功能
        cursor.execute("set global audit_log=1")
        cursor.execute("set global log_output='table'")
        cursor.execute("set global long_query_time=0")
        print('执行开启审计策略')
        # 创建审计策略
        cursor.execute("select name from cloud.audit_policy")
        audit_policys = cursor.fetchall()
        for audit_policy in audit_policys:
            if audit_policy[0] == 'audit_zztest':
                cursor.execute("DROP AUDIT POLICY audit_zztest")
                print('删除审计策略')
        cursor.execute("create audit policy audit_zztest(db='zztestaudit')")
        print('创建审计策略')
        # 清理gclusterdb.audit_log
        cursor.execute("use gclusterdb")
        cursor.execute("show tables")
        gclusterdb_tables = cursor.fetchall()
        for table in gclusterdb_tables:
            if table[0] == 'audit_log':
                cursor.execute("truncate table gclusterdb.audit_log")
                print("执行清理gclusterdb.audit_log")
        # 开始执行sql
        cursor.execute(sql1)
        databases = cursor.fetchall()
        if not any(database[0] == 'zztestaudit' for database in databases):
            cursor.execute(sql2)
            print('执行创建数据库')
        cursor.execute("use zztestaudit")
        cursor.execute(sql3)
        tables = cursor.fetchall()
        for table in tables:
            if table[0] == 'zztest_t1':
                cursor.execute("DROP table zztest_t1")
        # 创建表
        cursor.execute(sql4)
        print('执行创建表')
        # 插入数据
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        try:
            ssh.connect('10.10.55.74', username='gbase', password='gbase')
            insert_table = f"/opt/gccli_install/gcluster/server/bin/gccli -u{config.db_user} -p{config.db_passwd} -h{config.host} -P{config.port} -e \"insert into zztestaudit.zztest_t1(name, age, sex) values('James', 24, 'M'),('David', 28, 'M'),('Emily', 34, 'F')\""
            stdin, stdout, stderr = ssh.exec_command(insert_table)
            output = stdout.read().decode('utf-8')
            print("插入数据标准输出:", output)
            error = stderr.read().decode('utf-8')
            print("插入数据错误输出:", error)
            select_table = f"/opt/gccli_install/gcluster/server/bin/gccli -u{config.db_user} -p{config.db_passwd} -h{config.host} -P{config.port} -e \"select name, age from zztestaudit.zztest_t1 where sex = 'M'\""
            stdin, stdout, stderr = ssh.exec_command(select_table)
            update_table = f"/opt/gccli_install/gcluster/server/bin/gccli -u{config.db_user} -p{config.db_passwd} -h{config.host} -P{config.port} -e \"update zztestaudit.zztest_t1 set age=18  where name = 'Emily'\""
            stdin, stdout, stderr = ssh.exec_command(update_table)
            delete_table = f"/opt/gccli_install/gcluster/server/bin/gccli -u{config.db_user} -p{config.db_passwd} -h{config.host} -P{config.port} -e \"delete from zztestaudit.zztest_t1 where age = 28\""
            stdin, stdout, stderr = ssh.exec_command(delete_table)
        except Exception as e:
            print('paramiko 执行sql语句出现错误',e)
        finally:
            ssh.close()
        # cursor.execute("insert into zztestaudit.zztest_t1(name, age, sex) values('James', 24, 'M'),('David', 28, 'M'),('Emily', 34, 'F')")
        # print('执行插入数据', cursor.fetchall())
        
        # cursor.execute("select name, age from zztestaudit.zztest_t1 where sex = 'M'")
        # print('执行查询数据', cursor.fetchall())
        # cursor.execute("update zztestaudit.zztest_t1 set age=18  where name = 'Emily'")
        # cursor.execute("delete from zztestaudit.zztest_t1 where age = 28")
        # print('执行删除数据', cursor.fetchall())
        # 执行flush 刷新到gclusterdb.audit_log
        sleep(5)
        cursor.execute("flush audit logs force")
        db.commit()

    except Exception as e:
        print(e)
        db.rollback()
    cursor.close()
    db.close()


    print("setup方法")


def end():
    """
    用例执行后需要做的处理
    """


    print("teardown方法")


@pytest.mark.mark_10
@pytest.mark.skipif(condition=config.is_run("test_10_history_sql_detail"), reason='用例下线')
def test_10_history_sql_detail():
    try:    
        # 获取驱动
        driver = config.DRIVER
        # 进入History SQL模块
        WebDriverWait(driver, 30).until(EC.element_to_be_clickable((By.LINK_TEXT, "History SQL"))).click()
        # 获取query id下标
        table_header = driver.find_element(By.XPATH, "//*[@id='app']/section/main/section/main/div/div/div[2]/div/div[2]/table").find_element(By.TAG_NAME, "tr")
        cols = table_header.find_elements(By.TAG_NAME, "th")
        cols_texts = []
        for col in cols:
            cols_texts.append(col.text)
        Queryid_index = cols_texts.index("Query Id")
        # 等待表格加载
        table = WebDriverWait(driver, 30).until(EC.presence_of_all_elements_located((By.XPATH, "//*[@id='app']/section/main/section/main/div/div/div[2]/div/div[3]/table")))
        rows = table[0].find_elements(By.TAG_NAME, "tr")
        row_0_cells = rows[0].find_elements(By.TAG_NAME, "td")
        Queryid = row_0_cells[Queryid_index].text
        row_0_cells[Queryid_index].find_element(By.LINK_TEXT, Queryid).click()
        WebDriverWait(driver, 30).until(EC.presence_of_all_elements_located((By.XPATH, "//*[@id='app']/section/main/section/div[1]/div/div[1]/span")))
        
        print(Queryid)
        detail_table = driver.find_element(By.XPATH, "//*[@id='pane-first']/div/div[1]/div/div[2]/table")
        print(type(detail_table), detail_table)
        detail_table_rows = detail_table.find_elements(By.TAG_NAME, "tr")
        detail_dict = {}
        for detail_table_row in detail_table_rows:
            detail_table_row_cells = detail_table_row.find_elements(By.TAG_NAME, "td")
            detail_dict[detail_table_row_cells[0].text] = detail_table_row_cells[1].text
        sql = driver.find_element(By.XPATH, "//*[@id='pane-first']/div/div[2]/div/textarea").get_attribute('value')
        detail_dict['SQL'] = sql

        print(detail_dict)
        # 连接数据库
        db = conn_sql()
        # 创建游标
        cursor = db.cursor()
        # 执行语句
        sql_dict = {
            'Query Id': '', 'Command': '', 'Status': '', 'Info': '', 'Table List': '', 'Warehouse Name': '', 'Warehouse Size': '', 'User': '', 'Role': '',
            'Start Time': '', 'End Time': '', 'Query Time': '', 'Thread Id': '', 'Client Type': '', 'Client Ip': '', 'Coordinator Ip': '', 'SQL': ''
        }
        
        try:
            sql = f'select id, command, status, info, table_list, warehouse_name, warehouse_size, user, role, start_time, end_time, query_time, thread_id, client_type, client_ip, coor_ip, `sql` from gclusterdb.audit_log where id = {Queryid}'
            cursor.execute(sql)
            results = cursor.fetchall()
            for key, value in zip(sql_dict.keys(), results[0]):
                sql_dict[key] = str(value)
            
            # 处理info
            sql_dict['Info'] = sql_dict['Info'].replace('\n', ' ')
            print(sql_dict)
        except Exception as e:
            print(e)
        finally:
            cursor.close()
            db.close()

        

    except Exception as e:
        print(e)
    warehouse_size = detail_dict['Warehouse Size'].replace('-','').upper()
    if sql_dict['Warehouse Size'] in warehouse_size:
        detail_dict['Warehouse Size'] = sql_dict['Warehouse Size']
    detail_dict['Info'] = detail_dict['Info'].replace(' ', '')
    sql_dict['Info'] = sql_dict['Info'].replace(' ', '')
    detail_dict['Table List'] = detail_dict['Table List'].replace(' ', '')
    sql_dict['Table List'] = sql_dict['Table List'].replace(' ','')
    assert detail_dict == sql_dict, 'history sql detail 显示数据与数据库记录不完全一致'
    print('history sql detail 显示数据与数据库记录完全一致')