Route patch-event-id logic update;

calendar_backend/routes/event/event.py

@@ -8,9 +8,9 @@
 from fastapi_sqlalchemy import db
 from pydantic import TypeAdapter
 
-from calendar_backend.exceptions import NotEnoughCriteria
+from calendar_backend.exceptions import NotEnoughCriteria, ObjectNotFound
 from calendar_backend.methods import list_calendar
-from calendar_backend.models import Event, Group, Lecturer, Room
+from calendar_backend.models import Event, EventsGroups, EventsLecturers, EventsRooms, Group, Lecturer, Room
 from calendar_backend.routes.models import EventGet
 from calendar_backend.routes.models.event import (
     EventPatch,
@@ -199,7 +199,57 @@ async def patch_event_by_name(
 async def patch_event(
     id: int, event_inp: EventPatch, _=Depends(UnionAuth(scopes=["timetable.event.update"]))
 ) -> EventGet:
-    patched = Event.update(id, session=db.session, **event_inp.model_dump(exclude_unset=True))
+    # проверяем, что событие с таким id существует
+    Events = db.session.query(Event).filter(Event.id == id).one_or_none()
+    if not Events:
+        raise ObjectNotFound(type=type(Events), ids=id)
+    event_upd_data = event_inp.model_dump(exclude_unset=True)
+
+    upd_rooms = set(event_upd_data.get("room_id"))
+    db_rooms = {room.id for room in db.session.query(Room).filter(Room.id.in_(upd_rooms)).all()}
+    no_exist_rooms = upd_rooms - db_rooms
+    if no_exist_rooms:
+        raise ObjectNotFound(type=type(Room), ids=list(no_exist_rooms))
+
+    upd_groups = set(event_upd_data.get("group_id"))
+    db_groups = {group.id for group in db.session.query(Group).filter(Group.id.in_(upd_groups)).all()}
+    no_exist_groups = upd_groups - db_groups
+    if no_exist_groups:
+        raise ObjectNotFound(type=type(Group), ids=list(no_exist_groups))
+
+    upd_lecturers = set(event_upd_data.get("lecturer_id"))
+    db_lecturers = {lecturer.id for lecturer in db.session.query(Lecturer).filter(Lecturer.id.in_(upd_lecturers)).all()}
+    no_exist_lecturers = upd_lecturers - db_lecturers
+    if no_exist_lecturers:
+        raise ObjectNotFound(type=type(Lecturer), ids=list(no_exist_lecturers))
+
+    def add_new_remove_old_ids(dbsession, event_id, artefact: str, event_instance, negotiator_instance) -> None:
+        """Вспомогательная фукнция описывающая удаление старых и добавление новых (room_, group_, lecturer_)*id - артефактов"""
+        # получаем множество артефактов события
+        event_artefacts = {room.id for room in getattr(event_instance, f"{artefact}")}
+        # получаем множетсво страрых артефактов
+        old_artefacts = event_artefacts - set(event_upd_data.get(f"{artefact}_id"))
+        # получаем множество новых артефактов
+        new_artefacts = set(event_upd_data.get(f"{artefact}_id")) - event_artefacts
+        # получаем артефакты к удалению(уникальные старые)
+        to_remove = old_artefacts - new_artefacts
+        # получаем артефакты к созданию(уникальные новые)
+        to_add = new_artefacts - old_artefacts
+
+        if to_remove:
+            dbsession.query(negotiator_instance).filter(
+                negotiator_instance.event_id == event_id, getattr(negotiator_instance, f"{artefact}_id").in_(to_remove)
+            ).delete()
+        if to_add:
+            dbsession.bulk_insert_mappings(
+                negotiator_instance, [{"event_id": f"{event_id}", f"{artefact}_id": id_to_add} for id_to_add in to_add]
+            )
+
+    add_new_remove_old_ids(db.session, id, "room", Events, EventsRooms)
+    add_new_remove_old_ids(db.session, id, "group", Events, EventsGroups)
+    add_new_remove_old_ids(db.session, id, "lecturer", Events, EventsLecturers)
+
+    patched = Event.update(id, session=db.session, **event_upd_data)
     db.session.commit()
     return EventGet.model_validate(patched)
 

requirements.dev.txt

@@ -5,3 +5,4 @@ pytest-mock
 black==23.11.0
 isort
 autoflake
+testcontainers[postgres]

tests/conftest.py

@@ -1,27 +1,100 @@
 from datetime import datetime
+from functools import lru_cache
+from pathlib import Path
 
 import pytest
+from alembic import command
+from alembic.config import Config as AlembicConfig
 from fastapi.testclient import TestClient
+from pytest import MonkeyPatch
 from pytest_mock import MockerFixture
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, sessionmaker
 from starlette import status
+from testcontainers.postgres import PostgresContainer
 
 from calendar_backend.models.base import DeclarativeBase
 from calendar_backend.models.db import Event, Group, Lecturer, Room
-from calendar_backend.routes import app
-from calendar_backend.settings import get_settings
+from calendar_backend.settings import Settings
+
+
+class PostgresConfig:
+    """Дата-класс со значениями для контейнера с тестовой БД и для alembic-миграции."""
+
+    container_name: str = "timetable-api-test"
+    username: str = "postgres"
+    host: str = "localhost"
+    external_port: int = 5433
+    image: str = "postgres:15"
+    host_auth_method: str = "trust"
+    alembic_ini: str = Path(__file__).resolve().parent.parent / "alembic.ini"
+
+    @classmethod
+    def get_url(cls) -> str:
+        """Возвращает URI для подключения к БД."""
+        return f"postgresql://{cls.username}@{cls.host}:{cls.external_port}/postgres"
+
+
+@pytest.fixture(scope="session")
+def session_mp():
+    mp = MonkeyPatch()
+    yield mp
+    mp.undo()
+
+
+@pytest.fixture(scope="session")
+def get_settings_mock(session_mp):
+    """Переопределение get_settings в calendar_backend/settings.py."""
+
+    @lru_cache
+    def get_test_settings():
+        test_settings = Settings()
+        test_settings.DB_DSN = PostgresConfig.get_url()
+        return test_settings
+
+    dsn_mock = session_mp.setattr("calendar_backend.settings.get_settings", get_test_settings)
+    return dsn_mock
+
+
+@pytest.fixture(scope="session")
+def get_app_with_test_settings(get_settings_mock):
+    """Загрузка app с тестовыми настройками."""
+    from calendar_backend.routes import app
+
+    return app
+
+
+@pytest.fixture(scope="session")
+def db_container(get_settings_mock):
+    """Фикстура настройки БД для тестов в Docker-контейнере."""
+    container = (
+        PostgresContainer(
+            image=PostgresConfig.image, username=PostgresConfig.username, dbname=PostgresConfig.container_name
+        )
+        .with_bind_ports(5432, PostgresConfig.external_port)
+        .with_env("POSTGRES_HOST_AUTH_METHOD", PostgresConfig.host_auth_method)
+    )
+    container.start()
+    alembic_ini = PostgresConfig.alembic_ini
+    cfg = AlembicConfig(str(alembic_ini.resolve()))
+    cfg.set_main_option("script_location", "%(here)s/migrations")
+    command.upgrade(cfg, "head")
+    try:
+        yield PostgresConfig.get_url()
+    finally:
+        container.stop()
 
 
 @pytest.fixture()
-def client():
+def client(get_app_with_test_settings):
+    app = get_app_with_test_settings
     client = TestClient(app)
     return client
 
 
 @pytest.fixture()
-def client_auth(mocker: MockerFixture):
-    user_mock = mocker.patch('auth_lib.fastapi.UnionAuth.__call__')
+def client_auth(mocker: MockerFixture, get_app_with_test_settings):
+    user_mock = mocker.patch('auth_lib.fastapi.UnionAuth.__call__', autospec=True)
     user_mock.return_value = {
         "session_scopes": [{"id": 0, "name": "string", "comment": "string"}],
         "user_scopes": [{"id": 0, "name": "string", "comment": "string"}],
@@ -30,17 +103,17 @@ def client_auth(mocker: MockerFixture):
         "id": 0,
         "email": "string",
     }
+    app = get_app_with_test_settings
     client = TestClient(app)
     return client
 
 
 @pytest.fixture()
-def dbsession():
-    settings = get_settings()
-    engine = create_engine(str(settings.DB_DSN), isolation_level='AUTOCOMMIT')
+def dbsession(db_container):
+    engine = create_engine(str(db_container), isolation_level='AUTOCOMMIT')
     TestingSessionLocal = sessionmaker(bind=engine)
     DeclarativeBase.metadata.create_all(bind=engine)
-    return TestingSessionLocal()
+    yield TestingSessionLocal()
 
 
 @pytest.fixture()

tests/event/event.py

@@ -237,11 +237,16 @@ def test_update_all(client_auth: TestClient, dbsession: Session):
     assert response_obj["end_ts"][:20] == request_obj["end_ts"][:20]
 
     # Update
+    another_room = Room(name="5-07" + datetime.datetime.utcnow().isoformat(), direction="North")
+    another_lecturer = Lecturer(first_name="s", middle_name="s", last_name="s")
+    another_group = Group(name="", number="202" + datetime.datetime.utcnow().isoformat())
+    dbsession.add_all((another_room, another_group, another_lecturer))
+    dbsession.commit()
     request_obj_2 = {
         "name": "frfrf",
-        "room_id": [room.id],
-        "group_id": [group.id],
-        "lecturer_id": [lecturer.id],
+        "room_id": [another_room.id],
+        "group_id": [another_group.id],
+        "lecturer_id": [another_lecturer.id],
         "start_ts": "2022-08-26T22:32:38.575Z",
         "end_ts": "2022-08-26T22:32:38.575Z",
     }