Pytest-mock - 尝试模拟整个类时未使用 new_callable 类

如何解决Pytest-mock - 尝试模拟整个类时未使用 new_callable 类

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

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

编辑：已按照建议和最初更新我修补的位置，但仍然无法正常工作？

要测试的类：

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()

我的测试班：

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,'Date Now')
    ''',None)
    data_access.execute('''
        INSERT INTO Purchase_Product (purchase_id,product_id,quantity)
            VALUES(1,None)

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

解决方法

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

此外，我发现 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','1st Class',100.0,3,4)
                       ]

,

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

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

主数据访问类：

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)

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