modernize
Some checks failed
Python lint and test / linttest (push) Has been cancelled

This commit is contained in:
Piotr Domański 2026-01-03 20:36:46 +01:00
parent 29445273ca
commit af6c753e90
4 changed files with 27 additions and 14 deletions

View file

@ -1,6 +1,6 @@
from fooder.app import app
from fooder.tasks_app import app as tasks_app
from httpx import AsyncClient
from httpx import ASGITransport, AsyncClient
import pytest
import httpx
import os
@ -12,8 +12,12 @@ class Client:
username: str | None = None,
password: str | None = None,
):
self.client = AsyncClient(app=app, base_url="http://testserver/api")
self.client.headers["Accept"] = "application/json"
self.client = lambda: AsyncClient(
transport=ASGITransport(app=app),
base_url="http://testserver/api",
headers=self.headers,
)
self.headers = {"Accept": "application/json"}
def set_token(self, token: str) -> None:
"""set_token.
@ -22,7 +26,7 @@ class Client:
:type token: str
:rtype: None
"""
self.client.headers["Authorization"] = "Bearer " + token
self.headers["Authorization"] = "Bearer " + token
async def create_user(self, username: str, password: str) -> None:
data = {"username": username, "password": password}
@ -57,26 +61,33 @@ class Client:
self.set_token(result["access_token"])
async def get(self, path: str, **kwargs) -> httpx.Response:
return await self.client.get(path, **kwargs)
async with self.client() as client:
return await client.get(path, **kwargs)
async def delete(self, path: str, **kwargs) -> httpx.Response:
return await self.client.delete(path, **kwargs)
async with self.client() as client:
return await client.delete(path, **kwargs)
async def post(self, path: str, **kwargs) -> httpx.Response:
return await self.client.post(path, **kwargs)
async with self.client() as client:
return await client.post(path, **kwargs)
async def patch(self, path: str, **kwargs) -> httpx.Response:
return await self.client.patch(path, **kwargs)
async with self.client() as client:
return await client.patch(path, **kwargs)
class TasksClient(Client):
def __init__(self, authorized: bool = True):
super().__init__()
self.client = AsyncClient(app=tasks_app, base_url="http://testserver/api")
self.client.headers["Accept"] = "application/json"
self.client = lambda: AsyncClient(
transport=ASGITransport(app=tasks_app),
base_url="http://testserver/api",
headers=self.headers,
)
if authorized:
self.client.headers["Authorization"] = "Bearer " + self.get_token()
self.headers["Authorization"] = "Bearer " + self.get_token()
def get_token(self) -> str:
return os.getenv("API_KEY")

View file

@ -6,6 +6,7 @@ uvicorn[standard]
asyncpg
psycopg2-binary==2.9.3
python-jose[cryptography]
bcrypt<5.0.0
passlib[bcrypt]
fastapi-users
requests

View file

@ -4,6 +4,7 @@ pydantic_settings
sqlalchemy[postgresql_asyncpg]
uvicorn[standard]
python-jose[cryptography]
bcrypt<5.0.0
passlib[bcrypt]
fastapi-users
pytest

View file

@ -15,13 +15,13 @@ export SECRET_KEY=$(openssl rand -hex 32)
export REFRESH_SECRET_KEY=$(openssl rand -hex 32)
export API_KEY=$(openssl rand -hex 32)
python -m fooder --create-tables
python3 -m fooder --create-tables
# finally run tests
if [[ $# -eq 1 ]]; then
python -m pytest fooder --disable-warnings -sv -k "${1}"
python3 -m pytest fooder --disable-warnings -sv -k "${1}"
else
python -m pytest fooder --disable-warnings -sv
python3 -m pytest fooder --disable-warnings -sv
fi
status=$?