run teststep:sql request
(runner: HttpRunner, step: TStep)
| 42 | |
| 43 | |
| 44 | def run_step_sql_request(runner: HttpRunner, step: TStep) -> StepResult: |
| 45 | """run teststep:sql request""" |
| 46 | start_time = time.time() |
| 47 | |
| 48 | step_result = StepResult( |
| 49 | name=step.name, |
| 50 | step_type="sql", |
| 51 | success=False, |
| 52 | ) |
| 53 | step_variables = runner.merge_step_variables(step.variables) |
| 54 | # parse |
| 55 | request_dict = step.sql_request.dict() |
| 56 | parsed_request_dict = runner.parser.parse_data(request_dict, step_variables) |
| 57 | config = runner.get_config() |
| 58 | parsed_request_dict["db_config"]["psm"] = ( |
| 59 | parsed_request_dict["db_config"]["psm"] or config.db.psm |
| 60 | ) |
| 61 | parsed_request_dict["db_config"]["user"] = ( |
| 62 | parsed_request_dict["db_config"]["user"] or config.db.user |
| 63 | ) |
| 64 | parsed_request_dict["db_config"]["password"] = ( |
| 65 | parsed_request_dict["db_config"]["password"] or config.db.password |
| 66 | ) |
| 67 | parsed_request_dict["db_config"]["ip"] = ( |
| 68 | parsed_request_dict["db_config"]["ip"] or config.db.ip |
| 69 | ) |
| 70 | parsed_request_dict["db_config"]["port"] = ( |
| 71 | parsed_request_dict["db_config"]["port"] or config.db.port |
| 72 | ) |
| 73 | parsed_request_dict["db_config"]["database"] = ( |
| 74 | parsed_request_dict["db_config"]["database"] or config.db.database |
| 75 | ) |
| 76 | |
| 77 | if not runner.db_engine: |
| 78 | ensure_sql_ready() |
| 79 | from httprunner.database.engine import DBEngine |
| 80 | |
| 81 | runner.db_engine = DBEngine( |
| 82 | f'mysql+pymysql://{parsed_request_dict["db_config"]["user"]}:' |
| 83 | f'{parsed_request_dict["db_config"]["password"]}@{parsed_request_dict["db_config"]["ip"]}:' |
| 84 | f'{parsed_request_dict["db_config"]["port"]}/{parsed_request_dict["db_config"]["database"]}' |
| 85 | f"?charset=utf8mb4" |
| 86 | ) |
| 87 | |
| 88 | # parsed_request_dict["headers"].setdefault( |
| 89 | # "HRUN-Request-ID", |
| 90 | # f"HRUN-{self.__case_id}-{str(int(time.time() * 1000))[-6:]}", |
| 91 | # ) |
| 92 | |
| 93 | # setup hooks |
| 94 | if step.setup_hooks: |
| 95 | call_hooks(runner, step.setup_hooks, step_variables, "setup request") |
| 96 | |
| 97 | # log request |
| 98 | sql_request_print = "====== sql request details ======\n" |
| 99 | sql_request_print += f"sql: {step.sql_request.sql}\n" |
| 100 | for k, v in parsed_request_dict.items(): |
| 101 | v = utils.omit_long_data(v) |
no test coverage detected