1  from typing import Any, AsyncIterator
 2  from unittest.mock import AsyncMock, MagicMock
 3  
 4  import pytest
 5  from _pytest.monkeypatch import MonkeyPatch
 6  from httpx import AsyncClient
 7  from pytest_mock import MockerFixture
 8  from sqlalchemy.ext.asyncio import create_async_engine
 9  
10  from api.app import app
11  from api.database import db
12  
13  
14  @pytest.fixture(autouse=True)
15  async def database(monkeypatch: MonkeyPatch) -> None:
16      monkeypatch.setattr(db, "engine", create_async_engine("sqlite+aiosqlite:///:memory:"))
17      await db.create_tables()
18  
19  
20  @pytest.fixture
21  async def session(request: Any) -> AsyncIterator[MagicMock]:
22      session = MagicMock()
23      session.user.enabled = True
24      session.user.admin = False
25      if marker := request.node.get_closest_marker("user_params"):
26          for k, v in marker.kwargs.items():
27              setattr(session.user, k, v)
28              if k == "id":
29                  session.user_id = v
30      yield session
31  
32  
33  @pytest.fixture
34  async def client() -> AsyncIterator[AsyncClient]:
35      async with AsyncClient(app=app, base_url="http://test") as client:
36          yield client
37  
38  
39  @pytest.fixture
40  async def auth_client(client: AsyncClient, mocker: MockerFixture) -> AsyncIterator[AsyncClient]:
41      mocker.patch("api.auth.StaticTokenAuth._check_token", AsyncMock(return_value=True))
42      mocker.patch("api.auth.JWTAuth.__call__", AsyncMock(return_value={"foo": "bar"}))
43      yield client
44  
45  
46  @pytest.fixture
47  async def user_client(client: AsyncClient, session: MagicMock, mocker: MockerFixture) -> AsyncIterator[AsyncClient]:
48      mocker.patch("api.auth.Session.from_access_token", AsyncMock(return_value=session))
49      yield client
50  
51  
52  @pytest.fixture
53  async def admin_client(user_client: AsyncClient, session: MagicMock) -> AsyncIterator[AsyncClient]:
54      session.user.admin = True
55      yield user_client