python - 使用 pytest 模拟 psycopg2 的正确方法是什么？

Question

我需要在没有实际连接的情况下模拟数据库连接。我发现的所有答案都试图以不同的方式模拟方法，连接到 docker db，连接到本地运行的实际 PostgreSQL。我相信我需要模拟变体，但我无法在脑海中制定我应该如何模拟。我错过了什么吗？我是否朝着错误的方向前进？

我使用 PostgreSQL 和 psycopg2。包psycopg2-binary

数据库连接:

import os

import psycopg2
from loguru import logger
from psycopg2.extensions import parse_dsn


def init_currency_history_table(cursor):
    create_users_table_query = """
        CREATE TABLE IF NOT EXISTS history(
          id BIGINT PRIMARY KEY NOT NULL,
          event TEXT,
          creation_date TIMESTAMPTZ DEFAULT NOW()
        );
    """
    cursor.execute(create_users_table_query)


def load_db(db_url):
    db = psycopg2.connect(**db_url)
    db.autocommit = True
    return db


class PostgresqlApi(object):

    def __init__(self, load=load_db):
        logger.info(os.environ.get('DATABASE_URL'))
        db_url = parse_dsn(os.environ.get('DATABASE_URL'))
        db_url['sslmode'] = 'require'
        logger.info('HOST: {0}'.format(db_url.get('host')))
        self.db = load_db(db_url)
        self.cursor = self.db.cursor()

        init_currency_history_table(self.cursor)
        self.db.commit()

    def add_event(self, *, event):
        insert_event_table = """
            INSERT INTO history (event) VALUES (%s);
        """
        self.cursor.execute(insert_event_table, (event))

    def events(self):
        select_event_table = """SELECT * FROM event;"""
        self.cursor.execute(select_event_table)
        return self.cursor.fetchall()

    def close(self):
        self.cursor.close()
        self.db.close()

我使用 DB for Falcon API。

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from decimal import Decimal, getcontext

from db import PostgresqlApi

app = FastAPI()
security = HTTPBasic()
database = None


def db_connection():
    global database
    if not database:
        database = PostgresqlApi()
    return database

def check_basic_auth_creds(credentials: HTTPBasicCredentials = Depends(security)):
    correct_username = secrets.compare_digest(credentials.username, os.environ.get('APP_USERNAME'))
    correct_password = secrets.compare_digest(credentials.password, os.environ.get('APP_PASSWORD'))
    if not (correct_username and correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username and password",
            headers={'WWW-Authenticate': 'Basic'}
        )
    return credentials

@app.get("/currencies")
def read_currencies(credentials: HTTPBasicCredentials = Depends(check_basic_auth_creds)):
    db = db_connection()
    return {'get events': 'ok'}

我尝试过不同的方法和插件。其中包括pytest-pgsql、pytest-postgresql。

Answer 1

我找到的解决方案如下。

• 创建了具有与 PostgresqlApi 结构完全相同的假类。 (参见下面的实现)
• 为 db_connection 方法创建了固定装置。 (参见下面的实现)

假类实现

class FakePostgresqlApi(PostgresqlApi):

    event_list = []

    def __init__(self):
        pass

    def add_event(self, *, event):
        self.event_list.append([1, 'magic trick', 1653630607])

    def events(self):
        return self.event_list

    def close(self):
        self.event_list.clear()

fixture

from unittest.mock import MagicMock

@pytest.fixture
def mock_db_connection(mocker):
    mocker.patch('src.main.db_connection', MagicMock(return_value=FakePostgresqlApi()))

测试本身是:

def test_read_events(mock_db_connection):
   # Do whatever I need here, in my case call Falcon API test client