1. 目录结构

2. session.py
import logging
import requestslogging.basicConfig(level=logging.INFO, filename='requests.log', filemode="a", encoding='utf-8')
logger = logging.getLogger("mylogger")class MySession(requests.Session):"""支持BaseURL支持日志文件记录"""def __init__(self, base_url = ""):self.base_url = base_urlsuper().__init__()def request(self, method, url, **kwargs):if not url.startswith("http"):url = self.base_url + urlresponse = super().request(method, url, **kwargs)return responsedef send(self, request: requests.PreparedRequest, **kwargs):logger.info(f"发送请求>>> 接口地址:{request.method}-{request.url}")logger.info(f"发送请求>>> 请求头:{request.headers}")logger.info(f"发送请求>>> 请求体:{request.body}")response = super().send(request, **kwargs)logger.info(f"接收响应<<< 响应状态码:{response.status_code}")logger.info(f"接收响应<<< 响应头:{response.headers}")logger.info(f"接收响应<<< 响应内容:{response.content}")return response
3. test_api.py
import pytest@pytest.mark.parametrize("data, code",[({}, 422),({"email": "1341935532@qq.com", "password": "123456"}, 422),({"username": "admin", "password": "123456"}, 200),({"username": "admin", "password": "wrongpassword"}, 401),]
)
def test_login(session, data, code):resp = session.post("/login/get_token", json=data)assert resp.status_code == codeif code == 200:print("=" * 100)print("token: ", resp.json()["token"])session.headers = {"Authorization": resp.json()["token"]}def test_create_task(session):resp = session.get("/todo", json={})assert resp.status_code == 200def test_get_task(session):resp = session.get(f"/todo/1")assert resp.status_code == 200def test_delete_task(session):resp = session.delete("/todo/1")assert resp.status_code == 204
4. app.py
from typing import Annotatedfrom fastapi import FastAPI, Request, Form, Body, Response
from fastapi.responses import JSONResponseapp = FastAPI()@app.post("/login/with_form")
async def login_form(username: Annotated[str, Form()],password: Annotated[str, Form()],
):return {"username": username, "password": password}@app.post("/login/with_json")
async def login_json(username: Annotated[str, Body()],password: Annotated[str, Body()]
):return {"username": username, "password": password}@app.post("/set_cookie")
async def set_cookie():response = {"message": "Cookie set"}response_obj = JSONResponse(response)response_obj.set_cookie("cookie", "123789")return response_obj@app.get("/get_cookie")
async def get_cookie(request: Request):cookie = request.cookies.get("cookie")if cookie:return {"cookie": cookie}else:return JSONResponse(status_code=401, content={"message": "Cookie not found"})@app.post("/login/get_token")
async def get_token(username: Annotated[str, Body()],password: Annotated[str, Body()]
):if username == "admin" and password == "123456":return {"token": "123789"}else:return JSONResponse(status_code=401, content={"message": "Invalid credentials"})@app.get("/todo")
async def get_todo(request: Request):token = request.headers.get("Authorization")if token != "123789":return JSONResponse(status_code=401, content={"message": "Invalid or missing token"})return {"todos": ["task1", "task2", "task3"]}@app.delete("/todo/{item_id}")
async def delete_todo(request: Request, item_id: int):token = request.headers.get("Authorization")if token != "123789":return JSONResponse(status_code=401, content={"message": "Invalid or missing token"})return Response(status_code=204)@app.get("/todo/{item_id}")
async def get_todo(request: Request, item_id: int):token = request.headers.get("Authorization")if token != "123789":return JSONResponse(status_code=401, content={"message": "Invalid or missing token"})if item_id != 1:return JSONResponse(status_code=404, content={"message": "Task not found"})return {"task": f"task{item_id}"}
5. conftest.py
import _pytest.hookspec
import pytestfrom commons.session import MySession
@pytest.fixture(scope="session")
def session():yield MySession("http://127.0.0.1:8000")def pytest_configure():print("pytest_configure: 测试会话开始前执行")def pytest_unconfigure():print("pytest_unconfigure: 测试会话结束后执行")def pytest_yaml_run_step(item):print("正在执行第几个步骤", item)return True
6. main.py
import osimport pytest
pytest.main()
os.system("allure generate -o report ./allure_results --clean")
7. my_yaml.py
from pprint import pprint
import yamlwith open("test.yaml", "r", encoding="utf-8") as f:data = yaml.safe_load(f)pprint(data)
8. pytest.ini
[pytest]
addopts = -vs --alluredir=./allure_results --clean-alluredirlog_file = pytest.log
log_file_level = debug
log_file_format = %(levelname)-8s %(asctime)s [%(name)s:%(lineno)s] : %(message)s
log_file_date_format = %Y-%m-%d %H:%M:%S
result_log_level_verbose = info;run_yaml_case = truemarkers =login: 登录相关db: 数据库相关upload: 上传相关maya: maya相关