65 lines
2.5 KiB
Python
65 lines
2.5 KiB
Python
import httpx
|
|
from typing import Optional
|
|
from tests.base.config import USER_EMAIL, USER_PASSWORD, BASE_URL
|
|
|
|
|
|
class AuthenticationWeb:
|
|
def __init__(self, user_email: str = USER_EMAIL, password: str = USER_PASSWORD, base_url: str = BASE_URL):
|
|
self.user_email = user_email
|
|
self.password = password
|
|
self.base_url = base_url
|
|
self.token: Optional[str] = None
|
|
|
|
def login(self):
|
|
"""Login and store JWT token"""
|
|
with httpx.Client(base_url=self.base_url) as client:
|
|
resp = self.do_login(self.user_email, self.password)
|
|
self.token = resp.json()["access_token"]
|
|
return resp
|
|
|
|
def do_login(self, user_email: str = USER_EMAIL, password: str = USER_PASSWORD):
|
|
"""Login and store JWT token"""
|
|
with httpx.Client(base_url=self.base_url) as client:
|
|
resp = client.post("/api/auth/signin/signin-with-email-and-password", json={
|
|
"email": user_email,
|
|
"password": password
|
|
})
|
|
return resp
|
|
|
|
async def request(self, method: str, url: str, **kwargs):
|
|
"""Send authenticated request"""
|
|
headers = kwargs.pop("headers", {})
|
|
if self.token:
|
|
headers["Authorization"] = f"Bearer {self.token}"
|
|
async with httpx.AsyncClient(base_url=self.base_url) as client:
|
|
resp = await client.request(method, url, headers=headers, **kwargs)
|
|
return resp
|
|
|
|
async def create_role(self, role_data: dict):
|
|
"""Create a new role via API"""
|
|
return await self.request("POST", "/api/auth/role/create", json=role_data)
|
|
|
|
async def delete_role(self, role_data: dict):
|
|
"""Delete role via API"""
|
|
return await self.request("POST", "/api/auth/role/delete", json=role_data)
|
|
|
|
async def update_role(self, role_data: dict):
|
|
"""Update role via API"""
|
|
return await self.request("POST", "/api/auth/role/update", json=role_data)
|
|
|
|
async def query_roles(self, params: dict = None):
|
|
"""Query roles via API"""
|
|
if params is None:
|
|
params = {}
|
|
return await self.request("POST", "/api/auth/role/query", json=params)
|
|
|
|
async def create_permission(self, perm_data: dict):
|
|
"""Create a new permission via API"""
|
|
return await self.request("POST", "/api/auth/permission/create", json=perm_data)
|
|
|
|
async def query_permissions(self, params: dict = None):
|
|
"""Query permissions via API"""
|
|
return await self.request("GET", "/api/auth/permission/query", params=params)
|
|
|
|
|