Test/add tests for groups notes services

modal_backend/routes/notes.py

@@ -33,13 +33,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 | None = Query(
         enum=["active", "archived"],
         default=None,
     ),
     asc_order: bool = False,
-    limit: int = 10,
-    offset: int = 0,
+    limit: int | None = Query(10, ge=0, description="Лимит записией"),
+    offset: int | None = Query(0, ge=0, description="Смещение записей на N+offset, где N - первая запись"),
     user=Depends(UnionAuth()),
 ) -> list[NoteGet]:
     """

tests/conftest.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timedelta
 from functools import lru_cache
 from pathlib import Path
 
@@ -10,7 +11,14 @@
 from sqlalchemy.orm import sessionmaker
 from testcontainers.postgres import PostgresContainer
 
-from modal_backend.models.db import NoteType
+from modal_backend.models.db import Group, ModalStatus, Note, NoteType, Service
+from modal_backend.schemas.models import (
+    NoteChoicePost,
+    NoteImagePost,
+    NoteInfoPost,
+    NoteRatingPost,
+    NoteTextPost,
+)
 from modal_backend.settings import Settings
 
 
@@ -145,15 +153,11 @@ def create_note_type(name: str, type_id: int):
 def note_types(dbsession):
     """Создает три разных типа модалок."""
     note_type_data = [
-        (
-            "Name 1",
-            1,
-        ),
-        (
-            "Name 2",
-            2,
-        ),
-        ("Name 3", 3),
+        ("info", 1),
+        ("rating", 2),
+        ("text", 3),
+        ("choice", 4),
+        ("image", 5),
     ]
 
     note_types = [create_note_type(*note_type) for note_type in note_type_data]
@@ -165,3 +169,139 @@ def note_types(dbsession):
     for note_type in note_types:
         dbsession.delete(note_type)
     dbsession.commit()
+
+
+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")]
+    groupes = [create_group(*group) for group in group_data]
+    for group in groupes:
+        dbsession.add(group)
+    dbsession.commit()
+    yield groupes
+    for group in groupes:
+        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()
+
+
+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,
+    note_types,
+    groups,
+    services,
+    authlib_user_data,
+):
+    note_data = [
+        {
+            "type_id": note_types[0].type_id,
+            "schema": NoteInfoPost(
+                header="header_1",
+                is_always=False,
+                frequency=10,
+                group_ids=[group.group_id for group in groups if group.group_id == 3],
+                service_ids=[service.service_id for service in services if service.service_id == 3],
+            ),
+            "admin_id": authlib_user_data.get("id"),
+            "status": ModalStatus.ACTIVE,
+        },
+        {
+            "type_id": note_types[1].type_id,
+            "schema": NoteRatingPost(
+                header="header_2",
+                is_always=False,
+                frequency=10,
+                group_ids=[group.group_id for group in groups if group.group_id < 3],
+                service_ids=[service.service_id for service in services if service.service_id < 3],
+            ),
+            "admin_id": authlib_user_data.get("id"),
+            "status": ModalStatus.ACTIVE,
+        },
+        {
+            "type_id": note_types[2].type_id,
+            "schema": NoteTextPost(
+                header="header_3",
+                is_always=False,
+                frequency=10,
+                group_ids=[group.group_id for group in groups if group.group_id == 3],
+                service_ids=[service.service_id for service in services if service.service_id == 3],
+            ),
+            "admin_id": authlib_user_data.get("id"),
+            "status": ModalStatus.ACTIVE,
+        },
+        {
+            "type_id": note_types[3].type_id,
+            "schema": NoteChoicePost(
+                header="header_4",
+                is_always=False,
+                frequency=10,
+                group_ids=[group.group_id for group in groups if group.group_id < 3],
+                service_ids=[service.service_id for service in services if service.service_id < 3],
+            ),
+            "admin_id": authlib_user_data.get("id"),
+            "status": ModalStatus.ARCHIVED,
+        },
+        {
+            "type_id": note_types[4].type_id,
+            "schema": NoteImagePost(
+                header="header_5",
+                is_always=False,
+                frequency=10,
+                group_ids=[group.group_id for group in groups if group.group_id == 3],
+                service_ids=[service.service_id for service in services if service.service_id == 3],
+            ),
+            "admin_id": authlib_user_data.get("id"),
+            "status": ModalStatus.ARCHIVED,
+        },
+    ]
+    for offset, d in enumerate(note_data):
+        d["schema"].start_ts = datetime.now() + offset * timedelta(hours=1)
+        d["schema"].end_ts = datetime.now() + offset * timedelta(hours=1)
+
+    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()

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",
+    [
+        (status.HTTP_200_OK),
+    ],
+)
+def test_get_group(client, groups, status_code):
+    response = client.get(url)
+    assert response.status_code == status_code
+
+
+@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 == response.status_code
+
+    if status_code == status.HTTP_200_OK:
+        response_data = response.json()
+        response_model = GroupGet(**response_data)
+        exist_group = dbsession.query(Group).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)
+
+
+@pytest.mark.parametrize(
+    "status_code, group_n",
+    [
+        (
+            status.HTTP_200_OK,
+            1,
+        ),
+        (
+            status.HTTP_404_NOT_FOUND,
+            999,
+        ),
+        (
+            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
+
+
+@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 = dbsession.query(Group).filter(Group.id == response_model.id).populate_existing().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)

tests/test_routes/test_note_type.py

@@ -29,7 +29,7 @@ def test_get_notification_type(client, note_types, status_code):
         (
             status.HTTP_200_OK,
             {
-                "type_id": 4,
+                "type_id": 6,
                 "name": "No_exist_type",
             },
         ),
@@ -43,7 +43,7 @@ def test_get_notification_type(client, note_types, status_code):
         (
             status.HTTP_422_UNPROCESSABLE_ENTITY,
             {
-                "type_id": 4,
+                "type_id": 6,
                 "name": 123,
             },
         ),
@@ -64,5 +64,8 @@ def test_post_create_notification_type(client, dbsession, note_types, status_cod
         response_model = NoteTypeGet(**response.json())
         exist_note_type = dbsession.query(NoteType).filter(NoteType.type_id == response_model.type_id).one_or_none()
         assert exist_note_type
-        assert exist_note_type.name == body.get("name")
-        assert exist_note_type.type_id == body.get("type_id")
+        try:
+            assert exist_note_type.name == body.get("name")
+            assert exist_note_type.type_id == body.get("type_id")
+        finally:
+            dbsession.delete(exist_note_type)