FastAPI（43）- 基於 pytest + requests 進行單元測試

 FastAPI 的單元測試

• 對於服務端來說，通常會對功能進行單元測試，也稱白盒測試
• FastAPI 集成了第三方庫，讓我們可以快捷的編寫單元測試
• FastAPI 的單元測試是基於 Pytest + Request 的 

 

Pytest 學習

https://www.cnblogs.com/poloyy/tag/Pytest/

 

TestClient 簡單的栗子

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠蘿測試筆記
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/29 10:55 下午
# file: 37_pytest.py
"""
import uvicorn
from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}


# 聲明一個 TestClient，把 FastAPI() 實例對象傳進去
client = TestClient(app)


# 測試用
def test_read_main():
    # 請求 127.0.0.1:8080/
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}


if __name__ == '__main__':
    uvicorn.run(app="37_pytest:app", reload=True, host="127.0.0.1", port=8080)

  

在該文件夾下的命令行敲

pytest 37_pytest.py

 

運行結果

 

TestClient 的源碼解析

繼承了 requests 庫的 Session

所以可以像使用 requests 庫一樣使用 TestClient，擁有 requests 所有方法、屬性

 

重寫了 Session.requests 方法

重寫了 requests 方法，不過只是加了一句 url = urljoin(self.base_url, url) url 拼接代碼，還有給函數參數都加了類型指示，更加完善啦~

 

自定義 websocket 連接方法

后面學到 webSocket 再詳細講他

 

重寫了 __enter__、__exit__ 方法

• Session 的這兩個方法還是比較簡陋的，TestClient 做了一次重寫，主要是為了添加異步的功能（異步測試后面詳解，這篇舉栗子的都是普通函數 def）
• 前面講過有 __enter__、__exit__ 方法的對象都是上下文管理器，可以用 with .. as .. 語句來調用上下文管理器哦

 

.get() 方法

上面代碼 client.get()，直接調用的就是 Session 提供的 get() 方法啦！

 

復雜的測試場景

服務端

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠蘿測試筆記
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/29 10:55 下午
# file: s37_pytest.py
"""
import uvicorn
from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}


# 聲明一個 TestClient，把 FastAPI() 實例對象傳進去
client = TestClient(app)


# 測試用
def test_read_main():
    # 請求 127.0.0.1:8080/
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}


from typing import Optional

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

# 模擬真實 token
fake_secret_token = "coneofsilence"

# 模擬真實數據庫
fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Optional[str] = None


# 接口一：查詢數據
@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header(...)):
    # 1、校驗 token 失敗
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="x-token 錯誤")

    # 2、若數據庫沒有對應數據
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="找不到 item_id")
    # 3、找到數據則返回
    return fake_db[item_id]


# 接口二：創建數據
@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: str = Header(...)):
    # 1、校驗 token 失敗
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="x-token 錯誤")

    # 2、若數據庫已經存在相同 id 的數據
    if item.id in fake_db:
        raise HTTPException(status_code=400, detail="找不到 item_id")

    # 3、添加數據到數據庫
    fake_db[item.id] = item

    # 4、返回添加的數據
    return item


if __name__ == '__main__':
    uvicorn.run(app="s37_test_pytest:app", reload=True, host="127.0.0.1", port=8080)

 

單元測試 

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠蘿測試筆記
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/29 10:55 下午
# file: s37_pytest.py
"""
from fastapi.testclient import TestClient
from .s37_test_pytest import app

client = TestClient(app)


def test_read_item():
    expect = {"id": "foo", "title": "Foo", "description": "There goes my hero"}
    headers = {"x-token": "coneofsilence"}
    resp = client.get("/items/foo", headers=headers)
    assert resp.status_code == 200
    assert resp.json() == expect


def test_read_item_error_header():
    expect = {"detail": "x-token 錯誤"}
    headers = {"x-token": "test"}
    resp = client.get("/items/foo", headers=headers)
    assert resp.status_code == 400
    assert resp.json() == expect


def test_read_item_error_id():
    expect = {"detail": "找不到 item_id"}
    headers = {"x-token": "coneofsilence"}
    resp = client.get("/items/foos", headers=headers)
    assert resp.status_code == 404
    assert resp.json() == expect


def test_create_item():
    body = {"id": "foos", "title": "Foo", "description": "There goes my hero"}
    headers = {"x-token": "coneofsilence"}
    resp = client.post("/items/", json=body, headers=headers)
    assert resp.status_code == 200
    assert resp.json() == body


def test_create_item_error_header():
    body = {"id": "foo", "title": "Foo", "description": "There goes my hero"}
    expect = {"detail": "x-token 錯誤"}
    headers = {"x-token": "test"}
    resp = client.post("/items/", json=body, headers=headers)
    assert resp.status_code == 400
    assert resp.json() == expect


def test_create_item_error_id():
    expect = {"detail": "找不到 item_id"}
    body = {"id": "foo", "title": "Foo", "description": "There goes my hero"}
    headers = {"x-token": "coneofsilence"}
    resp = client.post("/items/", json=body, headers=headers)
    assert resp.status_code == 400
    assert resp.json() == expect

 

命令行運行

pytest test.py -sq

  

運行結果

> pytest s37_pytest.py -sq
......
6 passed in 0.40s