如何通过覆写或Mock实现无修改的数据库单元测试？

当然有简单靠谱的实现方法！核心是通过依赖注入解耦你的代码与真实数据库，再配合内存数据库或Mock工具，就能完全隔离真实环境，测试时不会对pos.db产生任何改动。下面结合你的代码给你两种实用方案：

先修复代码里的小问题（重要！） #

你的DataChecker代码里有个明显的bug：方法里直接用cursor和conn，但这些是实例的属性，应该用self.cursor和self.conn，不然运行时会报未定义错误。另外查询操作不需要调用commit()，也一起修正：

import sqlite3
from customer import Customer

class DataChecker:
    def __init__(self, conn=None):
        # 新增支持传入连接的参数，方便测试时替换
        self.conn = conn or sqlite3.connect('pos.db')
        self.cursor = self.conn.cursor()

    def customer_has_equipment_attached(self, customerID):
        # 修正为使用实例属性self.cursor/self.conn
        self.cursor.execute("""SELECT * FROM Customers WHERE ID == ?;""", (customerID,))
        customers = self.cursor.fetchall()
        if len(customers) == 0:
            print("Customer ID not found in DB")
            return False
        customer = customers[0]
        tmp_customer = Customer(
            ID=customer[0], Firstname=customer[1], Lastname=customer[2], Age=customer[3], 
            Sex=customer[4], Street=customer[5], Zip=customer[6], City=customer[7], 
            Nationality=customer[8], IMSIPtr=customer[9], IMEIPtr=customer[10], 
            SubscriptionPtr=customer[11], Email=customer[12], Password=customer[13]
        )
        # 获取设备
        if tmp_customer.IMEIPtr is not None:
            self.cursor.execute("""SELECT * FROM Equipment WHERE ID = ?;""", (tmp_customer.IMEIPtr,))
            equipment = self.cursor.fetchone()
            if not equipment:  # 更简洁的空值判断
                print('Customer has equipment ID but ID cant be found in DB.')
                return False
            return True  # 设备存在时返回True
        else:
            print('Customer has no equipment attached to profile.')
            return False

方案一：用SQLite内存数据库（适合验证SQL逻辑） #

SQLite支持创建内存中的数据库（:memory:），测试时我们可以预先创建好测试用的表和数据，完全和真实数据库隔离，还能验证SQL语句的正确性。

测试用例示例（基于unittest） #

import unittest
import sqlite3
from your_module import DataChecker, Customer

class TestDataChecker(unittest.TestCase):
    def setUp(self):
        # 创建内存数据库连接
        self.conn = sqlite3.connect(':memory:')
        cursor = self.conn.cursor()
        # 创建与真实库一致的测试表结构
        cursor.execute("""
            CREATE TABLE Customers (
                ID INTEGER PRIMARY KEY, Firstname TEXT, Lastname TEXT, Age INTEGER,
                Sex TEXT, Street TEXT, Zip TEXT, City TEXT, Nationality TEXT,
                IMSIPtr INTEGER, IMEIPtr INTEGER, SubscriptionPtr INTEGER,
                Email TEXT, Password TEXT
            )
        """)
        cursor.execute("""
            CREATE TABLE Equipment (
                ID INTEGER PRIMARY KEY, Name TEXT  # 按真实表结构补充字段
            )
        """)
        # 插入测试数据
        cursor.execute("INSERT INTO Customers VALUES (1, 'John', 'Doe', 30, 'M', 'Street 1', '12345', 'City', 'US', null, 100, null, 'john@test.com', 'pass')")
        cursor.execute("INSERT INTO Equipment VALUES (100, 'Test Phone')")
        self.conn.commit()
        # 初始化DataChecker时传入内存连接
        self.checker = DataChecker(conn=self.conn)

    def tearDown(self):
        self.conn.close()

    def test_customer_has_valid_equipment(self):
        result = self.checker.customer_has_equipment_attached(1)
        self.assertTrue(result)

    def test_customer_not_found(self):
        result = self.checker.customer_has_equipment_attached(999)
        self.assertFalse(result)

    def test_equipment_not_found(self):
        # 修改客户的IMEIPtr为不存在的ID
        cursor = self.conn.cursor()
        cursor.execute("UPDATE Customers SET IMEIPtr = 999 WHERE ID = 1")
        self.conn.commit()
        result = self.checker.customer_has_equipment_attached(1)
        self.assertFalse(result)

if __name__ == '__main__':
    unittest.main()

方案二：用unittest.mock直接Mock数据库操作（适合快速验证业务逻辑） #

如果不需要验证SQL语句的正确性，只是想快速覆盖业务逻辑分支，可以直接Mocksqlite3的连接和游标，模拟各种查询结果。

测试用例示例 #

import unittest
from unittest.mock import patch, MagicMock
from your_module import DataChecker, Customer

class TestDataCheckerWithMock(unittest.TestCase):
    @patch('sqlite3.connect')
    def test_customer_has_equipment_attached(self, mock_connect):
        # Mock数据库连接和游标
        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_connect.return_value = mock_conn
        mock_conn.cursor.return_value = mock_cursor

        # 模拟客户存在且设备存在的场景
        mock_cursor.fetchall.return_value = [
            (1, 'John', 'Doe', 30, 'M', 'Street 1', '12345', 'City', 'US', None, 100, None, 'john@test.com', 'pass')
        ]
        mock_cursor.fetchone.return_value = (100, 'Test Phone')

        checker = DataChecker()
        result = checker.customer_has_equipment_attached(1)
        self.assertTrue(result)

        # 验证游标执行了正确的SQL语句
        mock_cursor.execute.assert_any_call("""SELECT * FROM Customers WHERE ID == ?;""", (1,))
        mock_cursor.execute.assert_any_call("""SELECT * FROM Equipment WHERE ID = ?;""", (100,))

    @patch('sqlite3.connect')
    def test_customer_not_found(self, mock_connect):
        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_connect.return_value = mock_conn
        mock_conn.cursor.return_value = mock_cursor

        # 模拟客户不存在的场景
        mock_cursor.fetchall.return_value = []

        checker = DataChecker()
        result = checker.customer_has_equipment_attached(999)
        self.assertFalse(result)

if __name__ == '__main__':
    unittest.main()

总结 #

• 如果需要验证SQL语句、表结构的兼容性，选内存数据库方案；
• 如果只是快速覆盖业务逻辑分支，选Mock方案，更轻便；
• 核心是通过依赖注入让DataChecker支持外部传入连接，这是让代码可测试的关键一步。

内容的提问来源于stack exchange，提问作者coder