64 lines
1.3 KiB
Python
64 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import AsyncIterator, Iterator
|
|
from pathlib import Path
|
|
|
|
import psycopg
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from config import Settings
|
|
from db import Database
|
|
|
|
TEST_DATABASE_URL = os.getenv("TEST_DATABASE_URL", "").strip()
|
|
|
|
TABLES = (
|
|
"requests",
|
|
"blocked_requests",
|
|
"pending_payments",
|
|
"users",
|
|
"user_contexts",
|
|
"app_settings",
|
|
)
|
|
|
|
|
|
def database_url() -> str:
|
|
if not TEST_DATABASE_URL:
|
|
pytest.skip("TEST_DATABASE_URL не задан")
|
|
return TEST_DATABASE_URL
|
|
|
|
|
|
def drop_tables(url: str) -> None:
|
|
with psycopg.connect(url, autocommit=True) as connection:
|
|
for table in TABLES:
|
|
connection.execute(f"DROP TABLE IF EXISTS {table} CASCADE")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db() -> AsyncIterator[Database]:
|
|
url = database_url()
|
|
drop_tables(url)
|
|
database = Database(url)
|
|
await database.connect()
|
|
try:
|
|
yield database
|
|
finally:
|
|
await database.close()
|
|
drop_tables(url)
|
|
|
|
|
|
@pytest.fixture
|
|
def web_settings(tmp_path: Path) -> Iterator[Settings]:
|
|
url = database_url()
|
|
drop_tables(url)
|
|
settings = Settings(
|
|
bot_token="",
|
|
database_url=url,
|
|
storage_dir=tmp_path / "storage",
|
|
)
|
|
try:
|
|
yield settings
|
|
finally:
|
|
drop_tables(url)
|