AutoFramework/Runner/API/DefaultRunner.py

import unittest
from Business.Class.HTMLTestRunner import HTMLTestRunner
from Business.Class.ExcelUtils import *
from Business.Class.JsonOrYaml import *
from Base.Class.Http import *
from Base.Class.Database import *
from Base.Class.Logger import *
log = Logger(name='test', level='DEBUG')

class TestCase:
    def __init__(self, workbook, data='数据配置', case='测试用例', envs='测试环境', level_list=None):
        from pandas.core.frame import DataFrame
        self._g = {}
        self._l = {}
        self.case = case
        self.envs = envs
        self.http = Request
        init = Excel().open(workbook, read_only=True)
        init_data = auto_decode(init.select(data).cellGet(0, 0))
        init_case = auto_decode(init.select(case).cellGet(0, 0))
        self.cell_list = {
            'item':
                auto_decode(init.select(case).cellGet(cell=init_case['fixeds']['item'])),
            'mode':
                auto_decode(init.select(case).cellGet(cell=init_case['fixeds']['mode']))
        }
        self.view_list = {
            'http':
                DataFrame(read_view_dict(filename=workbook, sheet=data, area=init_data['tables']['http']['views'], fields=init_data['tables']['http']['field'], auto_truncate=True)),
            'base':
                DataFrame(read_view_dict(filename=workbook, sheet=data, area=init_data['tables']['base']['views'], fields=init_data['tables']['base']['field'], auto_truncate=True)),
            'data':
                DataFrame(read_view_dict(filename=workbook, sheet=data, area=init_data['tables']['data']['views'], fields=init_data['tables']['data']['field'], auto_truncate=True)),
            'case':
                DataFrame(read_view_dict(filename=workbook, sheet=case, area=init_case['tables']['case']['views'], fields=init_case['tables']['case']['field'], auto_truncate=True))
        }
        self.i = self.cell_list['item']
        self.m = self.cell_list['mode']
        self.h = self.view_list['http']
        self.b = self.view_list['base']
        self.d = self.view_list['data']
        self.c = self.view_list['case']
        init.exit()
        self._setLevels(level_list)
        self._setRequestMode(self.m)
        self.dirs = os.path.dirname(workbook)

    def _setRequestMode(self, value=None):
        self.http = [Request, Session][value in ['会话模式', 'Session', 'session']]()
        return True

    def _setLevels(self, level_list=None):
        if isinstance(level_list, (list, tuple, str)):
            self.leve_list = level_list

    @staticmethod
    def _matchCaseType(value=None):
        match value:
            case '正常用例':
                return 10
            case '前置用例':
                return 20
            case _:
                return 50

    @staticmethod
    def _matchParmType(value=None):
        match value:
            case 'Json':
                return 10
            case 'Form':
                return 20
            case _:
                return 50

    def _matchLevelRun(self, value=None):
        if self.leve_list is None:
            return True
        return value in self.leve_list

    @staticmethod
    def _sub_variable(text='', vars_dict=None):
        if not isinstance(text, str):
            return text
        for variable_name in re.findall('\${(.*?)}', text):
            try:
                value = vars_dict[variable_name]
            except:
                value = ''
            text = text.replace('${%s}' % variable_name, str(value))
        return text

    def _sub_variable_auto(self, data=None, vars_dict=None):
        if isinstance(vars_dict, list):
            d = {}
            for v in vars_dict:
                d.update(v)
            vars_dict = d
        if isinstance(data, list):
            for i in range(len(data)):
                if not isinstance(data[i], str):
                    continue
                data[i] = self._sub_variable(text=data[i], vars_dict=vars_dict)
            return data
        if isinstance(data, dict):
            for k in data.keys():
                if not isinstance(data[k], str):
                    continue
                data[k] = self._sub_variable(text=data[k], vars_dict=vars_dict)
            return data
        if isinstance(data, str): return self._sub_variable(data, vars_dict=vars_dict)
        return data

    def test_unit(self, data):
        log.d('正在执行：%s' % data['CaseTitle'])
        HTTPCh = self.h.where((self.h['Env'] == self.envs) & (self.h['Name'] == data['HTTPCh']), inplace=False).dropna(how='all').loc[0].to_dict()
        res_kwargs = {
            'debug': True,
            'method': data['HTTPMethod'],
            'url':
                self._sub_variable_auto(
                    (HTTPCh['HostWithScheme'] or '') + (data['HTTPUri'] or '/'),
                    [self._g, self._l]
                ),
            'query':
                self._sub_variable_auto(auto_decode(data['HTTPQuery']),
                    [self._g, self._l]),
            'header':
                self._sub_variable_auto(auto_decode(data['HTTPHeaders']),
                    [self._g, self._l]),
            'cookie':
                self._sub_variable_auto(auto_decode(data['HTTPCookies']),
                    [self._g, self._l]),
            'auto_redirect':
                data['HTTPRedirect'] in ['是', 'True', 'true']
        }
        HTTPParamType = self._matchParmType(data['HTTPParamType'])
        match HTTPParamType:
            case 10:
                res_kwargs['data_json'] = self._sub_variable_auto(auto_decode(data['HTTPParamContent']))
            case 20:
                res_kwargs['data'] = self._sub_variable_auto(auto_decode(data['HTTPParamContent']))
            case _:
                res_kwargs['data'] = self._sub_variable_auto(data['HTTPParamContent'])
        HTTPParamOfFile = self._sub_variable_auto(auto_decode(data['HTTPParamOfFile']))
        if isinstance(HTTPParamOfFile, dict):
            for k, v in HTTPParamOfFile:
                HTTPParamOfFile[k] = open(os.path.abspath(os.path.join(self.dirs, './%s' % v)), 'rb')
            res_kwargs['data_file'] = HTTPParamOfFile

        res = self.http.http(**res_kwargs)
        log.d('\n%s' % (res['brief'] or ''))
        # print(res)
        # print(HTTPCh)

    def test(self, index):
        # print(view_case.loc[index])
        main_case = self.c.loc[index].to_dict()
        self.test_unit(data=main_case)

    def main(self):
        class ClassTestCase(unittest.TestCase):
            pass
        ClassTestCase.__qualname__ = self.case
        ClassTestCase._test_ = self.test
        serial = 0
        for i in range(len(self.c)):
            thisCase = self.c.loc[i].to_dict()
            if not self._matchCaseType(thisCase['Flag']) == 10 and self._matchLevelRun(thisCase['CaseLevel']): continue
            serial += 1
            def foo(self, index=i): return self._test_(index)
            foo.__doc__ = thisCase['CaseTitle']
            exec('ClassTestCase.test%s_%s = foo' % (str('%04d' % serial), str(thisCase['CaseId'])))
        return ClassTestCase


if __name__ == '__main__':
    test_suite = unittest.TestSuite()
    test_suite.addTest(unittest.makeSuite(testCaseClass=TestCase(
        workbook='D:/Desktop/接口自动化测试用例.xlsx',
        data='数据配置',
        case='测试用例',
        envs='测试环境',
        level_list=['P0']
    ).main(), prefix='test'))
    HTMLTestRunner(verbosity=5, report=os.path.abspath(os.path.join(os.path.dirname(__file__), './1.html'))).run(
        test_suite)