diff --git a/migrations/versions/bd5dc5ad8627_changed_logic.py b/migrations/versions/bd5dc5ad8627_changed_logic.py index 3c1171f..3f203d2 100644 --- a/migrations/versions/bd5dc5ad8627_changed_logic.py +++ b/migrations/versions/bd5dc5ad8627_changed_logic.py @@ -5,9 +5,9 @@ Create Date: 2026-08-17 12:55:02.773293 """ -from alembic import op -import sqlalchemy as sa +import sqlalchemy as sa +from alembic import op # revision identifiers, used by Alembic. revision = 'bd5dc5ad8627' diff --git a/modal_backend/models/db.py b/modal_backend/models/db.py index efa666d..e05adcc 100644 --- a/modal_backend/models/db.py +++ b/modal_backend/models/db.py @@ -11,7 +11,6 @@ Integer, String, cast, - func, or_, true, ) diff --git a/modal_backend/routes/notes.py b/modal_backend/routes/notes.py index 8b56ae5..7b32245 100644 --- a/modal_backend/routes/notes.py +++ b/modal_backend/routes/notes.py @@ -34,13 +34,13 @@ async def get_notes( type_id: int = Query(None), groups_id: list[int] = Query(None), services_id: list[int] = Query(None), - status: str = Query( + status: ModalStatus = Query( enum=["active", "archived"], default=None, ), asc_order: bool = False, - limit: int = 10, - offset: int = 0, + limit: int = Query(10, ge=0, description="Лимит записей"), + offset: int = Query(0, ge=0, description="Смещение записей на N+offset, где N - первая запись"), user=Depends(UnionAuth()), ) -> list[NoteGet]: """ diff --git a/modal_backend/utils/services.py b/modal_backend/utils/services.py index a4b1a05..f77c8b0 100644 --- a/modal_backend/utils/services.py +++ b/modal_backend/utils/services.py @@ -23,7 +23,7 @@ async def get_notes_by_filters( type_id: int = None, groups_id: list[int] = None, services_id: list[int] = None, - status: str = None, + status: ModalStatus | None = None, ): notes_query = ( Note.query(session=db.session) diff --git a/tests/conftest.py b/tests/conftest.py index e123f4d..8e5148f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,3 +1,4 @@ +from datetime import datetime, timedelta, timezone from functools import lru_cache from pathlib import Path @@ -10,8 +11,18 @@ from sqlalchemy.orm import sessionmaker from testcontainers.postgres import PostgresContainer +from modal_backend.models.db import Group, ModalStatus, Note, NoteTypeEnum, Service +from modal_backend.schemas.models import ( + NoteChoicePost, + NoteImagePost, + NoteInfoPost, + NoteRatingPost, + NoteTextPost, +) from modal_backend.settings import Settings +NOW = datetime.now(timezone.utc).replace(tzinfo=None) + class PostgresConfig: """Дата-класс со значениями для контейнера с тестовой БД и для alembic-миграции.""" @@ -133,3 +144,200 @@ def client(get_app_with_test_settings, user_mock): app = get_app_with_test_settings client = TestClient(app) return client + + +def create_group(group_id: int, name: str) -> Group: + """Вспомогательная функция-мини-фабрика для создания разных групп в фикстуре groups.""" + return Group(group_id=group_id, name=name) + + +@pytest.fixture() +def groups(dbsession): + """Создает три группы.""" + group_data = [(1, "Group_1"), (2, "Group_2"), (3, "Group_3")] + groups = [create_group(*group) for group in group_data] + for group in groups: + dbsession.add(group) + dbsession.commit() + yield groups + for group in groups: + dbsession.delete(group) + dbsession.commit() + + +def create_service(service_id: int, name: str) -> Service: + """Вспомогательная функция-мини-фабрика для создания разных сервисов в фикстуре services.""" + return Service(service_id=service_id, name=name) + + +@pytest.fixture() +def services(dbsession): + """Создает три сервиса.""" + service_data = [(1, "Service_1"), (2, "Service_2"), (3, "Service_3")] + services = [create_service(*service) for service in service_data] + for service in services: + dbsession.add(service) + dbsession.commit() + yield services + for service in services: + dbsession.delete(service) + dbsession.commit() + + +@pytest.fixture() +def mock_datetime_now(mocker): + mock_datetime = mocker.patch("modal_backend.utils.services.datetime") + mock_datetime.now.return_value = NOW + yield + + +def create_note( + type_id: int, + schema: NoteChoicePost | NoteImagePost | NoteInfoPost | NoteRatingPost | NoteTextPost, + admin_id: int, + status: ModalStatus, +): + return Note( + type_id=type_id, + **schema.model_dump(), + admin_id=admin_id, + status=status, + ) + + +@pytest.fixture() +def notes( + dbsession, + groups, + services, + authlib_user_data, +): + """Создает 8 модалок: + indexes: descriprion: + (0, 1, 2, 3, 4) 5 с разными типами + (0, 1, 2) 3 активные + (3, 4, 5, 6, 7) 5 архивных + (0, 2, 4, 6, 7) group_ids = [3] service_ids = [3] + (1, 3) group_ids = [1, 2] service_ids = [1, 2] + (5) group_ids - [1, 2, 3] service_ids = [1, 2, 3] + (6, 7) просроченные, одна с is_always=True + """ + note_data = [ + { + "type_id": NoteTypeEnum.INFO, + "schema": NoteInfoPost( + header="header_1", + is_always=False, + frequency=10, + group_ids=[group.id for group in groups][2:3], + service_ids=[service.id for service in services][2:3], + ), + "admin_id": authlib_user_data.get("id"), + "status": ModalStatus.ACTIVE, + }, + { + "type_id": NoteTypeEnum.RATING, + "schema": NoteRatingPost( + header="header_2", + is_always=False, + frequency=10, + group_ids=[group.id for group in groups][:2], + service_ids=[service.id for service in services][:2], + ), + "admin_id": authlib_user_data.get("id"), + "status": ModalStatus.ACTIVE, + }, + { + "type_id": NoteTypeEnum.TEXT, + "schema": NoteTextPost( + header="header_3", + is_always=False, + frequency=10, + group_ids=[group.id for group in groups][2:3], + service_ids=[service.id for service in services][2:3], + ), + "admin_id": authlib_user_data.get("id"), + "status": ModalStatus.ACTIVE, + }, + { + "type_id": NoteTypeEnum.CHOICE, + "schema": NoteChoicePost( + header="header_4", + is_always=False, + frequency=10, + group_ids=[group.id for group in groups][:2], + service_ids=[service.id for service in services][:2], + ), + "admin_id": authlib_user_data.get("id"), + "status": ModalStatus.ARCHIVED, + }, + { + "type_id": NoteTypeEnum.IMAGE, + "schema": NoteImagePost( + header="header_5", + is_always=False, + frequency=10, + group_ids=[group.id for group in groups][2:3], + service_ids=[service.id for service in services][2:3], + ), + "admin_id": authlib_user_data.get("id"), + "status": ModalStatus.ARCHIVED, + }, + { + "type_id": NoteTypeEnum.IMAGE, + "schema": NoteImagePost( + header="header_5", + is_always=False, + frequency=10, + group_ids=[group.id for group in groups], + service_ids=[service.id for service in services], + ), + "admin_id": authlib_user_data.get("id"), + "status": ModalStatus.ARCHIVED, + }, + ] + + for offset, d in enumerate(note_data): + d["schema"].start_ts = NOW + offset * timedelta(hours=1) + d["schema"].end_ts = NOW + offset * timedelta(hours=1) + timedelta(hours=1) + note_data.extend( + [ + { # просроченная модалка index 6 + "type_id": NoteTypeEnum.IMAGE, + "schema": NoteImagePost( + header="header_6", + is_always=False, + frequency=10, + group_ids=[group.id for group in groups][2:3], + service_ids=[service.id for service in services][2:3], + start_ts=NOW + timedelta(hours=2) * 5, + end_ts=NOW - timedelta(hours=1) * 5, + ), + "admin_id": authlib_user_data.get("id"), + "status": ModalStatus.ARCHIVED, + }, + { # просроченная модалка c is_always=True index 7 + "type_id": NoteTypeEnum.IMAGE, + "schema": NoteImagePost( + header="header_7", + is_always=True, + frequency=10, + group_ids=[group.id for group in groups][2:3], + service_ids=[service.id for service in services][2:3], + start_ts=NOW + timedelta(hours=3) * 5, + end_ts=NOW - timedelta(hours=1) * 5, + ), + "admin_id": authlib_user_data.get("id"), + "status": ModalStatus.ARCHIVED, + }, + ] + ) + notes = [create_note(**note) for note in note_data] + + for note in notes: + dbsession.add(note) + dbsession.commit() + yield notes + for note in notes: + dbsession.delete(note) + dbsession.commit() diff --git a/tests/test_routes/test_groups.py b/tests/test_routes/test_groups.py new file mode 100644 index 0000000..9260cde --- /dev/null +++ b/tests/test_routes/test_groups.py @@ -0,0 +1,130 @@ +import pytest +from starlette import status + +from modal_backend.models import Group +from modal_backend.schemas.models import GroupGet +from modal_backend.settings import get_settings + +url: str = "/group" +settings = get_settings() + + +@pytest.mark.parametrize( + "status_code, body", + [ + (status.HTTP_200_OK, {"group_id": 4, "name": "Group_3"}), + (status.HTTP_409_CONFLICT, {"group_id": 2, "name": "Group_2"}), + (status.HTTP_422_UNPROCESSABLE_CONTENT, {"group_id": "abc", "name": "Group_2"}), + (status.HTTP_422_UNPROCESSABLE_CONTENT, {"group_id": 4, "name": 123}), + ], +) +def test_post_group(client, dbsession, groups, status_code, body): + response = client.post(url, json=body) + assert response.status_code == status_code + + if status_code == status.HTTP_200_OK: + response_data = response.json() + response_model = GroupGet(**response_data) + exist_group = Group.query(session=dbsession).filter(Group.id == response_model.id).one_or_none() + assert exist_group + try: + assert exist_group.group_id == body.get("group_id") + assert exist_group.name == body.get("name") + finally: + dbsession.delete(exist_group) + dbsession.commit() + + +@pytest.mark.parametrize( + "status_code", + [ + (status.HTTP_200_OK), + ], +) +def test_get_groups(client, status_code): + response = client.get(url) + assert response.status_code == status_code + + +@pytest.mark.parametrize( + "status_code, body, group_n", + [ + ( + status.HTTP_200_OK, + {"group_id": 999, "name": "New_group_name"}, + 1, + ), + ( + status.HTTP_200_OK, + {"group_id": 2, "name": "New_group_name"}, + 1, + ), + ( + status.HTTP_200_OK, + {"group_id": 999, "name": "Group_2"}, + 1, + ), + ( + status.HTTP_404_NOT_FOUND, + {"group_id": 1, "name": "New_group_name"}, + 999, + ), + ( + status.HTTP_409_CONFLICT, + {"group_id": 2, "name": "Group_2"}, + 1, + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + {"group_id": "abc", "name": "Group_2"}, + 1, + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + {"group_id": 2, "name": 999}, + 1, + ), + ], +) +def test_update_group(client, dbsession, groups, status_code, body, group_n): + group_indexes = list(range(len(groups))) + response = client.patch(f"{url}/{groups[group_n].id if group_n in group_indexes else group_n}", json=body) + assert response.status_code == status_code + + if status_code == status.HTTP_200_OK: + response_data = response.json() + response_model = GroupGet(**response_data) + exist_group = ( + Group.query(session=dbsession).filter(Group.id == response_model.id).populate_existing().one_or_none() + ) + assert exist_group + assert exist_group.group_id == body.get("group_id") + assert exist_group.name == body.get("name") + + +@pytest.mark.parametrize( + "status_code, group_n", + [ + ( + status.HTTP_200_OK, + 1, + ), + ( + status.HTTP_404_NOT_FOUND, + 99, + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + "abc", + ), + ], +) +def test_delete_group(client, dbsession, groups, status_code, group_n): + group_indexes = list(range(len(groups))) + response = client.delete(f"{url}/{groups[group_n].id if group_n in group_indexes else group_n}") + assert response.status_code == status_code + if status_code == status.HTTP_200_OK: + none_exist_group = ( + dbsession.query(Group).filter(Group.id == groups[group_n].id).populate_existing().one_or_none() + ) + assert none_exist_group.is_deleted diff --git a/tests/test_routes/test_notes.py b/tests/test_routes/test_notes.py new file mode 100644 index 0000000..1f9ea39 --- /dev/null +++ b/tests/test_routes/test_notes.py @@ -0,0 +1,654 @@ +from typing import Any + +import pytest +from starlette import status + +from modal_backend.models.db import Group, ModalStatus, Note, NoteTypeEnum, Service +from modal_backend.schemas.models import NoteChoiceGet, NoteImageGet, NoteInfoGet, NoteRatingGet, NoteTextGet +from modal_backend.settings import get_settings + +url: str = "/notification" +settings = get_settings() + + +def resolve_items(n_list: list | Any, items: list) -> list | Any: + """ + Вспомогательная функция для перевода индексов списков групп и сервисов + с учётом возможности указывать невалидные значения(для негативных кейсов) + """ + indexes = range(len(items)) + list_ids = [] + if isinstance(n_list, list): + for n in n_list: + if n in indexes: + list_ids.append(items[n].id) + else: + list_ids.append(n) + else: + list_ids = n_list + return list_ids + + +@pytest.mark.parametrize( + "status_code, body, type_model, path", + [ + ( + status.HTTP_200_OK, + { + "header": "string", + "group_ids": [0], # индексы + "service_ids": [0], # индексы + "frequency": 0, + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": False, + "info_text": "string", + }, + NoteInfoGet, + "/info", + ), + ( + status.HTTP_200_OK, + { + "header": "string", + "group_ids": [0], # индексы + "service_ids": [0], # индексы + "frequency": 0, + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": True, + "rating_max": 0, + }, + NoteRatingGet, + "/rating", + ), + ( + status.HTTP_200_OK, + { + "header": "string", + "group_ids": [0], # индексы + "service_ids": [0], # индексы + "frequency": 0, + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": False, + "text": "string", + "max_length": 6, + }, + NoteTextGet, + "/text", + ), + ( + status.HTTP_200_OK, + { + "header": "string", + "group_ids": [0], # индексы + "service_ids": [0], # индексы + "frequency": 0, + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": True, + "choice_options": [{"id": 0, "text": "string"}], + "is_multiple": True, + }, + NoteChoiceGet, + "/choice", + ), + ( + status.HTTP_200_OK, + { + "header": "string", + "group_ids": [0], # индексы + "service_ids": [0], # индексы + "frequency": 0, + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": False, + "images": ["string"], + }, + NoteImageGet, + "/image", + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + { + "header": "string", + "group_ids": "two", # индексы + "service_ids": [1, "two", 3], # индексы + "frequency": "digit", + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": None, + "info_text": "string", + }, + NoteInfoGet, + "/info", + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + { + "header": "string", + "group_ids": "two", # индексы + "service_ids": [1, "two", 3], # индексы + "frequency": "digit", + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": None, + "max_rating": "string", + }, + NoteRatingGet, + "/rating", + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + { + "header": "string", + "group_ids": "two", # индексы + "service_ids": [1, "two", 3], # индексы + "frequency": "digit", + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": None, + "text": "string", + "max_length": "string", + }, + NoteTextGet, + "/text", + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + { + "header": "string", + "group_ids": "two", # индексы + "service_ids": [1, "two", 3], # индексы + "frequency": "digit", + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": None, + "choice_options": [{"id": 0, "text": "string"}], + "is_multiple": None, + }, + NoteChoiceGet, + "/choice", + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + { + "header": "string", + "group_ids": "two", # индексы + "service_ids": [1, "two", 3], # индексы + "frequency": "digit", + "start_ts": "2026-08-04T17:21:45.694Z", + "end_ts": "2026-08-04T17:22:45.694Z", + "is_always": None, + "images": ["string"], + }, + NoteImageGet, + "/image", + ), + ], +) +def test_create_all_type_of_note(client, dbsession, groups, services, status_code, body, type_model, path): + + json_body = body + group_n_list = json_body.get("group_ids") + service_n_list = json_body.get("service_ids") + groups_id = resolve_items(group_n_list, groups) + services_id = resolve_items(service_n_list, services) + + json_body["group_ids"] = groups_id + json_body["service_ids"] = services_id + + response = client.post(f"{url}{path}", json=json_body) + assert response.status_code == status_code + + if status_code == status.HTTP_200_OK: + response_model = type_model.model_validate(response.json(), extra="forbid") + note = Note.query(session=dbsession).filter(Note.id == response_model.id).one_or_none() + assert note + assert response_model.type_id == note.type_id + assert response_model.header == note.header + assert response_model.status == note.status + assert response_model.admin_id == note.admin_id + + if type_model is NoteTextGet: + assert len(note.text) <= note.max_length + dbsession.delete(note) + + +def calculate_expected_len( + all_notes, + type_id: int | None, + groups_id: list[int] | None, + services_id: list[int] | None, + status: str | None, + limit: int, + offset: int, + asc_order: bool | None, +) -> list[Note]: + """ + Вспомогательная функция для подсчета корректной длины списка модалок + документирующая контракт пагинации и фильтрации + в modal_backend/utils/service.py::NoteService.get_note_by_filters + """ + filtered_notes = all_notes + if type_id is not None: + filtered_notes = [note for note in filtered_notes if note.type_id == type_id] + if groups_id is not None: + filtered_notes = [note for note in filtered_notes if any(g in note.group_ids for g in groups_id)] + if services_id is not None: + filtered_notes = [note for note in filtered_notes if any(s in note.service_ids for s in services_id)] + if status is not None: + filtered_notes = [note for note in filtered_notes if note.status == status] + + filtered_notes = sorted(filtered_notes, key=lambda note: note.start_ts, reverse=(asc_order is not True)) + + return [note.id for note in filtered_notes[offset : limit + offset]] + + +@pytest.mark.parametrize( + "status_code, type_id, group_n_list, service_n_list, modal_status, asc_order, limit, offset", + [ + # позитивные кейсы(объединенные проверки) + ( # все модалки + status.HTTP_200_OK, + None, # type_id + None, # group_n_list + None, # service_n_list + None, # modal_status + None, # asc_order + None, # limit + None, # offset + ), + ( # активные - ограничение по лимиту и смещению + порядок + группы + status.HTTP_200_OK, + None, # type_id + [2], # group_n_list + [2], # service_n_list + ModalStatus.ACTIVE.value, # modal_status + True, # asc_order + 2, # limit + 1, # offset + ), + ( # архив - ограничение по лимиту и смещению + порядок + status.HTTP_200_OK, + None, # type_id + [0, 1, 2], # group_n_list + [0, 1, 2], # service_n_list + ModalStatus.ARCHIVED.value, # modal_status + False, # asc_order + 999, # limit + 1, # offset + ), + ( + status.HTTP_404_NOT_FOUND, + None, # type_id + [0, 1, 2], # group_n_list + [0, 1, 2], # service_n_list + ModalStatus.ARCHIVED.value, # modal_status + False, # asc_order + None, # limit + 5, # offset + ), + ( # ограничение по группам и сервисам + status.HTTP_200_OK, + None, # type_id + [0, 1], # group_n_list + [2], # service_n_list + None, # modal_status + None, # asc_order + None, # limit + None, # offset + ), + ( # не существующие группы и сервисы + status.HTTP_404_NOT_FOUND, + None, # type_id + [4], # group_n_list + [4], # service_n_list + None, # modal_status + None, # asc_order + None, # limit + None, # offset + ), + ( # ограничение по типу модалки + status.HTTP_200_OK, + NoteTypeEnum.CHOICE.value, # type_id + None, # group_n_list + None, # service_n_list + None, # modal_status + None, # asc_order + None, # limit + None, # offset + ), + ( # нулевой лимит + status.HTTP_404_NOT_FOUND, + None, # type_id + None, # group_n_list + None, # service_n_list + None, # modal_status + None, # asc_order + 0, # limit + None, # offset + ), + # негативные кейсы + ( # отрицательный лимит + не валидный лимит + status.HTTP_422_UNPROCESSABLE_CONTENT, + None, # type_id + None, # group_n_list + None, # service_n_list + None, # modal_status + None, # asc_order + -1, # limit + None, # offset + ), + ( # отрицательное смещение + не валидное смещение + status.HTTP_422_UNPROCESSABLE_CONTENT, + None, # type_id + None, # group_n_list + None, # service_n_list + None, # modal_status + None, # asc_order + None, # limit + -1, # offset + ), + ( # offset превышающее lwc и limit + status.HTTP_404_NOT_FOUND, + None, # type_id + None, # group_n_list + None, # service_n_list + None, # modal_status + None, # asc_order + 4, # limit + 999, # offset + ), + ( # не существующий type_id + status.HTTP_404_NOT_FOUND, + 999, # type_id + None, # group_n_list + None, # service_n_list + None, # modal_status + None, # asc_order + None, # limit + None, # offset + ), + ( # не валидный type_id + status.HTTP_422_UNPROCESSABLE_CONTENT, + "abc", # type_id + None, # group_n_list + None, # service_n_list + None, # modal_status + None, # asc_order + None, # limit + None, # offset + ), + ( # не валидные group_n_list + status.HTTP_422_UNPROCESSABLE_CONTENT, + None, # type_id + "two", # group_n_list + None, # service_n_list + None, # modal_status + None, # asc_order + None, # limit + None, # offset + ), + ( # не валидные service_ids + status.HTTP_422_UNPROCESSABLE_CONTENT, + None, # type_id + None, # group_n_list + "two", # service_n_list + None, # modal_status + None, # asc_order + None, # limit + None, # offset + ), + ( # не валидный status + status.HTTP_422_UNPROCESSABLE_CONTENT, + None, # type_id + None, # group_n_list + None, # service_n_list + 999, # modal_status + None, # asc_order + None, # limit + None, # offset + ), + ( # не валидные asc_order + status.HTTP_422_UNPROCESSABLE_CONTENT, + None, # type_id + None, # group_n_list + None, # service_n_list + None, # modal_status + 999, # asc_order + None, # limit + None, # offset + ), + ], +) +def test_get_notes( + client, + notes, + groups, + services, + status_code, + type_id, + group_n_list, + service_n_list, + modal_status, + asc_order, + limit, + offset, +): + # Добавление id групп и сервисов с проверкой типа, чтобы можно было указать не валидные query-параметры. + groups_id = resolve_items(group_n_list, groups) + services_id = resolve_items(service_n_list, services) + + dict_of_params = { + "type_id": type_id if type_id is not None else None, + "groups_id": groups_id if groups_id is not None else None, + "services_id": services_id if services_id is not None else None, + "status": modal_status if modal_status is not None else None, + "asc_order": asc_order if asc_order is not None else None, + "limit": limit if limit is not None else None, + "offset": offset if offset is not None else None, + } + query = {k: v for k, v in dict_of_params.items() if v is not None} + response = client.get(url, params=query) + + assert response.status_code == status_code + + if status_code == status.HTTP_200_OK: + response_data = response.json() + + ids_in_response = [note.get("id") for note in response_data] + # проверяем порядок + start_ts_by_id = {note.id: note.start_ts for note in notes} + ts_data = [start_ts_by_id[note_id] for note_id in ids_in_response] + if query.get("asc_order"): + assert ts_data == sorted(ts_data) + else: + assert ts_data == sorted(ts_data, reverse=True) + + # проверка контракта фильтрации и пагинации(содержимого и длины после урезания limit-ом и offset-ом) + expected_note_ids = calculate_expected_len( + notes, + type_id=query.get("type_id"), + groups_id=query.get("groups_id"), + services_id=query.get("services_id"), + status=query.get("status"), + limit=query.get("limit", 10), + offset=query.get("offset", 0), + asc_order=query.get("asc_order"), + ) + + assert ids_in_response == expected_note_ids + # проверка корректности данных отфильтрованных модалок + for resp_obj in response_data: + if type_id: + assert resp_obj.get("type_id") == type_id + if modal_status: + assert resp_obj.get("status") == modal_status + + +@pytest.mark.parametrize( + "status_code, note_n, type_model", + [ + ( + status.HTTP_200_OK, + 0, + NoteInfoGet, + ), + ( + status.HTTP_200_OK, + 1, + NoteRatingGet, + ), + ( + status.HTTP_200_OK, + 2, + NoteTextGet, + ), + ( + status.HTTP_200_OK, + 3, + NoteChoiceGet, + ), + ( + status.HTTP_200_OK, + 4, + NoteImageGet, + ), + ( + status.HTTP_404_NOT_FOUND, + -1, + None, + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + "abc", + None, + ), + ], +) +def test_get_note_by_id(client, notes, status_code, note_n, type_model): + notes_indexes = range(len(notes)) + id_of_note = notes[note_n].id if note_n in notes_indexes else note_n + response = client.get(f"{url}/{id_of_note}") + + assert response.status_code == status_code + + if status_code == status.HTTP_200_OK: + type_model.model_validate(response.json(), extra="forbid") + + +@pytest.mark.parametrize( + "status_code, note_n, modal_status, deleted_group_id_flag, deleted_service_id_flag", + [ + ( + status.HTTP_200_OK, + 0, + ModalStatus.ARCHIVED, + False, + False, + ), + ( + status.HTTP_404_NOT_FOUND, + -1, + ModalStatus.ACTIVE, + False, + False, + ), + ( + status.HTTP_200_OK, + 3, + ModalStatus.ACTIVE, + False, + False, + ), + ( + status.HTTP_403_FORBIDDEN, + 3, + ModalStatus.ACTIVE, + True, + False, + ), + ( + status.HTTP_403_FORBIDDEN, + 3, + ModalStatus.ACTIVE, + False, + True, + ), + ( + status.HTTP_403_FORBIDDEN, + 3, + ModalStatus.ACTIVE, + True, + True, + ), + ( + status.HTTP_403_FORBIDDEN, + 6, + ModalStatus.ACTIVE, + False, + False, + ), # просроченная модалка + ( + status.HTTP_200_OK, + 7, + ModalStatus.ACTIVE, + False, + False, + ), # просроченная модалка с is_always=True + ], +) +def test_update_note_status( + client, + dbsession, + mock_datetime_now, + notes, + note_n, + status_code, + modal_status, + deleted_group_id_flag, + deleted_service_id_flag, +): + + notes_indexes = range(len(notes)) + id_of_note = notes[note_n].id if note_n in notes_indexes else note_n + + if deleted_group_id_flag: + for group_id in notes[note_n].group_ids: + group = Group.query(session=dbsession).filter(Group.id == group_id).one() + group.is_deleted = True + dbsession.commit() + if deleted_service_id_flag: + for service_id in notes[note_n].service_ids: + service = Service.query(session=dbsession).filter(Service.id == service_id).one() + service.is_deleted = True + dbsession.commit() + + response = client.patch(f"{url}/{id_of_note}/status") + assert response.status_code == status_code + + if status_code == status.HTTP_200_OK: + response_data = response.json() + note = Note.query(session=dbsession).filter(Note.id == response_data.get("id")).populate_existing().one() + assert note.status == modal_status + + +@pytest.mark.parametrize( + "status_code, note_n", + [ + (status.HTTP_200_OK, 0), + (status.HTTP_404_NOT_FOUND, -1), + (status.HTTP_422_UNPROCESSABLE_CONTENT, "one"), + ], +) +def test_delete_note(client, dbsession, notes, status_code, note_n): + note_indexes = range(len(notes)) + id_of_note = notes[note_n].id if note_n in note_indexes else note_n + + response = client.delete(f"{url}/{id_of_note}") + assert response.status_code == status_code + if status_code == status.HTTP_200_OK: + deleted_note = dbsession.query(Note).filter(Note.id == id_of_note).populate_existing().one_or_none() + assert deleted_note.is_deleted diff --git a/tests/test_routes/test_services.py b/tests/test_routes/test_services.py new file mode 100644 index 0000000..b324192 --- /dev/null +++ b/tests/test_routes/test_services.py @@ -0,0 +1,129 @@ +import pytest +from starlette import status + +from modal_backend.models import Service +from modal_backend.schemas.models import ServiceGet +from modal_backend.settings import get_settings + +url: str = "/service" +settings = get_settings() + + +@pytest.mark.parametrize( + "status_code, body", + [ + (status.HTTP_200_OK, {"service_id": 4, "name": "Service_3"}), + (status.HTTP_409_CONFLICT, {"service_id": 2, "name": "Service_2"}), + (status.HTTP_422_UNPROCESSABLE_CONTENT, {"service_id": "abc", "name": "Service_2"}), + (status.HTTP_422_UNPROCESSABLE_CONTENT, {"service_id": 4, "name": 123}), + ], +) +def test_post_service(client, dbsession, services, status_code, body): + response = client.post(url, json=body) + assert response.status_code == status_code + + if status_code == status.HTTP_200_OK: + response_data = response.json() + response_model = ServiceGet(**response_data) + exist_service = Service.query(session=dbsession).filter(Service.id == response_model.id).one_or_none() + assert exist_service + try: + assert exist_service.service_id == body.get("service_id") + assert exist_service.name == body.get("name") + finally: + dbsession.delete(exist_service) + dbsession.commit() + + +@pytest.mark.parametrize( + "status_code", + [ + (status.HTTP_200_OK), + ], +) +def test_get_services(client, status_code): + response = client.get(url) + assert response.status_code == status_code + + +@pytest.mark.parametrize( + "status_code, body, service_n", + [ + ( + status.HTTP_200_OK, + {"service_id": 999, "name": "New_service_name"}, + 1, + ), + ( + status.HTTP_200_OK, + {"service_id": 2, "name": "New_service_name"}, + 1, + ), + ( + status.HTTP_200_OK, + {"service_id": 999, "name": "Service_2"}, + 1, + ), + ( + status.HTTP_404_NOT_FOUND, + {"service_id": 1, "name": "New_service_name"}, + 999, + ), + ( + status.HTTP_409_CONFLICT, + {"service_id": 2, "name": "Service_2"}, + 1, + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + {"service_id": "abc", "name": "Service_2"}, + 1, + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + {"service_id": 2, "name": 999}, + 1, + ), + ], +) +def test_update_service(client, dbsession, services, status_code, body, service_n): + service_indexes = list(range(len(services))) + response = client.patch(f"{url}/{services[service_n].id if service_n in service_indexes else service_n}", json=body) + assert response.status_code == status_code + + if status_code == status.HTTP_200_OK: + response_model = ServiceGet.model_validate(response.json(), extra="forbid") + exist_service = ( + Service.query(session=dbsession).filter(Service.id == response_model.id).populate_existing().one_or_none() + ) + assert exist_service + assert exist_service.service_id == body.get("service_id") + assert exist_service.name == body.get("name") + + +@pytest.mark.parametrize( + "status_code, service_n", + [ + ( + status.HTTP_200_OK, + 1, + ), + ( + status.HTTP_404_NOT_FOUND, + 999, + ), + ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + "abc", + ), + ], +) +def test_delete_service(client, dbsession, services, status_code, service_n): + service_indexes = list(range(len(services))) + response = client.delete(f"{url}/{services[service_n].id if service_n in service_indexes else service_n}") + assert response.status_code == status_code + if status_code == status.HTTP_200_OK: + none_exist_service = ( + dbsession.query(Service).filter(Service.id == services[service_n].id).populate_existing().one_or_none() + ) + assert none_exist_service.is_deleted