Pytest-mock

Question

在我正在测试的 class 中，我想模拟用作成员变量的整个 DataAccess class。 DataAccess class 只是抽象 SQLite 数据库连接。

我已经创建了一个连接到测试数据库的替换 MockDataAccess class，但似乎仍然调用了主数据库 - 我做错了什么？

编辑：已经按照建议更新了我正在打补丁的地方，但仍然没有用？

Class 测试：

from .data_access import DataAccess


class Query:
    def __init__(self):
        self._data_access = DataAccess()

    def get_all_data(self):
        queryset = self._data_access.execute('''
            SELECT
                Purchase_Product.purchase_id,
                Product.product_name,
                Purchase_Product.quantity,
                Customer.first_name,
                Customer.last_name,
                Customer.email,
                Address.line_one,
                Address.line_two,
                Address.city,
                Address.postcode,
                Status.status_description,
                Purchase.created_date,
                Purchase.dispatched_date,
                Purchase.completed_date,
                Postage.postage_description,
                Product.individual_price,
                Product.id,
                Product.aisle,
                Product.shelf
            FROM
                Product
                INNER JOIN Purchase_Product ON Purchase_Product.product_id = Product.id
                INNER JOIN Purchase ON Purchase.id = Purchase_Product.purchase_id
                INNER JOIN Status ON Status.id = Purchase.status_id
                INNER JOIN Postage ON Postage.id = Purchase.postage_id
                INNER JOIN Customer ON Customer.id = Purchase.customer_id
                INNER JOIN Customer_Address ON Customer_Address.customer_id = Customer.id
                INNER JOIN Address ON Address.id = Customer_Address.address_id
            ORDER BY
                Status.id
        ''', None)
        return queryset.fetchall()

我的测试class:

import os
import sqlite3
from sqlite3.dbapi2 import Connection, Cursor
import pytest
import mock
from src.main.order_management.data.query import Query

class MockDataAccess:
    def __init__(self):
        self._conn = self._create_connection()
        self._cur = self._conn.cursor()
    
    def _create_connection(self):
        THIS_DIR = os.path.dirname(__file__)
        TEST_DATABASE = os.path.join(THIS_DIR, 'database', 'TestOnlineStore.db')
        return sqlite3.connect(TEST_DATABASE)

    def execute(self, query, data):
        if data is None:
            self._cur.execute(query)
        else:
            self._cur.execute(query, data)
        self._conn.commit()
        return self._cur

@pytest.fixture
def data_access():
    return MockDataAccess()

@pytest.fixture
def query():
    return Query()

@pytest.fixture
def setup_database(data_access):
    data_access.execute('''
        CREATE TABLE IF NOT EXISTS Address (
            id integer PRIMARY KEY AUTOINCREMENT,
            line_one text NOT NULL,
            line_two text,
            city text NOT NULL,
            postcode text
    );''', None)
    data_access.execute('''
        CREATE TABLE IF NOT EXISTS Customer (
            id integer PRIMARY KEY AUTOINCREMENT,
            first_name text NOT NULL,
            last_name text NOT NULL,
            email text
    );''', None)
    data_access.execute('''
        CREATE TABLE IF NOT EXISTS Customer_Address (
            customer_id integer,
            address_id integer,

            FOREIGN KEY (customer_id)
                REFERENCES Customer (id)
                    ON DELETE CASCADE
            FOREIGN KEY (address_id)
                REFERENCES Address (id)
                    ON DELETE CASCADE
            PRIMARY KEY (customer_id, address_id)
    );''', None)
    data_access.execute('''
        CREATE TABLE IF NOT EXISTS Platform (
            id integer PRIMARY KEY AUTOINCREMENT,
            platform_name integer NOT NULL,
            user_token text NOT NULL
    );''', None)
    data_access.execute('''
        CREATE TABLE IF NOT EXISTS Status (
            id integer PRIMARY KEY AUTOINCREMENT,
            status_description text NOT NULL
    );''', None)
    data_access.execute('''
        CREATE TABLE IF NOT EXISTS Postage (
            id integer PRIMARY KEY AUTOINCREMENT,
            postage_description text NOT NULL
    );''', None)
    data_access.execute('''
        CREATE TABLE IF NOT EXISTS Purchase (
            id integer PRIMARY KEY AUTOINCREMENT,
            platform_id integer,
            customer_id integer,
            status_id integer,
            postage_id integer,
            created_date text NOT NULL,
            dispatched_date text,
            completed_date text,
            
            FOREIGN KEY (platform_id)
                REFERENCES Platform (id)
                    ON DELETE CASCADE
            FOREIGN KEY (customer_id)
                REFERENCES Customer (id)
                    ON DELETE CASCADE
            FOREIGN KEY (status_id)
                REFERENCES Status (id)
                    ON DELETE CASCADE
    );''', None)
    data_access.execute('''
        CREATE TABLE IF NOT EXISTS Product (
            id integer PRIMARY KEY AUTOINCREMENT,
            product_name integer,
            product_description integer,
            individual_price real,
            stock_count integer,
            aisle integer,
            shelf integer
    );''', None)
    data_access.execute('''
        CREATE TABLE IF NOT EXISTS Purchase_Product (
            purchase_id integer,
            product_id integer,
            quantity integer,

            FOREIGN KEY (purchase_id)
                REFERENCES Purchase (id)
                    ON DELETE CASCADE
            FOREIGN KEY (product_id)
                REFERENCES Product (id)
                    ON DELETE CASCADE
            PRIMARY KEY (purchase_id, product_id)
    );''', None)
    
