Byte Ebi's Logo

Byte Ebi 🍤

每天一小口,蝦米變鯨魚

[Pytest 101] 03 Mock SQL Query

將單元測試從資料庫連線中解放出來

Ray

透過 Mock,我們不需要真的連線到資料庫來執行測試,就可以模擬資料庫操作的結果。
而不需要準備一個測試用的資料庫,並在測試後清除資料庫。

定義 Mock 物件類別

在開始之前先談談怎麼定義抽換的物件類別,因為預設的 Mock / MagicMock 在存取任何屬性,呼叫任何方法時都會回傳一個 mock,就算屬性和方法不存在。雖然使用上相當方便,但是這樣的行為表現就顯得不真實。

可以藉由 spec 來定義 Mock 物件的規格,讓 mock 盡可能地按照規格模仿,試圖取得不在規格內的屬性或者方法,就拋出 AttributeError,如此一來可避免 mock 預設的特性造成的困擾,測試案例的 mock 就更加嚴謹。

最上方不要忘記引入對應的類別!

from model.user_table import User

接著就可以使用 spec 指定 mock 出來的物件類別

mock_user = mocker.MagicMock(spec=User)

Mock 鍊接呼叫

透過以下特性,可以簡化不需要深入測試的 Mock 物件

在被呼叫前取得其 return_value,一個新的 Mock 就會被建立。

當有個查詢為

user = session.query(User).filter(User.id == user_id).first()

這是逐個物件進行 Mock 的寫法:

# 模擬 User 物件
mock_user = mocker.MagicMock(spec=User)

# 模擬 session
mock_session = mocker.MagicMock()

# 模擬 session.query(...) 回傳 mock_query
mock_query = mocker.MagicMock()
mock_session.query.return_value = mock_query

# 模擬 query.filter(...) 回傳 mock_filter
mock_filter = mocker.MagicMock()
mock_query.filter.return_value = mock_filter

# 模擬 filter.first() 回傳 mock_user
mock_filter.first.return_value = mock_user

# 呼叫被測函式
user = get_user_by_id(mock_session, user_id=1)

在這段程式碼中 MagicMock 被用來模擬:

  • mock_session:模擬資料庫 session 物件。
  • mock_query:模擬資料庫 query 物件。
  • mock_filter:模擬 query 的 filter 方法的回傳值。
  • mock_user:模擬一個 User 物件。

在這個測試中不在乎資料庫查詢時各階段的錯誤,所以可以用鍊接呼叫直接將查詢簡化為

mock_user = mocker.MagicMock(spec=User)

mock_session = mocker.MagicMock()
mock_session.query.return_value.filter.return_value.first.return_value = mock_user

# 呼叫被測函式
user = get_user_by_id(mock_session, user_id=1)

資料查詢

抽換資料庫的查詢,藉由 mock 一個 user 讓函式回傳,避免資料庫連線意外對測試結果造成影響。

# my_project/service/user_service.py

def get_user_by_id(session, user_id):
    user = session.query(User).filter(User.id == user_id).first()
    return user

如果使用者存在,斷言判斷回傳的 user 物件要和我們 mock 的 user 物件相同。

# my_project/test/service/test_user_service.py

import pytest
from pytest_mock import MockFixture

from model.user_table import User
from service.user_service import (
    get_user_by_id,
)

def test_get_user_by_id_success(mocker: MockFixture):
	mock_user = mocker.MagicMock(spec=User)  # 建立一個 mock 的 User 物件
    mock_user.id = 1

    mock_session = mocker.MagicMock()        # 建立一個 mock 的 session 物件
    mock_session.query.return_value.filter.return_value.first.return_value = mock_user

    user = get_user_by_id(mock_session, user_id=1)
    assert user == mock_user

資料操作

在使用 SQLALchemy 對資料庫進行操作時,不管是都會使用 session.commit()來提交變更。
所以可以針對 session.commit() 有沒有被呼叫,甚至呼叫幾次來驗證對資料庫的操作
並且可以設定 commit 的 return_value 作為資料庫操作事件回傳判斷

也可以針對 session.addsession.delete 更近一步驗證呼叫的方法是否正確,這邊用delete 做範例

# my_project/service/user_service.py

def delete_user(session, user_id):
    user = session.query(User).filter(User.id == user_id).first()
    session.delete(user)
    session.commit()
    return user.id

斷言式說明:

  • called:該方法有被呼叫
  • call_count:該方法被呼叫的次數,回傳一個 int
# my_project/test/service/test_user_service.py

import pytest
from pytest_mock import MockFixture

from model.user_table import User
from service.user_service import (
    delete_user,
)

def test_delete_user_success(mocker: MockFixture):
    mock_user = mocker.MagicMock(spec=User)
    mock_user.id = 1

    mock_session = mocker.MagicMock()
    mock_session.query.return_value.filter.return_value.first.return_value = mock_user
    mock_session.delete.return_value = None

    user_id = delete_user(mock_session, mock_user.id)

    assert mock_session.delete.called
    assert mock_session.delete.call_count == 1
    assert mock_session.commit.called
    assert user_id == mock_user.id

資料庫錯誤

當然資料庫不可能永遠都是穩定的,也可以藉由測試來模擬錯誤。
例如使用 SQLAlchemy 的 one_or_none() 時,如果有多筆資料會回傳 MultipleResultsFound 類型的錯誤。

為了方便測試,以下範例不做額外處理,直接將查詢結果回傳。
當發生錯誤時便會回傳 MultipleResultsFound 類型的錯誤。

# my_project/service/user_service.py

def get_user_by_id(session, id) -> User:
    return session.query(User).filter(User.id == id).one_or_none()

案例說明:

  1. 引入 sqlalchemy.orm.excMultipleResultsFound 物件。
  2. 抽換待測函式中的 one_or_none() 查詢結果,回傳 MultipleResultsFound 類型錯誤。
  3. 使用斷言判斷取得的資料類別符合預期。
# my_project/test/service/test_user_service.py

import pytest
from pytest_mock import MockFixture
from sqlalchemy.orm.exc import MultipleResultsFound

from model.user_table import User
from service.user_service import (
    get_user_by_id,
)

def test_get_user_by_id_multiple(mocker: MockFixture):
    mock_session = mocker.MagicMock()
    mock_session.query.return_value.filter.return_value.one_or_none.return_value = (
        MultipleResultsFound()
    )

    user = get_service_info_by_id(mock_session, 1)
    assert isinstance(user, MultipleResultsFound)

最新文章

Category

Tag