使用数据库查询结果参数化httprunner

创建读取数据库数据的read_pg.py，里面创建getsessionsofsyncable方法如下：

import psycopg2
import json
from itertools import groupby
import os
import requests

def getsessionsofsyncable():
    conn = psycopg2.connect(database='xx',user='xx',password='xx',host='xx.xx.xx.xx',port='xxxx')
    cur = conn.cursor()
    sql1 = "select c_session_id from t_xx"
    sql2 = "delete from t_xx"
    # 查询所有可以供httprunner使用的id,并放到sessions列表。
    cur.execute(sql1)
    rows = cur.fetchall()
    sessions = []
    for row in rows:
        tem = []
        tem.append(row[0])
        sessions.append(tem)
    print(sessions)
    # 查询之后删除数据，因为接口会验证id，已有的将不再执行
    cur.execute(sql2)
    #删除操作需要commit
    conn.commit()
    conn.close()
    return sessions

下面是charles抓包后使用har2case转的xx_test.py文件，变大加粗的字体是我自己加的内容：

# NOTE: Generated By HttpRunner v3.1.6
# FROM: xxx.har

import pytest
from httprunner import HttpRunner, Config, Step, RunRequest, RunTestCase,Parameters
from read_pg.py import getsessionsofsyncable 

#sessions = getsessionsofsyncable()
class xxx(HttpRunner):
    config = (
        Config("testcase description")
        .verify(False)
    )


    teststeps = [
        Step(
            RunRequest("/review-syncdata")
            .with_variables(
                **{"sessionids": "$sessionids",
                   }
            )
            .post("http://xxx.xxx.x.xxx:xxxx/xx-xx")
            .with_headers(
                **{
                    "Host": ".....",
                    "Content-Length": "104",
                    "Accept": "application/json, text/javascript, */*; q=0.01",
                    "X-Requested-With": "XMLHttpRequest",
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36",
                    "Content-Type": "application/json",
                    "Origin": "http://...",
                    "Referer": "http://...",
                    "Accept-Encoding": "gzip, deflate",
                    "Accept-Language": "zh-CN,zh;q=0.9",
                    "Cookie": "...",
                }
            )
            .with_cookies(
                **{
                    "_ga": "..",
                    "screenResolution": "..",
                    "JSESSIONID.fd7222a4": "...",
                    "session_read": "...",
                }
            )
            .with_json(
                {
                    "user_id": "...", #接口需要的另一个参数
                    "sessionIds": "$sessionids",
                }
            )
            .validate()
            .assert_equal("status_code", 200)
            .assert_equal("body.code", 0)
            .assert_contains("body.message", "同步完成！")
        ),
    ]
# main里调用同步方法，并把参数传给httprunner的test_start方法

if __name__ == "__main__":
      print('哈哈哈哈')
    sessions = getsessionsofsyncable()
    xxx.test_start(param={'sessionids':sessions})