diff --git a/src/api/tests/schemas/test_utils.py b/src/api/tests/schemas/test_utils.py index e836c2c9..f3a8d25e 100644 --- a/src/api/tests/schemas/test_utils.py +++ b/src/api/tests/schemas/test_utils.py @@ -8,6 +8,8 @@ from sqlalchemy import func from sqlmodel import select +from qualicharge.auth.factories import GroupFactory, UserFactory +from qualicharge.auth.schemas import GroupOperationalUnit, UserGroup from qualicharge.db import SAQueryCounter from qualicharge.exceptions import ( DatabaseQueryException, @@ -28,6 +30,7 @@ from qualicharge.schemas.utils import ( build_statique, get_or_create, + is_pdc_allowed_for_user, list_statique, pdc_to_statique, save_schema_from_statique, @@ -423,3 +426,58 @@ def test_list_statique_operational_units_filtering(db_session): assert {s.id_pdc_itinerance for s in statiques} == { s.id_pdc_itinerance for s in expected_statiques } + + +def test_is_pdc_allowed_for_user(db_session): + """Test the is_pdc_allowed_for_user utility.""" + UserFactory.__session__ = db_session + GroupFactory.__session__ = db_session + + id_pdc_itinerance = "FRS63E0001" + id_station_itinerance = "FRS63P0001" + + # Superuser + user = UserFactory.create_sync(is_superuser=True) + assert is_pdc_allowed_for_user(id_pdc_itinerance, user) is True + + # Normal user with no assigned operational units + user = UserFactory.create_sync(is_superuser=False) + assert user.operational_units == [] + assert is_pdc_allowed_for_user(id_pdc_itinerance, user) is False + + # Normal user with matching assigned operational units + save_statique( + db_session, + StatiqueFactory.build( + id_pdc_itinerance=id_pdc_itinerance, + id_station_itinerance=id_station_itinerance, + ), + ) + group = GroupFactory.create_sync() + db_session.add( + GroupOperationalUnit( + group_id=group.id, + operational_unit_id=db_session.exec( + select(OperationalUnit.id).where( + OperationalUnit.code == id_pdc_itinerance[:5] + ) + ).one(), + ) + ) + user = UserFactory.create_sync(is_superuser=False) + db_session.add(UserGroup(user_id=user.id, group_id=group.id)) + assert is_pdc_allowed_for_user(id_pdc_itinerance, user) is True + + # Normal user with assigned operational units not linked to the point of charge + group = GroupFactory.create_sync() + db_session.add( + GroupOperationalUnit( + group_id=group.id, + operational_unit_id=db_session.exec( + select(OperationalUnit.id).where(OperationalUnit.code == "FRS72") + ).one(), + ) + ) + user = UserFactory.create_sync(is_superuser=False) + db_session.add(UserGroup(user_id=user.id, group_id=group.id)) + assert is_pdc_allowed_for_user(id_pdc_itinerance, user) is False