id: 用例ID
用例名稱: 用例的描述
是否需要token (0為需要, 1為不需要，默認(rèn)為0)
請(qǐng)求方式 (0為GET, 1為POST)
請(qǐng)求數(shù)據(jù)格式 (0為application/json, 1為application/x-www-form-urlencoded)
請(qǐng)求數(shù)據(jù) (統(tǒng)一存放格式為JSON)
返回?cái)?shù)據(jù) (測(cè)試用例執(zhí)行后回寫)
depends_on (依賴的用例ID)

import mysql.connector
from mysql.connector import Error
class DB:
    def __init__(self, host, database, user, password):
        self.host = host
        self.database = database
        self.user = user
        self.password = password
        self.connection = None
    def connect(self):
        try:
            self.connection = mysql.connector.connect(
                host=self.host,
                database=self.database,
                user=self.user,
                password=self.password
            )
            if self.connection.is_connected():
                return True
        except Error as e:
            print(f"Error while connecting to MySQL: {e}")
        return False
    def close(self):
        if self.connection and self.connection.is_connected():
            self.connection.close()
    def get_test_cases(self):
        cursor = self.connection.cursor(dictinotallow=True)
        query = "SELECT * FROM testdata"
        cursor.execute(query)
        results = cursor.fetchall()
        cursor.close()
        return results
    def update_test_case(self, id, response):
        cursor = self.connection.cursor()
        query = "UPDATE testdata SET 返回?cái)?shù)據(jù) = %s WHERE id = %s"
        cursor.execute(query, (response, id))
        self.connection.commit()
        cursor.close()

import requests
import json
from db_operations import DB
def send_request(test_case, context):
    headers = {}
    if test_case['請(qǐng)求數(shù)據(jù)格式'] == 0:
        headers['Content-Type'] = 'application/json'
        data = json.loads(test_case['請(qǐng)求數(shù)據(jù)'])
    else:
        headers['Content-Type'] = 'application/x-www-form-urlencoded'
        data = test_case['請(qǐng)求數(shù)據(jù)']
    # 參數(shù)化：替換請(qǐng)求數(shù)據(jù)中的占位符
    for key, value in data.items():
        if isinstance(value, str) and value.startswith('{{') and value.endswith('}}'):
            var_name = value[2:-2]
            if var_name in context:
                data[key] = context[var_name]
    url = "http://your_api_url_here"  # 替換為實(shí)際的API URL
    if test_case['請(qǐng)求方式'] == 0:
        response = requests.get(url, params=data, headers=headers)
    else:
        if test_case['是否需要token'] == 0:
            headers['Authorization'] = 'Bearer your_token_here'  # 替換為實(shí)際的Token
        response = requests.post(url, data=json.dumps(data) if headers['Content-Type'] == 'application/json' else data, headers=headers)
    return response.text
def run_tests(host, database, user, password):
    db = DB(host, database, user, password)
    if not db.connect():
        return
    test_cases = db.get_test_cases()
    context = {}  # 用于存儲(chǔ)變量值
    for test_case in test_cases:
        # 檢查是否存在依賴
        depends_on = test_case.get('depends_on')
        if depends_on:
            # 獲取依賴用例的結(jié)果
            dependency = next((tc for tc in test_cases if tc['id'] == depends_on), None)
            if dependency and '返回?cái)?shù)據(jù)' in dependency:
                # 將依賴用例的結(jié)果放入上下文中
                context.update(json.loads(dependency['返回?cái)?shù)據(jù)']))
        # 執(zhí)行當(dāng)前用例
        response = send_request(test_case, context)
        db.update_test_case(test_case['id'], response)
        # 更新上下文
        context.update(json.loads(response))
    db.close()
if __name__ == "__main__":
    # 這里可以添加參數(shù)解析器來動(dòng)態(tài)獲取數(shù)據(jù)庫連接信息等
    run_tests('localhost', 'your_database', 'your_user', 'your_password')

# main.py
from test_executor import run_tests
if __name__ == "__main__":
    # 可以在這里添加額外的初始化代碼、日志記錄等
    run_tests('localhost', 'your_database', 'your_user', 'your_password')