import json import unittest from .utils import * class TestUtils(unittest.TestCase): def test_convert_to_tuples(self): l = ["q", "q", 1, 2] t = tuple(l) print(t) def test_split(self): """测试 split 函数""" content = '''2023-03-07T07:31:43.483695315Z;ip-10-250-8-35.ec2.internal;tidb1379661944643794099;db-tidb-0; slowlog # Time: 2023-03-07T07:31:42.828651591Z # Txn_start_ts: 0 # Query_time: 0.908763888 # Parse_time: 0 # Compile_time: 0.000023634 # Rewrite_time: 0.000005453 # Optimize_time: 0 # Wait_TS: 0.000267572 # DB: mysql # Is_internal: true # Digest: 0a22979ad136f741465cf206b13e5717b9758dda34f7cae5a7a925f91711d6a0 # Num_cop_tasks: 0 # Prepared: false # Plan_from_cache: false # Plan_from_binding: false # Has_more_results: false # KV_total: 0 # PD_total: 0 # Backoff_total: 0 # Write_sql_response_total: 0 # Result_rows: 0 # Succ: true # IsExplicitTxn: false # IsSyncStatsFailed: false CREATE DATABASE IF NOT EXISTS test; 2023-03-07T07:31:50.484295876Z;ip-10-250-8-35.ec2.internal;tidb1379661944643794099;db-tidb-0;slowlog # Time: 2023-03-07T07:31:49.685517608Z # Txn_start_ts: 0 # Query_time: 1.07960149 # Parse_time: 0.000239078 # Compile_time: 0.000537771 # Rewrite_time: 0.000002751 # Optimize_time: 0.000256457 # Wait_TS: 0.000284771 # Is_internal: true # Digest: 2e5b6f591b9fbbfab3e57d191b2252c2a439182e8fbeb90d688cdf38ed93398c # Num_cop_tasks: 0 # Prepared: false # Plan_from_cache: false # Plan_from_binding: false # Has_more_results: false # KV_total: 0 # PD_total: 0 # Backoff_total: 0 # Write_sql_response_total: 0 # Result_rows: 0 # Succ: true # IsExplicitTxn: false # IsSyncStatsFailed: false CREATE TABLE IF NOT EXISTS mysql.tidb_audit_table_access ( id int not null auto_increment, user varchar(32) NOT NULL comment "username need be audit(regex)", db varchar(32) NOT NULL comment "db name need be audit(regex)", tbl varchar(32) NOT NULL comment "table name need be audit(regex)", access_type varchar(1024) NOT NULL comment "table access types split by comma", create_time timestamp NULL DEFAULT CURRENT_TIMESTAMP comment "record create time", update_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP comment "record last update time", UNIQUE KEY(user, db, tbl), PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;''' logs = split(content, ";\n") self.assertEqual(2, len(logs)) print(logs) def test_parse(self): content = '''2023-03-07T07:31:43.483695315Z;ip-10-250-8-35.ec2.internal;tidb1379661944643794099;db-tidb-0; slowlog # Time: 2023-03-07T07:31:42.828651591Z # Txn_start_ts: 0 # Query_time: 0.908763888 # Parse_time: 0 # Compile_time: 0.000023634 # Rewrite_time: 0.000005453 # Optimize_time: 0 # Wait_TS: 0.000267572 # DB: mysql # Is_internal: true # Digest: 0a22979ad136f741465cf206b13e5717b9758dda34f7cae5a7a925f91711d6a0 # Num_cop_tasks: 0 # Prepared: false # Plan_from_cache: false # Plan_from_binding: false # Has_more_results: false # KV_total: 0 # PD_total: 0 # Backoff_total: 0 # Write_sql_response_total: 0 # Result_rows: 0 # Succ: true # IsExplicitTxn: false # IsSyncStatsFailed: false CREATE DATABASE IF NOT EXISTS test''' result1 = parse(content, "[^;]+") print(result1) self.assertEqual("db-tidb-0", result1[3]) result2 = parse(content, "#\s*(\w+)\s*:\s*(.*?)\n") self.assertEqual("Parse_time", result2[3][0]) # result3 = parse(content, "(?<=\n\n)(.*?)") # todo # print("\n------re3", result3) def test_convert_to_dict(self): li = [('Time', '2023-03-07T07:31:42.828651591Z'), ('Txn_start_ts', '0'), ('Query_time', '0.908763888'), ('Parse_time', '0'), ('Compile_time', '0.000023634'), ('Rewrite_time', '0.000005453'), ('Optimize_time', '0'), ('Wait_TS', '0.000267572'), ('DB', 'mysql'), ('Is_internal', 'true'), ('Digest', '0a22979ad136f741465cf206b13e5717b9758dda34f7cae5a7a925f91711d6a0'), ('Num_cop_tasks', '0'), ('Prepared', 'false'), ('Plan_from_cache', 'false'), ('Plan_from_binding', 'false'), ('Has_more_results', 'false'), ('KV_total', '0'), ('PD_total', '0'), ('Backoff_total', '0'), ('Write_sql_response_total', '0'), ('Result_rows', '0'), ('Succ', 'true'), ('IsExplicitTxn', 'false'), ('IsSyncStatsFailed', 'false')] result = convert_to_dict(li) self.assertEqual("2023-03-07T07:31:42.828651591Z", result["Time"]) def test_mapping(self): li = ['2023-03-07T07:31:43.483695315Z', 'db-tidb-0'] result = mapping(li, ["time", "tidb_instance"], [0, 1]) self.assertEqual("db-tidb-0", result["tidb_instance"]) dic = {'Time': '2023-03-07T07:31:42.828651591Z', 'Txn_start_ts': '0', 'Query_time': '0.908763888', 'Parse_time': '0', 'Compile_time': '0.000023634', 'Rewrite_time': '0.000005453', 'Optimize_time': '0', 'Wait_TS': '0.000267572', 'DB': 'mysql', 'Is_internal': 'true', 'Digest': '0a22979ad136f741465cf206b13e5717b9758dda34f7cae5a7a925f91711d6a0', 'Num_cop_tasks': '0', 'Prepared': 'false', 'Plan_from_cache': 'false', 'Plan_from_binding': 'false', 'Has_more_results': 'false', 'KV_total': '0', 'PD_total': '0', 'Backoff_total': '0', 'Write_sql_response_total': '0', 'Result_rows': '0', 'Succ': 'true', 'IsExplicitTxn': 'false', 'IsSyncStatsFailed': 'false'} result = mapping(dic, ["time", "query_time"], ['Time', 'Query_time']) self.assertEqual("0.908763888", result["query_time"]) def test_merge_map(self): dict1 = {'Time': '2023-03-07T07:31:42.828651591Z', 'Txn_start_ts': '0', 'Query_time': '0.908763888'} dict2 = {'Parse_time': '0', 'Compile_time': '0.000023634', 'Rewrite_time': '0.000005453'} result = merge_map(dict1, dict2) self.assertEqual(6, len(result)) def test_parse_multi(self): content1 = '''2023-03-07T07:31:43.483695315Z;ip-10-250-8-35.ec2.internal;tidb1379661944643794099;db-tidb-0; slowlog # Time: 2023-03-07T07:31:42.828651591Z # Txn_start_ts: 0 # Query_time: 0.908763888 # Parse_time: 0 # Compile_time: 0.000023634 # Rewrite_time: 0.000005453 # Optimize_time: 0 # Wait_TS: 0.000267572 # DB: mysql # Is_internal: true # Digest: 0a22979ad136f741465cf206b13e5717b9758dda34f7cae5a7a925f91711d6a0 # Num_cop_tasks: 0 # Prepared: false # Plan_from_cache: false # Plan_from_binding: false # Has_more_results: false # KV_total: 0 # PD_total: 0 # Backoff_total: 0 # Write_sql_response_total: 0 # Result_rows: 0 # Succ: true # IsExplicitTxn: false # IsSyncStatsFailed: false CREATE DATABASE IF NOT EXISTS test ''' content2 = '''2023-03-07T07:31:50.484295876Z;ip-10-250-8-35.ec2.internal;tidb1379661944643794099;db-tidb-0; slowlog # Time: 2023-03-07T07:31:49.685517608Z # Txn_start_ts: 0 # Query_time: 1.07960149 # Parse_time: 0.000239078 # Compile_time: 0.000537771 # Rewrite_time: 0.000002751 # Optimize_time: 0.000256457 # Wait_TS: 0.000284771 # Is_internal: true # Digest: 2e5b6f591b9fbbfab3e57d191b2252c2a439182e8fbeb90d688cdf38ed93398c # Num_cop_tasks: 0 # Prepared: false # Plan_from_cache: false # Plan_from_binding: false # Has_more_results: false # KV_total: 0 # PD_total: 0 # Backoff_total: 0 # Write_sql_response_total: 0 # Result_rows: 0 # Succ: true # IsExplicitTxn: false # IsSyncStatsFailed: false CREATE TABLE IF NOT EXISTS mysql.tidb_audit_table_access ( id int not null auto_increment, user varchar(32) NOT NULL comment "username need be audit(regex)", db varchar(32) NOT NULL comment "db name need be audit(regex)", tbl varchar(32) NOT NULL comment "table name need be audit(regex)", access_type varchar(1024) NOT NULL comment "table access types split by comma", create_time timestamp NULL DEFAULT CURRENT_TIMESTAMP comment "record create time", update_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP comment "record last update time", UNIQUE KEY(user, db, tbl), PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;''' content = [content1, content2] result = parse_multi(content, "[^;]+") self.assertEqual("2023-03-07T07:31:50.484295876Z", result[1][0]) self.assertEqual("ip-10-250-8-35.ec2.internal", result[0][1]) def test_convert_to_dict_multi(self): content = [[('Time', '2023-03-07T07:31:42.828651591Z'), ('Txn_start_ts', '0'), ('Query_time', '0.908763888'), ('Parse_time', '0'), ('Compile_time', '0.000023634')], [('Time', '2023-03-07T08:31:42.828651591Z'), ('Txn_start_ts', '0'), ('Query_time', '0.708763888'), ('Parse_time', '0'), ('Compile_time', '0.000123634')]] result = convert_to_dict_multi(content) self.assertEqual(2, len(result)) self.assertEqual("2023-03-07T07:31:42.828651591Z", result[0]['Time']) self.assertEqual("2023-03-07T08:31:42.828651591Z", result[1]['Time']) self.assertEqual("0.000023634", result[0]['Compile_time']) print(result) def test_mapping_multi(self): content = [ ['2023-03-07T07:31:43.483695315Z', 'ip-10-250-8-35.ec2.internal', 'tidb1379661944643794099', 'db-tidb-0'], ['2023-03-07T07:31:50.484295876Z', 'ip-10-250-8-35.ec2.internal', 'tidb1379661944643794099', 'db-tidb-1']] result = mapping_multi(content, ["time", "tidb_instance"], [0, 3]) self.assertEqual(2, len(result)) self.assertEqual("2023-03-07T07:31:43.483695315Z", result[0]['time']) self.assertEqual("2023-03-07T07:31:50.484295876Z", result[1]['time']) self.assertEqual("db-tidb-0", result[0]['tidb_instance']) self.assertEqual("db-tidb-1", result[1]['tidb_instance']) print("111", result) content1 = [{'Time': '2023-03-07T07:31:42.828651591Z', 'Txn_start_ts': '0', 'Query_time': '0.908763888', 'Parse_time': '0', 'Compile_time': '0.000023634'}, {'Time': '2023-03-07T08:31:42.828651591Z', 'Txn_start_ts': '0', 'Query_time': '0.708763888', 'Parse_time': '0', 'Compile_time': '0.000123634'}] result1 = mapping_multi(content1, ["query_time", "compile_time"], ["Query_time", "Compile_time"]) self.assertEqual(2, len(result)) self.assertEqual("0.908763888", result1[0]['query_time']) self.assertEqual("0.708763888", result1[1]['query_time']) self.assertEqual("0.000023634", result1[0]['compile_time']) self.assertEqual("0.000123634", result1[1]['compile_time']) print("222", result1) def test_merge_map_multi(self): data1 = [{'time': '2023-03-07T07:31:43.483695315Z', 'tidb_instance': 'db-tidb-0'}, {'time': '2023-03-07T07:31:50.484295876Z'}] data2 = [{'query_time': '0.908763888', 'compile_time': '0.000023634'}, {'query_time': '0.708763888', 'compile_time': '0.000123634'}] result = merge_map_multi(data1, data2) self.assertEqual(2, len(result)) self.assertEqual(4, len(result[0])) self.assertEqual(3, len(result[1])) self.assertEqual("0.908763888", result[0]['query_time']) self.assertEqual("0.000123634", result[1]['compile_time']) def test_convert_to_metrics(self): data = { "digest1": [7, 2, 1, 9, 3], "digest2": [2, 4, 6], "digest3": [1, 3, 5, 7, 9], "digest4": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] } result = convert_to_metrics(data, 2, 9, 2) print(result) rows = build_data_for_sqlite(result, 2, 9, 2) print(rows) def test_calculate_count(self): data = [1, 2, 3, 4] a, b = calculate_count(data, 2, 2, 0) print(a, b) def test_build_create_sql(self): data = [("digest1", "digest2", "digest3"), (1, 0, 0, 1111), (0, 1, 0, 1112), (0, 0, 1, 1113)] create, insert = build_create_sql(data, "test") print(create, insert) def test_create_slowlog_count_sqlite(self): data = [{"time": "2023-03-07T07:31:42.828651591Z", "tidb_instance": "tidb:db-0", "digest": "digest1"}, {"time": "2023-03-07T07:32:42.828651591Z", "tidb_instance": "tidb:db-0", "digest": "digest1"}, {"time": "2023-03-07T07:31:42.828651591Z", "tidb_instance": "tidb:db-1", "digest": "digest1"}, {"time": "2023-03-07T07:31:42.828651591Z", "tidb_instance": "tidb:db-0", "digest": "digest2"}, {"time": "2023-03-07T07:32:42.828651591Z", "tidb_instance": "tidb:db-0", "digest": "digest2"}] create_slowlog_count_sqlite(data, 1678145500, 1678145600, 20, "test_1", "slowlog_count") def test_create_slowlog_sqlite(self): data = [{'time': '2023-04-04T05:56:31.456041802Z', 'db': 'adwords', 'digest': 'digest1', 'query_time': '1.357866546', 'mem_max': '120895', 'tidb_instance': 'tidb-db-0'}, {'time': '2023-04-04T05:56:31.390116921Z', 'db': 'adwords', 'digest': 'digest2', 'query_time': '1.199196262', 'mem_max': '128031', 'tidb_instance': 'tidb-db-0'}] create_slowlog_sqlite(data, "./test.db", "slow_log") def test_convert_utc_time_to_timestamp(self): time_str = "2023-04-04T16:00:00.000000Z" ts = convert_utc_time_to_timestamp(time_str) self.assertEqual(1680624000, ts) def test_clean_metrics_header(self): file_name = "/Users/xie/GolandProjects/drhouse/diagnosis/preprocess/utils/1379661944621804065_tidb_server_execute_error_total_405.csv" with open(file_name) as csvfile: csv_reader = csv.reader(csvfile) header = next(csv_reader) # 读取第一行的标题 print(header) result = clean_metrics_header(header) print(result) if __name__ == '__main__': unittest.main()