Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions modal_backend/routes/notes.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,16 +9,21 @@
from modal_backend.schemas.base import StatusResponseModel
from modal_backend.schemas.models import (
NoteChoiceGet,
NoteChoicePatch,
NoteChoicePost,
NoteGet,
NoteImageGet,
NoteImagePatch,
NoteImagePost,
NoteInfoGet,
NoteInfoPatch,
NoteInfoPost,
NoteRatingGet,
NoteRatingPatch,
NoteRatingPost,
NoteStatus,
NoteTextGet,
NoteTextPatch,
NoteTextPost,
NotificationGet,
)
Expand Down Expand Up @@ -209,6 +214,46 @@ async def create_note_images(
return NoteImageGet.model_validate(new_note)


@note.patch("/info/{id}", response_model=NoteInfoGet)
async def update_note_info(
id: int, note_info: NoteInfoPatch, user=Depends(UnionAuth(scopes=["modal.note.patch"]))
) -> NoteInfoGet:
updated_note = await NoteService.update_note(db, id, note_info, NoteTypeEnum.INFO)
return NoteInfoGet.model_validate(updated_note)


@note.patch("/rating/{id}", response_model=NoteRatingGet)
async def update_note_rating(
id: int, note_info: NoteRatingPatch, user=Depends(UnionAuth(scopes=["modal.note.patch"]))
) -> NoteRatingGet:
updated_note = await NoteService.update_note(db, id, note_info, NoteTypeEnum.RATING)
return NoteRatingGet.model_validate(updated_note)


@note.patch("/text/{id}", response_model=NoteTextGet)
async def update_note_text(
id: int, note_info: NoteTextPatch, user=Depends(UnionAuth(scopes=["modal.note.patch"]))
) -> NoteTextGet:
updated_note = await NoteService.update_note(db, id, note_info, NoteTypeEnum.TEXT)
return NoteTextGet.model_validate(updated_note)


@note.patch("/choice/{id}", response_model=NoteChoiceGet)
async def update_note_choice(
id: int, note_info: NoteChoicePatch, user=Depends(UnionAuth(scopes=["modal.note.patch"]))
) -> NoteChoiceGet:
updated_note = await NoteService.update_note(db, id, note_info, NoteTypeEnum.CHOICE)
return NoteChoiceGet.model_validate(updated_note)


@note.patch("/image/{id}", response_model=NoteImageGet)
async def update_note_image(
id: int, note_info: NoteImagePatch, user=Depends(UnionAuth(scopes=["modal.note.patch"]))
) -> NoteImageGet:
updated_note = await NoteService.update_note(db, id, note_info, NoteTypeEnum.IMAGE)
return NoteImageGet.model_validate(updated_note)


@note.patch("/{id}/status", response_model=NoteStatus)
async def update_note_status(id: int, user=Depends(UnionAuth(scopes=["modal.note.patch"]))) -> NoteStatus:
"""
Expand Down
41 changes: 41 additions & 0 deletions modal_backend/schemas/models.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import datetime

from pydantic import model_validator

from modal_backend.models.db import ModalStatus
from modal_backend.schemas.base import Base

Expand Down Expand Up @@ -77,28 +79,67 @@ class NotificationPost(Base):
is_always: bool


class NotificationPatch(Base):
type_id: int | None = None
header: str | None = None
group_ids: list[int] | None = None
service_ids: list[int] | None = None
frequency: int | None = None
start_ts: datetime.datetime | None = None
end_ts: datetime.datetime | None = None
is_always: bool | None = None

@model_validator(mode="after")
def validate_period(self):
if self.start_ts is not None and self.end_ts is not None and self.start_ts >= self.end_ts:
raise ValueError("start_ts must be earlier than end_ts")
return self


class NoteInfoPost(NotificationPost): # type_id=1
info_text: str | None = None


class NoteInfoPatch(NotificationPatch): # type_id=1
info_text: str | None = None


class NoteRatingPost(NotificationPost): # type_id=2
rating_max: int | None = None


class NoteRatingPatch(NotificationPatch): # type_id=2
rating_max: int | None = None


class NoteTextPost(NotificationPost): # type_id=3
text: str | None = None
max_length: int | None = None


class NoteTextPatch(NotificationPatch): # type_id=3
text: str | None = None
max_length: int | None = None


class NoteChoicePost(NotificationPost): # type_id=4
choice_options: list[ChoiceOption] | None = None
is_multiple: bool | None = None


class NoteChoicePatch(NotificationPatch): # type_id=4
choice_options: list[ChoiceOption] | None = None
is_multiple: bool | None = None


class NoteImagePost(NotificationPost): # type_id=5
images: list[str] | None = None


class NoteImagePatch(NotificationPatch): # type_id=5
images: list[str] | None = None


class GroupGet(Base):
id: int
group_id: int
Expand Down
32 changes: 31 additions & 1 deletion modal_backend/utils/services.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
from modal_backend.exceptions import AlreadyExists, ForbiddenAction, ObjectNotFound
from modal_backend.models.db import Group, ModalStatus, Note, Service
from modal_backend.schemas.base import StatusResponseModel
from modal_backend.schemas.models import GroupPost, ServicePost
from modal_backend.schemas.models import GroupPost, NotificationPatch, ServicePost


class NoteService:
Expand Down Expand Up @@ -85,6 +85,36 @@ async def update_status(cls, db: Session, id: int) -> Note:
)
return updated_note

@classmethod
async def update_note(cls, db: Session, id: int, note_info: NotificationPatch, note_type: int) -> Note:
note = Note.get(session=db.session, id=id)
if note.type_id != note_type:
raise ForbiddenAction(Note)

values = note_info.model_dump(exclude_unset=True)
if values.get("type_id") is not None:
raise ForbiddenAction(Note)
values.pop("type_id", None)

content_fields = {
"info_text",
"rating_max",
"text",
"max_length",
"choice_options",
"is_multiple",
"images",
}
if note.status == ModalStatus.ACTIVE and content_fields.intersection(values):
raise ForbiddenAction(Note)

start_ts = values.get("start_ts", note.start_ts)
end_ts = values.get("end_ts", note.end_ts)
if start_ts is not None and end_ts is not None and start_ts >= end_ts:
raise ForbiddenAction(Note)

return Note.update(id=id, session=db.session, **values)


class ServiceManager:
"""
Expand Down
159 changes: 159 additions & 0 deletions tests/test_routes/test_notes.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from datetime import datetime, timedelta, timezone
from typing import Any

import pytest
Expand Down Expand Up @@ -635,6 +636,164 @@ def test_update_note_status(
assert note.status == modal_status


@pytest.mark.parametrize(
"status_code, path, note_n, payload, type_model, expected_field",
[
(
status.HTTP_200_OK,
"/info",
0,
{
"header": "updated_header_info",
"frequency": 11,
"group_ids": [1],
"service_ids": [1],
"start_ts": "2026-08-28T19:10:29.567Z",
"end_ts": "2026-08-30T19:10:29.567Z",
"is_always": False,
},
NoteInfoGet,
"header",
),
(
status.HTTP_200_OK,
"/rating",
1,
{
"header": "updated_header_rating",
"frequency": 12,
"group_ids": [1],
"service_ids": [1],
"start_ts": "2026-08-28T19:10:29.567Z",
"end_ts": "2026-08-30T19:10:29.567Z",
"is_always": False,
},
NoteRatingGet,
"header",
),
(
status.HTTP_200_OK,
"/text",
2,
{
"header": "updated_header_text",
"frequency": 13,
"group_ids": [1],
"service_ids": [1],
"start_ts": "2026-08-28T19:10:29.567Z",
"end_ts": "2026-08-30T19:10:29.567Z",
"is_always": False,
},
NoteTextGet,
"header",
),
(
status.HTTP_200_OK,
"/choice",
3,
{
"header": "updated_header_choice",
"choice_options": [{"id": 1, "text": "A"}, {"id": 2, "text": "B"}],
"is_multiple": True,
"frequency": 14,
"group_ids": [1],
"service_ids": [1],
"start_ts": "2026-08-28T19:10:29.567Z",
"end_ts": "2026-08-30T19:10:29.567Z",
"is_always": False,
},
NoteChoiceGet,
"choice_options",
),
(
status.HTTP_200_OK,
"/image",
4,
{
"header": "updated_header_image",
"images": ["img2.jpg", "img3.jpg"],
"frequency": 15,
"group_ids": [1],
"service_ids": [1],
"start_ts": "2026-08-28T19:10:29.567Z",
"end_ts": "2026-08-30T19:10:29.567Z",
"is_always": False,
},
NoteImageGet,
"images",
),
],
)
def test_update_note_by_type(client, notes, status_code, path, note_n, payload, type_model, expected_field):
id_of_note = notes[note_n].id

response = client.patch(f"{url}{path}/{id_of_note}", json=payload)
assert response.status_code == status_code

if status_code == status.HTTP_200_OK:
response_model = type_model.model_validate(response.json(), extra="forbid")
assert response_model.id == id_of_note

if expected_field == "choice_options":
actual_value = [{"id": item.id, "text": item.text} for item in response_model.choice_options]
assert actual_value == payload.get(expected_field)
else:
assert getattr(response_model, expected_field) == payload.get(expected_field)


@pytest.mark.parametrize(
"path, note_n, payload",
[
("/info", 1, {"header": "wrong_type_info"}),
("/rating", 0, {"header": "wrong_type_rating"}),
("/text", 1, {"header": "wrong_type_text"}),
("/choice", 0, {"header": "wrong_type_choice"}),
("/image", 0, {"header": "wrong_type_image"}),
("/text", 2, {"text": "changed_main_content"}),
("/rating", 1, {"rating_max": 10}),
],
)
def test_update_note_by_type_forbidden(client, notes, path, note_n, payload):
id_of_note = notes[note_n].id

response = client.patch(f"{url}{path}/{id_of_note}", json=payload)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert response.json()["status"] == "Error"


@pytest.mark.parametrize(
"path, payload",
[
("/choice", {"choice_options": [{"id": 1, "text": "Y"}]}),
("/image", {"images": ["new.jpg"]}),
],
)
def test_update_note_by_type_forbidden_active_content(client, dbsession, groups, services, path, payload):
note = Note(
type_id=NoteTypeEnum.CHOICE if path == "/choice" else NoteTypeEnum.IMAGE,
header="active_choice_or_image",
choice_options=[{"id": 1, "text": "A"}] if path == "/choice" else None,
images=["old.jpg"] if path == "/image" else None,
group_ids=[group.id for group in groups][:1],
service_ids=[service.id for service in services][:1],
frequency=10,
start_ts=datetime.now(timezone.utc).replace(tzinfo=None),
end_ts=datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(hours=1),
is_always=False,
admin_id=0,
status=ModalStatus.ACTIVE,
)
dbsession.add(note)
dbsession.commit()

response = client.patch(f"{url}{path}/{note.id}", json=payload)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert response.json()["status"] == "Error"

dbsession.delete(note)
dbsession.commit()


@pytest.mark.parametrize(
"status_code, note_n",
[
Expand Down
Loading