Programmiererforum

from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from contextlib import asynccontextmanager
import os

class Database:
def __init__(self) -> None:
self.engine = None

def connect(self) -> None:
if not self.engine:
self.engine = create_async_engine(os.environ.get("DATABASE_URL", "sqlite:///db.sqlite")
)
else:
raise ConnectionAbortedError("Database is already connected")

@asynccontextmanager
async def get_async_session(self):
if not self.engine:
raise ConnectionError("Database is not connected")

async with AsyncSession(self.engine) as session:
try:
yield session
finally:
await session.close()

async def test_register_user_mutation(self, gql_schema: Schema, engine):
with patch("sqlmodel.create_engine") as mock_create_engine:
mock_create_engine.return_value = engine

@pytest.fixture(scope="session", autouse=True)
def app() -> FastAPI:
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
SQLModel.metadata.create_all(engine)

with patch("sqlmodel.create_engine") as mock_create_engine:
mock_create_engine.return_value = engine

from daytistics.main import app

return app
< /code>
und die Sitzung: < /p>
async def test_register_user_mutation(self, gql_schema: Schema, engine):
with patch("sqlmodel.Session", autospec=True) as mock_session_class:
mock_session_class.return_value = Session(engine)
< /code>
Aber jedes Mal, wenn sie in die Produktionsdatenbank geschrieben wurde.def create_app() -> FastAPI:
dotenv.load_dotenv()
app = FastAPI()

register_dependencies()

with container.sync_context() as ctx:
db = ctx.resolve(Database)
db.connect()

graphql_router = create_graphql_router()
app.include_router(graphql_router, prefix="/graphql")

return app

app = create_app()
< /code>
Ein MonkeyPatch funktionierte auch nicht: < /p>
@pytest.fixture(scope="function")
async def mock_db(monkeypatch):
test_engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
echo=False,
future=True,
connect_args={"check_same_thread": False},
)

async with test_engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)

def mock_connect(self):
self.engine = test_engine

@asynccontextmanager
async def mock_get_async_session(self):
async with AsyncSession(test_engine) as session:
try:
yield session
finally:
await session.close()

monkeypatch.setattr(Database, "connect", mock_connect)
monkeypatch.setattr(Database, "get_async_session", mock_get_async_session)

return test_engine

[url=viewtopic.php?t=14917]Ich möchte[/url] meinen GraphQL -Endpunkt testen. Dazu muss ich die Datenbank verspotten. < /P>
[code]from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from contextlib import asynccontextmanager
import os

class Database:
def __init__(self) -> None:
self.engine = None

def connect(self) -> None:
if not self.engine:
self.engine = create_async_engine(os.environ.get("DATABASE_URL", "sqlite:///db.sqlite")
)
else:
raise ConnectionAbortedError("Database is already connected")

@asynccontextmanager
async def get_async_session(self):
if not self.engine:
raise ConnectionError("Database is not connected")

async with AsyncSession(self.engine) as session:
try:
yield session
finally:
await session.close()
[/code]
Ich habe versucht, in meinem Test zu create create_engine verspottet:
[code]async def test_register_user_mutation(self, gql_schema: Schema, engine):
with patch("sqlmodel.create_engine") as mock_create_engine:
mock_create_engine.return_value = engine
[/code]
In der App Fixture:
[code]@pytest.fixture(scope="session", autouse=True)
def app() -> FastAPI:
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
SQLModel.metadata.create_all(engine)

with patch("sqlmodel.create_engine") as mock_create_engine:
mock_create_engine.return_value = engine

from daytistics.main import app

return app
< /code>
und die Sitzung: < /p>
async def test_register_user_mutation(self, gql_schema: Schema, engine):
with patch("sqlmodel.Session", autospec=True) as mock_session_class:
mock_session_class.return_value = Session(engine)
< /code>
Aber jedes Mal, wenn sie in die Produktionsdatenbank geschrieben wurde.def create_app() -> FastAPI:
dotenv.load_dotenv()
app = FastAPI()

register_dependencies()

with container.sync_context() as ctx:
db = ctx.resolve(Database)
db.connect()

graphql_router = create_graphql_router()
app.include_router(graphql_router, prefix="/graphql")

return app

app = create_app()
< /code>
Ein MonkeyPatch funktionierte auch nicht: < /p>
@pytest.fixture(scope="function")
async def mock_db(monkeypatch):
test_engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
echo=False,
future=True,
connect_args={"check_same_thread": False},
)

async with test_engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)

def mock_connect(self):
self.engine = test_engine

@asynccontextmanager
async def mock_get_async_session(self):
async with AsyncSession(test_engine) as session:
try:
yield session
finally:
await session.close()

monkeypatch.setattr(Database, "connect", mock_connect)
monkeypatch.setattr(Database, "get_async_session", mock_get_async_session)

return test_engine
[/code]