对依赖于数据库的函数进行单元测试 [英] Unit testing a function that depends on database

问题描述

我正在对某些功能进行测试.我有一个使用数据库查询的函数.所以，我浏览了一些博客和文档，这些博客和文档说我们必须在内存或测试数据库中创建一个才能使用这些功能.下面是我的函数，

I am running tests on some functions. I have a function that uses database queries. So, I have gone through the blogs and docs that say we have to make an in memory or test database to use such functions. Below is my function,

def already_exists(story_data,c):
    # TODO(salmanhaseeb): Implement de-dupe functionality by checking if it already
    # exists in the DB.
    c.execute("""SELECT COUNT(*) from posts where post_id = ?""", (story_data.post_id,))
    (number_of_rows,)=c.fetchone()
    if number_of_rows > 0:
        return True
    return False

此函数命中生产数据库.我的问题是，在测试时，我创建了一个内存数据库并在那里填充我的值，我将查询该数据库(测试数据库).但是我想测试我的 already_exists() 函数，在从测试中调用我的 already_exists 函数后，我的生产数据库将被命中.测试此功能时如何使我的测试数据库命中?

This function hits the production database. My question is that, when in testing, I create an in memory database and populate my values there, I will be querying that database (test DB). But I want to test my already_exists() function, after calling my already_exists function from test, my production db will be hit. How do I make my test DB hit while testing this function?

推荐答案

问题在于确保您的代码始终使用相同的数据库连接.然后您可以将其设置为适合当前环境的任何值.

The issue is ensuring that your code consistently uses the same database connection. Then you can set it once to whatever is appropriate for the current environment.

与其在方法之间传递数据库连接，不如使其成为单例.

Rather than passing the database connection around from method to method, it might make more sense to make it a singleton.

def already_exists(story_data):
    # Here `connection` is a singleton which returns the database connection.
    connection.execute("""SELECT COUNT(*) from posts where post_id = ?""", (story_data.post_id,))
    (number_of_rows,) = connection.fetchone()
    if number_of_rows > 0:
        return True
    return False

或者让connection成为每个类的一个方法，将already_exists变成一个方法.无论如何，它应该是一种方法.

Or make connection a method on each class and turn already_exists into a method. It should probably be a method regardless.

def already_exists(self):
    # Here the connection is associated with the object.
    self.connection.execute("""SELECT COUNT(*) from posts where post_id = ?""", (self.post_id,))
    (number_of_rows,) = self.connection.fetchone()
    if number_of_rows > 0:
        return True
    return False

但实际上您不应该自己滚动此代码.相反，您应该使用 ORM，例如 SQLAlchemy 为您处理基本查询和连接管理.它只有一个连接，"session".

But really you shouldn't be rolling this code yourself. Instead you should use an ORM such as SQLAlchemy which takes care of basic queries and connection management like this for you. It has a single connection, the "session".

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from sqlalchemy_declarative import Address, Base, Person

engine = create_engine('sqlite:///sqlalchemy_example.db')
Base.metadata.bind = engine

DBSession = sessionmaker(bind=engine)
session = DBSession()

然后你用它来进行查询.例如，它有一个 存在方法.

Then you use that to make queries. For example, it has an exists method.

session.query(Post.id).filter(q.exists()).scalar()

使用 ORM 将极大地简化您的代码.这是基础简短教程和更长更完整的教程.

Using an ORM will greatly simplify your code. Here's a short tutorial for the basics, and a longer and more complete tutorial.

这篇关于对依赖于数据库的函数进行单元测试的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！