72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
from collections.abc import AsyncGenerator
|
|
from urllib.parse import urlencode
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.api.deps import get_current_telegram_user
|
|
from app.core.config import settings
|
|
from app.db.base import Base
|
|
from app.db.session import get_session
|
|
from app.main import app
|
|
from app.models import car, expense, gamification, push, user # noqa: F401
|
|
|
|
TEST_BOT_TOKEN = "123456:test-token"
|
|
TEST_INTERNAL_TOKEN = "internal-test-token"
|
|
|
|
|
|
def make_init_data(telegram_id: int, first_name: str = "Test") -> str:
|
|
user_payload = json.dumps(
|
|
{"id": telegram_id, "first_name": first_name, "username": str(telegram_id)},
|
|
separators=(",", ":"),
|
|
)
|
|
values = {"auth_date": str(int(time.time())), "user": user_payload}
|
|
data_check_string = "\n".join(f"{key}={values[key]}" for key in sorted(values))
|
|
secret = hmac.new(b"WebAppData", TEST_BOT_TOKEN.encode(), hashlib.sha256).digest()
|
|
values["hash"] = hmac.new(secret, data_check_string.encode(), hashlib.sha256).hexdigest()
|
|
return urlencode(values)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def configure_settings() -> None:
|
|
settings.bot_token = TEST_BOT_TOKEN
|
|
settings.internal_api_token = TEST_INTERNAL_TOKEN
|
|
settings.app_env = "test"
|
|
settings.allow_dev_auth = False
|
|
yield
|
|
|
|
|
|
@pytest.fixture()
|
|
async def client() -> AsyncGenerator[AsyncClient, None]:
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
async def override_session() -> AsyncGenerator:
|
|
async with session_factory() as session:
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = override_session
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://testserver") as test_client:
|
|
yield test_client
|
|
app.dependency_overrides.pop(get_session, None)
|
|
app.dependency_overrides.pop(get_current_telegram_user, None)
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture()
|
|
def auth_headers() -> dict[str, str]:
|
|
return {"X-Telegram-Init-Data": make_init_data(1001)}
|
|
|
|
|
|
@pytest.fixture()
|
|
def other_auth_headers() -> dict[str, str]:
|
|
return {"X-Telegram-Init-Data": make_init_data(2002)}
|