如何用pytest测试含SQL调用的POST请求？解决表不存在报错

问题核心 #

你遇到的“表不存在”错误本质是测试环境与生产环境使用的数据库不一致：正常运行时用的是MySQL，而pytest测试时配置了SQLite（sqlite:////tmp/test.db），且你mock了create_all方法，导致SQLite中从未创建过movies表。

待测试代码与配置回顾 #

业务逻辑与接口 #

def getMovieBudget(movieId):
    engine = create_engine(app.config['SQLALCHEMY_DATABASE_URI'])
    query = """
            SELECT *
            FROM movies
            WHERE id = '{}';
            """.format(movieId)
    data = pd.read_sql(query, engine)
    return data.iloc[0]['building_id']

@app.route('/upload', methods=['POST'])
def test_post():
    # .. 其他逻辑
    movieId = request.form['movieId']
    movieBudget = getMovieBudget(movieId)

测试代码 #

def test_post(client, capsys):
    url = f'/{helpers.get_domain()}/{helpers.get_version()}/movies/'

    data = {
        'movieId': 144,
    }

    response = client.post(url, data=data)

    assert response.status_code == 200

测试配置 #

os.environ['TEST_ENV'] = 'TRUE'

@pytest.fixture(autouse=True)
def mock_patch(monkeypatch):
    monkeypatch.setattr(SQLAlchemy, "create_all", mock.Mock(return_value=True))

@pytest.fixture
def client():
    from movie_handler import app

    app.config['TESTING'] = True
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
    return app.test_client()

解决办法 #

方法1：初始化SQLite测试表 #

取消对create_all的mock，让SQLAlchemy自动创建表（若用ORM），或手动执行建表语句（原生SQL场景）：

场景A：使用SQLAlchemy ORM

# 删除或注释掉mock create_all的fixture
# @pytest.fixture(autouse=True)
# def mock_patch(monkeypatch):
#     monkeypatch.setattr(SQLAlchemy, "create_all", mock.Mock(return_value=True))

@pytest.fixture
def client():
    from movie_handler import app, db  # 引入你的SQLAlchemy db实例

    app.config['TESTING'] = True
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
    
    # 在测试前创建表
    with app.app_context():
        db.create_all()
    
    yield app.test_client()
    
    # 测试完成后清理表
    with app.app_context():
        db.drop_all()

场景B：使用原生SQL

@pytest.fixture
def client():
    from movie_handler import app
    from sqlalchemy import create_engine

    app.config['TESTING'] = True
    db_uri = 'sqlite:////tmp/test.db'
    app.config['SQLALCHEMY_DATABASE_URI'] = db_uri
    
    # 手动创建movies表并插入测试数据
    engine = create_engine(db_uri)
    with engine.connect() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS movies (
                id VARCHAR PRIMARY KEY,
                building_id VARCHAR
                -- 补充其他必要字段
            )
        """)
        conn.execute("INSERT INTO movies (id, building_id) VALUES ('144', 'b123')")
        conn.commit()
    
    yield app.test_client()
    
    # 测试后清理
    with engine.connect() as conn:
        conn.execute("DROP TABLE movies")
        conn.commit()

方法2：Mock子函数，跳过真实数据库调用 #

如果测试重点是验证POST接口的参数处理、流程逻辑，而非SQL查询本身，直接mockgetMovieBudget：

def test_post(client, monkeypatch):
    # 替换原函数为mock实现
    def mock_getMovieBudget(movieId):
        return 'b123'
    
    monkeypatch.setattr('movie_handler.getMovieBudget', mock_getMovieBudget)
    
    url = f'/{helpers.get_domain()}/{helpers.get_version()}/movies/'
    data = {'movieId': 144}
    
    response = client.post(url, data=data)
    
    assert response.status_code == 200

方法3：测试环境使用MySQL（不推荐） #

将测试配置改为使用真实MySQL数据库，但需注意测试数据隔离：

@pytest.fixture
def client():
    from movie_handler import app

    app.config['TESTING'] = True
    # 替换为你的测试MySQL连接串
    app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://username:password@localhost/test_movie_db'
    return app.test_client()

注意：必须使用专门的测试数据库，测试前后清理数据，避免污染业务数据。

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