0

在我正在测试的类中，我想模拟用作成员变量的整个 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, 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")
4

2 回答 2

0

我发现我的问题是它没有修补 data_access，因为当我使用 pytest 夹具创建测试实例时已经声明了 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)
                       ]
于 2021-03-29T18:14:51.950 回答
0

我还认为使用继承创建我的 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)

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

于 2021-03-30T13:03:47.537 回答