75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest_asyncio
|
|
import requests_async
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
class Env(BaseSettings):
|
|
HOST:str
|
|
PORT:str
|
|
TEST_USERNAME:str
|
|
TEST_PASSWORD:str
|
|
|
|
model_config=SettingsConfigDict(env_file="configs/.e2e.env", extra=None)
|
|
|
|
e2e_settings=Env() # type: ignore[call-arg]
|
|
|
|
class MySession(requests_async.AsyncSession):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.headers = {}
|
|
self.token = None
|
|
|
|
async def request(self, method:str, url:str, **kwargs):
|
|
if self.token:
|
|
self.headers['Authorization'] = f"Bearer {self.token}"
|
|
kwargs.setdefault('headers', self.headers)
|
|
return await super().request(method, url, **kwargs)
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def auth_fixture():
|
|
|
|
payload = {"username": e2e_settings.TEST_USERNAME, "password": e2e_settings.TEST_PASSWORD}
|
|
url = f'http://{e2e_settings.HOST}:{e2e_settings.PORT}'
|
|
|
|
async with MySession() as session:
|
|
response = await session.post(url + "/protected/token", data=payload)
|
|
response.raise_for_status()
|
|
session.token = response.json()["access_token"]
|
|
yield session
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_user_fixture(request, auth_fixture: MySession):
|
|
|
|
url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
|
test_id=uuid4()
|
|
|
|
direct_permission_param, group_param = request.param
|
|
|
|
new_user_record={
|
|
"first_name":f"TEST_{test_id}",
|
|
"last_name":f"TEST_{test_id}",
|
|
"middle_name":f"TEST_{test_id}",
|
|
"email":f"TEST_{test_id}@d.d",
|
|
"plain_password":"Test1234!",
|
|
"direct_permissions":direct_permission_param,
|
|
"group":group_param
|
|
}
|
|
response=await auth_fixture.post(f"{url}/user/create_user", json=new_user_record)
|
|
response.raise_for_status()
|
|
|
|
|
|
async with MySession() as session:
|
|
|
|
payload={"username": new_user_record.get("email"), "password": new_user_record.get("plain_password")}
|
|
|
|
response = await session.post(url + "/protected/token", data=payload)
|
|
response.raise_for_status()
|
|
session.token = response.json()["access_token"]
|
|
|
|
yield (session, new_user_record)
|
|
|
|
response=await auth_fixture.post(f"{url}/user/delete_user_hard", params={"email":new_user_record.get("email")})
|
|
response.raise_for_status() |