import uuid
from datetime import datetime, timedelta, timezone
from typing import AsyncGenerator

import pytest
from pydantic import UUID4
from sqlalchemy import exc
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import Session, SQLModel, create_engine

from fastapi_users_db_sqlmodel import SQLModelUserDatabase, SQLModelUserDatabaseAsync
from fastapi_users_db_sqlmodel.access_token import (
    SQLModelAccessTokenDatabase,
    SQLModelAccessTokenDatabaseAsync,
    SQLModelBaseAccessToken,
)
from tests.conftest import UserDB


class AccessToken(SQLModelBaseAccessToken, table=True):
    pass


@pytest.fixture
def user_id() -> UUID4:
    return uuid.UUID("a9089e5d-2642-406d-a7c0-cbc641aca0ec")


async def init_sync_session(url: str) -> AsyncGenerator[Session, None]:
    engine = create_engine(url, connect_args={"check_same_thread": False})
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session
    SQLModel.metadata.drop_all(engine)


async def init_async_session(url: str) -> AsyncGenerator[AsyncSession, None]:
    engine = create_async_engine(url, connect_args={"check_same_thread": False})
    make_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
        async with make_session() as session:
            yield session
        await conn.run_sync(SQLModel.metadata.drop_all)


@pytest.fixture(
    params=[
        (
            init_sync_session,
            "sqlite:///./test-sqlmodel-access-token.db",
            SQLModelAccessTokenDatabase,
            SQLModelUserDatabase,
        ),
        (
            init_async_session,
            "sqlite+aiosqlite:///./test-sqlmodel-access-token.db",
            SQLModelAccessTokenDatabaseAsync,
            SQLModelUserDatabaseAsync,
        ),
    ],
    ids=["sync", "async"],
)
async def sqlmodel_access_token_db(
    request, user_id: UUID4
) -> AsyncGenerator[SQLModelAccessTokenDatabase, None]:
    create_session = request.param[0]
    database_url = request.param[1]
    access_token_database_class = request.param[2]
    user_database_class = request.param[3]
    async for session in create_session(database_url):
        user = UserDB(
            id=user_id, email="<EMAIL>", hashed_password="<PASSWORD>"
        )
        user_db = user_database_class(UserDB, session)
        await user_db.create(user)
        yield access_token_database_class(AccessToken, session)


@pytest.mark.asyncio
@pytest.mark.db
async def test_queries(
    sqlmodel_access_token_db: SQLModelAccessTokenDatabase[AccessToken],
    user_id: UUID4,
):
    access_token = AccessToken(token="TOKEN", user_id=user_id)

    # Create
    access_token_db = await sqlmodel_access_token_db.create(access_token)
    assert access_token_db.token == "TOKEN"
    assert access_token_db.user_id == user_id

    # Update
    access_token_db.created_at = datetime.now(timezone.utc)
    await sqlmodel_access_token_db.update(access_token_db)

    # Get by token
    access_token_by_token = await sqlmodel_access_token_db.get_by_token(
        access_token_db.token
    )
    assert access_token_by_token is not None

    # Get by token expired
    access_token_by_token = await sqlmodel_access_token_db.get_by_token(
        access_token_db.token, max_age=datetime.now(timezone.utc) + timedelta(hours=1)
    )
    assert access_token_by_token is None

    # Get by token not expired
    access_token_by_token = await sqlmodel_access_token_db.get_by_token(
        access_token_db.token, max_age=datetime.now(timezone.utc) - timedelta(hours=1)
    )
    assert access_token_by_token is not None

    # Get by token unknown
    access_token_by_token = await sqlmodel_access_token_db.get_by_token(
        "NOT_EXISTING_TOKEN"
    )
    assert access_token_by_token is None

    # Delete token
    await sqlmodel_access_token_db.delete(access_token_db)
    deleted_access_token = await sqlmodel_access_token_db.get_by_token(
        access_token_db.token
    )
    assert deleted_access_token is None


@pytest.mark.asyncio
@pytest.mark.db
async def test_insert_existing_token(
    sqlmodel_access_token_db: SQLModelAccessTokenDatabase[AccessToken], user_id: UUID4
):
    access_token = AccessToken(token="TOKEN", user_id=user_id)
    await sqlmodel_access_token_db.create(access_token)

    with pytest.raises(exc.IntegrityError):
        await sqlmodel_access_token_db.create(
            AccessToken(token="TOKEN", user_id=user_id)
        )

