生成基于basemodel，pytest，yaml，jsonpath，mysql，缓存机制的python接口自动化，分模块生成

好的，我将为您分模块生成基于 basemodel、pytest、yaml、jsonpath、mysql、缓存机制的 Python 接口自动化代码。

### 1. 基于 basemodel 的接口请求封装

首先，我们使用基于 basemodel 的接口请求封装，方便我们发送各种类型的接口请求。

from typing import Dict, Union
from pydantic import BaseModel
import requests


class BaseResponse(BaseModel):
    status: int
    message: str
    data: Union[Dict, str]


class BaseRequest(BaseModel):
    url: str
    method: str = "GET"
    headers: Dict[str, str] = {}
    params: Dict[str, Union[str, int]] = {}
    data: Dict[str, Union[str, int]] = {}

    def send(self) -> BaseResponse:
        resp = requests.request(
            method=self.method,
            url=self.url,
            headers=self.headers,
            params=self.params,
            json=self.data
        )

        return BaseResponse(
            status=resp.status_code,
            message=resp.reason,
            data=resp.json()
        )

### 2. pytest 测试用例编写

接下来，我们编写 pytest 测试用例，利用 BaseRequest 封装的接口请求，完成接口测试。

import pytest
from base_request import BaseRequest, BaseResponse


@pytest.fixture(scope="module")
def api():
    return BaseRequest(url="https://jsonplaceholder.typicode.com/posts")


@pytest.mark.parametrize("params, expected", [
    ({}, "200 OK"),
    ({"userId": 1}, "200 OK"),
    ({"id": 1}, "200 OK"),
    ({"id": 101}, "404 Not Found"),
])
def test_get_posts(api, params, expected):
    api.params = params
    resp: BaseResponse = api.send()

    assert resp.status == 200
    assert resp.message == expected

### 3. 使用 YAML 存储测试数据

为了方便管理测试数据，我们使用 YAML 文件存储测试数据。

- name: "get all posts"
  params: {}
  expected: "200 OK"

- name: "get posts by userId"
  params:
    userId: 1
  expected: "200 OK"

- name: "get posts by id"
  params:
    id: 1
  expected: "200 OK"

- name: "get posts by invalid id"
  params:
    id: 101
  expected: "404 Not Found"

### 4. 使用 jsonpath 提取响应数据

接下来，我们使用 jsonpath 提取响应数据，并进行断言。

import pytest
from base_request import BaseRequest, BaseResponse
from jsonpath import jsonpath
import yaml


@pytest.fixture(scope="module")
def api():
    return BaseRequest(url="https://jsonplaceholder.typicode.com/posts")


@pytest.fixture(scope="module")
def test_data():
    with open("test_data.yaml", encoding="utf-8") as f:
        data = yaml.safe_load(f)

    return data


@pytest.mark.parametrize("test_case", test_data())
def test_get_posts(api, test_case):
    api.params = test_case["params"]
    resp: BaseResponse = api.send()

    assert resp.status == 200
    assert resp.message == test_case["expected"]

    data = jsonpath(resp.data, "$..title")
    assert len(data) > 0

### 5. 使用 MySQL 存储测试结果

最后，我们使用 MySQL 存储测试结果，方便后续数据分析。

import pytest
from base_request import BaseRequest, BaseResponse
from jsonpath import jsonpath
import yaml
import pymysql


@pytest.fixture(scope="module")
def api():
    return BaseRequest(url="https://jsonplaceholder.typicode.com/posts")


@pytest.fixture(scope="module")
def test_data():
    with open("test_data.yaml", encoding="utf-8") as f:
        data = yaml.safe_load(f)

    return data


@pytest.fixture(scope="module")
def mysql():
    conn = pymysql.connect(
        host="localhost",
        user="root",
        password="password",
        database="test"
    )

    yield conn

    conn.close()


@pytest.mark.parametrize("test_case", test_data())
def test_get_posts(api, test_case, mysql):
    api.params = test_case["params"]
    resp: BaseResponse = api.send()

    assert resp.status == 200
    assert resp.message == test_case["expected"]

    data = jsonpath(resp.data, "$..title")
    assert len(data) > 0

    cursor = mysql.cursor()
    sql = """
        INSERT INTO test_result (api, params, status, message)
        VALUES (%s, %s, %s, %s)
    """
    cursor.execute(sql, (api.url, str(api.params), resp.status, resp.message))
    mysql.commit()
    cursor.close()

### 6. 使用缓存机制提高测试效率

如果我们的接口请求中有一些不经常变动的数据，我们可以使用缓存机制提高测试效率。

import pytest
from base_request import BaseRequest, BaseResponse
from jsonpath import jsonpath
import yaml
import pymysql
import pickle


@pytest.fixture(scope="module")
def api():
    return BaseRequest(url="https://jsonplaceholder.typicode.com/posts")


@pytest.fixture(scope="module")
def test_data():
    with open("test_data.yaml", encoding="utf-8") as f:
        data = yaml.safe_load(f)

    return data


@pytest.fixture(scope="module")
def mysql():
    conn = pymysql.connect(
        host="localhost",
        user="root",
        password="password",
        database="test"
    )

    yield conn

    conn.close()


@pytest.fixture(scope="module")
def cache():
    try:
        with open("cache.pickle", "rb") as f:
            data = pickle.load(f)
    except:
        data = {}

    yield data

    with open("cache.pickle", "wb") as f:
        pickle.dump(data, f)


@pytest.mark.parametrize("test_case", test_data())
def test_get_posts(api, test_case, mysql, cache):
    if api.url in cache and str(api.params) in cache[api.url]:
        resp_data = cache[api.url][str(api.params)]
    else:
        resp: BaseResponse = api.send()
        resp_data = jsonpath(resp.data, "$..title")
        cache.setdefault(api.url, {})[str(api.params)] = resp_data

    assert len(resp_data) > 0

    cursor = mysql.cursor()
    sql = """
        INSERT INTO test_result (api, params, status, message)
        VALUES (%s, %s, %s, %s)
    """
    cursor.execute(sql, (api.url, str(api.params), 200, "OK"))
    mysql.commit()
    cursor.close()

这样，我们就完成了基于 basemodel、pytest、yaml、jsonpath、mysql、缓存机制的 Python 接口自动化代码。