read_excel.py

# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName:   test62
# FileName:      read_excel.py
# Author:       lao_zhao
# Datetime:     2024/9/4 14:53
# Description:
# 
# ---------------------------------------------------------------------------
import openpyxl
from apiAutoTest_v3 import log
from apiAutoTest_v3.common.read_json import read_json
from apiAutoTest_v3.common.read_project_ini import ReadProjectIni
from apiAutoTest_v3.common.read_user_ini import ReadUserIni
from apiAutoTest_v3.data_config.settings import *


# read_excel.py文件的功能：就是获取每个用户的测试数据，并将测试数据存放在一个二维列表中


class ReadExcel:
    def __init__(self, user_dir_name):
        """获取所有json文件的路径，并读取json文件，再获取excel文件的路径，加载excel的工作簿，并获取工作表的名称，再获取工作表"""
        # 被测系统的域名存放在项目的ini配置文件中，所有需要ReadProjectIni对象，调用get_host方法获取被测系统的域名
        self.pro_ini = ReadProjectIni()
        # 先读取每个用户的配置文件的对象。ReadUserIni对象。因为每个用户的数据存放的文件信息放在每个用户的ini配置文件中。
        user_ini = ReadUserIni(user_dir_name)
        # 根据ReadUserIni对象获取每个用户的数据文件路径
        case_data_path = user_ini.get_file_path(FILE_CASE)
        expect_data_path = user_ini.get_file_path(FILE_EXPECT)
        sql_data_path = user_ini.get_file_path(FILE_SQL)

        excel_path = user_ini.get_file_path(FILE_EXCEL)
        table_name = user_ini.get_table_name(TABLE_kEY)

        self.case_data_dict = read_json(case_data_path)
        self.expect_data_dict = read_json(expect_data_path)
        self.sql_data_dict = read_json(sql_data_path)

        try:
            wb = openpyxl.load_workbook(excel_path)
            self.ws = wb[table_name]
        except Exception as e:
            log.error(f"加载excel的工作簿或获取excel的工作表时产生错误，错误为：{e}, excel的路径为：{excel_path}, 工作表的名称为：{table_name}")
            raise e

    def __get_cell_value(self, column, row):
        """根据列号和行号获取指定单元格数据"""
        try:
            cell_value = self.ws[column+str(row)].value
        except Exception as e:
            log.error(f"获取指定单元格数据，出错，错误为：{e}, 传入的列号：{column}和行号：{row}")
            raise e
        else:
            if cell_value is None:
                return None
            elif cell_value.strip():
                return cell_value.strip()

    def module_name(self, row):
        """根据行号，获取模块名称"""
        return self.__get_cell_value(MODULE, row)

    def api_name(self, row):
        """根据行号，获取接口名称"""
        return self.__get_cell_value(API, row)

    def case_url(self, row):
        """根据行号，获取用例的url"""
        path = self.__get_cell_value(URL, row)
        if path:
            # 被测系统的域名存放在项目的ini配置文件中，所有需要ReadProjectIni对象，调用get_host方法获取被测系统的域名
            return self.pro_ini.get_host(HOST_KEY) + path

    def case_req_method(self, row):
        """根据行号，获取用例的请求方法"""
        return self.__get_cell_value(REQ_METHOD, row)

    def case_mime(self, row):
        """根据行号，获取用例请求的媒体类型"""
        mime = self.__get_cell_value(MIME, row)
        if mime:
            return mime.lower()

    def case_data(self, row):
        """根据行号，获取用例数据"""
        case_data_key = self.__get_cell_value(CASE_DATA, row)
        if case_data_key:
            module_name = self.module_name(row)
            api_name = self.api_name(row)
            return self.case_data_dict[module_name][api_name][case_data_key]

    def expect_data(self, row):
        """根据行号，获取期望数据"""
        expect_data_key = self.__get_cell_value(EXPECT_DATA, row)
        if expect_data_key:
            module_name = self.module_name(row)
            api_name = self.api_name(row)
            return self.expect_data_dict[module_name][api_name][expect_data_key]

    def sql_data(self, row):
        """根据行号，获取sql语句"""
        sql_data_key = self.__get_cell_value(SQL_DATA, row)
        if sql_data_key:
            module_name = self.module_name(row)
            api_name = self.api_name(row)
            return self.sql_data_dict[module_name][api_name][sql_data_key]

    def sql_type(self, row):
        """根据行号，获取sql语句的类型"""
        value = self.__get_cell_value(SQL_TYPE, row)
        if value:
            return value.lower()

    def update_key(self, row):
        """根号行号，获取更新的key"""
        return self.__get_cell_value(UPDATE_KEY, row)

    def case_title(self, row):
        """根据行号，获取用例的标题"""
        return self.__get_cell_value(TITLE, row)

    def case_level(self, row):
        """根号行号，获取用例的等级"""
        return self.__get_cell_value(LEVEL, row)

    def get_data(self):
        """获取所有的测试数据，并存放在一个二维列表中"""
        list_data = []
        for row in range(2, self.ws.max_row+1):
            module_name = self.module_name(row)
            api_name = self.api_name(row)
            title = self.case_title(row)
            level = self.case_level(row)
            url = self.case_url(row)
            method = self.case_req_method(row)
            mime = self.case_mime(row)
            case_data = self.case_data(row)
            expect_data = self.expect_data(row)
            sql_type = self.sql_type(row)
            sql_data = self.sql_data(row)
            update_key = self.update_key(row)
            if url is not None and method is not None and expect_data is not None:
                list_data.append([module_name, api_name, level, title, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key])
        else:
            return list_data


if __name__ == '__main__':
    excel = ReadExcel("lao_zhang")
    print(excel.get_data())