8b925263a2
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
42 lines
1.1 KiB
Python
42 lines
1.1 KiB
Python
import pytest
|
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from mqtt_home.database import Base
|
|
from mqtt_home.config import Settings
|
|
|
|
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
|
|
|
|
|
@pytest.fixture
|
|
def settings():
|
|
return Settings(
|
|
mqtt_host="localhost",
|
|
mqtt_port=1883,
|
|
emqx_api_url="http://localhost:18083/api/v5",
|
|
emqx_api_key="test-key",
|
|
emqx_api_secret="test-secret",
|
|
database_url=TEST_DATABASE_URL,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def engine():
|
|
engine = create_async_engine(
|
|
TEST_DATABASE_URL,
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield engine
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_session(engine):
|
|
async_session = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with async_session() as session:
|
|
yield session |