Correct read events rights

This commit is contained in:
Sebastiaan
2025-06-09 17:01:37 +02:00
parent bf3fa03db0
commit 9bc19e6a68
3 changed files with 93 additions and 10 deletions

View File

@@ -51,17 +51,19 @@ def read_events(
count_statement = ( count_statement = (
select(func.count()) select(func.count())
.select_from(Event) .select_from(Event)
.join(EventUserLink) # Join with EventUserLink to check user permissions
.where( .where(
EventUserLink.user_id == current_user.id, EventUserLink.user_id == current_user.id,
(EventUserLink.rights & PermissionRight.READ) == PermissionRight.READ, # FIXME: (EventUserLink.rights & PermissionRight.READ) == PermissionRight.READ,
) )
) )
count = session.exec(count_statement).one() count = session.exec(count_statement).one()
statement = ( statement = (
select(Event) select(Event)
.join(EventUserLink) # Join with EventUserLink to check user permissions
.where( .where(
EventUserLink.user_id == current_user.id, EventUserLink.user_id == current_user.id,
(EventUserLink.rights & PermissionRight.READ) == PermissionRight.READ, # FIXME: (EventUserLink.rights & PermissionRight.READ) == PermissionRight.READ,
) )
.offset(skip) .offset(skip)
.limit(limit) .limit(limit)

View File

@@ -5,12 +5,16 @@ from sqlmodel import Session
from app.core.config import settings from app.core.config import settings
from app.models.user import PermissionRight from app.models.user import PermissionRight
from app.tests.conftest import EventUserHeader
from app.tests.utils.event import create_random_event from app.tests.utils.event import create_random_event
from app.tests.utils.user import create_random_user, authentication_token_from_user from app.tests.utils.user import create_random_user, authentication_token_from_user
def test_create_event(client: TestClient, superuser_token_headers: dict[str, str]) -> None: def test_create_event(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
data = {"name": "Foo", "contact": "Someone"} data = {
"name": "Foo",
"contact": "Someone",
}
response = client.post( response = client.post(
f"{settings.API_V1_STR}/events/", f"{settings.API_V1_STR}/events/",
@@ -70,6 +74,24 @@ def test_read_event_not_enough_permissions(
assert content["detail"] == "Not enough permissions" assert content["detail"] == "Not enough permissions"
def test_read_event_with_event_user(
client: TestClient, event_user_token_headers: EventUserHeader, db: Session
) -> None:
event = event_user_token_headers.event
response = client.get(
f"{settings.API_V1_STR}/events/{event.id}",
headers=event_user_token_headers.headers,
)
assert response.status_code == 200
content = response.json()
assert content["name"] == event.name
assert content["contact"] == event.contact
assert content["id"] == str(event.id)
assert content["is_active"] == event.is_active
assert str(content["start_at"]) == str(event.start_at)
assert str(content["end_at"]) == str(event.end_at)
def test_read_events( def test_read_events(
client: TestClient, superuser_token_headers: dict[str, str], db: Session client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None: ) -> None:
@@ -81,14 +103,41 @@ def test_read_events(
) )
assert response.status_code == 200 assert response.status_code == 200
content = response.json() content = response.json()
assert len(content["data"]) >= 2 assert "count" in content
assert content["count"] >= 2
assert "data" in content
assert isinstance(content["data"], list)
assert len(content["data"]) <= content["count"]
def test_read_events_with_event_user(
client: TestClient, db: Session
) -> None:
event = create_random_event(db)
user = create_random_user(db)
event.add_user(user=user, rights=PermissionRight.READ, session=db)
response = client.get(
f"{settings.API_V1_STR}/events/",
headers=authentication_token_from_user(db=db, user=user, client=client),
)
assert response.status_code == 200
content = response.json()
assert "count" in content
assert content["count"] == 1
assert "data" in content
assert isinstance(content["data"], list)
assert len(content["data"]) <= content["count"]
def test_update_event( def test_update_event(
client: TestClient, superuser_token_headers: dict[str, str], db: Session client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None: ) -> None:
event = create_random_event(db) event = create_random_event(db)
data = {"name": "Updated name", "contact": "Updated contact"} data = {
"name": "Updated name",
"contact": "Updated contact",
}
response = client.put( response = client.put(
f"{settings.API_V1_STR}/events/{event.id}", f"{settings.API_V1_STR}/events/{event.id}",
headers=superuser_token_headers, headers=superuser_token_headers,

View File

@@ -9,6 +9,7 @@ from app.core.config import settings
from app.tests.conftest import EventUserHeader from app.tests.conftest import EventUserHeader
from app.tests.utils.event import create_random_event from app.tests.utils.event import create_random_event
from app.tests.utils.team import create_random_team from app.tests.utils.team import create_random_team
from app.tests.utils.user import create_random_user, authentication_token_from_user
def test_create_team(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None: def test_create_team(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
@@ -109,9 +110,11 @@ def test_read_teams(client: TestClient, superuser_token_headers: dict[str, str],
) )
assert response.status_code == 200 assert response.status_code == 200
content = response.json() content = response.json()
assert "count" in content
assert content["count"] >= 2
assert "data" in content assert "data" in content
assert isinstance(content["data"], list) assert isinstance(content["data"], list)
assert content["count"] >= 2 assert len(content["data"]) <= content["count"]
def test_read_teams_with_normal_user(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None: def test_read_teams_with_normal_user(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
@@ -123,22 +126,51 @@ def test_read_teams_with_normal_user(client: TestClient, normal_user_token_heade
) )
assert response.status_code == 200 assert response.status_code == 200
content = response.json() content = response.json()
assert "count" in content
assert content["count"] == 0 assert content["count"] == 0
assert "data" in content
assert isinstance(content["data"], list)
assert len(content["data"]) == 0
def test_read_teams_with_event_user(client: TestClient, event_user_token_headers: EventUserHeader, db: Session) -> None: def test_read_teams_with_event_user_readonly(client: TestClient, db: Session) -> None:
create_random_team(db, event=event_user_token_headers.event) event = create_random_event(db)
user = create_random_user(db)
event.add_user(user=user, rights=PermissionRight.READ, session=db)
create_random_team(db, event=event)
response = client.get( response = client.get(
f"{settings.API_V1_STR}/teams/", f"{settings.API_V1_STR}/teams/",
headers=event_user_token_headers.headers, headers=authentication_token_from_user(db=db, user=user, client=client),
) )
assert response.status_code == 200 assert response.status_code == 200
content = response.json() content = response.json()
assert "count" in content
assert content["count"] == 1
assert "data" in content assert "data" in content
assert isinstance(content["data"], list) assert isinstance(content["data"], list)
assert content["count"] >= 1 assert len(content["data"]) <= content["count"]
def test_read_teams_with_event_user_team_manager(client: TestClient, db: Session) -> None:
event = create_random_event(db)
user = create_random_user(db)
event.add_user(user=user, rights=PermissionRight.MANAGE_TEAMS, session=db)
create_random_team(db, event=event)
response = client.get(
f"{settings.API_V1_STR}/teams/",
headers=authentication_token_from_user(db=db, user=user, client=client),
)
assert response.status_code == 200
content = response.json()
assert "count" in content
assert content["count"] == 1
assert "data" in content
assert isinstance(content["data"], list)
assert len(content["data"]) <= content["count"]
def test_update_team_name(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None: def test_update_team_name(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None: