为电影应用add_user功能编写测试用例遇问题求助

问题描述 #

我在开发电影网站时编写测试用例，测试add_user方法遇到异常：每次调用add_user()后，SQLite数据库中确实创建了用户，但方法总是返回「用户已存在」的失败提示。

相关代码 #

add_user方法实现 #

def add_user(self, name):
    param = {'name': name}
    query = text('Insert into users (name) values (:name)')
    check_usr_query = text('Select name from users '
                           'Where name = :name')

    check_usr = self._execute_query(check_usr_query, param).fetchone()

    if check_usr is not None:
       return 'User Already Exists in Database'
    else:
        with self.engine.connect() as conn:
            conn.execute(query, param)
            conn.commit()
        return 'Successfully added user'

测试代码 #

import pytest
from DataManager.sql_data_manager import SQLiteDataManager

dataManager = SQLiteDataManager('DataManager/movie.db')

def test_add_user():
    name = 'Esth'
    test_data = dataManager.add_user(name)
    check_usr = dataManager.get_user_by_name(name)
    if test_data == 'Successfully added user':
        assert test_data == 'Successfully added user'
        print(name + ' added to db')
    elif test_data == 'User Already Exists in Database':
        assert check_usr is not None
        print(name + ' Already exists in the db')

问题原因 #

1. 测试环境残留数据：测试重复运行时，之前插入的用户未被清理，导致后续调用add_user时，检查步骤查到已存在的用户，返回错误提示。
2. 数据库连接不一致：检查用户存在的_execute_query方法与插入用户的engine.connect()使用了不同的数据库连接，极端情况下可能出现连接隔离导致的状态不同步（比如未提交的事务）。

解决方案 #

方案1：清理测试前置数据 #

修改测试代码，每次运行测试前先删除测试用户，确保测试环境干净：

import pytest
from sqlalchemy import text
from DataManager.sql_data_manager import SQLiteDataManager

def test_add_user():
    dataManager = SQLiteDataManager('DataManager/movie.db')
    name = 'Esth'
    
    # 测试前清理残留用户
    delete_query = text('DELETE FROM users WHERE name = :name')
    with dataManager.engine.connect() as conn:
        conn.execute(delete_query, {'name': name})
        conn.commit()
    
    # 测试首次添加
    test_data = dataManager.add_user(name)
    check_usr = dataManager.get_user_by_name(name)
    
    assert test_data == 'Successfully added user'
    assert check_usr is not None
    print(f'{name} added to db')
    
    # 测试重复添加场景
    test_data_dup = dataManager.add_user(name)
    assert test_data_dup == 'User Already Exists in Database'
    assert check_usr is not None
    print(f'{name} Already exists in the db')

方案2：统一数据库连接 #

修改add_user方法，在同一个连接中完成用户检查与插入操作，避免连接隔离问题：

def add_user(self, name):
    param = {'name': name}
    query = text('Insert into users (name) values (:name)')
    check_usr_query = text('Select name from users '
                           'Where name = :name')

    with self.engine.connect() as conn:
        # 同一连接内执行检查与插入
        check_usr = conn.execute(check_usr_query, param).fetchone()
        if check_usr is not None:
            return 'User Already Exists in Database'
        conn.execute(query, param)
        conn.commit()
        return 'Successfully added user'

方案3：数据库层面添加唯一约束（推荐） #

在users表的name字段添加唯一约束，从数据库层避免重复用户，同时捕获插入异常增强健壮性：

1. 执行SQL添加约束：

ALTER TABLE users ADD CONSTRAINT unique_username UNIQUE (name);

2. 修改add_user方法：

from sqlalchemy.exc import IntegrityError

def add_user(self, name):
    param = {'name': name}
    query = text('Insert into users (name) values (:name)')

    try:
        with self.engine.connect() as conn:
            conn.execute(query, param)
            conn.commit()
            return 'Successfully added user'
    except IntegrityError:
        return 'User Already Exists in Database'

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