Enforce signup priority correctness

routes/user_router.py

@@ -19,11 +19,10 @@
 from helpers.image_checker import validate_image
 from helpers.rate_limit import rate_limit
 from helpers.types import ALLOWED_EXT, ALLOWED_IMG_SIZES, ALLOWED_IMG_TYPES, ASSETS_BASE_PATH
-from db_models.nollning_model import Nollning_DB
 from services import user as user_service
+from services.nollning_service import get_user_nollning_priorities
 from user.permission import Permission
 from api_schemas.post_schemas import PostRead
-import datetime
 
 user_router = APIRouter()
 
@@ -231,29 +230,6 @@ def get_my_priorities(me: Annotated[User_DB, Permission.member()], db: DB_depend
     for post in me.posts:
         priorities.append(post.name_sv)
 
-    now = datetime.datetime.now(datetime.timezone.utc)
-
-    nollning = db.query(Nollning_DB).filter(Nollning_DB.year == now.year).one_or_none()
-
-    found = False
-
-    if nollning:
-        for nollningGroup in nollning.nollning_groups:
-            for groupUser in nollningGroup.group.group_users:
-                if me.id == groupUser.user_id:
-                    if nollningGroup.group.group_type == "Mentor":
-                        if groupUser.group_user_type == "Mentor":
-                            priorities.append("Gruppfadder")
-                        elif groupUser.group_user_type == "Mentee":
-                            priorities.append("Nolla")
-                    elif nollningGroup.group.group_type == "Mission":
-                        if groupUser.group_user_type == "Mentor":
-                            priorities.append("Uppdragsfadder")
-                        elif groupUser.group_user_type == "Mentee":
-                            priorities.append("Nolla")
-                    found = True
-                    break
-            if found:
-                break
+    priorities.extend(get_user_nollning_priorities(db, me))
 
     return priorities

services/event_signup_service.py

@@ -10,6 +10,26 @@
 from api_schemas.event_signup_schemas import EventSignupCreate, EventSignupUpdate
 from helpers.constants import DEFAULT_USER_PRIORITY, NOLLNING_PRIORITIES
 from helpers.types import GROUP_TYPE
+from services.nollning_service import get_user_nollning_priorities
+
+
+def get_allowed_signup_priorities(event: Event_DB, user: User_DB, db: Session) -> set[str]:
+    """The priorities the user may sign up to the event with: the event's own priorities which the
+    user actually holds, plus the default priority which everyone falls back on."""
+    allowed = {DEFAULT_USER_PRIORITY}
+    if not event.priorities:
+        return allowed
+
+    user_priorities = {post.name_sv for post in user.posts} | get_user_nollning_priorities(db, user)
+
+    return allowed | ({p.priority for p in event.priorities} & user_priorities)
+
+
+def check_priority_allowed(event: Event_DB, user: User_DB, priority: str, db: Session):
+    """Signups are only ever stored with a priority the user really has, so that the stored
+    signups can be trusted."""
+    if priority not in get_allowed_signup_priorities(event, user, db):
+        raise HTTPException(status.HTTP_403_FORBIDDEN, detail="User cannot sign up with this priority")
 
 
 def user_matches_existing_event_post_priorities(user: User_DB, event: Event_DB):
@@ -48,6 +68,9 @@ def signup_to_event(event: Event_DB, user: User_DB, data: EventSignupCreate, man
     if manage_permission == False and not is_group_allowed(event, user, data.group_name):
         raise HTTPException(status.HTTP_403_FORBIDDEN, detail="User cannot sign up with this group")
 
+    if manage_permission == False and data.priority:  # a falsy priority just means the default one
+        check_priority_allowed(event, user, data.priority, db)
+
     signup = EventUser_DB(user=user, user_id=user.id, event=event, event_id=event.id)
 
     for var, value in vars(data).items():
@@ -105,14 +128,17 @@ def update_event_signup(event: Event_DB, data: EventSignupUpdate, user_id: int,
     if (
         manage_permission == False
         and "group_name" in updates
-        and not is_group_allowed(event, db.query(User_DB).filter(User_DB.id == user_id).one(), updates["group_name"])
+        and not is_group_allowed(event, signup.user, updates["group_name"])
     ):
         raise HTTPException(status.HTTP_403_FORBIDDEN, detail="User cannot sign up with this group")
 
     # priority and drinkPackage are not nullable in the database, so a null means "back to default"
     if "priority" in updates and not updates["priority"]:  # if falsy
         updates["priority"] = DEFAULT_USER_PRIORITY
 
+    if manage_permission == False and "priority" in updates and updates["priority"] != signup.priority:
+        check_priority_allowed(event, signup.user, updates["priority"], db)
+
     if "drinkPackage" in updates and updates["drinkPackage"] is None:
         del updates["drinkPackage"]  # None is not a valid value ("None" is), so leave the old value in place
 
@@ -174,6 +200,9 @@ def is_group_allowed(event: Event_DB, user: User_DB, group_name: str | None):
 
 
 def _is_group_allowed(gu: GroupUser_DB, e: Event_DB, agt: list[GROUP_TYPE]):
+    if not any(nollning.year == datetime.now(UTC).year for nollning in gu.group.nollnings):
+        return False
+
     if gu.group.group_type in agt:
         return True
     elif (gu.group_user_type == "Mentor") and e.allow_other_mentors:

services/nollning_service.py

@@ -1,12 +1,36 @@
+from datetime import UTC, datetime
 from fastapi import HTTPException
 from sqlalchemy.orm import Session
 from api_schemas.nollning_schema import NollningAddGroup, NollningCreate
 from db_models.group_model import Group_DB
 from db_models.nollning_group_model import NollningGroup_DB
 from db_models.nollning_model import Nollning_DB
+from db_models.user_model import User_DB
 from sqlalchemy.exc import IntegrityError
 
 
+def get_user_nollning_priorities(db: Session, user: User_DB) -> set[str]:
+    """The nollning priorities the user actually holds through this year's nollning."""
+    year = datetime.now(UTC).year
+
+    priorities: set[str] = set()
+
+    for group_user in user.group_users:
+        if not any(nollning.year == year for nollning in group_user.group.nollnings):
+            continue
+
+        group_type = group_user.group.group_type
+        if group_type not in ("Mentor", "Mission"):
+            continue
+
+        if group_user.group_user_type == "Mentee":
+            priorities.add("Nolla")
+        elif group_user.group_user_type == "Mentor":
+            priorities.add("Gruppfadder" if group_type == "Mentor" else "Uppdragsfadder")
+
+    return priorities
+
+
 def create_nollning(db: Session, data: NollningCreate):
     if data.year < 1960:
         raise HTTPException(400, "Cannot have nollning year be less than 1960")

tests/basic_factories.py

@@ -163,6 +163,30 @@ def event_data_factory(**kwargs):
     return {**default_data, **kwargs}
 
 
+def add_group_to_current_nollning(db_session, group):
+    """Put a group in this year's nollning (created if missing). Returns the group."""
+
+    return add_group_to_nollning(db_session, group, datetime.datetime.now(timezone.utc).year)
+
+
+def add_group_to_nollning(db_session, group, year):
+    """Put a group in specified year's nollning (created if missing). Returns the group."""
+
+    from db_models.nollning_model import Nollning_DB
+    from db_models.nollning_group_model import NollningGroup_DB
+
+    nollning = db_session.query(Nollning_DB).filter_by(year=year).one_or_none()
+    if nollning is None:
+        nollning = Nollning_DB(name=f"Nollning {year}", description="Test nollning", year=year)
+        db_session.add(nollning)
+        db_session.commit()
+
+    db_session.add(NollningGroup_DB(nollning_id=nollning.id, group_id=group.id))
+    db_session.commit()
+
+    return group
+
+
 def add_user_to_group(db_session, user, name, group_type, group_user_type="Mentee"):
     """Create a group of a given type and put the user in it. Returns the group."""
     from db_models.group_model import Group_DB

tests/basic_fixtures.py

@@ -301,7 +301,12 @@ def nollning_event(client, admin_token, admin_council_id):
 
     response = client.post(
         "/events/",
-        json=event_data_factory(council_id=admin_council_id, is_nollning_event=True, mentor_group_types=["Mentor"]),
+        json=event_data_factory(
+            council_id=admin_council_id,
+            is_nollning_event=True,
+            mentor_group_types=["Mentor"],
+            priorities=["Nolla", "Gruppfadder", "Uppdragsfadder"],
+        ),
         headers=auth_headers(admin_token),
     )
     assert response.status_code in (200, 201), response.text
@@ -311,13 +316,23 @@ def nollning_event(client, admin_token, admin_council_id):
 
 @pytest.fixture()
 def mentor_group(db_session, membered_user):
-    """Create a group of type "Mentor" with the member user as a mentee."""
+    """Create a group of type "Mentor" in this year's nollning, with the member user as a mentee."""
 
-    return add_user_to_group(db_session, membered_user, "Fadderiet", "Mentor", "Mentee")
+    group = add_user_to_group(db_session, membered_user, "Fadderiet", "Mentor", "Mentee")
+    return add_group_to_current_nollning(db_session, group)
 
 
 @pytest.fixture()
 def mission_group(db_session, membered_user):
-    """Create a group of type "Mission" with the member user as a mentee."""
+    """Create a group of type "Mission" in this year's nollning, with the member user as a mentee."""
 
-    return add_user_to_group(db_session, membered_user, "Uppdraget", "Mission", "Mentee")
+    group = add_user_to_group(db_session, membered_user, "Uppdraget", "Mission", "Mentee")
+    return add_group_to_current_nollning(db_session, group)
+
+
+@pytest.fixture()
+def last_years_mentor_group(db_session, membered_user):
+    """Create a group of type "Mentor" in last year's nollning, with the member user as a mentee."""
+
+    group = add_user_to_group(db_session, membered_user, "Förfadderiet", "Mentor", "Mentee")
+    return add_group_to_nollning(db_session, group, datetime.now(timezone.utc).year - 1)

tests/test_event_signup.py

@@ -1,7 +1,7 @@
 # type: ignore
 import pytest
 from helpers.constants import DEFAULT_USER_PRIORITY
-from .basic_factories import add_user_to_group, auth_headers, event_data_factory
+from .basic_factories import add_user_to_group, add_group_to_current_nollning, auth_headers, event_data_factory
 
 
 def test_signup_with_allowed_group_type(client, member_token, membered_user, nollning_event, mentor_group):
@@ -27,6 +27,19 @@ def test_signup_with_disallowed_group_type(client, member_token, membered_user,
     assert response.status_code == 403
 
 
+def test_signup_with_group_type_from_wrong_year(
+    client, member_token, membered_user, nollning_event, last_years_mentor_group
+):
+    """Signup with a group with the correct type but the wrong year is rejected."""
+    response = client.post(
+        f"/event-signup/{nollning_event['id']}",
+        json={"user_id": membered_user.id, "group_name": last_years_mentor_group.name},
+        headers=auth_headers(member_token),
+    )
+
+    assert response.status_code == 403
+
+
 def test_signup_with_group_the_user_is_not_in(client, member_token, membered_user, nollning_event, mentor_group):
     """A group name the user isn't a member of is rejected, even if the type is allowed."""
     response = client.post(
@@ -109,6 +122,8 @@ def test_signup_as_mentor_of_group_with_other_type(
 ):
     """A mentor may sign up with a group of a disallowed type only if the event allows it."""
     group = add_user_to_group(db_session, membered_user, "Uppdraget", "Mission", "Mentor")
+    add_group_to_current_nollning(db_session, group)
+
     data = event_data_factory(
         council_id=admin_council_id,
         is_nollning_event=True,
@@ -200,12 +215,11 @@ def test_non_nollning_event_update_signup_remove_group_name_is_allowed(
 
     response = client.patch(
         f"/event-signup/{event['id']}",
-        json={"priority": "Nolla", "group_name": sent_group_name},
+        json={"group_name": sent_group_name},
         headers=auth_headers(member_token),
     )
 
     assert response.status_code == 200, response.text
-    assert response.json()["priority"] == "Nolla"
     assert response.json()["group_name"] is None
 
 
@@ -255,6 +269,70 @@ def test_update_signup_with_null_priority_resets_to_default(
     assert response.json()["group_name"] == mentor_group.name
 
 
+def test_signup_with_priority_the_user_does_not_have(client, member_token, membered_user, nollning_event, mentor_group):
+    """A mentee is not a gruppfadder, so we reject the signup even if the event allows the priority"""
+    response = client.post(
+        f"/event-signup/{nollning_event['id']}",
+        json={"user_id": membered_user.id, "group_name": mentor_group.name, "priority": "Gruppfadder"},
+        headers=auth_headers(member_token),
+    )
+
+    assert response.status_code == 403, response.text
+
+
+def test_signup_with_priority_not_on_the_event(
+    client, admin_token, admin_council_id, member_token, membered_user, member_post
+):
+    """Signups with a post priority the user holds but which the event does not ask for is rejected."""
+    data = event_data_factory(council_id=admin_council_id, priorities=["Nolla"])
+    event = client.post("/events/", json=data, headers=auth_headers(admin_token)).json()
+
+    response = client.post(
+        f"/event-signup/{event['id']}",
+        json={"user_id": membered_user.id, "priority": member_post.name_sv},
+        headers=auth_headers(member_token),
+    )
+
+    assert response.status_code == 403, response.text
+
+
+def test_admin_can_sign_up_user_with_any_priority(client, admin_token, membered_user, nollning_event, mentor_group):
+    """Admins make the final call and are not restricted by the priorities."""
+    response = client.post(
+        f"/event-signup/{nollning_event['id']}",
+        json={"user_id": membered_user.id, "group_name": mentor_group.name, "priority": "Gruppfadder"},
+        headers=auth_headers(admin_token),
+    )
+
+    assert response.status_code in (200, 201), response.text
+    assert response.json()["priority"] == "Gruppfadder"
+
+
+def test_update_signup_to_priority_the_user_does_not_have(
+    client, member_token, membered_user, nollning_event, mentor_group
+):
+    """Switching to a priority the user does not have is rejected, and nothing else is changed."""
+    signup = client.post(
+        f"/event-signup/{nollning_event['id']}",
+        json={"user_id": membered_user.id, "group_name": mentor_group.name, "priority": "Nolla"},
+        headers=auth_headers(member_token),
+    )
+    assert signup.status_code in (200, 201), signup.text
+
+    response = client.patch(
+        f"/event-signup/{nollning_event['id']}",
+        json={"priority": "Uppdragsfadder", "drinkPackage": "Alcohol"},
+        headers=auth_headers(member_token),
+    )
+
+    assert response.status_code == 403, response.text
+    signup_after = client.get(
+        f"/event-signup/me-signup/{nollning_event['id']}", headers=auth_headers(member_token)
+    ).json()
+    assert signup_after["priority"] == "Nolla"
+    assert signup_after["drinkPackage"] == "None"
+
+
 def test_update_signup_to_disallowed_group(
     client, member_token, membered_user, nollning_event, mentor_group, mission_group
 ):