所以我使用 FastAPI 来创建我的应用程序服务器，其中有一些 url，我对数据库进行了一些粗略的操作。例如：当我点击一个/createuser/带有 json 正文的 url 时，它会在 db 中插入一条记录。所以我正在使用 pytest 来测试这些东西，但 pytest 似乎使用了我的 FastAPI 应用程序正在使用的相同数据库。我想创建一个模拟数据库，以便它使用另一个数据库来运行测试。我也用过pytest.fixtures，但我想有什么问题。请帮忙解决这个问题。我也使用 postgres 作为我的数据库。
- 创建 FastAPI 应用程序的主应用程序
from mock_server.database import engine, SessionLocal, Base
def create_app(connection):
"""This function creates the FastAPI app server.
Returns:
app[FastAPI] -- The main app server
"""
Base.metadata.create_all(bind=connection)
print("from main:app",connection)
app = FastAPI(
title="Masquerader - Mock Server",
description="""This project allows you to mock any system,
or service that you wish to."""
)
return app
app = create_app(engine)
def get_db():
"""To Get the current db connection.
Yields:
[db] -- [The current connection]
"""
try:
db = SessionLocal()
yield db
finally:
db.close()
@app.post("/createuser/", response_model=schemas.User)
async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
"""Endpoint for creating a user.
Arguments:
user {schemas.UserCreate} -- JSON Body with user details to create
Keyword Arguments:
db {Session} -- Current db connection
"""
user = crud.create_user(db=db, user=user)
if user is None:
raise HTTPException(status_code=400, detail="User Already Exists")
return user
- 我的数据库类，其中定义了 db 连接
from config import current_config
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# sqlalchemy_db_url = "postgresql://fyndlocal:fynd@123@localhost:5432/mockdb"
if os.environ.get("ENV") == "development":
engine = create_engine(current_config.POSTGRES_MASQUERADER)
if os.environ.get("ENV") is None:
print("Lol test hai yeh")
engine = create_engine(current_config.MASQUERADER_LOCAL)
if os.environ.get("ENV") == "pre-production":
os.environ.__setitem__("POSTGRES_USER", "runner")
os.environ.__setitem__("POSTGRES_PASS", "")
os.environ.__setitem__("POSTGRES_HOST", "postgres")
engine = create_engine(current_config.POSTGRES_TEST_GITLAB)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
- 我的 CRUD 操作课
def create_user(db: Session, user: schemas.UserCreate):
"""Operation to create a user in the db.
Arguments:
db {Session} -- gets current db session
user {schemas.UserCreate} -- JSON Body that contains
user details to be created.
Returns:
db_user[dict] -- Details of created user.
"""
hashed_pass = common.hash_generate(user.password)
old_user = (
db.query(models.users).filter(models.users.name == user.name).first()
)
if old_user is not None:
return None
db_user = models.users(
name=user.name,
password=hashed_pass,
is_active=user.is_active,
is_admin=user.is_admin,
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
- 我的测试类（我希望它使用模拟数据库）
from main import app
import pytest
from sqlalchemy import create_engine
from starlette.testclient import TestClient
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
@pytest.yield_fixture(scope="module")
def application():
"""Yiest TestClient from FastAPI.
Yields:
app[TestClient] -- Testing based application
"""
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=cleanup_database)
db = SessionLocal()
yield TestClient(app)
@pytest.fixture(scope="module")
def cleanup_database():
"""Creates a mock database for testing purposes.
Creates a mock database on server for testing and deletes once done.
"""
username = os.environ.get("POSTGRES_USER", "fyndlocal")
password = os.environ.get("POSTGRES_PASS", "fynd@123")
postgres_host = os.environ.get("POSTGRES_HOST", "localhost")
postgres_port = os.environ.get("POSTGRES_PORT_5432_TCP_PORT", 5432)
postgres_db = os.environ.get("POSTGRES_DB", "mockdb")
if not password:
db_dict = {
"username": username,
"password": password,
"host": postgres_host,
"port": postgres_port,
"db": postgres_db,
}
default_db_url = current_config.POSTGRES_NOPASS_DSN.format(**db_dict)
print("if no pass", default_db_url)
else:
db_dict = {
"username": username,
"password": password,
"host": postgres_host,
"port": postgres_port,
"db": postgres_db,
}
default_db_url = current_config.POSTGRES_PASS_DSN.format(**db_dict)
print("if pass", default_db_url)
test_masqueraderdb_url = current_config.POSTGRES_TEST_DSN.format(**db_dict)
print("POSTGRES Config")
print(db_dict)
print(test_masqueraderdb_url)
db_name = test_masqueraderdb_url.split("/")[-1]
engine = create_engine(default_db_url)
conn = engine.connect()
# db.drop_all(engine)
# db.create_all(engine)
try:
conn.execution_options(isolation_level="AUTOCOMMIT").execute(
f"CREATE DATABASE {db_name}"
)
except Exception as e:
print("this stage", e)
rv = create_engine(test_masqueraderdb_url)
db.create_all(rv)
yield rv
db.drop_all(rv)
def test_create_user(application,cleanup_database):
"""Test to create user exists in db."""
response = application.post(
"/createuser/",
json={
"name": "test",
"is_active": True,
"is_admin": True,
"password": "test123",
},
)
expected_resp = {
"name": "test",
"is_active": True,
"is_admin": True,
"id": 1,
"urls": [],
}
assert response.json() == expected_resp
以下测试失败，因为它总是让用户出现在我当前的数据库中。
谢谢你。