diff --git a/.gitignore b/.gitignore index 9e570b4..85bdcac 100644 --- a/.gitignore +++ b/.gitignore @@ -1,7 +1,11 @@ +test.db +test.db-journal + .vscode/ *.jpg *.PNG infra/.env +test.db # Byte-compiled / optimized / DLL files __pycache__/ diff --git a/app/api/endpoints/achievement.py b/app/api/endpoints/achievement.py index c0a80e5..6639e04 100644 --- a/app/api/endpoints/achievement.py +++ b/app/api/endpoints/achievement.py @@ -1,12 +1,16 @@ -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, HTTPException, Response, status from sqlalchemy.ext.asyncio import AsyncSession from app.api.validators import check_name_duplicate from app.core.db import get_async_session from app.core.user import current_superuser, current_user from app.crud import achievement_crud -from app.schemas.achievement import AchievementCreate, AchievementRead +from app.models import Achievement, User +from app.schemas.achievement import (AchievementCreate, AchievementRead, + AchievementUpdate) from app.services.endpoints_services import delete_obj +from app.services.utils import (Pagination, add_response_headers, + get_pagination_params, paginated) router = APIRouter() @@ -14,19 +18,66 @@ @router.get( "/", response_model=list[AchievementRead], - dependencies=[Depends(current_user)] + dependencies=[Depends(current_superuser)] ) async def get_all_achievements( + response: Response, + pagination: Pagination = Depends(get_pagination_params), session: AsyncSession = Depends(get_async_session), ) -> list[AchievementRead]: """Возвращает все achievement.""" - return await achievement_crud.get_multi(session) + achievements = await achievement_crud.get_multi(session) + response = add_response_headers( + response, achievements, pagination + ) + return paginated(achievements, pagination) + + +@router.get( + '/me', + response_model=list[AchievementRead], + dependencies=[Depends(current_user)] +) +async def get_self_achievements( + user: User = Depends(current_user), + session: AsyncSession = Depends(get_async_session) +): + """Возвращает ачивментс юзера.""" + return await achievement_crud.get_users_obj(user.id, session) + + +@router.get( + '/me/{achievement_id}', + response_model=AchievementRead, + dependencies=[Depends(current_user)] +) +async def get_self_achievement_by_id( + achievement_id: int, + user: User = Depends(current_user), + session: AsyncSession = Depends(get_async_session) +): + """Возвращает ачивмент юзера по id.""" + achievement: Achievement = await achievement_crud.get( + achievement_id, session + ) + if achievement is None: + raise HTTPException( + status_code=404, + detail='Achievement не существует.' + ) + if user.id not in [_.id for _ in achievement.profiles]: + raise HTTPException( + status_code=403, + detail='У выс нет этого achievement.' + ) + return achievement @router.post( "/", response_model=AchievementRead, - dependencies=[Depends(current_superuser)] + dependencies=[Depends(current_superuser)], + status_code=status.HTTP_201_CREATED ) async def create_achievement( achievement: AchievementCreate, @@ -37,7 +88,25 @@ async def create_achievement( return await achievement_crud.create(obj_in=achievement, session=session) -@router.delete("/{obj_id}", dependencies=[Depends(current_superuser)]) +@router.patch( + '/{achievement_id}', + response_model=AchievementRead, + dependencies=[Depends(current_superuser)] +) +async def update_achievement( + achievement_id: int, + data: AchievementUpdate, + session: AsyncSession = Depends(get_async_session) +): + """Апдейт ачивмент.""" + _achievement = await achievement_crud.get(achievement_id, session) + return await achievement_crud.update( + _achievement, data, session + ) + + +@router.delete("/{obj_id}", dependencies=[Depends(current_superuser)], + status_code=status.HTTP_204_NO_CONTENT) async def delete_achievement( obj_id: int, session: AsyncSession = Depends(get_async_session), diff --git a/app/api/endpoints/group.py b/app/api/endpoints/group.py index cad05f2..04825ff 100644 --- a/app/api/endpoints/group.py +++ b/app/api/endpoints/group.py @@ -1,12 +1,15 @@ -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, HTTPException, Response, status from sqlalchemy.ext.asyncio import AsyncSession from app.api.validators import check_name_duplicate from app.core.db import get_async_session from app.core.user import current_superuser, current_user from app.crud import group_crud -from app.schemas.group import GroupCreate, GroupRead +from app.models import Group, User +from app.schemas.group import GroupCreate, GroupRead, GroupUpdate from app.services.endpoints_services import delete_obj +from app.services.utils import (Pagination, add_response_headers, + get_pagination_params, paginated) router = APIRouter() @@ -14,19 +17,77 @@ @router.get( "/", response_model=list[GroupRead], - dependencies=[Depends(current_user)] + dependencies=[Depends(current_superuser)] ) async def get_all_groups( + response: Response, session: AsyncSession = Depends(get_async_session), + pagination: Pagination = Depends(get_pagination_params) ) -> list[GroupRead]: """Возвращает все группы.""" - return await group_crud.get_multi(session) + groups = await group_crud.get_multi(session) + response = add_response_headers( + response, groups, pagination + ) + return paginated(groups, pagination) + + +@router.get( + '/me', + response_model=list[GroupRead], + dependencies=[Depends(current_user)] +) +async def get_self_groups( + user: User = Depends(current_user), + session: AsyncSession = Depends(get_async_session) +): + """Получение групп юзером.""" + return await group_crud.get_users_obj(user.id, session) + + +@router.get( + '/me/{group_id}', + response_model=GroupRead, + dependencies=[Depends(current_user)] +) +async def get_self_group_by_id( + group_id: int, + user: User = Depends(current_user), + session: AsyncSession = Depends(get_async_session) +): + """Получение группы по id юзером.""" + group: Group | None = await group_crud.get(group_id, session) + if group is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail='Такой группы не существует.' + ) + if user not in group.users: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail='Вы не состоите в этой группе.' + ) + return group + + +@router.get( + '/{group_id}', + response_model=GroupRead, + dependencies=[Depends(current_superuser)] +) +async def get_group( + group_id: int, + session: AsyncSession = Depends(get_async_session) +): + """Получение группы по id""" + return await group_crud.get(group_id, session) @router.post( "/", response_model=GroupRead, - dependencies=[Depends(current_superuser)] + dependencies=[Depends(current_superuser)], + status_code=status.HTTP_201_CREATED ) async def create_group( group: GroupCreate, session: AsyncSession = Depends(get_async_session) @@ -36,9 +97,25 @@ async def create_group( return await group_crud.create(obj_in=group, session=session) +@router.patch( + '/{group_id}', + dependencies=[Depends(current_superuser)], + response_model=GroupRead +) +async def update_group( + group_id: int, + group: GroupUpdate, + session: AsyncSession = Depends(get_async_session) +): + """Апдейт группы.""" + _group = await group_crud.get(group_id, session) + return await group_crud.update(_group, group, session) + + @router.delete( "/{obj_id}", - dependencies=[Depends(current_superuser)] + dependencies=[Depends(current_superuser)], + status_code=status.HTTP_204_NO_CONTENT ) async def delete_group( obj_id: int, diff --git a/app/api/endpoints/profile.py b/app/api/endpoints/profile.py index f2865bd..94af5b5 100644 --- a/app/api/endpoints/profile.py +++ b/app/api/endpoints/profile.py @@ -49,10 +49,16 @@ async def get_current_user_profile( ) -@router.get( - '/me//photo', - dependencies=[Depends(current_user)] -) +@router.get('/{profile_id}', response_model=ProfileRead, + dependencies=[Depends(current_superuser)]) +async def get_profile( + profile_id: int, + session: AsyncSession = Depends(get_async_session) +): + return await profile_crud.get(profile_id, session) + + +@router.get('/me/photo', dependencies=[Depends(current_user)]) async def get_user_photo( user: User = Depends(current_user), session: AsyncSession = Depends(get_async_session) @@ -76,7 +82,7 @@ async def update_profile( @router.patch( - '/me//update_photo', + '/me/update_photo', response_model=ProfileRead, dependencies=[Depends(current_user)] ) @@ -109,7 +115,7 @@ def create_profile(): @router.delete('/{obj_id}', deprecated=True) -def delete_profile(obg_id: str): +def delete_profile(): """Удалить объект""" raise HTTPException( status_code=status.HTTP_405_METHOD_NOT_ALLOWED, diff --git a/app/crud/achievement.py b/app/crud/achievement.py index ec44edc..46c6240 100644 --- a/app/crud/achievement.py +++ b/app/crud/achievement.py @@ -1,9 +1,33 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + from app.crud.base import CRUDBase -from app.models import Achievement +from app.models import Achievement, Profile class CRUDAchievement(CRUDBase): - pass + async def get(self, obj_id: int, session: AsyncSession): + stmt = ( + select(Achievement) + .where(Achievement.id == obj_id) + .options( + selectinload(Achievement.profiles) + ) + ) + achievement = await session.execute(stmt) + achievement = achievement.scalars().first() + return achievement + + async def get_users_obj(self, user_id: int, session: AsyncSession): + stmt = ( + select(Achievement) + .options( + selectinload(Achievement.profiles) + ).where(Achievement.profiles.any(Profile.user_id == user_id)) + ) + db_obj = await session.execute(stmt) + return db_obj.scalars().all() achievement_crud = CRUDAchievement(Achievement) diff --git a/app/crud/group.py b/app/crud/group.py index 54cda8a..3ce0e3d 100644 --- a/app/crud/group.py +++ b/app/crud/group.py @@ -1,9 +1,34 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + from app.crud.base import CRUDBase -from app.models import Group +from app.models import Group, User class CRUDGroup(CRUDBase): - pass + async def get(self, group_id: int, session: AsyncSession): + stmt = ( + select(Group) + .where(Group.id == group_id) + .options( + selectinload(Group.users) + ) + ) + group = await session.execute(stmt) + group = group.scalars().first() + return group + + async def get_users_obj(self, user_id: int, session: AsyncSession): + stmt = ( + select(Group) + # .where(Group.users == user_id) + .options( + selectinload(Group.users) + ).where(Group.users.any(User.id == user_id)) + ) + db_obj = await session.execute(stmt) + return db_obj.scalars().all() group_crud = CRUDGroup(Group) diff --git a/app/crud/profile.py b/app/crud/profile.py index aa470b8..df70e2f 100644 --- a/app/crud/profile.py +++ b/app/crud/profile.py @@ -35,6 +35,15 @@ async def get_multi(self, session: AsyncSession): ) return db_objs.scalars().all() + async def get(self, obj_id: int, session: AsyncSession): + profile = await session.execute( + select(self.model) + .options( + selectinload(Profile.achievements) + ) + ) + return profile.scalars().first() + async def get_user_photo( self, user_id: int, diff --git a/app/models/profile.py b/app/models/profile.py index 5adedfa..333651c 100644 --- a/app/models/profile.py +++ b/app/models/profile.py @@ -38,7 +38,7 @@ class Profile(Base): ForeignKey('user.id'), unique=True ) user: Mapped[User] = relationship(back_populates='profile') - achievements: Mapped[Achievement] = relationship( + achievements: Mapped[list[Achievement]] = relationship( secondary=achievement_profile_association, back_populates="profiles" ) image: Mapped[str] = Column( @@ -48,4 +48,6 @@ class Profile(Base): ) def __repr__(self): - return self.first_name + if self.first_name: + return self.first_name + return str(self.user_id) diff --git a/app/models/user.py b/app/models/user.py index a5e87df..d931314 100644 --- a/app/models/user.py +++ b/app/models/user.py @@ -39,7 +39,7 @@ class User(SQLAlchemyBaseUserTable[int], Base): ) tariff: Mapped[Tariff] = relationship(back_populates="users") profile: Mapped[Profile] = relationship(back_populates="user") - groups: Mapped[Group] = relationship( + groups: Mapped[list[Group]] = relationship( secondary=group_user_association, back_populates="users" ) notifications: Mapped[Notification] = relationship( diff --git a/app/schemas/achievement.py b/app/schemas/achievement.py index bc47044..f4b67fb 100644 --- a/app/schemas/achievement.py +++ b/app/schemas/achievement.py @@ -1,6 +1,6 @@ from typing import Optional -from pydantic import BaseModel +from pydantic import BaseModel, Field class AchievementCreate(BaseModel): @@ -15,3 +15,8 @@ class AchievementRead(BaseModel): class Config: from_attributes = True + + +class AchievementUpdate(BaseModel): + name: Optional[str] = Field(None,) + description: Optional[str] = Field(None,) diff --git a/app/schemas/group.py b/app/schemas/group.py index f1b55d3..f260634 100644 --- a/app/schemas/group.py +++ b/app/schemas/group.py @@ -1,6 +1,6 @@ from typing import Optional -from pydantic import BaseModel +from pydantic import BaseModel, Field class GroupCreate(BaseModel): @@ -15,3 +15,8 @@ class GroupRead(BaseModel): class Config: from_attributes = True + + +class GroupUpdate(BaseModel): + name: Optional[str] = Field(None,) + description: Optional[str] = Field(None,) diff --git a/app/schemas/profile.py b/app/schemas/profile.py index 71fd9f1..fdad29a 100644 --- a/app/schemas/profile.py +++ b/app/schemas/profile.py @@ -2,8 +2,6 @@ from pydantic import BaseModel, Field -from app.schemas.achievement import AchievementRead - class ProfileRead(BaseModel): id: int @@ -12,7 +10,6 @@ class ProfileRead(BaseModel): age: Optional[int] user_id: int image: Optional[str] - achievements: Optional[list[AchievementRead]] class Config: from_attributes = True diff --git a/tests/conftest.py b/tests/conftest.py index e1b5cb1..73670d9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,6 +16,9 @@ pytest_plugins = [ 'tests.fixtures.user', + 'tests.fixtures.profile', + 'tests.fixtures.group', + 'tests.fixtures.achievement' ] engine_test = create_async_engine( diff --git a/tests/fixtures/achievement.py b/tests/fixtures/achievement.py new file mode 100644 index 0000000..0c48cfe --- /dev/null +++ b/tests/fixtures/achievement.py @@ -0,0 +1,18 @@ +import pytest_asyncio +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models import Achievement + + +@pytest_asyncio.fixture +async def moc_achievements( + db_session: AsyncSession +) -> None: + moc_achievements = [ + Achievement( + name=f'Achievement_{i}', + description=f'Description for Achievemetnt_{i}' + ) for i in range(1, 6) + ] + db_session.add_all(moc_achievements) + await db_session.commit() diff --git a/tests/fixtures/group.py b/tests/fixtures/group.py new file mode 100644 index 0000000..96edd61 --- /dev/null +++ b/tests/fixtures/group.py @@ -0,0 +1,18 @@ +import pytest_asyncio +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models import Group + + +@pytest_asyncio.fixture +async def moc_groups( + db_session: AsyncSession +) -> None: + moc_groups = [ + Group( + name=f'Group_{i}', + description=f'Description for Group_{i}' + ) for i in range(1, 6) + ] + db_session.add_all(moc_groups) + await db_session.commit() diff --git a/tests/fixtures/profile.py b/tests/fixtures/profile.py new file mode 100644 index 0000000..c6ce10f --- /dev/null +++ b/tests/fixtures/profile.py @@ -0,0 +1,90 @@ +from typing import AsyncGenerator + +import pytest_asyncio +from passlib.hash import bcrypt +from sqlalchemy import select + +from app.models import Profile, User +from tests.conftest import AsyncSessionLocalTest + + +@pytest_asyncio.fixture +async def moc_users( + db_session: AsyncSessionLocalTest +) -> AsyncGenerator: + """Фикстура заполнения базы юзерами с профилями.""" + hashed_password = bcrypt.hash('qwerty') + moc_users = [ + User( + email=f'user_{i}@example.com', + hashed_password=hashed_password, + role='user', + username=f'user_{i}' + ) for i in range(1, 6) + ] + db_session.add_all(moc_users) + await db_session.commit() + moc_users = await db_session.execute(select(User)) + moc_users = moc_users.scalars().all() + moc_profiles = [ + Profile( + first_name=f'name_{user.username}', + last_name=f'surname_{user.username}', + age=i + 20, + user_id=user.id + ) for i, user in enumerate(moc_users) + ] + db_session.add_all(moc_profiles) + await db_session.commit() + + +# @pytest_asyncio.fixture +# async def user_1( +# prepare_database: FastAPI, +# db_session: AsyncSessionLocalTest +# ) -> AsyncGenerator: +# """Фикстура зарегистрированного клиента.""" +# hashed_password = bcrypt.hash('qwerty') +# user_1 = User( +# email='user_1@example.com', +# hashed_password=hashed_password, +# role='user', +# username='user_1' +# ) +# db_session.add(user_1) +# await db_session.commit() +# await db_session.refresh(user_1) +# profile_1 = Profile( +# first_name='user_1_fn', +# last_name='user_1_ln', +# age=25, +# user_id=user_1.id +# ) +# db_session.add(profile_1) +# await db_session.commit() + + +# @pytest_asyncio.fixture +# async def user_2( +# prepare_database: FastAPI, +# db_session: AsyncSessionLocalTest +# ) -> AsyncGenerator: +# """Фикстура зарегистрированного клиента.""" +# hashed_password = bcrypt.hash('qwerty') +# user_2 = User( +# email='user_2@example.com', +# hashed_password=hashed_password, +# role='user', +# username='user_2' +# ) +# db_session.add(user_2) +# await db_session.commit() +# await db_session.refresh(user_2) +# profile_2 = Profile( +# first_name='user_2_fn', +# last_name='user_2_ln', +# age=47, +# user_id=user_2.id +# ) +# db_session.add(profile_2) +# await db_session.commit() diff --git a/tests/fixtures/user.py b/tests/fixtures/user.py index 6a1392c..37e573e 100644 --- a/tests/fixtures/user.py +++ b/tests/fixtures/user.py @@ -59,3 +59,57 @@ async def auth_client( access_token = response.json().get('access_token') new_client.headers.update({'Authorization': f'Bearer {access_token}'}) yield new_client + + +@pytest_asyncio.fixture +async def superuser( + db_session: AsyncSessionLocalTest +): + """Фикстура суперюзера.""" + hashed_password = bcrypt.hash('admin') + superuser = User( + email='admin@admin.com', + hashed_password=hashed_password, + role='admin', + username='admin', + is_superuser=True + ) + db_session.add(superuser) + await db_session.commit() + await db_session.refresh(superuser) + yield superuser + + +# @pytest_asyncio.fixture +# async def user_1( +# prepare_database: FastAPI, +# db_session: AsyncSessionLocalTest +# ) -> AsyncGenerator: +# """Фикстура зарегистрированного клиента.""" +# hashed_password = bcrypt.hash('qwerty') +# user_1 = User( +# email='user_1@example.com', +# hashed_password=hashed_password, +# role='user', +# username='user_1' +# ) +# db_session.add(user_1) +# await db_session.commit() +# await db_session.refresh(user_1) +# yield user_1 + + +@pytest_asyncio.fixture +async def auth_superuser( + new_client, + superuser, + # user_1, +) -> AsyncGenerator | TestClient: + """Фикстура для суперюзера, вошедшего в систему.""" + response = new_client.post( + '/auth/jwt/login', + data={'username': 'admin@admin.com', 'password': 'admin'}) + assert response.status_code == status.HTTP_200_OK + access_token = response.json().get('access_token') + new_client.headers.update({'Authorization': f'Bearer {access_token}'}) + yield new_client diff --git a/tests/test_achievement.py b/tests/test_achievement.py new file mode 100644 index 0000000..9059cea --- /dev/null +++ b/tests/test_achievement.py @@ -0,0 +1,320 @@ +from fastapi import status +from fastapi.testclient import TestClient +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from app.models import Achievement, Profile, User + +from .utils import get_obj_count + +CREATE_SCHEME = { + 'name': 'Achievment name', + 'description': 'Achievment description' +} +WRONG_CREATE_SCHEME = { + 'description': 'Achievment description' +} +UPDATE_SCHEME = { + 'name': 'new achievement name' +} + + +async def _get_achievement_by_id( + index: int, + db_session: AsyncSession +): + """Возвращает ачивмент по id.""" + stmt = ( + select(Achievement) + .where(Achievement.id == index) + .options( + selectinload(Achievement.profiles) + ) + ) + achievement = await db_session.execute(stmt) + return achievement.scalar() + + +async def _get_user_by_id( + index: int, + db_session: AsyncSession +): + """Возвращает юзера по id.""" + user = await db_session.execute( + select(User) + .where(User.id == index) + .options( + selectinload(User.profile) + .selectinload(Profile.achievements) + ) + ) + return user.scalar() + + +async def _get_profile_by_user_id( + index: int, + db_session: AsyncSession +): + """Возвращает профиль юзера.""" + profile = await db_session.execute( + select(Profile) + .where(Profile.user_id == index) + ) + return profile.scalar() + + +class TestCreateAchievement: + async def test_create_achievement_superuser( + self, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Тест создания ачивмент.""" + achievements = await get_obj_count(Achievement, db_session) + response = auth_superuser.post( + '/achievements', + json=CREATE_SCHEME + ) + assert response.status_code == status.HTTP_201_CREATED + new_achievements = await get_obj_count(Achievement, db_session) + assert new_achievements == achievements + 1 + + async def test_wrong_create_scheme( + self, + db_session: AsyncSession, + auth_superuser: TestClient + ): + achievements = await get_obj_count(Achievement, db_session) + response = auth_superuser.post( + '/achievements', + json=WRONG_CREATE_SCHEME + ) + assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY + new_achievements = await get_obj_count(Achievement, db_session) + assert new_achievements == achievements + + async def test_forbidden_create_achievement_user( + self, + db_session: AsyncSession, + auth_client: TestClient + ): + """Тест запрета создания ачивмент юзером.""" + achievements = await get_obj_count(Achievement, db_session) + response = auth_client.post( + '/achievements', + json=CREATE_SCHEME + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + new_achievements = await get_obj_count(Achievement, db_session) + assert new_achievements == achievements + + async def test_forbidden_create_achievement_nonauth( + self, + db_session: AsyncSession, + new_client: TestClient + ): + """Тест запрета создания ачивмент юзером.""" + achievements = await get_obj_count(Achievement, db_session) + response = new_client.post( + '/achievements', + json=CREATE_SCHEME + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + new_achievements = await get_obj_count(Achievement, db_session) + assert new_achievements == achievements + + +class TestGetAchievement: + async def test_get_all_achievements_superuser( + self, + moc_achievements, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Тест получения всех ачивмент суперюзером.""" + achievements = await get_obj_count(Achievement, db_session) + response = auth_superuser.get('/achievements') + assert response.status_code == status.HTTP_200_OK + assert len(response.json()) == achievements + + async def test_forbidden_get_all_achievements_user( + self, + moc_achievements, + db_session: AsyncSession, + auth_client: TestClient + ): + """Тест запрета получения всех ачивмент юзером.""" + response = auth_client.get('/achievements') + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_forbidden_get_all_achievements_nonauth( + self, + moc_achievements, + new_client: TestClient + ): + """Тест запрета получения ачивментс неавторизованным.""" + response = new_client.get('/achievements') + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + async def test_get_self_achievements_user( + self, + register_client, + moc_users, + moc_achievements, + db_session: AsyncSession, + auth_client: TestClient + ): + """Тест получения юзером своих ачивментс.""" + user = await _get_user_by_id(register_client.id, db_session) + profile = await _get_profile_by_user_id(user.id, db_session) + achievement = await _get_achievement_by_id(1, db_session) + achiv = await _get_achievement_by_id(2, db_session) + achiv.profiles.append(profile) + achievement.profiles.append(profile) + await db_session.commit() + response = auth_client.get('achievements/me') + assert response.status_code == status.HTTP_200_OK + result = response.json() + assert len(result) == 2 + assert result[0]['id'] in (1, 2) + assert result[1]['id'] in (1, 2) + assert result[0]['id'] != result[1]['id'] + + async def test_get_self_achievement_by_id_user( + self, + register_client, + moc_users, + moc_achievements, + db_session: AsyncSession, + auth_client: TestClient + ): + """Тест получения юзером своего ачивмент по id.""" + user = await _get_user_by_id(register_client.id, db_session) + profile = await _get_profile_by_user_id(user.id, db_session) + achievement = await _get_achievement_by_id(1, db_session) + achiv = await _get_achievement_by_id(2, db_session) + achiv.profiles.append(profile) + achievement.profiles.append(profile) + await db_session.commit() + response = auth_client.get('/achievements/me/1') + assert response.status_code == status.HTTP_200_OK + assert response.json()['id'] == 1 + response = auth_client.get('/achievements/me/3') + assert response.status_code == status.HTTP_403_FORBIDDEN + response = auth_client.get('/achievements/me/22') + assert response.status_code == status.HTTP_404_NOT_FOUND + + +class TestUpdateAchievement: + async def test_forbidden_update_achievement_user( + self, + moc_achievements, + db_session: AsyncSession, + auth_client: TestClient + ): + """Тест запрета апдейта ачивмент юзером.""" + response = auth_client.patch( + '/achievements/1' + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_forbidden_update_achievement_nonauth( + self, + moc_achievements, + db_session: AsyncSession, + new_client: TestClient + ): + """Тест запрета апдейта ачивмент неавторизованным.""" + response = new_client.patch( + '/achievements/1' + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + async def test_update_achievement_superuser( + self, + moc_achievements, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Тест апдейта ачивмент суперюзером.""" + achievement = await _get_achievement_by_id(1, db_session) + response = auth_superuser.patch( + '/achievements/1', + json=UPDATE_SCHEME + ) + assert response.status_code == status.HTTP_200_OK + assert response.json()['name'] == UPDATE_SCHEME['name'] + check_achievement = await _get_achievement_by_id(1, db_session) + assert check_achievement.name == UPDATE_SCHEME['name'] + assert check_achievement.description == achievement.description + + +class TestDeleteAchievement: + async def test_forbidden_delete_acievement_user( + self, + moc_achievements, + db_session: AsyncSession, + auth_client: TestClient + ): + """Тест запрета удаления ачивмент юзером.""" + achievements = await get_obj_count(Achievement, db_session) + assert achievements > 0 + response = auth_client.delete('/achievements/1') + assert response.status_code == status.HTTP_403_FORBIDDEN + check_achiv = await get_obj_count(Achievement, db_session) + assert check_achiv == achievements + + async def test_forbidden_delete_acievement_nonauth( + self, + moc_achievements, + db_session: AsyncSession, + new_client: TestClient + ): + """Тест запрета удаления ачивмент неавторизованным.""" + achievements = await get_obj_count(Achievement, db_session) + assert achievements > 0 + response = new_client.delete('/achievements/1') + assert response.status_code == status.HTTP_401_UNAUTHORIZED + check_achiv = await get_obj_count(Achievement, db_session) + assert check_achiv == achievements + + async def test_delete_achievement_superuser( + self, + moc_achievements, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Тест удаления ачивмент суперюзером.""" + achievement = await _get_achievement_by_id(1, db_session) + assert achievement is not None + achiv_count = await get_obj_count(Achievement, db_session) + response = auth_superuser.delete('/achievements/1') + assert response.status_code == status.HTTP_204_NO_CONTENT + removed_achiv = await _get_achievement_by_id(1, db_session) + assert removed_achiv is None + check_achiv_count = await get_obj_count(Achievement, db_session) + assert check_achiv_count == achiv_count - 1 + + +class TestPaginationGroup: + async def test_pagination( + self, + moc_achievements, + auth_superuser: TestClient + ): + """Тест пагинации профилей""" + response = auth_superuser.get( + '/achievements/?limit=2' + ) + result = response.json() + assert len(result) == 2 + assert result[0]['id'] == 1 + assert result[1]['id'] == 2 + response = auth_superuser.get( + '/achievements/?offset=2&limit=2' + ) + result = response.json() + assert len(result) == 2 + assert result[0]['id'] == 3 + assert result[1]['id'] == 4 diff --git a/tests/test_group.py b/tests/test_group.py new file mode 100644 index 0000000..1869bb7 --- /dev/null +++ b/tests/test_group.py @@ -0,0 +1,297 @@ +from fastapi import status +from fastapi.testclient import TestClient +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from app.models import Group, User + +from .utils import get_obj_count + +GROUP_SCHEME = { + 'name': 'Test Group', + 'description': 'Test Group Description' +} +UPDATE_SCHEME = { + 'name': 'Updated name', + 'description': 'Updated description', +} + + +async def _get_user( + db_session: AsyncSession +): + """Возвращает юзера.""" + stmt = (select(User) + .where(User.username == 'testuser') + .options(selectinload(User.groups))) + user = await db_session.execute(stmt) + return user.scalar() + + +async def _get_group_by_id( + index: int, + db_session: AsyncSession +): + """Возвращает группу по id.""" + stmt = (select(Group).where(Group.id == index) + .options(selectinload(Group.users))) + group = await db_session.execute(stmt) + return group.scalar() + + +class TestCreateGroup: + async def test_create_group( + self, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Создание группы суперюзером""" + groups = await get_obj_count(Group, db_session) + response = auth_superuser.post( + '/groups', + json=GROUP_SCHEME + ) + assert response.status_code == status.HTTP_201_CREATED + new_groups = await get_obj_count(Group, db_session) + assert new_groups == groups + 1 + + async def test_forbidden_create_group_by_user( + self, + auth_client: TestClient + ): + """Тест запрета создания группы юзером""" + response = auth_client.post( + '/groups', + json=GROUP_SCHEME + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_forbidden_create_group_nonauth( + self, + new_client: TestClient + ): + """Тест запрета создания группы неавторизованным.""" + response = new_client.post( + '/groups', + json=GROUP_SCHEME + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + +class TestGetGroup: + async def test_get_all_groups_superuser( + self, + moc_groups, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Получение всех групп суперюзером.""" + groups = await get_obj_count(Group, db_session) + response = auth_superuser.get( + '/groups' + ) + assert response.status_code == status.HTTP_200_OK + assert len(response.json()) == groups + + async def test_forbidden_get_all_groups_user( + self, + moc_groups, + db_session: AsyncSession, + auth_client: TestClient + ): + """Тест запрета получения всех групп юзером.""" + response = auth_client.get('/groups') + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_forbidden_get_all_groups_nonauth( + self, + moc_groups, + new_client: TestClient + ): + """Тест запрета получения групп неавторизованным.""" + response = new_client.get('/groups') + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + async def test_get_group_by_id_superuser( + self, + moc_groups, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Получение суперюзером группы по id.""" + response = auth_superuser.get( + '/groups/1' + ) + assert response.json()['id'] == 1 + + async def test_forbidden_get_group_by_id_nonauth( + self, + moc_groups, + new_client: TestClient + ): + """Тест запрета получения группы по id неавторизованным.""" + response = new_client.get('/groups/1') + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + async def test_forbidden_get_group_by_id_user( + self, + auth_client: TestClient + ): + """Тест запрета получения группы по id юзером.""" + response = auth_client.get('/groups/1') + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_get_self_groups_user( + self, + moc_groups, + db_session: AsyncSession, + auth_client: TestClient + ): + """Тест получения юзером своих групп.""" + user = await _get_user(db_session) + group = await _get_group_by_id(1, db_session) + group.users.append(user) + group_2 = await _get_group_by_id(2, db_session) + group_2.users.append(user) + await db_session.commit() + response = auth_client.get( + '/groups/me', + ) + assert response.status_code == status.HTTP_200_OK + result = response.json() + assert len(result) == 2 + assert result[0]['id'] in (1, 2) + assert result[1]['id'] in (1, 2) + assert result[0]['id'] != result[1]['id'] + + async def test_get_group_by_id_user( + self, + register_client, + moc_groups, + moc_users, + db_session: AsyncSession, + auth_client: TestClient + ): + """Тест получения юзером группы по id.""" + user = await _get_user(db_session) + group = await _get_group_by_id(1, db_session) + group.users.append(user) + group_2 = await _get_group_by_id(2, db_session) + group_2.users.append(user) + await db_session.commit() + response = auth_client.get('groups/me/1') + assert response.status_code == status.HTTP_200_OK + assert response.json()['id'] == 1 + response = auth_client.get('/groups/me/3') + assert response.status_code == status.HTTP_403_FORBIDDEN + response = auth_client.get('/groups/me/22') + assert response.status_code == status.HTTP_404_NOT_FOUND + + +class TestDeleteGroup: + async def test_delete_group_superuser( + self, + moc_groups, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Тест удаления группы суперюзером.""" + groups = await get_obj_count(Group, db_session) + response = auth_superuser.delete( + '/groups/1' + ) + assert response.status_code == status.HTTP_204_NO_CONTENT + groups_after_remove = await get_obj_count(Group, db_session) + assert groups_after_remove == groups - 1 + removed_group = await _get_group_by_id(1, db_session) + assert removed_group is None + + async def test_forbidden_delete_group_user( + self, + moc_groups, + auth_client: TestClient + ): + """Тест запрета удаления группы юзером.""" + response = auth_client.delete( + '/groups/1' + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_formidden_delete_group_nonauth( + self, + new_client: TestClient + ): + """Тест запрета удаления группы неавторизованным пользователем.""" + response = new_client.delete( + '/groups/1' + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + +class TestUpdateGroup: + async def test_update_group_superuser( + self, + moc_groups, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Тест апдейта группы суперюзером.""" + group = await _get_group_by_id(1, db_session) + assert group.id == 1 + assert group.name != UPDATE_SCHEME['name'] + assert group.description != UPDATE_SCHEME['description'] + response = auth_superuser.patch( + '/groups/1', + json=UPDATE_SCHEME + ) + assert response.status_code == status.HTTP_200_OK + upated_group = await _get_group_by_id(1, db_session) + assert group.id == upated_group.id + assert upated_group.name == UPDATE_SCHEME['name'] + assert upated_group.description == UPDATE_SCHEME['description'] + + async def test_forbidden_update_group_user( + self, + auth_client: TestClient + ): + """Тест запрета апдейта группы юзером.""" + response = auth_client.patch( + '/groups/1', + json=UPDATE_SCHEME + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_forbidden_update_group_nonauth( + self, + new_client: TestClient + ): + """Тест запрета апдейта неавторизованным пользователем.""" + response = new_client.patch( + '/groups/1', + json=UPDATE_SCHEME + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + +class TestPaginationGroup: + async def test_pagination( + self, + moc_groups, + auth_superuser: TestClient + ): + """Тест пагинации профилей""" + response = auth_superuser.get( + '/groups/?limit=2' + ) + result = response.json() + assert len(result) == 2 + assert result[0]['id'] == 1 + assert result[1]['id'] == 2 + response = auth_superuser.get( + '/groups/?offset=2&limit=2' + ) + result = response.json() + assert len(result) == 2 + assert result[0]['id'] == 3 + assert result[1]['id'] == 4 diff --git a/tests/test_profile.py b/tests/test_profile.py new file mode 100644 index 0000000..749c412 --- /dev/null +++ b/tests/test_profile.py @@ -0,0 +1,283 @@ +import base64 +import io +from pathlib import Path +from typing import AsyncGenerator + +from fastapi import Response, status +from fastapi.testclient import TestClient +from PIL import Image +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from app.core.config import settings +from app.models import Profile, User +from tests.fixtures.user import USER_EMAIL, USER_PASSWORD, USER_USERNAME + +from .utils import get_obj_count + +REGISTRATION_SCHEMA = { + 'email': USER_EMAIL, + 'password': USER_PASSWORD, + 'role': 'user', + 'username': USER_USERNAME, +} + + +def delete_tmpdir(path: Path): + for sub in path.iterdir(): # type: Path + if sub.is_dir(): + delete_tmpdir(sub) + else: + sub.unlink() + path.rmdir() + + +async def _get_user( + user_id: int, + db_session, +): + user = await db_session.execute( + select(User).filter(User.id == user_id) + .options(selectinload(User.profile)) + ) + user: User = user.scalars().first() + return user + + +class TestProfile: + async def test_create_profile_with_create_user( + self, new_client, + db_session: AsyncSession + ): + """Тест создания профиля при регистрации пользователя.""" + profiles = await get_obj_count(Profile, db_session) + users = await get_obj_count(User, db_session) + response: Response = new_client.post( + '/auth/register', json=REGISTRATION_SCHEMA + ) + assert response.status_code == status.HTTP_201_CREATED + check_profiles = await get_obj_count(Profile, db_session) + assert check_profiles == profiles + 1 + check_users = await get_obj_count(User, db_session) + assert check_users == users + 1 + + async def test_get_all_profiles_superuser( + self, + moc_users, + db_session: AsyncSession, + auth_superuser: TestClient + ): + """Тест получения всех профилей суперюзером.""" + profiles = await get_obj_count(Profile, db_session) + response: Response = auth_superuser.get( + '/profiles/' + ) + assert response.status_code == status.HTTP_200_OK + assert len(response.json()) == profiles + + async def test_forbidden_get_all_profiles_user( + self, + moc_users, + auth_client: TestClient, + ): + """Тест запрета получения профилей простым пользователем.""" + response = auth_client.get( + '/profiles/' + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_filter_profiles( + self, + moc_users, + db_session, + auth_superuser + ): + """Тест фильтрации профилей.""" + response = auth_superuser.get( + '/profiles/?age__gte=22&age__lte=23' + ) + assert len(response.json()) == 2 + response = auth_superuser.get( + '/profiles/?first_name__ilike=3' + ) + assert len(response.json()) == 1 + response = auth_superuser.get( + '/profiles/?last_name__ilike=4' + ) + assert len(response.json()) == 1 + + async def test_pagination( + self, + moc_users, + auth_superuser + ): + """Тест пагинации профилей""" + response = auth_superuser.get( + '/profiles/?limit=2' + ) + result = response.json() + assert len(result) == 2 + assert result[0]['user_id'] == 1 + assert result[1]['user_id'] == 2 + response = auth_superuser.get( + '/profiles/?offset=2&limit=2' + ) + result = response.json() + assert len(result) == 2 + assert result[0]['user_id'] == 3 + assert result[1]['user_id'] == 4 + + async def test_get_self_profile( + self, + moc_users, + new_client, + db_session + ): + """Тест получения своего профиля текущим юзером.""" + user = await _get_user(1, db_session) + response: Response = new_client.post( + '/auth/jwt/login', + data={'username': user.email, 'password': 'qwerty'}, + ) + access_token = response.json().get('access_token') + new_client.headers.update({'Authorization': f'Bearer {access_token}'}) + response: Response = new_client.get( + '/profiles/me/' + ) + result = response.json() + assert result['first_name'] == user.profile.first_name + + async def test_forbidden_get_other_user_profile_for_user( + self, + moc_users, + new_client, + db_session + ): + """Тест запрета получения чужого профиля текущим юзером.""" + user: User = await _get_user(1, db_session) + other_user: User = await _get_user(2, db_session) + response: Response = new_client.post( + '/auth/jwt/login', + data={'username': user.email, 'password': 'qwerty'}, + ) + access_token = response.json().get('access_token') + new_client.headers.update({'Authorization': f'Bearer {access_token}'}) + response = new_client.get( + f'/profiles/{other_user.profile.id}' + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_get_photo_from_self_profile( + self, + moc_users, + db_session, + new_client, + ): + """Тест получения фото своего профиля юзером.""" + user: User = await _get_user(1, db_session) + response: Response = new_client.post( + '/auth/jwt/login', + data={'username': user.email, 'password': 'qwerty'}, + ) + access_token = response.json().get('access_token') + new_client.headers.update({'Authorization': f'Bearer {access_token}'}) + response = new_client.get( + '/profiles/me/photo/' + ) + assert response.status_code == status.HTTP_200_OK + + async def test_update_self_profile( + self, + moc_users, + db_session, + new_client: TestClient + ): + """Тест апдейта своего профиля.""" + user: User = await _get_user(1, db_session) + response: Response = new_client.post( + '/auth/jwt/login', + data={'username': user.email, 'password': 'qwerty'}, + ) + access_token = response.json().get('access_token') + new_client.headers.update({'Authorization': f'Bearer {access_token}'}) + data = {'first_name': 'new_first_name'} + response = new_client.patch( + '/profiles/me', + json=data + ) + assert response.status_code == status.HTTP_200_OK + user: User = await _get_user(1, db_session) + assert user.profile.first_name == 'new_first_name' + + async def test_update_photo( + self, + moc_users, + db_session, + new_client: TestClient + ): + """Тест апдейта фото своего профиля.""" + user: User = await _get_user(1, db_session) + photo = user.profile.image + tmp_image = Image.new('RGB', (640, 480)) + buffer = io.BytesIO() + Path( + settings.base_dir / 'tmp_for_load' + ).mkdir(parents=True, exist_ok=True) + tmp_image.save(settings.base_dir / 'tmp_for_load' / 'img.jpeg', 'jpeg') + tmp_image.save(buffer, format='JPEG') + img_str = base64.b64encode(buffer.getvalue()) + response: Response = new_client.post( + '/auth/jwt/login', + data={'username': user.email, 'password': 'qwerty'}, + ) + access_token = response.json().get('access_token') + new_client.headers.update({'Authorization': f'Bearer {access_token}'}) + with open(settings.base_dir / 'media' / photo, 'rb') as f: + current_photo = base64.b64encode(f.read()) + response = new_client.patch( + '/profiles/me/update_photo', + files={ + 'file': ( + 'img.jpeg', + open(f'{settings.base_dir}/tmp_for_load/img.jpeg', 'rb'), + 'image/jpeg' + ) + } + ) + user: User = await _get_user(1, db_session) + with open(settings.base_dir / 'media' / user.profile.image, 'rb') as f: + new_photo = base64.b64encode(f.read()) + assert new_photo == img_str + assert current_photo != new_photo + path = Path(f'{settings.base_dir}/media/{user.profile.image}') + Path.unlink(path) + delete_tmpdir(settings.base_dir / 'tmp_for_load') + + async def test_create_profile_deprecated( + self, + new_client: TestClient + ): + """Тест запрета создания профиля без создания юзера.""" + response = new_client.post( + '/profiles/', + json={ + 'first_name': 'test_first_name', + 'last_name': 'test_last_name', + 'age': 47 + } + ) + assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED + + async def test_forbidden_delete_profile( + self, + moc_users, + db_session, + auth_superuser: AsyncGenerator | TestClient + ): + """Тест запрета удаления профиля.""" + user = await _get_user(1, db_session) + response = auth_superuser.delete( + f'/profiles/{user.profile.id}' + ) + assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED diff --git a/tests/utils.py b/tests/utils.py new file mode 100644 index 0000000..06e17e8 --- /dev/null +++ b/tests/utils.py @@ -0,0 +1,12 @@ +from sqlalchemy import func +from sqlalchemy.ext.asyncio import AsyncSession + + +async def get_obj_count( + model, + session: AsyncSession +) -> int: + """Возвращает количество объектов в базе.""" + stmt = func.count(model.id) + count = await session.execute(stmt) + return count.scalar()