e2e tests 0.1
This commit is contained in:
@@ -0,0 +1,61 @@
|
||||
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 get(self, url:str, **kwargs):
|
||||
if self.token:
|
||||
self.headers['Authorization'] = f"Bearer {self.token}"
|
||||
kwargs.setdefault('headers', self.headers)
|
||||
return await super().get(url, **kwargs)
|
||||
|
||||
@pytest_asyncio.fixture(scope="session", autouse=True)
|
||||
async def auth_fixture()->None:
|
||||
|
||||
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"]
|
||||
|
||||
@pytest_asyncio.fixture(scope="function", autouse=True)
|
||||
async def test_user_fixture():
|
||||
|
||||
url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
test_id=uuid4
|
||||
|
||||
new_user_record={
|
||||
"first_name":f"TEST_{test_id}",
|
||||
"last_name":f"TEST_{test_id}",
|
||||
"middle_name":f"TEST_{test_id}",
|
||||
"email":"TEST@TEST.TEST",
|
||||
"plain_password":"1234",
|
||||
"direct_permissions":[],
|
||||
"group":[]
|
||||
}
|
||||
response=await requests_async.post(f"{url}/create_user", data=new_user_record)
|
||||
print(response.json())
|
||||
yield
|
||||
new_user_to_delete={"email":"TEST@TEST.TEST"}
|
||||
response=await requests_async.post(f"{url}/delete_user_hard", data=new_user_to_delete)
|
||||
print(response.json())
|
||||
Reference in New Issue
Block a user