pytest.fixture sqlmodel.SQLModel.metadata.create_all sqlalchemy.ext.asyncio.create_async_engine tests.conftest.UserDB pytest.raises uuid.UUID datetime.datetime.now sqlmodel.SQLModel.metadata.drop_all sqlmodel.Session datetime.timedelta sqlmodel.create_engine sqlalchemy.orm.sessionmaker
[((1537, 1859), 'pytest.fixture', 'pytest.fixture', ([], {'params': "[(init_sync_session, 'sqlite:///./test-sqlmodel-access-token.db',\n    SQLModelAccessTokenDatabase, SQLModelUserDatabase), (init_async_session,\n    'sqlite+aiosqlite:///./test-sqlmodel-access-token.db',\n    SQLModelAccessTokenDatabaseAsync, SQLModelUserDatabaseAsync)]", 'ids': "['sync', 'async']"}), "(params=[(init_sync_session,\n    'sqlite:///./test-sqlmodel-access-token.db',\n    SQLModelAccessTokenDatabase, SQLModelUserDatabase), (init_async_session,\n    'sqlite+aiosqlite:///./test-sqlmodel-access-token.db',\n    SQLModelAccessTokenDatabaseAsync, SQLModelUserDatabaseAsync)], ids=[\n    'sync', 'async'])\n", (1551, 1859), False, 'import pytest\n'), ((725, 774), 'uuid.UUID', 'uuid.UUID', (['"""a9089e5d-2642-406d-a7c0-cbc641aca0ec"""'], {}), "('a9089e5d-2642-406d-a7c0-cbc641aca0ec')\n", (734, 774), False, 'import uuid\n'), ((862, 923), 'sqlmodel.create_engine', 'create_engine', (['url'], {'connect_args': "{'check_same_thread': False}"}), "(url, connect_args={'check_same_thread': False})\n", (875, 923), False, 'from sqlmodel import Session, SQLModel, create_engine\n'), ((928, 964), 'sqlmodel.SQLModel.metadata.create_all', 'SQLModel.metadata.create_all', (['engine'], {}), '(engine)\n', (956, 964), False, 'from sqlmodel import Session, SQLModel, create_engine\n'), ((1028, 1062), 'sqlmodel.SQLModel.metadata.drop_all', 'SQLModel.metadata.drop_all', (['engine'], {}), '(engine)\n', (1054, 1062), False, 'from sqlmodel import Session, SQLModel, create_engine\n'), ((1156, 1223), 'sqlalchemy.ext.asyncio.create_async_engine', 'create_async_engine', (['url'], {'connect_args': "{'check_same_thread': False}"}), "(url, connect_args={'check_same_thread': False})\n", (1175, 1223), False, 'from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine\n'), ((1243, 1308), 'sqlalchemy.orm.sessionmaker', 'sessionmaker', (['engine'], {'class_': 'AsyncSession', 'expire_on_commit': '(False)'}), '(engine, class_=AsyncSession, expire_on_commit=False)\n', (1255, 1308), False, 'from sqlalchemy.orm import sessionmaker\n'), ((3037, 3063), 'datetime.datetime.now', 'datetime.now', (['timezone.utc'], {}), '(timezone.utc)\n', (3049, 3063), False, 'from datetime import datetime, timedelta, timezone\n'), ((974, 989), 'sqlmodel.Session', 'Session', (['engine'], {}), '(engine)\n', (981, 989), False, 'from sqlmodel import Session, SQLModel, create_engine\n'), ((2349, 2414), 'tests.conftest.UserDB', 'UserDB', ([], {'id': 'user_id', 'email': '"""<EMAIL>"""', 'hashed_password': '"""<PASSWORD>"""'}), "(id=user_id, email='<EMAIL>', hashed_password='<PASSWORD>')\n", (2355, 2414), False, 'from tests.conftest import UserDB\n'), ((4475, 4508), 'pytest.raises', 'pytest.raises', (['exc.IntegrityError'], {}), '(exc.IntegrityError)\n', (4488, 4508), False, 'import pytest\n'), ((3437, 3463), 'datetime.datetime.now', 'datetime.now', (['timezone.utc'], {}), '(timezone.utc)\n', (3449, 3463), False, 'from datetime import datetime, timedelta, timezone\n'), ((3466, 3484), 'datetime.timedelta', 'timedelta', ([], {'hours': '(1)'}), '(hours=1)\n', (3475, 3484), False, 'from datetime import datetime, timedelta, timezone\n'), ((3676, 3702), 'datetime.datetime.now', 'datetime.now', (['timezone.utc'], {}), '(timezone.utc)\n', (3688, 3702), False, 'from datetime import datetime, timedelta, timezone\n'), ((3705, 3723), 'datetime.timedelta', 'timedelta', ([], {'hours': '(1)'}), '(hours=1)\n', (3714, 3723), False, 'from datetime import datetime, timedelta, timezone\n')]