From e087722ecdfc51d4a3e96dfc1b52cc0611eb5e83 Mon Sep 17 00:00:00 2001 From: zhaoyafan Date: Wed, 10 Aug 2022 18:32:30 +0800 Subject: [PATCH] 220810 --- Runner/API/1.html | 71 ++++++------ Runner/API/DefaultRunner.py | 216 ++++++++++++++++++++---------------- 2 files changed, 150 insertions(+), 137 deletions(-) diff --git a/Runner/API/1.html b/Runner/API/1.html index 1ce7ea4..6d00061 100644 --- a/Runner/API/1.html +++ b/Runner/API/1.html @@ -229,11 +229,11 @@ series: [{ name: '通过', color: '#64bb64', - data: [5] + data: [6] }, { name: '失败', color: '#f16d7e', - data: [1] + data: [0] }, { name: '错误', color: '#fdc68c', @@ -291,7 +291,7 @@ innerSize: '80%', name: '比例', data: [ - ['通过', 5], ['失败', 1], ['错误', 0] + ['通过', 6], ['失败', 0], ['错误', 0] ] }] }, function(c) { @@ -438,13 +438,13 @@ function html_escape(s) {

测试报告

-

开始时间 : 2022-08-09 22:17:02

+

开始时间 : 2022-08-10 18:32:08

-

合计耗时 : 00:00:00

+

合计耗时 : 00:00:01

-

测试结果 : 总共 6，通过 5，失败 1，错误 0，通过率 83.33%

+

测试结果 : 总共 6，通过 6，失败 0，错误 0，通过率 100.00%

-

失败用例 : 点击查看

  1. main.测试用例.test0006_None

+

失败用例 : 无

错误用例 : 无

@@ -457,9 +457,9 @@ function html_escape(s) {

- 概要 83.33% - 通过 5 - 失败 1 + 概要 100.00% + 通过 6 + 失败 0 错误 0 全部 6

@@ -486,14 +486,14 @@ function html_escape(s) { 详细 - + 测试用例 6 - 5 - 1 + 6 0 - 0.732秒 + 0 + 1.154秒 查看全部 @@ -513,8 +513,15 @@ function html_escape(s) {
test0003_None
- - + + + +
+ 
pt1_3:
+{'var8': 'Windows、Linux修改连接数限制'}
+
+
+ @@ -532,35 +539,21 @@ function html_escape(s) { - -
test0006_None
- - - -
- 
ft1_6:
-Traceback (most recent call last):
-  File "D:\Project\AutoFramework\Runner\API\DefaultRunner.py", line 507, in foo
-    return self._test_(index)
-  File "D:\Project\AutoFramework\Runner\API\DefaultRunner.py", line 492, in test
-    self.test_unit(data=main_case)
-  File "D:\Project\AutoFramework\Runner\API\DefaultRunner.py", line 460, in test_unit
-    _assert(_assert_parm[0], _assert_parm[1])
-AssertionError: 'nginx' unexpectedly found in 'Server: nginx\nDate: Tue, 09 Aug 2022 14:17:03 GMT\nContent-Type: application/json\nContent-Length: 208\nLast-Modified: Thu, 04 Aug 2022 07:51:36 GMT\nConnection: keep-alive\nETag: "62eb7a88-d0"\nAccept-Ranges: bytes'
-
-
- - + +
test0006_None
+ + + 总计 6 - 5 - 1 + 6 0 - 0.732秒 - 通过：83.33% + 0 + 1.154秒 + 通过：100.00% diff --git a/Runner/API/DefaultRunner.py b/Runner/API/DefaultRunner.py index 9db9bef..63d8cd2 100644 --- a/Runner/API/DefaultRunner.py +++ b/Runner/API/DefaultRunner.py @@ -169,7 +169,7 @@ class TestCase: "HTTPAssertBody", "HTTPAssertJson", "DatabaseChannel", - "DatabaseSqlQuery", + "DatabaseSql", "DatabaseExtract", "DatabaseAssertData", "DatabaseAssertCount", @@ -365,104 +365,124 @@ class TestCase: return str(text) def test_unit(self, data): - http = self.h.where((self.h['Name'] == data['HTTPChannel']), inplace=False).dropna(how='all').reset_index(drop=True, inplace=False).loc[0].to_dict() - res_kwargs = { - 'method': data['HTTPMethod'], - 'url': self._sub_variable_auto((http['HostWithScheme'] or '') + (data['HTTPUri'] or '/'), [self._g, self._l]), - 'query': self._sub_variable_auto(auto_decode(data['HTTPQuery']), [self._g, self._l]), - 'data': None, - 'json': None, - 'file': None, - 'header': {}, - 'cookie': self._sub_variable_auto( - auto_decode(data['HTTPCookie']), - [self._g, self._l] - ), - 'auth': None, - 'timeout': 15, - 'proxy': None, - 'auto_redirect': self._match_bool(data['HTTPRedirect']), - 'ignore_cert_error': False, - 'debug': True - } - res_kwargs['header'].update((auto_decode(http['DefaultHeader']) or {})) - res_kwargs['header'].update((self._sub_variable_auto(auto_decode(data['HTTPHeader']), [self._g, self._l]) or {})) - match self._match_parm_type(data['HTTPParamType']): - case 10: - res_kwargs['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']) - res_kwargs['file'] = locals().setdefault('f', self._sub_variable_auto(auto_decode( - data['HTTPParamOfFile']))) and {k: open(os.path.abspath(os.path.join(self.dirs, './%s' % v)), 'rb') for k, v in locals().get('f').items()} - res = self.http.http(**res_kwargs) - for var_name, var_value in { - '_Status': ReText(res['status'] or ''), - '_Reason': ReText(res['reason'] or ''), - '_Header': ReDict(res['header'] or {}).update({'_': "\n".join([k + ': ' + (v or '') for k, v in dict(res['header'] or {}).items()])}), - '_Cookie': ReDict(res['cookie'] or {}).update({'_': "; ".join([k + '=' + (v or '') for k, v in dict(res['cookie'] or {}).items()])}), - '_Body': ReText(res['text'] or ''), - '_Json': ReDict(res['json'] or {}), - json_encode(None): None, - json_encode(bool(0)): bool(0), - json_encode(bool(1)): bool(1) - }.items(): - locals().__setitem__(var_name, var_value) - for k, v in (auto_decode(data['HTTPExtract']) or {}).items(): - d = re.findall('^([0-9A-Za-z_]+).*?', v)[0] - self._g[str(k)] = str(eval('_' + v.replace(d, d.title(), 1))) - for assert_item in [ - {'expect': data['HTTPAssertStatus'], 'actual': locals().get('_Status')}, - {'expect': data['HTTPAssertReason'], 'actual': locals().get('_Reason')}, - {'expect': data['HTTPAssertHeader'], 'actual': locals().get('_Header')}, - {'expect': data['HTTPAssertCookie'], 'actual': locals().get('_Cookie')}, - {'expect': data['HTTPAssertBody'], 'actual': locals().get('_Body')}, - {'expect': data['HTTPAssertJson'], 'actual': locals().get('_Json')}, - ]: - actual = assert_item['actual'] - if isinstance(actual, (int, bool, str)): - actual = ReText(actual) - if isinstance(actual, (tuple, list, set)): - actual = ReList(actual) - if isinstance(actual, (dict, )): - actual = ReDict(actual) - expect = self._to_string(assert_item['expect']) - if expect: - expect_nums = 0 - expect_flag = 0 - for line in list(filter(lambda x: x, expect.split("\n"))): - expect_nums += 1 - for operator in [ - {'sign': ' == ', 'method': 'self.unit.assertEqual', 'reverse': 0}, - {'sign': ' != ', 'method': 'self.unit.assertNotEqual', 'reverse': 0}, - {'sign': ' >= ', 'method': 'self.unit.assertGreaterEqual', 'reverse': 0}, - {'sign': ' <= ', 'method': 'self.unit.assertLessEqual', 'reverse': 0}, - {'sign': ' =~ ', 'method': 'self.unit.assertIn', 'reverse': 1}, - {'sign': ' !~ ', 'method': 'self.unit.assertNotIn', 'reverse': 1}, - {'sign': ' > ', 'method': 'self.unit.assertGreater', 'reverse': 0}, - {'sign': ' < ', 'method': 'self.unit.assertLess', 'reverse': 0}, - {'sign': ' not in ', 'method': 'self.unit.assertNotIn', 'reverse': 0}, - {'sign': ' in ', 'method': 'self.unit.assertIn', 'reverse': 0}, - ]: - formula_args = self._parse_formula(self._sub_variable_auto(line).replace('$', 'actual'), operator['sign']) - if formula_args: - expect_flag += 1 - operator['reverse'] and formula_args.reverse() - _assert = eval(operator['method']) - _assert_parm = [eval(formula_args[0]), eval(formula_args[1])] - for i in range(len(_assert_parm)): - try: - if isinstance(_assert_parm[i], ReDict) and _assert_parm[i]['_']: _assert_parm[i] = str(_assert_parm[i]) - except TypeError as e: - pass - _assert(_assert_parm[0], _assert_parm[1]) - break - if 1 <= expect_flag != expect_nums: - raise Exception('wrong assertion statement') - if 0 == expect_flag: - self.unit.assertEqual(str(actual), expect) + if data['HTTPUri']: + if not data['HTTPChannel']: + raise Exception('channel not set') + http = self.h.where((self.h['Name'] == data['HTTPChannel']), inplace=False).dropna(how='all').reset_index(drop=True, inplace=False).loc[0].to_dict() + res_kwargs = { + 'method': data['HTTPMethod'], + 'url': self._sub_variable_auto((http['HostWithScheme'] or '') + data['HTTPUri'], [self._g, self._l]), + 'query': self._sub_variable_auto(auto_decode(data['HTTPQuery']), [self._g, self._l]), + 'data': None, + 'json': None, + 'file': None, + 'header': {}, + 'cookie': self._sub_variable_auto(auto_decode(data['HTTPCookie']), [self._g, self._l]), + 'auth': None, + 'timeout': 15, + 'proxy': None, + 'auto_redirect': self._match_bool(data['HTTPRedirect']), + 'ignore_cert_error': False, + 'debug': True + } + res_kwargs['header'].update((auto_decode(http['DefaultHeader']) or {})) + res_kwargs['header'].update((self._sub_variable_auto(auto_decode(data['HTTPHeader']), [self._g, self._l]) or {})) + match self._match_parm_type(data['HTTPParamType']): + case 10: + res_kwargs['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']) + res_kwargs['file'] = locals().setdefault('f', self._sub_variable_auto(auto_decode( + data['HTTPParamOfFile']))) and {k: open(os.path.abspath(os.path.join(self.dirs, './%s' % v)), 'rb') for k, v in locals().get('f').items()} + res = self.http.http(**res_kwargs) + for var_name, var_value in { + '_Status': ReText(res['status'] or ''), + '_Reason': ReText(res['reason'] or ''), + '_Header': ReDict(res['header'] or {}).update({'_': "\n".join([k + ': ' + (v or '') for k, v in dict(res['header'] or {}).items()])}), + '_Cookie': ReDict(res['cookie'] or {}).update({'_': "; ".join([k + '=' + (v or '') for k, v in dict(res['cookie'] or {}).items()])}), + '_Body': ReText(res['text'] or ''), + '_Json': ReDict(res['json'] or {}), + json_encode(None): None, + json_encode(bool(0)): bool(0), + json_encode(bool(1)): bool(1) + }.items(): + locals().__setitem__(var_name, var_value) + for k, v in (auto_decode(data['HTTPExtract']) or {}).items(): + d = re.findall('^([0-9A-Za-z_]+).*?', v)[0] + self._g[str(k)] = str(eval('_' + v.replace(d, d.title(), 1))) + for assert_item in [ + {'expect': data['HTTPAssertStatus'], 'actual': locals().get('_Status')}, + {'expect': data['HTTPAssertReason'], 'actual': locals().get('_Reason')}, + {'expect': data['HTTPAssertHeader'], 'actual': locals().get('_Header')}, + {'expect': data['HTTPAssertCookie'], 'actual': locals().get('_Cookie')}, + {'expect': data['HTTPAssertBody'], 'actual': locals().get('_Body')}, + {'expect': data['HTTPAssertJson'], 'actual': locals().get('_Json')}, + ]: + actual = assert_item['actual'] + if isinstance(actual, (int, bool, str)): + actual = ReText(actual) + if isinstance(actual, (tuple, list, set)): + actual = ReList(actual) + if isinstance(actual, (dict, )): + actual = ReDict(actual) + expect = self._to_string(assert_item['expect']) + if expect: + expect_nums = 0 + expect_flag = 0 + for line in list(filter(lambda x: x, expect.split("\n"))): + expect_nums += 1 + for operator in [ + {'sign': ' == ', 'method': 'self.unit.assertEqual', 'reverse': 0}, + {'sign': ' != ', 'method': 'self.unit.assertNotEqual', 'reverse': 0}, + {'sign': ' >= ', 'method': 'self.unit.assertGreaterEqual', 'reverse': 0}, + {'sign': ' <= ', 'method': 'self.unit.assertLessEqual', 'reverse': 0}, + {'sign': ' =~ ', 'method': 'self.unit.assertIn', 'reverse': 1}, + {'sign': ' !~ ', 'method': 'self.unit.assertNotIn', 'reverse': 1}, + {'sign': ' > ', 'method': 'self.unit.assertGreater', 'reverse': 0}, + {'sign': ' < ', 'method': 'self.unit.assertLess', 'reverse': 0}, + {'sign': ' not in ', 'method': 'self.unit.assertNotIn', 'reverse': 0}, + {'sign': ' in ', 'method': 'self.unit.assertIn', 'reverse': 0}, + ]: + formula_args = self._parse_formula(self._sub_variable_auto(line).replace('$', 'actual'), operator['sign']) + if formula_args: + expect_flag += 1 + operator['reverse'] and formula_args.reverse() + _assert = eval(operator['method']) + _assert_parm = [eval(formula_args[0]), eval(formula_args[1])] + for i in range(len(_assert_parm)): + try: + if isinstance(_assert_parm[i], ReDict) and _assert_parm[i]['_']: _assert_parm[i] = str(_assert_parm[i]) + except TypeError as e: + pass + _assert(_assert_parm[0], _assert_parm[1]) + break + if 1 <= expect_flag != expect_nums: + raise Exception('wrong assertion statement') + if 0 == expect_flag: + self.unit.assertEqual(str(actual), expect) + if data['DatabaseSql']: + if not data['DatabaseChannel']: + raise Exception('channel not set') + base = self.b.where((self.b['Name'] == data['DatabaseChannel']), inplace=False).dropna(how='all').reset_index(drop=True, inplace=False).loc[0].to_dict() + # print(base) + conn = MySQL().connect( + host=base['Host'], port=int(base['Port']), + user=base['User'], password=base['Password'], database=base['Database'], charset=base['Charset'], cursor='Dict' + ) + conn.execute(data['DatabaseSql']) + # print(conn.fetchall()) + # print(conn.count()) + for var_name, var_value in { + '_Data': ReList(conn.fetchall() or []) + }.items(): + locals().__setitem__(var_name, var_value) + for k, v in (auto_decode(data['DatabaseExtract']) or {}).items(): + d = re.findall('^([0-9A-Za-z_]+).*?', v)[0] + self._g[str(k)] = str(eval('_' + v.replace(d, d.title(), 1))) + print(self._g) # expect = self._to_string(data['HTTPAssertStatus'])