path: root/support/packet_generator/ui_verification_deprecated.py

# -*- coding: utf-8 -*-
import os
import sys
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
import time
import configparser
from support.ui_utils.policies.edit_rules import EditRule
from support.packet_generator.workpath import workdir
from support.ui_utils.element_position.log_element_position import *
from support.ui_utils.element_position.map_element_position_library import *
from support.ui_utils.logs.search_log import SearchLog
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.by import By
from datetime import datetime

class UiVerification:
    def ui_verification(self, driver, parameter, test_data, api_verification_result, test_rule_element_position, traffic_result, cfg_list):
        my_driver = driver
        policy_type = test_data["policy_type"]
        rule_action = test_data["rule_action"]
        is_log = parameter["is_log"]
        metric_result = False
        log_result = False
        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Start to verify the effect of the policy rule by ui.", flush=True)
        
        edit_rule = EditRule(my_driver)
        edit_rule_code, _ = edit_rule.edit(test_data["policy_type"], test_data["rule_name"], test_rule_element_position)
        if edit_rule_code == 200:
            if policy_type == "security"  or policy_type == "proxy_intercept" or policy_type == "proxy_manipulation" or policy_type == "monitor" or policy_type == "statistics":
                # 判断api的metric_result为true，只是为了作为命中和生成log的基本判断，无其他特殊意义
                if api_verification_result[2] == True:
                    # 从ui上判断metric_result
                    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Get hit count from ui detail page.", flush=True)
                    hit_count = -1
                    if policy_type == "statistics":  # 如果为Statistics Rule无对应的Log类型，则从Session Records查找
                        hit_count = my_driver.find_element(By.XPATH, statisticsRulePage_rightInfo_72hoursHitCount_posXpath).text
                    if rule_action == "shunt":
                        hit_count = my_driver.find_element(By.XPATH, "//td[normalize-space(text())='Hit Count (72 hours)']/parent::tr//span").text
                    else:
                        hit_count = my_driver.find_element(By.XPATH, "//td[normalize-space(text())='Hit Count (72 hours)']/parent::tr//span[@class='hit-count']").text
                    # print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "hit_count:", hit_count, flush=True)
                    if int(hit_count) == test_data["counters"]["hits"] or int(hit_count) > 0 and test_data["counters"]["hits"] == "many":
                        metric_result = True
                    
                    # 从ui上判断log_result
                    if int(is_log) == 1 and policy_type != "statistics" and rule_action != "shunt":
                        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Get log value from log page.", flush=True)
                        my_driver.find_element(By.XPATH, "//td[normalize-space(text())='Hit Count (72 hours)']/parent::tr//span[@class='hit-count']").click()
                        is_save_element = "//button[contains(@class, 'el-button el-button--default el-button--small el-button--primary operation-confirm-Leave')]"
                        element_exist_flag = my_driver.is_element_exist(is_save_element)
                        if element_exist_flag == True:
                            my_driver.find_element(By.XPATH, is_save_element).click()
                        time.sleep(3)
                        for _ in range(9):    # # 有可能日志出的慢，需要等待3分钟，最大等待日志3分钟
                            log_number_str = my_driver.find_element(By.XPATH, "//div[@class='search-addition-info search-row']//div[1]").text
                            log_number = log_number_str.split(": ")[-1]
                            if int(hit_count) > int(log_number):
                                time.sleep(20)
                                my_driver.back()   # 后退页面到策略页面
                                my_driver.find_element(By.XPATH, "//td[normalize-space(text())='Hit Count (72 hours)']/parent::tr//span[@class='hit-count']", find_before_wait_time=0.5).click()
                                #my_driver.find_element(By.XPATH, "//*[@id='facedQuery']").click()
                                continue
                            break
                        # print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "log_number:", log_number, flush=True)
                        
                        # 如果condition的范围很广，test_data["counters"]["hits"]设置为many，意味着不能进行精准的log判断。如果test_data["counters"]["hits"]不为many，则表示可以准确判断。
                        # 策略非要带着sip，目的是减少更繁琐的判断，可复用下面的流程，如果是negate sip，建议app选择除http和ssl外的协议
                        if test_data["counters"]["hits"] != "many":
                            # 等于0，则是bug，不需要进一步判断，直接返回false
                            if int(hit_count) == 0 or int(log_number) == 0:
                                if len(test_data['log_query_param']) == 0:
                                    log_result = "no_set"
                                else:
                                    log_result = False
                                    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "The value of hit_count or log_number is 0. Maybe it's a bug.", flush=True)
                            elif int(hit_count) == int(log_number) and 0 < int(hit_count) <= 2:
                                # ui日志校验，以server ip为例，在log_query_list.ini中通过log api查找log ui的值
                                parse = configparser.ConfigParser()
                                parse_dir = os.path.join(workdir, "support", "packet_generator", "log_query_list.ini")
                                parse.read(parse_dir, encoding="utf-8")
                                temp_log_result_list = []
                                # 点击展开
                                time.sleep(2)
                                my_driver.find_element(By.XPATH, "//i[@class='vxe-table--expand-btn vxe-icon-arrow-right']").click()
                                for t in range(len(test_data["log_query_param"])):
                                    log_query_key = test_data["log_query_param"][t]["query_field_key"]
                                    log_str = parse.get("log", log_query_key)
                                    if log_query_key == "server_ip" or log_query_key == "decoded_as" or log_query_key == "client_ip":
                                        log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(replaceValue=log_str)
                                        temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                                        if temp_value == test_data["log_query_param"][t]["query_value"]:
                                            temp_log_result = True
                                        else:
                                            temp_log_result = False
                                        temp_log_result_list.append(temp_log_result)
                                    # elif log_query_key == "ssl_cn" or log_query_key == "ssl_san" or log_query_key == "ssl_sni" or log_query_key == "ftp_account" or log_query_key == "security_action" or log_query_key == "ftp_url" or log_query_key == "ftp_link_type":
                                    elif log_query_key in ["ssl_cn", "ssl_san", "ssl_sni", "ftp_account", "security_action", "ftp_url", "ftp_link_type", "mail_account", "mail_from","mail_attachment_name","imsi", "phone_number", "server_fqdn", "ip_protocol", "mail_to", "mail_to_cmd", "mail_from_cmd", "mail_subject", "dns_qname"]:
                                        log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(replaceValue=log_str)
                                        # 当不存在此定位时,value为空,日志查询与预期不符,置为false
                                        log_item_pos_exist = my_driver.is_element_exist(log_item_pos_xpath)
                                        if log_item_pos_exist == True:
                                            temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                                            if temp_value == test_data["log_query_param"][t]["query_value"]:
                                                temp_log_result = True
                                            else:
                                                temp_log_result = False
                                        else:
                                            temp_log_result = False
                                        temp_log_result_list.append(temp_log_result)
                                    elif log_query_key  in {"packet_capture_file","http_response_body,http_request_body"}:   #日志中非结构文件断言
                                        log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(replaceValue=log_str)
                                        element_exist_flag = my_driver.is_element_exist(log_item_pos_xpath)
                                        # 验证开关开启
                                        if test_data["log_query_param"][t]["query_value"] == True:
                                            # 有下载按钮，点击下载按钮，验证是否报错
                                            if element_exist_flag == True:
                                                # 连续点击下载按钮5次（防止下载异常的时候异常信息消失的太快）
                                                for _ in range(5):
                                                    log_item_element = my_driver.find_element(By.XPATH,log_item_pos_xpath)
                                                    my_driver.execute_script("arguments[0].scrollIntoView();", log_item_element)
                                                    my_driver.find_element(By.XPATH, log_item_pos_xpath).click()
                                                    time.sleep(1)
                                                log_item_element = my_driver.find_element(By.XPATH, log_item_pos_xpath)
                                                my_driver.execute_script("arguments[0].scrollIntoView();", log_item_element)
                                                my_driver.find_element(By.XPATH, log_item_pos_xpath).click()
                                                time.sleep(1)
                                                # 只有下载失败的时候界面才有此定位
                                                error_text_posXpath = "//div[contains(@class, 'el-message el-message--error')]"
                                                error_text_is_exist = my_driver.is_element_exist(error_text_posXpath)
                                                # 有错误提示，下载失败
                                                if error_text_is_exist == True:
                                                    temp_log_result = False
                                                # 下载成功时无错误提示.
                                                else:
                                                    temp_log_result = True
                                            else:
                                                # 没有下载按钮，false
                                                temp_log_result = False
                                        # 验证开关不开启
                                        elif test_data["log_query_param"][t]["query_value"] == False:
                                            if element_exist_flag == False:
                                                temp_log_result = True
                                            else:
                                                temp_log_result = False
                                    elif log_query_key in ["http_host", "http_url", "http_response_content_type", "http_user_agent", "http_version", "http_request_line", "http_response_line", "http_cookie", "http_set_cookie"]:
                                        log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]".format(replaceValue=log_str)
                                        temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text.strip()
                                        #print("实际值:{}".format(temp_value))
                                        #print("预期值:{}".format(test_data["log_query_param"][t]["query_value"]))
                                        if temp_value == test_data["log_query_param"][t]["query_value"]:
                                            temp_log_result = True
                                        else:
                                            temp_log_result = False
                                        temp_log_result_list.append(temp_log_result)
                                    elif log_query_key in ["proxy_pinning_status", "proxy_intercept_status", "proxy_passthrough_reason"]:
                                        log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]".format(replaceValue=log_str)
                                        temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text.strip()
                                        #print("实际值:{}".format(temp_value))
                                        #print("预期值:{}".format(test_data["log_query_param"][t]["query_value"]))
                                        if temp_value == test_data["log_query_param"][t]["query_value"]:
                                            temp_log_result = True
                                        else:
                                            temp_log_result = False
                                        temp_log_result_list.append(temp_log_result)
                                    elif log_query_key in ["server_port"]:
                                        log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(
                                            replaceValue=log_str)
                                        temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text.strip()
                                        if int(temp_value.replace(',', '')) == int(test_data["log_query_param"][t]["query_value"]) :
                                            temp_log_result = True
                                        else:
                                            temp_log_result = False
                                        temp_log_result_list.append(temp_log_result)
                                    elif log_query_key == "monitor_mirrored_pkts":
                                            log_pkt_value = test_data["log_query_param"][t]["query_value"]
                                            log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(replaceValue=log_str)
                                            log_pkt_temp = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                                            temp_value = int(log_pkt_temp.replace(",", ""))
                                            if temp_value == log_pkt_value:
                                                temp_log_result = True
                                            else:
                                                temp_log_result = False
                                            temp_log_result_list.append(temp_log_result)
                                    elif log_query_key == "monitor_mirrored_bytes":
                                            log_byte_value = test_data["log_query_param"][t]["query_value"]
                                            log_bytes_temp = "" #对于bytes，之后还要进行处理
                                            if 1024 <= log_byte_value < 1048576:
                                                log_bytes_temp = str(round(log_byte_value / 1024.0, 2)) + " KB"
                                            elif log_byte_value < 1024:
                                                log_bytes_temp = str(log_byte_value) + " B"
                                            elif 1048576 <= log_byte_value < 1073741824:
                                                log_bytes_temp = str(round(log_byte_value / 1048576.0, 2)) + " MB"
                                            log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(
                                                replaceValue=log_str)
                                            temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                                            if temp_value == log_bytes_temp:
                                                temp_log_result = True
                                            else:
                                                temp_log_result = False
                                            temp_log_result_list.append(temp_log_result)
                                    elif log_query_key in ["client_port"]:
                                        log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(
                                            replaceValue=log_str)
                                        temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text.strip()
                                        temp_value = int(temp_value.replace(',', ''))
                                        # test_data中的client_port为string类型时,进行port的范围验证
                                        if type(test_data["log_query_param"][t]["query_value"]) == str:
                                            start, end = map(int, test_data["log_query_param"][t]["query_value"].split('-'))
                                            if start <= temp_value <= end:
                                                temp_log_result = True
                                            else:
                                                temp_log_result = False
                                        temp_log_result_list.append(temp_log_result)
                                if False not in temp_log_result_list:
                                    log_result = True
                                else:
                                    log_result = False
                            elif int(hit_count) > int(log_number):
                                print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "hit count > log number,need check.")
                        else:
                            # 不能直接把log_result赋值，也要判断log，目前值判断第一条日志
                            log_result = True
                    elif int(is_log) == 1 and policy_type == "statistics":
                        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Get log value from session records page.", flush=True)
                        element_position_map = get_element_position("session_log")
                        search_log = SearchLog(driver)
                        sql = "has(statistics_rule_list,{})".format(test_data["rule_id"])
                        search_rule_code, total = search_log.search(sql, element_position_map)
                        if search_rule_code == 200:
                            log_number = total
                            log_result = True
                    elif int(is_log) == 1 and rule_action == "shunt":
                        log_result = "no_set"
                
                if metric_result == True:
                    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "metric_result" at ui page is {}.'.format(metric_result), flush=True)
                elif metric_result == False:
                    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "metric_result" at ui page is {}.'.format(metric_result), flush=True)

                if log_result == True:
                    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "log_result" at ui page is {}.'.format(log_result), flush=True)
                elif log_result == False:
                    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "log_result" at ui page is {}.'.format(log_result), flush=True)
            
                return api_verification_result[0], log_result, metric_result, my_driver
            elif policy_type == "service_chaining":
                print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Get value of sent pkts, received pkts, sent bytes, received bytes from UI detail page.", flush=True)
                time.sleep(4)
                sent_pkts_str = my_driver.find_element(By.XPATH, "//td[normalize-space(text())='Sent Packets']/parent::tr//td[2]").text
                if sent_pkts_str == "-":
                    sent_pkts = 0
                else:
                    sent_pkts, sent_pkts_unit = self.extract_value("packets", sent_pkts_str)
                
                sent_bytes_str = my_driver.find_element(By.XPATH, "//td[normalize-space(text())='Sent Bytes']/parent::tr//td[2]").text
                if sent_bytes_str == "-":
                    sent_bytes = 0
                else:
                    sent_bytes, sent_bytes_unit = self.extract_value("bytes", sent_bytes_str)
                   
                received_pkts_str = my_driver.find_element(By.XPATH, "//td[normalize-space(text())='Received Packets']/parent::tr//td[2]").text
                if received_pkts_str == "-":
                    received_pkts = 0
                else:
                    received_pkts, received_pkts_unit = self.extract_value("packets", received_pkts_str)
                
                received_bytes_str = my_driver.find_element(By.XPATH, "//td[normalize-space(text())='Received Bytes']/parent::tr//td[2]").text
                if received_bytes_str == "-":
                    received_bytes = 0
                else:
                    received_bytes, received_bytes_unit = self.extract_value("bytes", received_bytes_str)

                parse = configparser.ConfigParser()
                parse_dir = os.path.join(workdir, "configuration_file.ini")
                parse.read(parse_dir, encoding="utf-8")
                ip_str = parse.get("sc_active_dst_ip", "ip_list")
                sc_active_ip_list = ip_str.split(",")
                effective_device_tag = ["group-xxg-tsgx","center-xxg-tsgx"]
                # traffic_result = {'total_packets': 175, 'total_packets_sent': 45, 'total_packets_received': 130, 'total_bytes': 180025, 'total_bytes_sent': 5799, 'total_bytes_received': 174226, 'total_syn_pkt': 1}
                statistics_sc_info_list = cfg_list["statistics_sc_info_list"]
                for sc_info in statistics_sc_info_list:
                    if len(traffic_result) > 0:
                        if sent_pkts == 0 and received_pkts == 0:
                            if sc_info["sf_method"] == "vxlan_g":
                                if sc_info["sf_dest_ip"] not in sc_active_ip_list:
                                    metric_result = True
                                elif sc_info["sf_dest_ip"] in sc_active_ip_list and sc_info["admin_status"] == 0:
                                    metric_result = True
                                elif sc_info["device_tag"] not in effective_device_tag and sc_info["load_balance_localization"] == "nearby":
                                    metric_result = True
                                elif sc_info["is_negate"] == 1:
                                    metric_result = True    
                                else:
                                    metric_result = False
                            elif sc_info["sf_method"] == "layer2_switch":
                                if sc_info["admin_status"] == 0:
                                    metric_result = True
                                elif sc_info["device_tag"] not in effective_device_tag and sc_info["load_balance_localization"] == "nearby":
                                    metric_result = True
                                elif sc_info["is_negate"] == 1:
                                    metric_result = True
                                else:
                                    metric_result = False
                        else:
                            if sc_info["targeted_traffic"] == "raw":
                                byte_value = traffic_result["total_bytes"]
                                pkt_value = traffic_result["total_packets"]
                                if sc_info["sf_method"] == "vxlan_g" and sc_info["sf_dest_ip"] in sc_active_ip_list:
                                    if 1000 <= pkt_value < 10000:
                                        pkt_value_temp = round(pkt_value/1000, 2)
                                    elif pkt_value < 1000:
                                        pkt_value_temp = pkt_value
                                    if 1024 <= byte_value < 1048576:
                                        total_bytes_temp = round(byte_value/1024.0, 2)
                                    elif byte_value < 1024:
                                        total_bytes_temp = byte_value
                                    elif 1048576 <= byte_value < 1073741824:
                                        total_bytes_temp = round(byte_value/1048576.0, 2)
                                    if sc_info["type"] == 1:
                                        if sent_bytes == received_bytes == total_bytes_temp and sent_pkts == received_pkts == pkt_value_temp:
                                            metric_result = True
                                        else:
                                            metric_result = False
                                    elif sc_info["type"] == 2:
                                        if sent_bytes == total_bytes_temp and sent_pkts == pkt_value_temp and received_bytes == received_pkts == 0:
                                            metric_result = True
                                        else:
                                            metric_result = False
                                    else:
                                        metric_result = False
                                elif sc_info["sf_method"] == "layer2_switch":
                                    if 1000 <= pkt_value < 10000:
                                        pkt_value_temp = round(pkt_value/1000, 2)
                                    elif pkt_value < 1000: 
                                        pkt_value_temp = pkt_value
                                    if 1024 <= byte_value < 1048576:
                                        total_bytes_temp = round(byte_value/1024.0, 2)
                                    elif byte_value < 1024:
                                        total_bytes_temp = byte_value
                                    elif 1048576 <= byte_value < 1073741824:
                                        total_bytes_temp = round(byte_value/1048576.0, 2)
                                    if sc_info["type"] == 2:
                                        if sent_bytes == total_bytes_temp and sent_pkts == pkt_value_temp and received_bytes == received_pkts == 0:
                                            metric_result = True
                                        else:
                                            metric_result = False
                                    else:
                                        metric_result = False
                            elif sc_info["targeted_traffic"] == "decrypted":
                                if sc_info["sf_method"] == "vxlan_g" and sc_info["sf_dest_ip"] in sc_active_ip_list:
                                    if sc_info["type"] == 1:
                                        if sent_bytes == received_bytes and sent_pkts == received_pkts:
                                            metric_result = True
                                        else:
                                            metric_result = False
                                    elif sc_info["type"] == 2:
                                        if sent_bytes != 0 and sent_pkts != 0 and received_bytes == received_pkts == 0:
                                            metric_result = True
                                        else:
                                            metric_result = False
                                    else:
                                        metric_result = False
                                elif sc_info["sf_method"] == "layer2_switch":
                                    if sc_info["type"] == 2:
                                        if sent_bytes != 0 and sent_pkts != 0 and received_bytes == received_pkts == 0:
                                            metric_result = True
                                        else:
                                            metric_result = False
                                    else:
                                        metric_result = False   
                    if metric_result == True:
                        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "metric_result" at ui page is True', flush=True)  
                    else:
                        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "metric_result" at ui page is False', flush=True)      
                    if int(is_log) == 1:
                        # 跳转至session record路径
                        sc_rule_id = int(test_data["rule_id"])
                        my_driver.find_element(By.ID,"Log").click()
                        time.sleep(1)
                        my_driver.find_element(By.ID,"Log_SesssionRecords").click()
                        time.sleep(1.5)
                        # 以policy id作为filter查找命中策略的session
                        my_driver.find_element(By.ID,"addFilter").click()
                        time.sleep(1)
                        my_driver.find_element(By.XPATH,"//div[@class='el-input el-input--small el-input--suffix is-focus']//input").click()
                        my_driver.find_element(By.XPATH,"//div[@class='el-input el-input--small el-input--suffix is-focus']//input").send_keys("service")
                        time.sleep(1.5)
                        my_driver.find_element(By.ID,"Service Chaining Rule List26-_Label_UnrecognizedExp_Expression_groupItem_TagsSearch_FacedtedSearch_Search_VPanel_SessionLog_Home_App_anonymousComponent").click()
                        my_driver.find_element(By.ID,"operatorhas").click()
                        time.sleep(1)
                        my_driver.find_element(By.XPATH,"//div[@class='el-tooltip el-input el-input--mini']//input").send_keys(sc_rule_id)
                        btn = my_driver.find_element(By.ID,"facedQuery")
                        my_driver.execute_script("arguments[0].click()", btn)
                        time.sleep(1.5)
                        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Get log value from ui log page.", flush=True)
                        for i in range(50) :
                            log_number_str = my_driver.find_element(By.XPATH, "//div[@class='search-addition-info search-row']//div[1]").text
                            log_number = log_number_str.split(": ")[-1]
                            if int(log_number) == 0:
                                i += 1
                                my_driver.execute_script("arguments[0].click()", btn)
                                time.sleep(7)
                            elif int(log_number) > 0:
                                break
                        if int(log_number) == 0:
                            if sc_info["is_negate"] == 1:
                                log_result = True
                            else:
                                log_result = False
                            print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Fail to get log value from log page.", flush=True)
                        else:
                            time.sleep(0.5)
                            # 点击展开
                            btn = my_driver.find_element(By.XPATH, "//i[@class='vxe-table--expand-btn vxe-icon-arrow-right']")
                            my_driver.execute_script("arguments[0].click()", btn)
                            # my_driver.find_element(By.XPATH, "//i[@class='vxe-table--expand-btn vxe-icon-arrow-right']").click()
                            time.sleep(2)
                            parse = configparser.ConfigParser()
                            parse_dir = os.path.join(workdir, "support", "packet_generator", "log_query_list.ini")
                            parse.read(parse_dir, encoding="utf-8")
                            temp_log_result_list = []
                            for t in range(len(test_data["log_query_param"])):
                                log_query_key = test_data["log_query_param"][t]["query_field_key"]
                                if log_query_key == "sc_rsp_raw" or log_query_key == "sc_rsp_decrypted":
                                    test_data["log_query_param"][t]["query_value"] = sc_info["sf_id_list"]
                                elif log_query_key == "sent_pkts":
                                    test_data["log_query_param"][t]["query_value"] = traffic_result["total_packets_sent"]
                                elif log_query_key == "received_pkts":
                                    test_data["log_query_param"][t]["query_value"] = traffic_result["total_packets_received"]
                                elif log_query_key == "sent_bytes":
                                    test_data["log_query_param"][t]["query_value"] = traffic_result["total_bytes_sent"]
                                elif log_query_key == "received_bytes":
                                    test_data["log_query_param"][t]["query_value"] = traffic_result["total_bytes_received"] 
                                log_str = parse.get("log", log_query_key)
                                if log_query_key == "sc_rsp_raw" or log_query_key == "sc_rsp_decrypted":
                                    log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue1}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span \
                                        | //div[normalize-space(text())='{replaceValue2}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//div[@style='white-space: nowrap;']".format(replaceValue1=log_str,replaceValue2=log_str)
                                    temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                                    if temp_value == "[]":
                                        if sc_info["sf_method"] == "vxlan_g":
                                            if sc_info["sf_dest_ip"] not in sc_active_ip_list:
                                                temp_log_result = True
                                            elif sc_info["sf_dest_ip"] in sc_active_ip_list and sc_info["admin_status"] == 0:
                                                temp_log_result = True
                                            elif sc_info["device_tag"] not in effective_device_tag and sc_info["load_balance_localization"] == "nearby":
                                                temp_log_result = True
                                            else:
                                                temp_log_result = False
                                        elif sc_info["sf_method"] == "layer2_switch":
                                            if sc_info["admin_status"] == 0:
                                                temp_log_result = True
                                            elif sc_info["device_tag"] not in effective_device_tag and sc_info["load_balance_localization"] == "nearby":
                                                temp_log_result = True
                                            else:
                                                temp_log_result = False 
                                    else:
                                        temp_value = temp_value.strip('[] ')   
                                        temp_value_list = temp_value.split(',')
                                        log_value_list = []
                                        for item in temp_value_list:
                                            log_value_list.append(int(item))
                                        if log_value_list == test_data["log_query_param"][t]["query_value"]:
                                            temp_log_result = True
                                        else:
                                            temp_log_result = False
                                elif log_query_key == "sent_pkts" or log_query_key == "received_pkts":
                                    log_pkt_value = test_data["log_query_param"][t]["query_value"]
                                    log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(replaceValue=log_str)
                                    log_pkt_temp = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                                    temp_value = int(log_pkt_temp.replace(",",""))
                                    if temp_value == log_pkt_value:
                                        temp_log_result = True
                                    else:
                                        temp_log_result = False
                                elif log_query_key == "sent_bytes" or log_query_key == "received_bytes":
                                    log_byte_value = test_data["log_query_param"][t]["query_value"]
                                    if 1024 <= log_byte_value < 1048576:
                                        log_bytes_temp = str(round(log_byte_value/1024.0, 2)) + " KB"
                                    elif log_byte_value < 1024:
                                        log_bytes_temp = str(log_byte_value) + " B"
                                    elif 1048576 <= log_byte_value < 1073741824:
                                        log_bytes_temp = str(round(log_byte_value/1048576.0, 2)) + " MB"
                                    log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(replaceValue=log_str)
                                    temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                                    if temp_value == log_bytes_temp:
                                        temp_log_result = True
                                    else:
                                        temp_log_result = False
                                elif log_query_key in ["ssl_esni_flag", "ssl_ech_flag", "ssl_sni", "ip_protocol", "imei", "imsi", "apn", "phone_number", "subscriber_id"]:
                                    log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(replaceValue=log_str)
                                    temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                                    if temp_value == test_data["log_query_param"][t]["query_value"]:
                                        temp_log_result = True
                                    else:
                                        temp_log_result = False
                                temp_log_result_list.append(temp_log_result)
                            if False not in temp_log_result_list:
                                    log_result = True
                            else:
                                    log_result = False
                        if log_result == True:
                            print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "log_result" at ui page is True', flush=True)    
                        else:
                            print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "log_result" at ui page is False', flush=True)
                return api_verification_result[0], log_result, metric_result, my_driver
            elif policy_type == "dos_protection":
                if api_verification_result[2] == True:
                    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Get hit count from ui detail page.", flush=True)
                    if int(is_log) == 1:
                        hit_count = my_driver.find_element(By.XPATH, "//td[normalize-space(text())='Hit Count (72 hours)']/following-sibling::td").text
                        my_driver.find_element(By.ID, "Log").click()
                        my_driver.find_element(By.ID, "Log_DosEventLogs").click()
                        time.sleep(1)
                        my_driver.find_element(By.ID, "addFilter").click()
                        my_driver.find_element(By.XPATH, "//div[@class='el-input el-input--small el-input--suffix is-focus']//input").click()
                        my_driver.find_element(By.XPATH, "//div[@class='el-input el-input--small el-input--suffix is-focus']//input").send_keys("destination")
                        my_driver.find_element(By.ID, "Destination IP8-_Label_UnrecognizedExp_Expression_groupItem_TagsSearch_FacedtedSearch_Search_VPanel_DosLog_Home_App_anonymousComponent").click()
                        my_driver.find_element(By.ID, "operator=").click()
                        my_driver.find_element(By.XPATH, "//div[@class='el-tooltip el-input el-input--mini']//input").send_keys(test_data['traffic']['servers_start_ip'])
                        btn = my_driver.find_element(By.ID, "facedQuery")
                        my_driver.execute_script("arguments[0].click()", btn)

                        #切换时间范围:仅调试代码时使用
                        my_driver.find_element(By.XPATH,"//p[normalize-space(text())='Last 1 hour']").click()
                        my_driver.find_element(By.XPATH,"//div[normalize-space(text())='Last 6 hours']").click()
                        
                        log_number_str = my_driver.find_element(By.XPATH, "//div[@class='search-addition-info search-row']//div[1]").text
                        log_number = log_number_str.split(": ")[-1]
                        if (int(hit_count) == test_data["counters"]["hits"])or (test_data["counters"]["hits"]== "many" and int(log_number) )!= 0:
                            metric_result = True
                        else:
                            metric_result = False
                        
                        parse = configparser.ConfigParser()
                        parse_dir = os.path.join(workdir, "support", "packet_generator", "log_query_list.ini")
                        parse.read(parse_dir, encoding="utf-8")
                        if int(log_number) == 2: #暂未添加tcp_syn_flood的校验
                            expand_element_postion = "//i[@class='vxe-table--expand-btn vxe-icon-arrow-right']"
                            log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(replaceValue="Attack Type")
                            temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                            if temp_value == "Custom Network Attack":
                                query_key_list = ["attack_type", "sessions", "session_rate", "packets", "packet_rate", "bytes", "bit_rate"]
                            elif temp_value == "TCP SYN Flood" or temp_value == "UDP Flood" or temp_value == "DNS Flood" or temp_value == "NTP Flood":
                                query_key_list = ["basic_attack_type", "basic_sessions", "basic_session_rate", "basic_packets", "basic_packet_rate", "basic_bytes", "basic_bit_rate"]
                            temp_log_result_list = self.verify_dos_events(my_driver, expand_element_postion, parse, test_data, query_key_list)
                        elif int(log_number) == 1:
                            query_key_list_1 = ["attack_type", "sessions", "session_rate", "packets", "packet_rate", "bytes", "bit_rate","rule_id","conditions","destination_ip"]
                            expand_element_postion_1 = "//tr[1]//i[@class='vxe-table--expand-btn vxe-icon-arrow-right']"
                            temp_log_result_list = self.verify_dos_events(my_driver, expand_element_postion_1, parse, test_data, query_key_list_1)
                            # query_key_list_2 = ["basic_attack_type", "basic_sessions", "basic_session_rate", "basic_packets", "basic_packet_rate", "basic_bytes", "basic_bit_rate"]
                            # expand_element_postion_2 = "//tr[2]//i[@class='vxe-table--expand-btn vxe-icon-arrow-right']"
                            # temp_log_result_list_2 = self.verify_dos_events(my_driver, expand_element_postion_2, parse, test_data, query_key_list_2)
                            # temp_log_result_list = temp_log_result_list_1 + temp_log_result_list_2
                        if False not in temp_log_result_list:
                            log_result = True
                        else:
                            log_result = False
                        
                        if metric_result == True:
                            print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "metric_result" at ui page is {}.'.format(metric_result), flush=True)
                        elif metric_result == False:
                            print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "metric_result" at ui page is {}.'.format(metric_result), flush=True)

                        if log_result == True:
                            print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "log_result" at ui page is {}.'.format(log_result), flush=True)
                        elif log_result == False:
                            print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The value of parameter "log_result" at ui page is {}.'.format(log_result), flush=True)
                        
                        return True, log_result, metric_result, my_driver
    
    def extract_value(self, type, value):
        if type == "packets":
            if " " in value:
                value_number = float(value.split(" ")[0])
                value_unit = value.split(" ")[-1]
            else:
                value_number = int(value)
                value_unit = ""
        elif type == "bytes":
            if " " in value:
                value_unit = value.split(" ")[-1]
                if value_unit == "B":
                    value_number = int(value.split(" ")[0])
                else:
                    value_number = float(value.split(" ")[0])
            else:
                value_number = 0
                value_unit = ""
        return value_number, value_unit
    
    def verify_dos_events(self, my_driver, expand_element_postion, parse, test_data, query_key_list):
        log_result_list = []
        my_driver.find_element(By.XPATH, expand_element_postion).click()
        for i in range(len(test_data["log_query_param"])):
            log_query_key = test_data["log_query_param"][i]["query_field_key"]
            if log_query_key in query_key_list:
                log_str = parse.get("log", log_query_key)
                log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(replaceValue=log_str)
                temp_value = my_driver.find_element(By.XPATH, log_item_pos_xpath).text
                log_query_value = test_data["log_query_param"][i]["query_value"]
                if log_query_key == "attack_type" or log_query_key == "basic_attack_type":
                    if log_query_value.replace("_", " ") == temp_value.lower():
                        log_result_list.append(True)
                    else:
                        log_result_list.append(False)
                elif log_query_key == "bit_rate" or log_query_key == "basic_bit_rate":
                    log_str = parse.get("log", log_query_key)
                    api_bit_rate_value = test_data["log_query_param"][i]["query_value"]

                    if api_bit_rate_value < 1024:
                        api_bit_rate_value = f"{round(api_bit_rate_value, 0)} b"
                    elif 1024 <= api_bit_rate_value < 1048576:
                        api_bit_rate_value = f"{round(api_bit_rate_value / 1000, 2)} K"
                    elif 1048576 <= api_bit_rate_value < 1073741824:
                        api_bit_rate_value = f"{round(api_bit_rate_value / 1048576, 2)} M"
                    else:
                        api_bit_rate_value = f"{round(api_bit_rate_value / 1073741824, 2)} G"

                    if api_bit_rate_value == temp_value: #String类型的比较
                        log_result_list.append(True)
                    else:
                        log_result_list.append(False)
                elif log_query_key in {"source_ip_list"}:  # 日志中非结构文件断言
                    log_item_pos_xpath = "//div[normalize-space(text())='{replaceValue}']/ancestor::div[contains(@class, 'FieldItem')]//div[contains(@class, 'field-value')]//span".format(
                        replaceValue=log_str)
                    element_exist_flag = my_driver.is_element_exist(log_item_pos_xpath)
                    # 验证开关开启
                    if len(test_data["log_query_param"][i]["query_value"]) != 0 :
                        # 有下载按钮，点击下载按钮，验证是否报错
                        if element_exist_flag == True:
                            # 连续点击下载按钮5次（防止下载异常的时候异常信息消失的太快）
                            for _ in range(5):
                                log_item_element = my_driver.find_element(By.XPATH, log_item_pos_xpath)
                                my_driver.execute_script("arguments[0].scrollIntoView();", log_item_element)
                                my_driver.find_element(By.XPATH, log_item_pos_xpath).click()
                                time.sleep(1)
                            log_item_element = my_driver.find_element(By.XPATH, log_item_pos_xpath)
                            my_driver.execute_script("arguments[0].scrollIntoView();", log_item_element)
                            my_driver.find_element(By.XPATH, log_item_pos_xpath).click()
                            time.sleep(1)
                            # 只有下载失败的时候界面才有此定位
                            error_text_posXpath = "//div[contains(@class, 'el-message el-message--error')]"
                            error_text_is_exist = my_driver.is_element_exist(error_text_posXpath)
                            # 有错误提示，下载失败
                            if error_text_is_exist == True:
                                temp_log_result = False
                            # 下载成功时无错误提示.
                            else:
                                temp_log_result = True
                        else:
                            # 没有下载按钮，false
                            temp_log_result = False
                    # 验证开关不开启
                    elif test_data["log_query_param"][i]["query_value"]:
                        if element_exist_flag == False:
                            temp_log_result = True
                        else:
                            temp_log_result = False
                    log_result_list.append(temp_log_result)
                else:
                    if temp_value == str(log_query_value):
                        log_result_list.append(True)
                    else:
                        log_result_list.append(False)

        return log_result_list
    
if __name__ == "__main__":
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "test")