@pytest.fixture
def clean_database(data_access):
    data_access.execute('''DROP TABLE IF EXISTS Address''', None)
    data_access.execute('''DROP TABLE IF EXISTS Customer''', None)
    data_access.execute('''DROP TABLE IF EXISTS Customer_Address''', None)
    data_access.execute('''DROP TABLE IF EXISTS Platform''', None)
    data_access.execute('''DROP TABLE IF EXISTS Postage''', None)
    data_access.execute('''DROP TABLE IF EXISTS Product''', None)
    data_access.execute('''DROP TABLE IF EXISTS Purchase''', None)
    data_access.execute('''DROP TABLE IF EXISTS Purchase_Product''', None)
    data_access.execute('''DROP TABLE IF EXISTS Status''', None)
    

@pytest.fixture
def setup_test_data1(setup_database, data_access):
    data_access.execute('''
        INSERT INTO Address (line_one, line_two, city) 
            VALUES('Test Line One', 'Test Line Two', 'Test City')
    ''', None)
    data_access.execute('''
        INSERT INTO Customer (first_name, last_name, email) 
            VALUES('Test First Name', 'Test Last Name', 'Test Email')
    ''', None)
    data_access.execute('''
        INSERT INTO Customer_Address (customer_id, address_id)
            VALUES(1, 1)
    ''', None)
    data_access.execute('''
        INSERT INTO Postage (postage_description)
            VALUES('1st Class')
    ''', None)
    data_access.execute('''
        INSERT INTO Product (product_name, individual_price, stock_count, aisle, shelf)
            VALUES('Test Product', 100.00, 2, 3, 4)
    ''', None)
    data_access.execute('''
        INSERT INTO Status (status_description)
            VALUES('Awaiting')
    ''', None)
    data_access.execute('''
        INSERT INTO Purchase (customer_id, status_id, postage_id, created_date)
            VALUES(1, 1, 1, 'Date Now')
    ''', None)
    data_access.execute('''
        INSERT INTO Purchase_Product (purchase_id, product_id, quantity)
            VALUES(1, 1, 1)
    ''', None)

@mock.patch('src.main.order_management.data.query.DataAccess', new_callable=MockDataAccess)
def test_get_all_data(mock_data_access, query, clean_database, setup_database, setup_test_data1, data_access):
    all_data = query.get_all_data()
    assert all_data == ("jdi", "fmn")

Answer 1

我发现我的问题是它没有修补 data_access，因为当我使用 pytest fixture 创建测试实例时已经声明了 data_access。

此外，我发现 new_callable 实际上并没有像我想象的那样运行，所以我改用 return_value 并传递了一个 MockDataAccess 实例。现在我的测试数据库正在按预期调用。

新 test_query.py（仅更改位）：


mock_data_access = MockDataAccess()

@mock.patch('src.main.order_management.data.query.DataAccess', return_value=mock_data_access)
def test_get_all_data(mock_data_access, clean_database,
                      setup_database, setup_test_data1):
    query = Query()
    all_data = query.get_all_data()
    assert all_data == [
                        (1,
                         'Test Product',
                         1,
                         'Test First Name',
                         'Test Last Name',
                         'Test Email',
                         'Test Line One',
                         'Test Line Two',
                         'Test City',
                         None,
                         'Awaiting',
                         'Date Now',
                         None,
                         None,
                         '1st Class',
                         100.0,
                         1,
                         3,
                         4)
                       ]

Answer 2

我还认为使用继承创建我的 MockDataAccess 更清晰、更合乎逻辑。

因此，我现在创建一个子 class 并设置父 class' (DataAccess) 连接属性，而不是复制和粘贴我的 MockDataAccess 并更改它指向的数据库.

主要数据访问class:

import sqlite3
from sqlite3.dbapi2 import Connection, Cursor
from ..config import DATABASE


class DataAccess:
    def __init__(self):
        self._connection = self._create_connection()
        self._cursor = self._connection.cursor()

    def _create_connection(self):
        return sqlite3.connect(DATABASE)

    def execute(self, query, data):
        if data is None:
            self._cursor.execute(query)
        else:
            self._cursor.execute(query, data)
        self._connection.commit()
        return self._cursor

模拟数据访问：

class MockDataAccess(DataAccess):
    def __init__(self):
        super(MockDataAccess, self).__init__()
        self._connection = self._create_connection()

    def _create_connection(self):
        return sqlite3.connect(TEST_DATABASE)

我是测试的新手，所以不知道这是否很明显 - 我想我会分享以防这对某人有帮助。

Pytest-mock - new_callable class 在尝试模拟整个 class 时未被使用

Pytest-mock - new_callable class not being used when trying to mock whole class

python

pytest

python-unittest