add tests for project #1
@@ -1,14 +1,14 @@
|
|||||||
import pytest
|
import pytest
|
||||||
|
from httpx import ASGITransport, AsyncClient
|
||||||
from sqlalchemy import NullPool, insert
|
from sqlalchemy import NullPool, insert
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||||||
|
|
||||||
from src.api.dependacies.db_dep import get_db
|
from src.api.dependacies.db_dep import get_db
|
||||||
from src.core.auth_manager import AuthManager
|
from src.core.auth_manager import AuthManager
|
||||||
from src.core.database import Base
|
from src.core.database import Base
|
||||||
|
from src.core.db_manager import DBManager
|
||||||
from src.main import app
|
from src.main import app
|
||||||
from src.models import * # noqa: F403
|
from src.models import * # noqa: F403
|
||||||
from src.repository.tasks import TasksRepo
|
|
||||||
from src.repository.users import UsersRepo
|
|
||||||
|
|
||||||
engine_null_pool = create_async_engine(
|
engine_null_pool = create_async_engine(
|
||||||
"sqlite+aiosqlite:///tests/test_db.db", poolclass=NullPool
|
"sqlite+aiosqlite:///tests/test_db.db", poolclass=NullPool
|
||||||
@@ -16,17 +16,9 @@ engine_null_pool = create_async_engine(
|
|||||||
test_session_maker = async_sessionmaker(engine_null_pool, expire_on_commit=False)
|
test_session_maker = async_sessionmaker(engine_null_pool, expire_on_commit=False)
|
||||||
|
|
||||||
|
|
||||||
class TestDBManager:
|
class TestDBManager(DBManager):
|
||||||
|
def __init__(self):
|
||||||
async def __aenter__(self) -> "TestDBManager":
|
self.session_factory = test_session_maker
|
||||||
self.session: AsyncSession = test_session_maker()
|
|
||||||
self.user = UsersRepo(self.session)
|
|
||||||
self.task = TasksRepo(self.session)
|
|
||||||
return self
|
|
||||||
|
|
||||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
|
||||||
await self.session.rollback()
|
|
||||||
await self.session.close()
|
|
||||||
|
|
||||||
|
|
||||||
async def get_test_db():
|
async def get_test_db():
|
||||||
@@ -40,24 +32,36 @@ async def db():
|
|||||||
yield db
|
yield db
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="function")
|
||||||
|
async def ac():
|
||||||
|
async with AsyncClient(
|
||||||
|
transport=ASGITransport(app=app), base_url="http://test"
|
||||||
|
) as ac:
|
||||||
|
yield ac
|
||||||
|
|
||||||
|
|
||||||
app.dependency_overrides[get_db] = get_test_db
|
app.dependency_overrides[get_db] = get_test_db
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="session", autouse=True)
|
@pytest.fixture(scope="session", autouse=True)
|
||||||
async def setup_database():
|
async def setup_database():
|
||||||
|
async with engine_null_pool.begin() as conn:
|
||||||
|
await conn.run_sync(Base.metadata.drop_all)
|
||||||
|
await conn.run_sync(Base.metadata.create_all)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session", autouse=True)
|
||||||
|
async def add_admin(setup_database):
|
||||||
hashed_pass = AuthManager.get_password_hash("admin")
|
hashed_pass = AuthManager.get_password_hash("admin")
|
||||||
user_admin = {
|
user_admin = {
|
||||||
"username": "admin",
|
"username": "admin",
|
||||||
"hashed_password": hashed_pass,
|
"hashed_password": hashed_pass,
|
||||||
"is_superuser": True,
|
"is_superuser": True,
|
||||||
}
|
}
|
||||||
async with engine_null_pool.begin() as conn:
|
|
||||||
await conn.run_sync(Base.metadata.drop_all)
|
|
||||||
await conn.run_sync(Base.metadata.create_all)
|
|
||||||
|
|
||||||
async with test_session_maker() as conn:
|
async with test_session_maker() as conn:
|
||||||
result = await conn.execute(
|
result = await conn.execute(
|
||||||
insert(UsersORM).values(user_admin).returning(UsersORM)) # noqa: F405
|
insert(UsersORM).values(user_admin).returning(UsersORM) # noqa: F405
|
||||||
|
)
|
||||||
await conn.commit()
|
await conn.commit()
|
||||||
admin = result.scalar_one()
|
admin = result.scalar_one()
|
||||||
assert admin.is_superuser is True
|
assert admin.is_superuser is True
|
||||||
|
|||||||
29
tests/integration_tests/test_auth_api.py
Normal file
29
tests/integration_tests/test_auth_api.py
Normal file
@@ -0,0 +1,29 @@
|
|||||||
|
from httpx import AsyncClient
|
||||||
|
|
||||||
|
from src.core.settings import settings
|
||||||
|
from src.schemas.users import User
|
||||||
|
|
||||||
|
|
||||||
|
async def test_registration(ac):
|
||||||
|
user = {"username": "kot", "email": "super@kot.ru", "password": "P@ssw0rd"}
|
||||||
|
result = await ac.post(
|
||||||
|
f"{settings.api.v1_login_url}/signup",
|
||||||
|
json=user,
|
||||||
|
)
|
||||||
|
assert result.status_code == 200
|
||||||
|
assert User.model_validate(result.json())
|
||||||
|
assert result.json()["is_active"]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_login(ac: AsyncClient):
|
||||||
|
result = await ac.post(
|
||||||
|
f"{settings.api.v1_login_url}/login",
|
||||||
|
data={
|
||||||
|
"grant_type": "password",
|
||||||
|
"username": "kot",
|
||||||
|
"password": "P@ssw0rd",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert result.status_code == 200
|
||||||
|
assert result.json().get("access_token")
|
||||||
|
|
||||||
@@ -1,10 +1,10 @@
|
|||||||
|
|
||||||
from src.core.auth_manager import AuthManager
|
from src.core.auth_manager import AuthManager
|
||||||
|
|
||||||
|
|
||||||
async def test_jwt():
|
async def test_jwt():
|
||||||
token = AuthManager.create_access_token(data={"id": 1, "sub": "testuser", "is_active": "True"})
|
token = AuthManager.create_access_token(
|
||||||
|
data={"id": 1, "sub": "testuser", "is_active": "True"}
|
||||||
|
)
|
||||||
assert token
|
assert token
|
||||||
encode_token = AuthManager.decode_access_token(token=token)
|
encode_token = AuthManager.decode_access_token(token=token)
|
||||||
assert encode_token['id'] == 1 and encode_token['sub'] == 'testuser'
|
assert encode_token["id"] == 1 and encode_token["sub"] == "testuser"
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user