Compare commits
10 Commits
2cec60cce3
...
7d524cf04d
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7d524cf04d | ||
|
|
479ca1986f | ||
|
|
1e6b138873 | ||
|
|
f2f0475859 | ||
|
|
7848578ebb | ||
|
|
9b4d32ffff | ||
|
|
9538b9067c | ||
|
|
56b503751a | ||
|
|
1d9e333ee0 | ||
|
|
13a1b4dd1e |
2
.env
2
.env
@@ -38,6 +38,8 @@ POSTGRES_DB=app
|
||||
POSTGRES_USER=postgres
|
||||
POSTGRES_PASSWORD=changethis
|
||||
|
||||
TZ=UTC
|
||||
|
||||
SENTRY_DSN=
|
||||
|
||||
# Configure these with your own Docker registry images
|
||||
|
||||
@@ -3,6 +3,9 @@ from fastapi import APIRouter
|
||||
from app.api.routes import (
|
||||
events,
|
||||
teams,
|
||||
associations,
|
||||
divisions,
|
||||
members,
|
||||
login,
|
||||
private,
|
||||
users,
|
||||
@@ -18,6 +21,9 @@ api_router.include_router(utils.router)
|
||||
|
||||
api_router.include_router(events.router)
|
||||
api_router.include_router(teams.router)
|
||||
api_router.include_router(associations.router)
|
||||
api_router.include_router(divisions.router)
|
||||
api_router.include_router(members.router)
|
||||
|
||||
|
||||
if settings.ENVIRONMENT == "local":
|
||||
|
||||
176
backend/app/api/routes/associations.py
Normal file
176
backend/app/api/routes/associations.py
Normal file
@@ -0,0 +1,176 @@
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, HTTPException, status
|
||||
from sqlmodel import func, select
|
||||
|
||||
from app.api.deps import CurrentUser, SessionDep
|
||||
from app.models.base import (
|
||||
ApiTags,
|
||||
Message,
|
||||
RowId,
|
||||
)
|
||||
from app.models.association import (
|
||||
Association,
|
||||
AssociationCreate,
|
||||
AssociationUpdate,
|
||||
AssociationPublic,
|
||||
AssociationsPublic,
|
||||
)
|
||||
from app.models.division import (
|
||||
Division,
|
||||
DivisionsPublic,
|
||||
)
|
||||
from app.models.user import (
|
||||
PermissionModule,
|
||||
PermissionPart,
|
||||
PermissionRight,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/associations", tags=[ApiTags.ASSOCIATIONS])
|
||||
|
||||
|
||||
# region # Associations ########################################################
|
||||
|
||||
@router.get("/", response_model=AssociationsPublic)
|
||||
def read_associations(
|
||||
session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
|
||||
) -> Any:
|
||||
"""
|
||||
Retrieve all associations.
|
||||
"""
|
||||
|
||||
if current_user.has_permissions(
|
||||
module=PermissionModule.ASSOCIATION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
):
|
||||
count_statement = select(func.count()).select_from(Association)
|
||||
count = session.exec(count_statement).one()
|
||||
statement = select(Association).offset(skip).limit(limit)
|
||||
associations = session.exec(statement).all()
|
||||
return AssociationsPublic(data=associations, count=count)
|
||||
|
||||
return AssociationsPublic(data=[], count=0)
|
||||
|
||||
|
||||
@router.get("/{id}", response_model=AssociationPublic)
|
||||
def read_association(session: SessionDep, current_user: CurrentUser, id: RowId) -> Any:
|
||||
"""
|
||||
Get association by ID.
|
||||
"""
|
||||
association = session.get(Association, id)
|
||||
if not association:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Association not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.ASSOCIATION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
return association
|
||||
|
||||
|
||||
@router.post("/", response_model=AssociationPublic)
|
||||
def create_association(
|
||||
*, session: SessionDep, current_user: CurrentUser, association_in: AssociationCreate
|
||||
) -> Any:
|
||||
"""
|
||||
Create new association.
|
||||
"""
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.ASSOCIATION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.CREATE,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
association = Association.create(create_obj=association_in, session=session)
|
||||
return association
|
||||
|
||||
|
||||
@router.put("/{id}", response_model=AssociationPublic)
|
||||
def update_association(
|
||||
*, session: SessionDep, current_user: CurrentUser, id: RowId, association_in: AssociationUpdate
|
||||
) -> Any:
|
||||
"""
|
||||
Update a association.
|
||||
"""
|
||||
association = session.get(Association, id)
|
||||
if not association:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Association not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.ASSOCIATION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.UPDATE,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
association = Association.update(db_obj=association, in_obj=association_in, session=session)
|
||||
return association
|
||||
|
||||
|
||||
@router.delete("/{id}")
|
||||
def delete_association(session: SessionDep,current_user: CurrentUser, id: RowId) -> Message:
|
||||
"""
|
||||
Delete a association.
|
||||
"""
|
||||
association = session.get(Association, id)
|
||||
if not association:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Association not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.ASSOCIATION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.DELETE,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
session.delete(association)
|
||||
session.commit()
|
||||
return Message(message="Association deleted successfully")
|
||||
|
||||
# endregion
|
||||
|
||||
|
||||
# region # Associations / Divisions ############################################
|
||||
|
||||
|
||||
@router.get("/{associations_id}/divisions/", response_model=DivisionsPublic)
|
||||
def read_association_division(
|
||||
session: SessionDep, current_user: CurrentUser, associations_id: RowId, skip: int = 0, limit: int = 100
|
||||
) -> Any:
|
||||
"""
|
||||
Retrieve all association divisions.
|
||||
"""
|
||||
|
||||
association = session.get(Association, associations_id)
|
||||
if not association:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Association not found")
|
||||
|
||||
if not current_user.has_permission(
|
||||
module=PermissionModule.ASSOCIATION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=(PermissionRight.MANAGE_DIVISIONS | PermissionRight.READ),
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
count_statement = (select(func.count())
|
||||
.select_from(Division)
|
||||
.where(Division.association_id == association.id)
|
||||
)
|
||||
count = session.exec(count_statement).one()
|
||||
statement = (select(Division)
|
||||
.where(Division.association_id == association.id)
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
)
|
||||
divisions = session.exec(statement).all()
|
||||
|
||||
return DivisionsPublic(data=divisions, count=count)
|
||||
|
||||
|
||||
# endregion
|
||||
132
backend/app/api/routes/divisions.py
Normal file
132
backend/app/api/routes/divisions.py
Normal file
@@ -0,0 +1,132 @@
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, HTTPException, status
|
||||
from sqlmodel import func, select
|
||||
|
||||
from app.api.deps import CurrentUser, SessionDep
|
||||
from app.models.base import (
|
||||
ApiTags,
|
||||
Message,
|
||||
RowId,
|
||||
)
|
||||
from app.models.division import (
|
||||
Division,
|
||||
DivisionCreate,
|
||||
DivisionUpdate,
|
||||
DivisionPublic,
|
||||
DivisionsPublic,
|
||||
)
|
||||
from app.models.user import (
|
||||
PermissionModule,
|
||||
PermissionPart,
|
||||
PermissionRight,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/divisions", tags=[ApiTags.DIVISIONS])
|
||||
|
||||
|
||||
# region # Divisions ###########################################################
|
||||
|
||||
@router.get("/", response_model=DivisionsPublic)
|
||||
def read_divisions(
|
||||
session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
|
||||
) -> Any:
|
||||
"""
|
||||
Retrieve all divisions.
|
||||
"""
|
||||
|
||||
if current_user.has_permissions(
|
||||
module=PermissionModule.DIVISION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
):
|
||||
count_statement = select(func.count()).select_from(Division)
|
||||
count = session.exec(count_statement).one()
|
||||
statement = select(Division).offset(skip).limit(limit)
|
||||
divisions = session.exec(statement).all()
|
||||
return DivisionsPublic(data=divisions, count=count)
|
||||
|
||||
return DivisionsPublic(data=[], count=0)
|
||||
|
||||
|
||||
@router.get("/{id}", response_model=DivisionPublic)
|
||||
def read_division(session: SessionDep, current_user: CurrentUser, id: RowId) -> Any:
|
||||
"""
|
||||
Get division by ID.
|
||||
"""
|
||||
division = session.get(Division, id)
|
||||
if not division:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Division not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.DIVISION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
return division
|
||||
|
||||
|
||||
@router.post("/", response_model=DivisionPublic)
|
||||
def create_division(
|
||||
*, session: SessionDep, current_user: CurrentUser, division_in: DivisionCreate
|
||||
) -> Any:
|
||||
"""
|
||||
Create new division.
|
||||
"""
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.DIVISION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.CREATE,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
division = Division.create(create_obj=division_in, session=session)
|
||||
return division
|
||||
|
||||
|
||||
@router.put("/{id}", response_model=DivisionPublic)
|
||||
def update_division(
|
||||
*, session: SessionDep, current_user: CurrentUser, id: RowId, division_in: DivisionUpdate
|
||||
) -> Any:
|
||||
"""
|
||||
Update a division.
|
||||
"""
|
||||
division = session.get(Division, id)
|
||||
if not division:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Division not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.DIVISION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.UPDATE,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
division = Division.update(db_obj=division, in_obj=division_in, session=session)
|
||||
return division
|
||||
|
||||
|
||||
@router.delete("/{id}")
|
||||
def delete_division(session: SessionDep,current_user: CurrentUser, id: RowId) -> Message:
|
||||
"""
|
||||
Delete a division.
|
||||
"""
|
||||
division = session.get(Division, id)
|
||||
if not division:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Division not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.ASSOCIATION,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.DELETE,
|
||||
):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
session.delete(division)
|
||||
session.commit()
|
||||
return Message(message="Division deleted successfully")
|
||||
|
||||
# endregion
|
||||
208
backend/app/api/routes/members.py
Normal file
208
backend/app/api/routes/members.py
Normal file
@@ -0,0 +1,208 @@
|
||||
from typing import Any
|
||||
|
||||
import sqlalchemy
|
||||
from fastapi import APIRouter, HTTPException, status
|
||||
from sqlmodel import func, select, and_, or_, SQLModel
|
||||
from sqlalchemy.orm import joinedload
|
||||
|
||||
from app.api.deps import CurrentUser, SessionDep
|
||||
from app.models.base import (
|
||||
ApiTags,
|
||||
Message,
|
||||
RowId,
|
||||
)
|
||||
from app.models.member import (
|
||||
Member,
|
||||
MemberCreate,
|
||||
MemberUpdate,
|
||||
MemberPublic,
|
||||
MembersPublic,
|
||||
MemberTeamLink,
|
||||
)
|
||||
|
||||
from app.models.event import Event, EventUserLink
|
||||
from app.models.team import Team
|
||||
from app.models.user import (
|
||||
PermissionModule,
|
||||
PermissionPart,
|
||||
PermissionRight,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/members", tags=[ApiTags.MEMBERS])
|
||||
|
||||
|
||||
# region # Members #############################################################
|
||||
|
||||
def load_member(
|
||||
session: SessionDep,
|
||||
current_user: CurrentUser,
|
||||
id: RowId | None = None,
|
||||
module: PermissionModule = PermissionModule.MEMBER,
|
||||
part: PermissionPart = PermissionPart.ADMIN,
|
||||
user_rights: PermissionRight | None = None,
|
||||
event_rights: PermissionRight | None = PermissionRight.MANAGE_MEMBERS,
|
||||
) -> Member | None:
|
||||
member = session.get(Member, id)
|
||||
|
||||
if id and not member:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Member not found")
|
||||
|
||||
no_links = True
|
||||
valid = False
|
||||
# Global member permissions
|
||||
if current_user.has_permissions(module=module, part=part, rights=user_rights):
|
||||
# Also valid for create new
|
||||
valid = True
|
||||
|
||||
# Own member items
|
||||
elif hasattr(member, "user") and member.user and member.user == current_user:
|
||||
valid = True
|
||||
|
||||
# Event member permissions
|
||||
elif hasattr(member, "team_links"):
|
||||
for link in member.team_links:
|
||||
team = link.team
|
||||
if team and team.event:
|
||||
no_links = False
|
||||
if team.event.user_has_rights(user=current_user, rights=event_rights):
|
||||
valid = True
|
||||
break
|
||||
|
||||
# Not yet linked, or unlinked member
|
||||
if no_links and hasattr(member, "created_by") and member.created_by == current_user.id:
|
||||
valid = True
|
||||
|
||||
if not valid:
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
return member
|
||||
|
||||
|
||||
@router.get("/", response_model=MembersPublic)
|
||||
def read_members(
|
||||
session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
|
||||
) -> Any:
|
||||
"""
|
||||
Retrieve all members.
|
||||
"""
|
||||
|
||||
if current_user.has_permissions(
|
||||
module=PermissionModule.MEMBER,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
):
|
||||
data_query = (
|
||||
select(Member)
|
||||
)
|
||||
else:
|
||||
data_query = (
|
||||
select(Member)
|
||||
.outerjoin(MemberTeamLink, MemberTeamLink.member_id == Member.id)
|
||||
.outerjoin(Team, MemberTeamLink.team_id == Team.id)
|
||||
.outerjoin(Event, Team.event_id == Event.id)
|
||||
.outerjoin(EventUserLink, EventUserLink.event_id == Event.id)
|
||||
.where(
|
||||
or_(
|
||||
# Own member
|
||||
Member.id == current_user.member_id,
|
||||
|
||||
# Created by user and unlinked
|
||||
and_(
|
||||
Member.created_by == current_user.id,
|
||||
MemberTeamLink.team_id == None
|
||||
),
|
||||
|
||||
# Event permissions via team -> event -> EventUserLink
|
||||
and_(
|
||||
EventUserLink.user_id == current_user.id,
|
||||
# FIXME: EventUserLink.rights.op("&")(PermissionRight.MANAGE_MEMBERS) != 0
|
||||
),
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
# Cache as subquery
|
||||
data_sub_query = data_query.subquery()
|
||||
aliased_member = sqlalchemy.orm.aliased(Member, data_sub_query)
|
||||
|
||||
# Count using subquery
|
||||
count = session.exec(
|
||||
select(func.count()).select_from(data_sub_query)
|
||||
).one()
|
||||
|
||||
# Paginated data query using same subquery
|
||||
data = session.exec(
|
||||
select(aliased_member).offset(skip).limit(limit)
|
||||
).all()
|
||||
|
||||
return MembersPublic(count=count, data=data)
|
||||
|
||||
|
||||
@router.get("/{id}", response_model=MemberPublic)
|
||||
def read_member(session: SessionDep, current_user: CurrentUser, id: RowId) -> Any:
|
||||
"""
|
||||
Get member by ID.
|
||||
"""
|
||||
member = load_member(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.READ,
|
||||
)
|
||||
|
||||
return member
|
||||
|
||||
|
||||
@router.post("/", response_model=MemberPublic)
|
||||
def create_member(
|
||||
*, session: SessionDep, current_user: CurrentUser, member_in: MemberCreate
|
||||
) -> Any:
|
||||
"""
|
||||
Create new member.
|
||||
"""
|
||||
|
||||
load_member(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
user_rights=PermissionRight.CREATE,
|
||||
)
|
||||
|
||||
member = Member.create(create_obj=member_in, session=session, user=current_user)
|
||||
return member
|
||||
|
||||
|
||||
@router.put("/{id}", response_model=MemberPublic)
|
||||
def update_member(
|
||||
*, session: SessionDep, current_user: CurrentUser, id: RowId, member_in: MemberUpdate
|
||||
) -> Any:
|
||||
"""
|
||||
Update a member.
|
||||
"""
|
||||
member = load_member(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.UPDATE,
|
||||
)
|
||||
|
||||
member = Member.update(db_obj=member, in_obj=member_in, session=session)
|
||||
return member
|
||||
|
||||
|
||||
@router.delete("/{id}")
|
||||
def delete_member(session: SessionDep,current_user: CurrentUser, id: RowId) -> Message:
|
||||
"""
|
||||
Delete a member.
|
||||
"""
|
||||
member = load_member(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.DELETE,
|
||||
)
|
||||
|
||||
session.delete(member)
|
||||
session.commit()
|
||||
return Message(message="Member deleted successfully")
|
||||
|
||||
# endregion
|
||||
@@ -17,7 +17,6 @@ router = APIRouter(tags=[ApiTags.PRIVATE], prefix="/private")
|
||||
class PrivateUserCreate(BaseModel):
|
||||
email: str
|
||||
password: str
|
||||
full_name: str
|
||||
is_verified: bool = False
|
||||
|
||||
|
||||
@@ -29,7 +28,6 @@ def create_user(user_in: PrivateUserCreate, session: SessionDep) -> Any:
|
||||
|
||||
user = User(
|
||||
email=user_in.email,
|
||||
full_name=user_in.full_name,
|
||||
hashed_password=get_password_hash(user_in.password),
|
||||
)
|
||||
|
||||
|
||||
@@ -4,11 +4,14 @@ from fastapi import APIRouter, HTTPException, status
|
||||
from sqlmodel import func, select
|
||||
|
||||
from app.api.deps import CurrentUser, SessionDep
|
||||
from app.api.routes.members import load_member
|
||||
from app.models.base import (
|
||||
ApiTags,
|
||||
Message,
|
||||
RowId,
|
||||
)
|
||||
from app.models.member import MemberTeamLink, MemberTeamLinkCreate, MemberTeamLinkUpdate, MemberTeamLinksPublic, \
|
||||
MemberTeamLinkPublic
|
||||
from app.models.team import (
|
||||
Team,
|
||||
TeamCreate,
|
||||
@@ -25,12 +28,46 @@ from app.models.user import (
|
||||
PermissionPart,
|
||||
PermissionRight,
|
||||
)
|
||||
from app.models.division import (
|
||||
DivisionTeamLink,
|
||||
DivisionTeamLinkCreate,
|
||||
DivisionTeamLinkUpdate,
|
||||
DivisionTeamLinkPublic,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/teams", tags=[ApiTags.TEAMS])
|
||||
|
||||
|
||||
# region # Teams ###############################################################
|
||||
|
||||
def load_team(
|
||||
session: SessionDep,
|
||||
current_user: CurrentUser,
|
||||
id: RowId | None = None,
|
||||
module: PermissionModule = PermissionModule.TEAM,
|
||||
part: PermissionPart = PermissionPart.ADMIN,
|
||||
user_rights: PermissionRight | None = None,
|
||||
event_rights: PermissionRight | None = PermissionRight.MANAGE_TEAMS,
|
||||
) -> Team | None:
|
||||
team = session.get(Team, id)
|
||||
|
||||
if id and not team:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")
|
||||
|
||||
valid = False
|
||||
if current_user.has_permissions(module=module, part=part, rights=user_rights):
|
||||
# Also valid for create new
|
||||
valid = True
|
||||
if hasattr(team, "event"):
|
||||
if team.event.user_has_rights(user=current_user, rights=event_rights):
|
||||
valid = True
|
||||
|
||||
if not valid:
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
return team
|
||||
|
||||
|
||||
@router.get("/", response_model=TeamsPublic)
|
||||
def read_teams(
|
||||
session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
|
||||
@@ -44,39 +81,22 @@ def read_teams(
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
):
|
||||
count_statement = select(func.count()).select_from(Team)
|
||||
count = session.exec(count_statement).one()
|
||||
statement = select(Team).offset(skip).limit(limit)
|
||||
teams = session.exec(statement).all()
|
||||
|
||||
else:
|
||||
# Only read teams that are connected to an event that the user can read
|
||||
count_statement = (
|
||||
select(func.count())
|
||||
.select_from(Team)
|
||||
.join(Event) # Join with Event to filter teams based on events
|
||||
.join(EventUserLink) # Join with EventUserLink to check user permissions
|
||||
.where(
|
||||
EventUserLink.user_id == current_user.id,
|
||||
# FIXME: (EventUserLink.rights & (PermissionRight.READ | PermissionRight.MANAGE_TEAMS)) > 0
|
||||
)
|
||||
)
|
||||
count = session.exec(count_statement).one()
|
||||
|
||||
statement = (
|
||||
data_query = (
|
||||
select(Team)
|
||||
.join(Event)
|
||||
.join(EventUserLink)
|
||||
)
|
||||
else:
|
||||
data_query = (
|
||||
select(Team)
|
||||
.join(EventUserLink, EventUserLink.event_id == Team.event_id)
|
||||
.where(
|
||||
EventUserLink.user_id == current_user.id,
|
||||
# FIXME: (EventUserLink.rights & (PermissionRight.READ | PermissionRight.MANAGE_TEAMS)) > 0
|
||||
)
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
)
|
||||
teams = session.exec(statement).all()
|
||||
|
||||
return TeamsPublic(data=teams, count=count)
|
||||
count = session.exec(select(func.count()).select_from(data_query.subquery())).one()
|
||||
data = session.exec(data_query.offset(skip).limit(limit)).all()
|
||||
return TeamsPublic(data=data, count=count)
|
||||
|
||||
|
||||
@router.get("/{id}", response_model=TeamPublic)
|
||||
@@ -84,20 +104,12 @@ def read_team(session: SessionDep, current_user: CurrentUser, id: RowId) -> Any:
|
||||
"""
|
||||
Get team by ID.
|
||||
"""
|
||||
team = session.get(Team, id)
|
||||
if not team:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")
|
||||
|
||||
event = session.get(Event, team.event_id)
|
||||
if not event: # pragma: no cover
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.TEAM,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
) and not (event.user_has_rights(user=current_user, rights=PermissionRight.MANAGE_TEAMS)):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.READ,
|
||||
)
|
||||
|
||||
return team
|
||||
|
||||
@@ -132,21 +144,12 @@ def update_team(
|
||||
"""
|
||||
Update a team.
|
||||
"""
|
||||
team = session.get(Team, id)
|
||||
if not team:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")
|
||||
|
||||
# Check user's permissions for the existing event
|
||||
event = session.get(Event, team.event_id)
|
||||
if not event: # pragma: no cover
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.TEAM,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.UPDATE,
|
||||
) and not (event.user_has_rights(user=current_user, rights=PermissionRight.MANAGE_TEAMS)):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.UPDATE,
|
||||
)
|
||||
|
||||
# Check rights for the new event data
|
||||
if team_in.event_id:
|
||||
@@ -171,23 +174,228 @@ def delete_team(session: SessionDep,current_user: CurrentUser, id: RowId) -> Mes
|
||||
"""
|
||||
Delete a team.
|
||||
"""
|
||||
team = session.get(Team, id)
|
||||
if not team:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")
|
||||
|
||||
event = session.get(Event, team.event_id)
|
||||
if not event: # pragma: no cover
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event not found")
|
||||
|
||||
if not current_user.has_permissions(
|
||||
module=PermissionModule.TEAM,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.DELETE,
|
||||
) and not (event.user_has_rights(user=current_user, rights=PermissionRight.MANAGE_TEAMS)):
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.DELETE,
|
||||
)
|
||||
|
||||
session.delete(team)
|
||||
session.commit()
|
||||
return Message(message="Team deleted successfully")
|
||||
|
||||
# endregion
|
||||
|
||||
|
||||
# region # Teams / Division ####################################################
|
||||
|
||||
@router.get("/{id}/division", response_model=DivisionTeamLinkPublic, tags=[ApiTags.DIVISIONS])
|
||||
def read_team_divisions(session: SessionDep, current_user: CurrentUser, id: RowId) -> Any:
|
||||
"""
|
||||
Get division from team by ID.
|
||||
"""
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.MANAGE_DIVISIONS,
|
||||
event_rights=PermissionRight.MANAGE_DIVISIONS,
|
||||
)
|
||||
|
||||
return team.division_link
|
||||
|
||||
|
||||
@router.post("/{id}/division", response_model=DivisionTeamLinkPublic, tags=[ApiTags.DIVISIONS])
|
||||
def create_team_division_link(
|
||||
*, session: SessionDep, current_user: CurrentUser, team_in: DivisionTeamLinkCreate, id: RowId
|
||||
) -> Any:
|
||||
"""
|
||||
Create new division link in team.
|
||||
"""
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.MANAGE_DIVISIONS,
|
||||
event_rights=PermissionRight.MANAGE_DIVISIONS,
|
||||
)
|
||||
|
||||
if team.division_link:
|
||||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Team already linked to division")
|
||||
|
||||
division_team_link = DivisionTeamLink.create(create_obj=team_in, session=session, team=team)
|
||||
return division_team_link
|
||||
|
||||
|
||||
@router.put("/{id}/division", response_model=DivisionTeamLinkPublic, tags=[ApiTags.DIVISIONS])
|
||||
def update_team_division_link(
|
||||
*, session: SessionDep, current_user: CurrentUser, id: RowId, team_in: DivisionTeamLinkUpdate
|
||||
) -> Any:
|
||||
"""
|
||||
Update division info inside team.
|
||||
"""
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.MANAGE_DIVISIONS,
|
||||
event_rights=PermissionRight.MANAGE_DIVISIONS,
|
||||
)
|
||||
|
||||
# Update the team
|
||||
division_team_link = DivisionTeamLink.update(db_obj=team.division_link, in_obj=team_in, session=session)
|
||||
return division_team_link
|
||||
|
||||
|
||||
@router.delete("/{id}/division", tags=[ApiTags.DIVISIONS])
|
||||
def delete_team_division_link(session: SessionDep, current_user: CurrentUser, id: RowId) -> Message:
|
||||
"""
|
||||
Delete a division link from a team.
|
||||
"""
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=id,
|
||||
user_rights=PermissionRight.MANAGE_DIVISIONS,
|
||||
event_rights=PermissionRight.MANAGE_DIVISIONS,
|
||||
)
|
||||
|
||||
session.delete(team.division_link)
|
||||
session.commit()
|
||||
return Message(message="Division deleted from team successfully")
|
||||
|
||||
# endregion
|
||||
|
||||
# region # Teams / Members #####################################################
|
||||
|
||||
def load_member_link(team: Team, member_id: RowId):
|
||||
link = next((link for link in team.member_links if link.member_id == member_id), None)
|
||||
if not link:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Member not found")
|
||||
|
||||
return link
|
||||
|
||||
|
||||
@router.get("/{team_id}/members", response_model=MemberTeamLinksPublic)
|
||||
def read_team_member_links(
|
||||
session: SessionDep, current_user: CurrentUser, team_id: RowId, skip: int = 0, limit: int = 100
|
||||
) -> Any:
|
||||
"""
|
||||
Retrieve all member links from a teams.
|
||||
"""
|
||||
|
||||
if current_user.has_permissions(
|
||||
module=PermissionModule.TEAM,
|
||||
part=PermissionPart.ADMIN,
|
||||
rights=PermissionRight.READ,
|
||||
):
|
||||
data_query = (
|
||||
select(MemberTeamLink)
|
||||
.where(MemberTeamLink.team_id == team_id)
|
||||
)
|
||||
else:
|
||||
data_query = (
|
||||
select(MemberTeamLink)
|
||||
.join(Team, Team.id == MemberTeamLink.team_id)
|
||||
.join(EventUserLink, EventUserLink.event_id == Team.event_id)
|
||||
.where(
|
||||
MemberTeamLink.team_id == team_id,
|
||||
EventUserLink.user_id == current_user.id,
|
||||
# FIXME: (EventUserLink.rights & (PermissionRight.MANAGE_MEMBERS)) > 0
|
||||
)
|
||||
)
|
||||
|
||||
count = session.exec(select(func.count()).select_from(data_query.subquery())).one()
|
||||
data = session.exec(data_query.offset(skip).limit(limit)).all()
|
||||
return MemberTeamLinksPublic(data=data, count=count)
|
||||
|
||||
|
||||
@router.get("/{team_id}/members/{member_id}", response_model=MemberTeamLinkPublic)
|
||||
def read_team_member_link(session: SessionDep, current_user: CurrentUser, team_id: RowId, member_id: RowId) -> Any:
|
||||
"""
|
||||
Get member link by member ID.
|
||||
"""
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=team_id,
|
||||
user_rights=PermissionRight.MANAGE_MEMBERS,
|
||||
event_rights=PermissionRight.MANAGE_MEMBERS,
|
||||
)
|
||||
|
||||
link = load_member_link(team=team, member_id=member_id)
|
||||
|
||||
return link
|
||||
|
||||
|
||||
@router.post("/{team_id}/members", response_model=MemberTeamLinkPublic)
|
||||
def create_team_member_link(
|
||||
*, session: SessionDep, current_user: CurrentUser, team_id: RowId, link_in: MemberTeamLinkCreate
|
||||
) -> Any:
|
||||
"""
|
||||
Create new team.
|
||||
"""
|
||||
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=team_id,
|
||||
user_rights=PermissionRight.MANAGE_MEMBERS,
|
||||
event_rights=PermissionRight.MANAGE_MEMBERS,
|
||||
)
|
||||
|
||||
# Check if user has rights for current status of the member
|
||||
load_member(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=link_in.member_id,
|
||||
user_rights=PermissionRight.MANAGE_MEMBERS,
|
||||
)
|
||||
|
||||
link = MemberTeamLink.create(session=session, create_obj=link_in, team=team)
|
||||
|
||||
return link
|
||||
|
||||
|
||||
@router.put("/{team_id}/members/{member_id}", response_model=MemberTeamLinkPublic)
|
||||
def update_team_member_link(
|
||||
*, session: SessionDep, current_user: CurrentUser, team_id: RowId, member_id: RowId, link_in: MemberTeamLinkUpdate
|
||||
) -> Any:
|
||||
"""
|
||||
Update a team member link.
|
||||
"""
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=team_id,
|
||||
user_rights=PermissionRight.MANAGE_MEMBERS,
|
||||
event_rights=PermissionRight.MANAGE_MEMBERS,
|
||||
)
|
||||
|
||||
link = load_member_link(team=team, member_id=member_id)
|
||||
link = MemberTeamLink.update(session=session, db_obj=link, in_obj=link_in)
|
||||
|
||||
return link
|
||||
|
||||
|
||||
@router.delete("/{team_id}/members/{member_id}")
|
||||
def delete_team_member_link(session: SessionDep,current_user: CurrentUser, team_id: RowId, member_id: RowId) -> Message:
|
||||
"""
|
||||
Delete a team member link.
|
||||
"""
|
||||
team = load_team(
|
||||
session=session,
|
||||
current_user=current_user,
|
||||
id=team_id,
|
||||
user_rights=PermissionRight.MANAGE_MEMBERS,
|
||||
event_rights=PermissionRight.MANAGE_MEMBERS,
|
||||
)
|
||||
|
||||
link = load_member_link(team=team, member_id=member_id)
|
||||
|
||||
session.delete(link)
|
||||
session.commit()
|
||||
return Message(message="Team member link deleted successfully")
|
||||
|
||||
# endregion
|
||||
|
||||
@@ -19,6 +19,7 @@ from app.models.apikey import (
|
||||
ApiKeysPublic,
|
||||
)
|
||||
from app.models.base import ApiTags, Message, RowId
|
||||
from app.models.member import MemberPublic, MemberUpdate, Member
|
||||
from app.models.user import (
|
||||
PermissionModule,
|
||||
PermissionPart,
|
||||
@@ -195,6 +196,37 @@ def read_user_me(current_user: CurrentUser) -> Any:
|
||||
return current_user
|
||||
|
||||
|
||||
@router.get("/me/member", response_model=MemberPublic, tags=[ApiTags.MEMBERS])
|
||||
def read_user_me_member(current_user: CurrentUser) -> Any:
|
||||
"""
|
||||
Get current user member.
|
||||
"""
|
||||
return current_user.member
|
||||
|
||||
|
||||
@router.put("/me/member", response_model=MemberPublic, tags=[ApiTags.MEMBERS])
|
||||
def update_user_me_member(
|
||||
*, session: SessionDep, current_user: CurrentUser, member_in: MemberUpdate
|
||||
) -> Any:
|
||||
"""
|
||||
Get current user member.
|
||||
"""
|
||||
member = session.get(Member, current_user.member_id)
|
||||
data_obj = member_in.model_dump(exclude_unset=True)
|
||||
|
||||
if not member:
|
||||
member = Member.model_validate(data_obj)
|
||||
current_user.member_id = member.id
|
||||
session.add(current_user)
|
||||
else:
|
||||
member.sqlmodel_update(data_obj)
|
||||
session.add(member)
|
||||
session.commit()
|
||||
session.refresh(member)
|
||||
|
||||
return member
|
||||
|
||||
|
||||
@router.delete("/me", response_model=Message)
|
||||
def delete_user_me(session: SessionDep, current_user: CurrentUser) -> Any:
|
||||
"""
|
||||
|
||||
@@ -14,6 +14,7 @@ from pydantic import (
|
||||
from pydantic_core import MultiHostUrl
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
from typing_extensions import Self
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
|
||||
def parse_cors(v: Any) -> list[str] | str:
|
||||
@@ -69,6 +70,12 @@ class Settings(BaseSettings):
|
||||
path=self.POSTGRES_DB,
|
||||
)
|
||||
|
||||
tz: str = "UTC"
|
||||
|
||||
@property
|
||||
def tz_info(self):
|
||||
return ZoneInfo(self.tz)
|
||||
|
||||
SMTP_TLS: bool = True
|
||||
SMTP_SSL: bool = False
|
||||
SMTP_PORT: int = 587
|
||||
|
||||
@@ -5,6 +5,12 @@ from app.models.event import (
|
||||
Event,
|
||||
EventCreate,
|
||||
)
|
||||
from app.models.association import (
|
||||
Association,
|
||||
)
|
||||
from app.models.division import (
|
||||
Division,
|
||||
)
|
||||
from app.models.team import (
|
||||
Team,
|
||||
TeamCreate,
|
||||
@@ -18,6 +24,9 @@ from app.models.user import (
|
||||
User,
|
||||
UserCreate,
|
||||
)
|
||||
from app.models.member import (
|
||||
Member, MemberCreate,
|
||||
)
|
||||
from app.models.apikey import (
|
||||
ApiKey,
|
||||
)
|
||||
@@ -109,6 +118,15 @@ def init_db(session: Session) -> None:
|
||||
)
|
||||
user = User.create(session=session, create_obj=user_in)
|
||||
user.add_role(db_obj=system_admin_role, session=session)
|
||||
|
||||
if not user.member_id:
|
||||
member_in = MemberCreate(
|
||||
name="Super Admin",
|
||||
)
|
||||
member = Member.create(session=session, create_obj=member_in, user=user)
|
||||
user.member = member
|
||||
session.add(user)
|
||||
|
||||
session.commit()
|
||||
|
||||
event = session.exec(
|
||||
@@ -124,11 +142,12 @@ def init_db(session: Session) -> None:
|
||||
event.add_user(user, PermissionRight.ADMIN, session=session)
|
||||
|
||||
team = session.exec(
|
||||
select(Event).where(Team.theme_name == "Laaiend vuur 熾熱的火 🔥")
|
||||
select(Event).where(Team.theme_name == "Lǎàǐend vuur 熾熱的火 🔥")
|
||||
).first()
|
||||
if not team:
|
||||
team_in = TeamCreate(
|
||||
theme_name="Laaiend vuur 熾熱的火 🔥",
|
||||
theme_name="Lǎàǐend vuur 熾熱的火 🔥",
|
||||
short_name="1",
|
||||
event_id=event.id,
|
||||
)
|
||||
team = Team.create(session=session, create_obj=team_in)
|
||||
|
||||
79
backend/app/models/association.py
Normal file
79
backend/app/models/association.py
Normal file
@@ -0,0 +1,79 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlmodel import (
|
||||
Session,
|
||||
Relationship,
|
||||
)
|
||||
|
||||
from . import mixin
|
||||
from .base import (
|
||||
BaseSQLModel,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .division import Division
|
||||
|
||||
# region # Association #########################################################
|
||||
|
||||
|
||||
class AssociationBase(
|
||||
mixin.Name,
|
||||
mixin.Contact,
|
||||
mixin.ScoutingId,
|
||||
BaseSQLModel,
|
||||
):
|
||||
pass
|
||||
|
||||
|
||||
# Properties to receive via API on creation
|
||||
class AssociationCreate(AssociationBase):
|
||||
pass
|
||||
|
||||
|
||||
# Properties to receive via API on update, all are optional
|
||||
class AssociationUpdate(AssociationBase):
|
||||
pass
|
||||
|
||||
|
||||
class Association(mixin.RowId, AssociationBase, table=True):
|
||||
# --- database only items --------------------------------------------------
|
||||
|
||||
# --- read only items ------------------------------------------------------
|
||||
|
||||
# --- back_populates links -------------------------------------------------
|
||||
divisions: list["Division"] = Relationship(back_populates="association", cascade_delete=True)
|
||||
|
||||
# --- CRUD actions ---------------------------------------------------------
|
||||
@classmethod
|
||||
def create(cls, *, session: Session, create_obj: AssociationCreate) -> "Association":
|
||||
data_obj = create_obj.model_dump(exclude_unset=True)
|
||||
|
||||
db_obj = cls.model_validate(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
@classmethod
|
||||
def update(
|
||||
cls, *, session: Session, db_obj: "Association", in_obj: AssociationUpdate
|
||||
) -> "Association":
|
||||
data_obj = in_obj.model_dump(exclude_unset=True)
|
||||
db_obj.sqlmodel_update(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
|
||||
# Properties to return via API, id is always required
|
||||
class AssociationPublic(mixin.RowIdPublic, AssociationBase):
|
||||
pass
|
||||
|
||||
|
||||
class AssociationsPublic(BaseSQLModel):
|
||||
data: list[AssociationPublic]
|
||||
count: int
|
||||
|
||||
|
||||
# endregion
|
||||
@@ -56,6 +56,9 @@ class ApiTags(DocumentedStrEnum):
|
||||
|
||||
EVENTS = "Events"
|
||||
TEAMS = "Teams"
|
||||
ASSOCIATIONS = "Associations"
|
||||
DIVISIONS = "Divisions"
|
||||
MEMBERS = "Members"
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
159
backend/app/models/division.py
Normal file
159
backend/app/models/division.py
Normal file
@@ -0,0 +1,159 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy.orm.sync import update
|
||||
from sqlmodel import (
|
||||
Session,
|
||||
Field,
|
||||
Relationship,
|
||||
)
|
||||
|
||||
from . import mixin
|
||||
from .base import (
|
||||
BaseSQLModel,
|
||||
RowId,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .association import Association
|
||||
from .team import Team
|
||||
|
||||
|
||||
# region # Divisions / Team Link ######################################
|
||||
|
||||
class DivisionTeamLinkBase(
|
||||
mixin.Name,
|
||||
BaseSQLModel,
|
||||
):
|
||||
pass
|
||||
|
||||
|
||||
class DivisionTeamLinkCreate(DivisionTeamLinkBase):
|
||||
division_id: RowId | None = Field(default=None)
|
||||
|
||||
|
||||
class DivisionTeamLinkUpdate(DivisionTeamLinkBase):
|
||||
pass
|
||||
|
||||
|
||||
class DivisionTeamLink(DivisionTeamLinkBase, table=True):
|
||||
# --- database only items --------------------------------------------------
|
||||
|
||||
# --- read only items ------------------------------------------------------
|
||||
team_id: RowId = Field(
|
||||
default=None,
|
||||
foreign_key="team.id",
|
||||
nullable=False,
|
||||
ondelete="CASCADE",
|
||||
primary_key=True,
|
||||
)
|
||||
|
||||
# --- back_populates links -------------------------------------------------
|
||||
division_id: RowId = Field(
|
||||
default=None,
|
||||
foreign_key="division.id",
|
||||
nullable=False,
|
||||
ondelete="CASCADE",
|
||||
)
|
||||
division: "Division" = Relationship(back_populates="team_links")
|
||||
team: "Team" = Relationship(back_populates="division_link")
|
||||
|
||||
# --- CRUD actions ---------------------------------------------------------
|
||||
@classmethod
|
||||
def create(cls, *, session: Session, create_obj: DivisionTeamLinkCreate, team: "Team") -> "DivisionTeamLink":
|
||||
data_obj = create_obj.model_dump(exclude_unset=True)
|
||||
|
||||
db_obj = cls.model_validate(data_obj, update={"team_id": team.id})
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
@classmethod
|
||||
def update(
|
||||
cls, *, session: Session, db_obj: "DivisionTeamLink", in_obj: DivisionTeamLinkUpdate
|
||||
) -> "DivisionTeamLink":
|
||||
data_obj = in_obj.model_dump(exclude_unset=True)
|
||||
db_obj.sqlmodel_update(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
|
||||
# Properties to return via API, id is always required
|
||||
class DivisionTeamLinkPublic(DivisionTeamLinkBase):
|
||||
team_id: RowId
|
||||
division_id: RowId
|
||||
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
|
||||
# region # Divisions ###########################################################
|
||||
|
||||
|
||||
class DivisionBase(
|
||||
mixin.Name,
|
||||
mixin.Contact,
|
||||
mixin.ScoutingId,
|
||||
BaseSQLModel,
|
||||
):
|
||||
association_id: RowId = Field(
|
||||
foreign_key="association.id", nullable=False, ondelete="CASCADE"
|
||||
)
|
||||
|
||||
|
||||
# Properties to receive via API on creation
|
||||
class DivisionCreate(DivisionBase):
|
||||
pass
|
||||
|
||||
|
||||
# Properties to receive via API on update, all are optional
|
||||
class DivisionUpdate(DivisionBase):
|
||||
association_id: RowId | None = Field(default=None)
|
||||
|
||||
|
||||
class Division(mixin.RowId, DivisionBase, table=True):
|
||||
# --- database only items --------------------------------------------------
|
||||
|
||||
# --- read only items ------------------------------------------------------
|
||||
|
||||
# --- back_populates links -------------------------------------------------
|
||||
association: "Association" = Relationship(back_populates="divisions")
|
||||
team_links: list["DivisionTeamLink"] = Relationship(back_populates="division", cascade_delete=True)
|
||||
|
||||
# --- CRUD actions ---------------------------------------------------------
|
||||
@classmethod
|
||||
def create(cls, *, session: Session, create_obj: DivisionCreate) -> "Division":
|
||||
data_obj = create_obj.model_dump(exclude_unset=True)
|
||||
|
||||
db_obj = cls.model_validate(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
@classmethod
|
||||
def update(
|
||||
cls, *, session: Session, db_obj: "Division", in_obj: DivisionUpdate
|
||||
) -> "Division":
|
||||
data_obj = in_obj.model_dump(exclude_unset=True)
|
||||
db_obj.sqlmodel_update(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
|
||||
# Properties to return via API, id is always required
|
||||
class DivisionPublic(mixin.RowIdPublic, DivisionBase):
|
||||
association_id: RowId
|
||||
|
||||
|
||||
class DivisionsPublic(BaseSQLModel):
|
||||
data: list[DivisionPublic]
|
||||
count: int
|
||||
|
||||
|
||||
# endregion
|
||||
@@ -102,7 +102,7 @@ class Event(mixin.RowId, EventBase, table=True):
|
||||
|
||||
# --- many-to-many links ---------------------------------------------------
|
||||
user_links: list["EventUserLink"] = Relationship(back_populates="event", cascade_delete=True)
|
||||
team_links: list["Team"] = Relationship(back_populates="event", cascade_delete=True)
|
||||
teams: list["Team"] = Relationship(back_populates="event", cascade_delete=True)
|
||||
|
||||
# --- CRUD actions ---------------------------------------------------------
|
||||
@classmethod
|
||||
|
||||
191
backend/app/models/member.py
Normal file
191
backend/app/models/member.py
Normal file
@@ -0,0 +1,191 @@
|
||||
from typing import TYPE_CHECKING, Optional
|
||||
|
||||
from sqlmodel import (
|
||||
Session,
|
||||
Field,
|
||||
Relationship,
|
||||
)
|
||||
|
||||
from . import mixin
|
||||
from .base import (
|
||||
BaseSQLModel,
|
||||
DocumentedIntFlag,
|
||||
auto_enum,
|
||||
RowId,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .user import User
|
||||
from .team import Team
|
||||
|
||||
|
||||
# region # Member / Teams ######################################################
|
||||
|
||||
|
||||
class MemberRank(DocumentedIntFlag):
|
||||
TEAM_MEMBER = auto_enum()
|
||||
|
||||
TEAM_ASSISTANT_LEADER = auto_enum()
|
||||
TEAM_LEADER = auto_enum()
|
||||
|
||||
DIVISION_LEADER = auto_enum()
|
||||
|
||||
VOLUNTEER = auto_enum()
|
||||
|
||||
|
||||
# ##############################################################################
|
||||
|
||||
# Shared properties
|
||||
class MemberTeamLinkBase(BaseSQLModel):
|
||||
rank: MemberRank = Field(default=MemberRank.TEAM_MEMBER, nullable=False)
|
||||
|
||||
|
||||
# Properties to receive via API on creation
|
||||
class MemberTeamLinkCreate(MemberTeamLinkBase):
|
||||
member_id: RowId = Field(default=None, nullable=False)
|
||||
|
||||
|
||||
# Properties to receive via API on update, all are optional
|
||||
class MemberTeamLinkUpdate(MemberTeamLinkBase):
|
||||
pass
|
||||
|
||||
|
||||
# Database model, database table inferred from class name
|
||||
class MemberTeamLink(MemberTeamLinkBase, table=True):
|
||||
# --- database only items --------------------------------------------------
|
||||
|
||||
# --- read only items ------------------------------------------------------
|
||||
|
||||
# --- back_populates links -------------------------------------------------
|
||||
member_id: RowId = Field(
|
||||
foreign_key="member.id",
|
||||
primary_key=True,
|
||||
nullable=False,
|
||||
ondelete="CASCADE",
|
||||
)
|
||||
|
||||
team_id: RowId = Field(
|
||||
foreign_key="team.id",
|
||||
primary_key=True,
|
||||
nullable=False,
|
||||
ondelete="CASCADE",
|
||||
)
|
||||
|
||||
member: "Member" = Relationship(back_populates="team_links")
|
||||
team: "Team" = Relationship(back_populates="member_links")
|
||||
|
||||
# --- CRUD actions ---------------------------------------------------------
|
||||
@classmethod
|
||||
def create(cls, *, session: Session, create_obj: MemberTeamLinkCreate, team: "Team") -> "MemberTeamLink":
|
||||
data_obj = create_obj.model_dump(exclude_unset=True)
|
||||
db_obj = cls.model_validate(data_obj, update={"team_id": team.id})
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
@classmethod
|
||||
def update(
|
||||
cls, *, session: Session, db_obj: "MemberTeamLink", in_obj: MemberTeamLinkUpdate
|
||||
) -> "MemberTeamLink":
|
||||
data_obj = in_obj.model_dump(exclude_unset=True)
|
||||
db_obj.sqlmodel_update(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
|
||||
# Properties to return via API
|
||||
class MemberTeamLinkPublic(MemberTeamLinkBase):
|
||||
member_id: RowId
|
||||
team_id: RowId
|
||||
|
||||
|
||||
class MemberTeamLinksPublic(BaseSQLModel):
|
||||
data: list[MemberTeamLinkPublic]
|
||||
count: int
|
||||
|
||||
|
||||
# endregion
|
||||
|
||||
|
||||
# region # Member ##############################################################
|
||||
|
||||
|
||||
# Shared properties
|
||||
class MemberBase(
|
||||
mixin.Name,
|
||||
mixin.Contact,
|
||||
mixin.ScoutingId,
|
||||
mixin.Comment,
|
||||
mixin.Allergy,
|
||||
mixin.Birthday,
|
||||
mixin.Canceled,
|
||||
BaseSQLModel,
|
||||
):
|
||||
pass
|
||||
|
||||
|
||||
# Properties to receive via API on creation
|
||||
class MemberCreate(MemberBase):
|
||||
pass
|
||||
|
||||
|
||||
# Properties to receive via API on update, all are optional
|
||||
class MemberUpdate(MemberBase):
|
||||
pass
|
||||
|
||||
|
||||
# Database model, database table inferred from class name
|
||||
class Member(mixin.RowId, mixin.Created, MemberBase, table=True):
|
||||
# --- database only items --------------------------------------------------
|
||||
|
||||
# --- read only items ------------------------------------------------------
|
||||
|
||||
# --- back_populates links -------------------------------------------------
|
||||
user: Optional["User"] = Relationship(
|
||||
back_populates="member",
|
||||
sa_relationship_kwargs={"foreign_keys": "User.member_id"},
|
||||
)
|
||||
team_links: list["MemberTeamLink"] = Relationship(back_populates="member", cascade_delete=True)
|
||||
|
||||
# --- CRUD actions ---------------------------------------------------------
|
||||
@classmethod
|
||||
def create(cls, *, session: Session, create_obj: MemberCreate, user: Optional["User"] = None) -> "Member":
|
||||
data_obj = create_obj.model_dump(exclude_unset=True)
|
||||
|
||||
extra_fields = {}
|
||||
if user:
|
||||
extra_fields["created_by"] = user.id
|
||||
|
||||
db_obj = cls.model_validate(data_obj, update=extra_fields)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
@classmethod
|
||||
def update(
|
||||
cls, *, session: Session, db_obj: "Member", in_obj: MemberUpdate
|
||||
) -> "Member":
|
||||
data_obj = in_obj.model_dump(exclude_unset=True)
|
||||
db_obj.sqlmodel_update(data_obj)
|
||||
session.add(db_obj)
|
||||
session.commit()
|
||||
session.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
|
||||
# Properties to return via API, id is always required
|
||||
class MemberPublic(mixin.RowIdPublic, MemberBase):
|
||||
# TODO: Return user_id
|
||||
pass
|
||||
|
||||
|
||||
class MembersPublic(BaseSQLModel):
|
||||
data: list[MemberPublic]
|
||||
count: int
|
||||
|
||||
|
||||
# endregion
|
||||
@@ -1,5 +1,5 @@
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from datetime import datetime, date
|
||||
|
||||
from pydantic import BaseModel, EmailStr
|
||||
from sqlmodel import (
|
||||
@@ -7,6 +7,7 @@ from sqlmodel import (
|
||||
)
|
||||
|
||||
from .base import RowId as RowIdType
|
||||
from ..core.config import settings
|
||||
|
||||
|
||||
class Name(BaseModel):
|
||||
@@ -25,6 +26,14 @@ class ThemeNameUpdate(ThemeName):
|
||||
theme_name: str | None = Field(default=None, max_length=255)
|
||||
|
||||
|
||||
class ShortName(BaseModel):
|
||||
short_name: str = Field(index=True, max_length=8)
|
||||
|
||||
|
||||
class ShortNameUpdate(ShortName):
|
||||
short_name: str | None = Field(default=None, max_length=8)
|
||||
|
||||
|
||||
class Contact(BaseModel):
|
||||
contact: str | None = Field(default=None, nullable=True, max_length=255)
|
||||
|
||||
@@ -54,7 +63,7 @@ class EmailUpdate(Email):
|
||||
|
||||
|
||||
class ScoutingId(BaseModel):
|
||||
scouting_id: str | None = Field(default=None, max_length=32)
|
||||
scouting_id: str | None = Field(default=None, max_length=32, description="Association registration number")
|
||||
|
||||
|
||||
class Password(BaseModel):
|
||||
@@ -81,6 +90,14 @@ class Description(BaseModel):
|
||||
description: str | None = Field(default=None, nullable=True, max_length=512)
|
||||
|
||||
|
||||
class Comment(BaseModel):
|
||||
comment: str | None = Field(default=None, nullable=True, max_length=512)
|
||||
|
||||
|
||||
class Allergy(BaseModel):
|
||||
allergy: str | None = Field(default=None, nullable=True, max_length=512)
|
||||
|
||||
|
||||
class StartEndDate:
|
||||
start_at: datetime | None = Field(default=None, nullable=True)
|
||||
end_at: datetime | None = Field(default=None, nullable=True)
|
||||
@@ -96,3 +113,10 @@ class CheckInCheckOut(BaseModel):
|
||||
checkout_at: datetime | None = Field(default=None, nullable=True)
|
||||
|
||||
|
||||
class Birthday(BaseModel):
|
||||
birthday_at: date | None = Field(default=None, nullable=True)
|
||||
|
||||
|
||||
class Created(BaseModel):
|
||||
created_at: datetime | None = Field(nullable=False, default_factory=lambda: datetime.now(settings.tz_info))
|
||||
created_by: RowIdType | None = Field(default=None, nullable=True, foreign_key="user.id", ondelete="SET NULL")
|
||||
|
||||
@@ -14,12 +14,15 @@ from .base import (
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .event import Event
|
||||
from .division import DivisionTeamLink
|
||||
from .member import MemberTeamLink
|
||||
|
||||
# region # Team ################################################################
|
||||
|
||||
|
||||
class TeamBase(
|
||||
mixin.ThemeName,
|
||||
mixin.ShortName,
|
||||
mixin.CheckInCheckOut,
|
||||
mixin.Canceled,
|
||||
BaseSQLModel
|
||||
@@ -27,9 +30,6 @@ class TeamBase(
|
||||
event_id: RowId = Field(
|
||||
foreign_key="event.id", nullable=False, ondelete="CASCADE"
|
||||
)
|
||||
# scouting_team_id: RowId | None = Field(
|
||||
# foreign_key="ScoutingTeam.id", nullable=False, ondelete="CASCADE"
|
||||
# )
|
||||
|
||||
|
||||
# Properties to receive via API on creation
|
||||
@@ -38,7 +38,7 @@ class TeamCreate(TeamBase):
|
||||
|
||||
|
||||
# Properties to receive via API on update, all are optional
|
||||
class TeamUpdate(mixin.ThemeNameUpdate, TeamBase):
|
||||
class TeamUpdate(mixin.ThemeNameUpdate, mixin.ShortNameUpdate, TeamBase):
|
||||
event_id: RowId | None = Field(default=None)
|
||||
|
||||
|
||||
@@ -48,8 +48,9 @@ class Team(mixin.RowId, TeamBase, table=True):
|
||||
# --- read only items ------------------------------------------------------
|
||||
|
||||
# --- back_populates links -------------------------------------------------
|
||||
event: "Event" = Relationship(back_populates="team_links")#, cascade_delete=True)
|
||||
# team: "ScoutingTeam" = Relationship(back_populates="event_links", cascade_delete=True)
|
||||
event: "Event" = Relationship(back_populates="teams")
|
||||
division_link: "DivisionTeamLink" = Relationship(back_populates="team", cascade_delete=True)
|
||||
member_links: list["MemberTeamLink"] = Relationship(back_populates="team", cascade_delete=True)
|
||||
|
||||
# --- CRUD actions ---------------------------------------------------------
|
||||
@classmethod
|
||||
@@ -84,4 +85,4 @@ class TeamsPublic(BaseSQLModel):
|
||||
count: int
|
||||
|
||||
|
||||
# endregion
|
||||
# endregion
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TYPE_CHECKING, Optional
|
||||
|
||||
from pydantic import EmailStr, field_validator
|
||||
from sqlmodel import Field, Relationship, Session, select
|
||||
@@ -17,6 +17,7 @@ from .base import (
|
||||
if TYPE_CHECKING:
|
||||
from .apikey import ApiKey
|
||||
from .event import EventUserLink
|
||||
from .member import Member
|
||||
|
||||
|
||||
# region # User ################################################################
|
||||
@@ -27,6 +28,9 @@ class PermissionModule(DocumentedStrEnum):
|
||||
USER = auto_enum()
|
||||
EVENT = auto_enum()
|
||||
TEAM = auto_enum()
|
||||
ASSOCIATION = auto_enum()
|
||||
DIVISION = auto_enum()
|
||||
MEMBER = auto_enum()
|
||||
|
||||
|
||||
class PermissionPart(DocumentedStrEnum):
|
||||
@@ -42,8 +46,18 @@ class PermissionRight(DocumentedIntFlag):
|
||||
|
||||
MANAGE_USERS = auto_enum()
|
||||
MANAGE_TEAMS = auto_enum()
|
||||
MANAGE_DIVISIONS = auto_enum()
|
||||
MANAGE_MEMBERS = auto_enum()
|
||||
|
||||
ADMIN = CREATE | READ | UPDATE | DELETE | MANAGE_USERS | MANAGE_TEAMS
|
||||
ADMIN = ( CREATE
|
||||
| READ
|
||||
| UPDATE
|
||||
| DELETE
|
||||
| MANAGE_USERS
|
||||
| MANAGE_TEAMS
|
||||
| MANAGE_DIVISIONS
|
||||
| MANAGE_MEMBERS
|
||||
)
|
||||
|
||||
|
||||
# ##############################################################################
|
||||
@@ -72,13 +86,16 @@ class UserRoleLink(BaseSQLModel, table=True):
|
||||
class UserBase(
|
||||
mixin.UserName,
|
||||
mixin.Email,
|
||||
mixin.FullName,
|
||||
mixin.ScoutingId,
|
||||
mixin.IsActive,
|
||||
mixin.IsVerified,
|
||||
BaseSQLModel,
|
||||
):
|
||||
pass
|
||||
member_id: RowId | None = Field(
|
||||
default=None,
|
||||
foreign_key="member.id",
|
||||
nullable=True,
|
||||
ondelete="SET NULL",
|
||||
)
|
||||
|
||||
|
||||
# Properties to receive via API on creation
|
||||
@@ -86,7 +103,7 @@ class UserCreate(mixin.Password, UserBase):
|
||||
pass
|
||||
|
||||
|
||||
class UserRegister(mixin.Password, mixin.FullName, BaseSQLModel):
|
||||
class UserRegister(mixin.Password, BaseSQLModel):
|
||||
email: EmailStr = Field(max_length=255)
|
||||
|
||||
|
||||
@@ -95,7 +112,7 @@ class UserUpdate(mixin.EmailUpdate, mixin.PasswordUpdate, UserBase):
|
||||
pass
|
||||
|
||||
|
||||
class UserUpdateMe(mixin.FullName, mixin.EmailUpdate, BaseSQLModel):
|
||||
class UserUpdateMe(mixin.EmailUpdate, BaseSQLModel):
|
||||
pass
|
||||
|
||||
|
||||
@@ -110,6 +127,10 @@ class User(mixin.RowId, UserBase, table=True):
|
||||
hashed_password: str
|
||||
|
||||
# --- back_populates links -------------------------------------------------
|
||||
member: Optional["Member"] = Relationship(
|
||||
back_populates="user",
|
||||
sa_relationship_kwargs={"foreign_keys": "User.member_id"},
|
||||
)
|
||||
api_keys: list["ApiKey"] = Relationship(back_populates="user", cascade_delete=True)
|
||||
|
||||
# --- many-to-many links ---------------------------------------------------
|
||||
|
||||
222
backend/app/tests/api/routes/test_association.py
Normal file
222
backend/app/tests/api/routes/test_association.py
Normal file
@@ -0,0 +1,222 @@
|
||||
import uuid
|
||||
|
||||
from fastapi import status
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.core.config import settings
|
||||
from app.tests.utils.association import create_random_association
|
||||
from app.tests.utils.division import create_random_division
|
||||
|
||||
|
||||
def test_create_association(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
data = {
|
||||
"name": "Scouting Maurits-Viool",
|
||||
"contact": "Sebas",
|
||||
"scouting_id": "2577",
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/associations/",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == data["name"]
|
||||
assert content["contact"] == data["contact"]
|
||||
assert content["scouting_id"] == data["scouting_id"]
|
||||
assert "id" in content
|
||||
|
||||
|
||||
def test_create_association_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
data = {
|
||||
"name": "Scouting Maurits-Viool",
|
||||
"contact": "Sebas",
|
||||
"scouting_id": "2577",
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/associations/",
|
||||
headers=normal_user_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_association(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/associations/{association.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(association.id)
|
||||
assert content["name"] == association.name
|
||||
assert content["contact"] == association.contact
|
||||
assert content["scouting_id"] == association.scouting_id
|
||||
|
||||
|
||||
def test_read_association_not_found(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/associations/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Association not found"
|
||||
|
||||
|
||||
def test_read_association_no_permission(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/associations/{association.id}",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_associations(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
create_random_association(db)
|
||||
create_random_association(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/associations/",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
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_associations_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
create_random_association(db)
|
||||
create_random_association(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/associations/",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert "count" in content
|
||||
assert content["count"] == 0
|
||||
assert "data" in content
|
||||
assert isinstance(content["data"], list)
|
||||
assert len(content["data"]) == 0
|
||||
|
||||
|
||||
def test_update_association(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
data = {
|
||||
"name": "Updated name",
|
||||
"contact": "Updated contact",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/associations/{association.id}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(association.id)
|
||||
assert content["name"] == data["name"]
|
||||
assert content["contact"] == data["contact"]
|
||||
assert content["scouting_id"] == association.scouting_id
|
||||
|
||||
|
||||
def test_update_association_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
data = {
|
||||
"name": "Not found",
|
||||
"contact": "Not found",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/associations/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Association not found"
|
||||
|
||||
|
||||
def test_update_association_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
data = {
|
||||
"name": "No permissions",
|
||||
"contact": "No permissions",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/associations/{association.id}",
|
||||
headers=normal_user_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_delete_association(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/associations/{association.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["message"] == "Association deleted successfully"
|
||||
|
||||
|
||||
def test_delete_association_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/associations/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Association not found"
|
||||
|
||||
|
||||
def test_delete_association_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/associations/{association.id}",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_association_divisions(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
create_random_division(db, association=association)
|
||||
create_random_division(db, association=association)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/associations/{association.id}/divisions",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
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_association_divisions_not_found(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/associations/{uuid.uuid4()}/divisions",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Association not found"
|
||||
|
||||
|
||||
def test_read_association_divisions_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
create_random_division(db, association=association)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/associations/{association.id}/divisions",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
471
backend/app/tests/api/routes/test_division.py
Normal file
471
backend/app/tests/api/routes/test_division.py
Normal file
@@ -0,0 +1,471 @@
|
||||
import uuid
|
||||
|
||||
from fastapi import status
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.core.config import settings
|
||||
from app.models.division import DivisionTeamLink, DivisionTeamLinkCreate
|
||||
from app.tests.conftest import EventUserHeader
|
||||
from app.tests.utils.division import create_random_division
|
||||
from app.tests.utils.association import create_random_association
|
||||
from app.tests.utils.team import create_random_team
|
||||
from app.tests.utils.utils import random_lower_string
|
||||
|
||||
|
||||
def test_create_division(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
data = {
|
||||
"name": "Verkenners",
|
||||
"scouting_id": "122314",
|
||||
"association_id": str(association.id),
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/divisions/",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == data["name"]
|
||||
assert content["scouting_id"] == data["scouting_id"]
|
||||
assert content["association_id"] == str(association.id)
|
||||
assert "contact" in content
|
||||
assert "id" in content
|
||||
|
||||
|
||||
def test_create_division_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
association = create_random_association(db)
|
||||
data = {
|
||||
"name": "Padvinsters",
|
||||
"contact": "-",
|
||||
"scouting_id": "122323",
|
||||
"association_id": str(association.id),
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/divisions/",
|
||||
headers=normal_user_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_division(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
division = create_random_division(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/divisions/{division.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(division.id)
|
||||
assert content["name"] == division.name
|
||||
assert content["contact"] == division.contact
|
||||
assert content["scouting_id"] == division.scouting_id
|
||||
assert content["association_id"] == str(division.association_id)
|
||||
|
||||
|
||||
def test_read_division_not_found(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/divisions/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Division not found"
|
||||
|
||||
|
||||
def test_read_division_no_permission(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
division = create_random_division(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/divisions/{division.id}",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_divisions(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
create_random_division(db)
|
||||
create_random_division(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/divisions/",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
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_divisions_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
create_random_division(db)
|
||||
create_random_division(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/divisions/",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert "count" in content
|
||||
assert content["count"] == 0
|
||||
assert "data" in content
|
||||
assert isinstance(content["data"], list)
|
||||
assert len(content["data"]) == 0
|
||||
|
||||
|
||||
def test_update_division(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
division = create_random_division(db)
|
||||
data = {
|
||||
"name": "Updated name",
|
||||
"contact": "Updated contact",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/divisions/{division.id}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(division.id)
|
||||
assert content["name"] == data["name"]
|
||||
assert content["contact"] == data["contact"]
|
||||
assert content["scouting_id"] == division.scouting_id
|
||||
assert content["association_id"] == str(division.association_id)
|
||||
|
||||
|
||||
def test_update_division_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
data = {
|
||||
"name": "Not found",
|
||||
"contact": "Not found",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/divisions/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Division not found"
|
||||
|
||||
|
||||
def test_update_division_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
division = create_random_division(db)
|
||||
data = {
|
||||
"name": "No permissions",
|
||||
"contact": "No permissions",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/divisions/{division.id}",
|
||||
headers=normal_user_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_delete_division(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
division = create_random_division(db)
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/divisions/{division.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["message"] == "Division deleted successfully"
|
||||
|
||||
|
||||
def test_delete_division_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/divisions/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Division not found"
|
||||
|
||||
|
||||
def test_delete_division_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
division = create_random_division(db)
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/divisions/{division.id}",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_create_team_division_link(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
division = create_random_division(db)
|
||||
data = {
|
||||
"name": "Otters",
|
||||
"division_id": str(division.id),
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == data["name"]
|
||||
assert content["team_id"] == str(team.id)
|
||||
assert content["division_id"] == str(division.id)
|
||||
|
||||
|
||||
def test_create_team_division_link_unknown_team(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
division = create_random_division(db)
|
||||
data = {
|
||||
"name": "Vossen",
|
||||
"division_id": str(division.id),
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/teams/{uuid.uuid4()}/division",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Team not found"
|
||||
|
||||
|
||||
def test_create_team_division_already_linked(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
|
||||
division1 = create_random_division(db)
|
||||
name = random_lower_string()
|
||||
DivisionTeamLink.create(session=db, create_obj=DivisionTeamLinkCreate(name=name, division_id=division1.id), team=team)
|
||||
|
||||
division2 = create_random_division(db)
|
||||
data = {
|
||||
"name": "Vossen",
|
||||
"division_id": str(division2.id),
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_409_CONFLICT
|
||||
assert response.json()["detail"] == "Team already linked to division"
|
||||
|
||||
|
||||
def test_create_team_division_link_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
division = create_random_division(db)
|
||||
data = {
|
||||
"name": "Arenden",
|
||||
"division_id": str(division.id),
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=normal_user_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_create_team_division_link_event_user(client: TestClient, event_user_token_headers: EventUserHeader, db: Session) -> None:
|
||||
team = create_random_team(db, event=event_user_token_headers.event)
|
||||
division = create_random_division(db)
|
||||
data = {
|
||||
"name": "Sperwers",
|
||||
"division_id": str(division.id),
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=event_user_token_headers.headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == data["name"]
|
||||
assert content["team_id"] == str(team.id)
|
||||
assert content["division_id"] == str(division.id)
|
||||
|
||||
|
||||
def test_read_team_division(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
|
||||
division = create_random_division(db)
|
||||
name = random_lower_string()
|
||||
DivisionTeamLink.create(session=db, create_obj=DivisionTeamLinkCreate(name=name, division_id=division.id), team=team)
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == name
|
||||
assert content["team_id"] == str(team.id)
|
||||
assert content["division_id"] == str(division.id)
|
||||
|
||||
|
||||
def test_read_team_division_unknown_team(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/teams/{uuid.uuid4()}/division",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Team not found"
|
||||
|
||||
|
||||
def test_read_team_division_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_team_division_event_user(client: TestClient, event_user_token_headers: EventUserHeader, db: Session) -> None:
|
||||
team = create_random_team(db, event=event_user_token_headers.event)
|
||||
|
||||
division = create_random_division(db)
|
||||
name = random_lower_string()
|
||||
DivisionTeamLink.create(session=db, create_obj=DivisionTeamLinkCreate(name=name, division_id=division.id), team=team)
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=event_user_token_headers.headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == name
|
||||
assert content["team_id"] == str(team.id)
|
||||
assert content["division_id"] == str(division.id)
|
||||
|
||||
|
||||
def test_update_team_division_link(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
|
||||
division = create_random_division(db)
|
||||
name = random_lower_string()
|
||||
DivisionTeamLink.create(session=db, create_obj=DivisionTeamLinkCreate(name=name, division_id=division.id), team=team)
|
||||
|
||||
data = {
|
||||
"name": "Muizenoor",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == data["name"]
|
||||
assert content["team_id"] == str(team.id)
|
||||
assert content["division_id"] == str(division.id)
|
||||
|
||||
|
||||
def test_update_team_division_link_unknown_team(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
data = {
|
||||
"name": "Leeuenbek",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/teams/{uuid.uuid4()}/division",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Team not found"
|
||||
|
||||
|
||||
def test_update_team_division_link_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
|
||||
division = create_random_division(db)
|
||||
name = random_lower_string()
|
||||
DivisionTeamLink.create(session=db, create_obj=DivisionTeamLinkCreate(name=name, division_id=division.id), team=team)
|
||||
|
||||
data = {
|
||||
"name": "Geitebaard",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=normal_user_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_update_team_division_link_event_user(client: TestClient, event_user_token_headers: EventUserHeader, db: Session) -> None:
|
||||
team = create_random_team(db, event=event_user_token_headers.event)
|
||||
|
||||
division = create_random_division(db)
|
||||
name = random_lower_string()
|
||||
DivisionTeamLink.create(session=db, create_obj=DivisionTeamLinkCreate(name=name, division_id=division.id), team=team)
|
||||
|
||||
data = {
|
||||
"name": "Berenklouw",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=event_user_token_headers.headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == data["name"]
|
||||
assert content["team_id"] == str(team.id)
|
||||
assert content["division_id"] == str(division.id)
|
||||
|
||||
|
||||
def test_delete_team_division_link(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
|
||||
division = create_random_division(db)
|
||||
name = random_lower_string()
|
||||
DivisionTeamLink.create(session=db, create_obj=DivisionTeamLinkCreate(name=name, division_id=division.id), team=team)
|
||||
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["message"] == "Division deleted from team successfully"
|
||||
|
||||
|
||||
def test_delete_team_division_link_unknown_team(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/teams/{uuid.uuid4()}/division",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Team not found"
|
||||
|
||||
|
||||
def test_delete_team_division_link_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
|
||||
division = create_random_division(db)
|
||||
name = random_lower_string()
|
||||
DivisionTeamLink.create(session=db, create_obj=DivisionTeamLinkCreate(name=name, division_id=division.id), team=team)
|
||||
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_delete_team_division_link_event_user(client: TestClient, event_user_token_headers: EventUserHeader, db: Session) -> None:
|
||||
team = create_random_team(db, event=event_user_token_headers.event)
|
||||
|
||||
division = create_random_division(db)
|
||||
name = random_lower_string()
|
||||
DivisionTeamLink.create(session=db, create_obj=DivisionTeamLinkCreate(name=name, division_id=division.id), team=team)
|
||||
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/division",
|
||||
headers=event_user_token_headers.headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["message"] == "Division deleted from team successfully"
|
||||
396
backend/app/tests/api/routes/test_member.py
Normal file
396
backend/app/tests/api/routes/test_member.py
Normal file
@@ -0,0 +1,396 @@
|
||||
import uuid
|
||||
|
||||
from fastapi import status
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.core.config import settings
|
||||
from app.models.member import Member, MemberTeamLink, MemberTeamLinkCreate, MemberRank
|
||||
from app.models.user import User
|
||||
from app.tests.conftest import EventUserHeader
|
||||
from app.tests.utils.team import create_random_team
|
||||
from app.tests.utils.member import create_random_member
|
||||
from app.tests.utils.user import create_random_user, authentication_token_from_user
|
||||
|
||||
|
||||
def test_create_member(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
data = {
|
||||
"name": "John Do",
|
||||
"scouting_id": "12345678",
|
||||
"allergy": "Do not feed Tomatoes",
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/members/",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["name"] == data["name"]
|
||||
assert content["scouting_id"] == data["scouting_id"]
|
||||
assert content["allergy"] == data["allergy"]
|
||||
assert "contact" in content
|
||||
assert "comment" in content
|
||||
assert "birthday_at" in content
|
||||
assert "canceled_at" in content
|
||||
assert "canceled_reason" in content
|
||||
assert "id" in content
|
||||
|
||||
member_query = select(Member).where(Member.id == content["id"])
|
||||
member_db = db.exec(member_query).first()
|
||||
assert member_db
|
||||
assert member_db.name == data["name"]
|
||||
assert member_db.scouting_id == data["scouting_id"]
|
||||
assert member_db.allergy == data["allergy"]
|
||||
|
||||
user = User.get_by_email(session=db, email=settings.FIRST_SUPERUSER)
|
||||
assert member_db.created_by == user.id
|
||||
|
||||
|
||||
def test_create_member_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
data = {
|
||||
"name": "No John",
|
||||
"scouting_id": "0",
|
||||
"comment": "Is not existing",
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/members/",
|
||||
headers=normal_user_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_member(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
member = create_random_member(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/members/{member.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(member.id)
|
||||
assert content["name"] == member.name
|
||||
assert content["contact"] == member.contact
|
||||
assert content["scouting_id"] == member.scouting_id
|
||||
assert content["comment"] == member.comment
|
||||
assert content["allergy"] == member.allergy
|
||||
assert content["birthday_at"] == member.birthday_at
|
||||
assert content["canceled_at"] == member.canceled_at
|
||||
assert content["canceled_reason"] == member.canceled_reason
|
||||
|
||||
|
||||
def test_read_member_event_user(client: TestClient, event_user_token_headers: EventUserHeader, db: Session) -> None:
|
||||
team = create_random_team(db, event=event_user_token_headers.event)
|
||||
|
||||
member = create_random_member(db)
|
||||
link = MemberTeamLinkCreate(member_id=member.id, rank=MemberRank.TEAM_MEMBER)
|
||||
MemberTeamLink.create(session=db, create_obj=link, team=team)
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/members/{member.id}",
|
||||
headers=event_user_token_headers.headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(member.id)
|
||||
assert content["name"] == member.name
|
||||
assert content["contact"] == member.contact
|
||||
assert content["scouting_id"] == member.scouting_id
|
||||
assert content["comment"] == member.comment
|
||||
assert content["allergy"] == member.allergy
|
||||
assert content["birthday_at"] == member.birthday_at
|
||||
assert content["canceled_at"] == member.canceled_at
|
||||
assert content["canceled_reason"] == member.canceled_reason
|
||||
|
||||
|
||||
def test_read_member_own_user(client: TestClient, db: Session) -> None:
|
||||
member = create_random_member(db)
|
||||
|
||||
user = create_random_user(db)
|
||||
user.member_id = member.id
|
||||
db.add(user)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/members/{member.id}",
|
||||
headers=authentication_token_from_user(client=client, db=db, user=user),
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(member.id)
|
||||
assert content["name"] == member.name
|
||||
assert content["contact"] == member.contact
|
||||
assert content["scouting_id"] == member.scouting_id
|
||||
assert content["comment"] == member.comment
|
||||
assert content["allergy"] == member.allergy
|
||||
assert content["birthday_at"] == member.birthday_at
|
||||
assert content["canceled_at"] == member.canceled_at
|
||||
assert content["canceled_reason"] == member.canceled_reason
|
||||
|
||||
|
||||
def test_read_member_not_found(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/members/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Member not found"
|
||||
|
||||
|
||||
def test_read_member_no_permission(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
member = create_random_member(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/members/{member.id}",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
assert response.json()["detail"] == "Not enough permissions"
|
||||
|
||||
|
||||
def test_read_members(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
create_random_member(db)
|
||||
create_random_member(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/members/",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
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_members_event_user(client: TestClient, event_user_token_headers: EventUserHeader, db: Session) -> None:
|
||||
team = create_random_team(db, event=event_user_token_headers.event)
|
||||
|
||||
member = create_random_member(db)
|
||||
link = MemberTeamLinkCreate(member_id=member.id, rank=MemberRank.TEAM_MEMBER)
|
||||
MemberTeamLink.create(session=db, create_obj=link, team=team)
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/members/",
|
||||
headers=event_user_token_headers.headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
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_read_members_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
create_random_member(db)
|
||||
create_random_member(db)
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/divisions/",
|
||||
headers=normal_user_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert "count" in content
|
||||
assert content["count"] == 0
|
||||
assert "data" in content
|
||||
assert isinstance(content["data"], list)
|
||||
assert len(content["data"]) == 0
|
||||
|
||||
|
||||
def test_read_members_self_created(client: TestClient, db: Session) -> None:
|
||||
user = create_random_user(db)
|
||||
member = create_random_member(db)
|
||||
member.created_by = user.id
|
||||
db.add(member)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/members/",
|
||||
headers=authentication_token_from_user(client=client, db=db, user=user),
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
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_member(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
member = create_random_member(db)
|
||||
member_id = member.id
|
||||
data = {
|
||||
"name": "Updated name",
|
||||
"contact": "Updated contact",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/members/{member_id}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["id"] == str(member_id)
|
||||
assert content["name"] == data["name"]
|
||||
assert content["contact"] == data["contact"]
|
||||
|
||||
member_query = select(Member).where(Member.id == member_id)
|
||||
member_db = db.exec(member_query).first()
|
||||
assert member_db
|
||||
db.refresh(member_db)
|
||||
|
||||
assert member_db.name == data["name"]
|
||||
assert member_db.contact == data["contact"]
|
||||
assert content["scouting_id"] == member_db.scouting_id
|
||||
assert content["comment"] == member_db.comment
|
||||
assert content["allergy"] == member_db.allergy
|
||||
assert content["birthday_at"] == member_db.birthday_at
|
||||
assert content["canceled_at"] == member_db.canceled_at
|
||||
assert content["canceled_reason"] == member_db.canceled_reason
|
||||
|
||||
|
||||
def test_delete_member(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
member = create_random_member(db)
|
||||
member_id = member.id
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/members/{member_id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["message"] == "Member deleted successfully"
|
||||
|
||||
member_query = select(Member).where(Member.id == member_id)
|
||||
member_db = db.exec(member_query).first()
|
||||
assert member_db is None
|
||||
|
||||
|
||||
def test_create_team_member_link(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
member = create_random_member(db)
|
||||
|
||||
data = {
|
||||
"member_id": str(member.id),
|
||||
"rank": MemberRank.TEAM_MEMBER
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/members",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["member_id"] == str(member.id)
|
||||
assert content["team_id"] == str(team.id)
|
||||
assert content["rank"] == data["rank"]
|
||||
|
||||
|
||||
def test_read_team_member_links(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
member = create_random_member(db)
|
||||
|
||||
link_in = MemberTeamLinkCreate(member_id=member.id, rank=MemberRank.TEAM_MEMBER)
|
||||
MemberTeamLink.create(session=db, create_obj=link_in, team=team)
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/members",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert "count" in content and content["count"] >= 1
|
||||
assert "data" in content and isinstance(content["data"], list)
|
||||
assert any(link["member_id"] == str(member.id) for link in content["data"])
|
||||
|
||||
|
||||
def test_read_team_member_link(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
member = create_random_member(db)
|
||||
|
||||
link_in = MemberTeamLinkCreate(member_id=member.id, rank=MemberRank.TEAM_MEMBER)
|
||||
MemberTeamLink.create(session=db, create_obj=link_in, team=team)
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/members/{member.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["member_id"] == str(member.id)
|
||||
assert content["team_id"] == str(team.id)
|
||||
|
||||
|
||||
def test_read_team_member_link_not_found(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/members/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert response.json()["detail"] == "Member not found"
|
||||
|
||||
|
||||
def test_read_team_member_link_event_user(client: TestClient, event_user_token_headers: EventUserHeader, db: Session) -> None:
|
||||
team = create_random_team(db, event=event_user_token_headers.event)
|
||||
member = create_random_member(db)
|
||||
|
||||
link_in = MemberTeamLinkCreate(member_id=member.id, rank=MemberRank.TEAM_MEMBER)
|
||||
MemberTeamLink.create(session=db, create_obj=link_in, team=team)
|
||||
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/members/{member.id}",
|
||||
headers=event_user_token_headers.headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["member_id"] == str(member.id)
|
||||
assert content["team_id"] == str(team.id)
|
||||
|
||||
|
||||
# TODO: Add event user test
|
||||
|
||||
# TODO: Add created_by test
|
||||
|
||||
|
||||
def test_update_team_member_link(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
member = create_random_member(db)
|
||||
|
||||
link_in = MemberTeamLinkCreate(member_id=member.id, rank=MemberRank.TEAM_MEMBER)
|
||||
MemberTeamLink.create(session=db, create_obj=link_in, team=team)
|
||||
|
||||
data = {"rank": MemberRank.TEAM_LEADER}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/members/{member.id}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["rank"] == data["rank"]
|
||||
|
||||
|
||||
def test_delete_team_member_link(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
member = create_random_member(db)
|
||||
|
||||
link_in = MemberTeamLinkCreate(member_id=member.id, rank=MemberRank.TEAM_MEMBER)
|
||||
MemberTeamLink.create(session=db, create_obj=link_in, team=team)
|
||||
|
||||
response = client.delete(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/members/{member.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["message"] == "Team member link deleted successfully"
|
||||
|
||||
# Verify deletion
|
||||
response = client.get(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}/members/{member.id}",
|
||||
headers=superuser_token_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
@@ -4,26 +4,25 @@ from sqlmodel import Session, select
|
||||
|
||||
from app.core.config import settings
|
||||
from app.models.user import User
|
||||
from app.tests.utils.utils import random_email
|
||||
|
||||
|
||||
def test_create_user(client: TestClient, db: Session) -> None:
|
||||
data = {
|
||||
"email": random_email(),
|
||||
"password": "password123",
|
||||
}
|
||||
|
||||
r = client.post(
|
||||
f"{settings.API_V1_STR}/private/users/",
|
||||
json={
|
||||
"email": "pollo@listo.com",
|
||||
"password": "password123",
|
||||
"full_name": "Pollo Listo",
|
||||
},
|
||||
json=data,
|
||||
)
|
||||
|
||||
assert r.status_code == status.HTTP_200_OK
|
||||
|
||||
data = r.json()
|
||||
|
||||
# TODO: Give user role
|
||||
|
||||
user = db.exec(select(User).where(User.id == data["id"])).first()
|
||||
|
||||
assert user
|
||||
assert user.email == "pollo@listo.com"
|
||||
assert user.full_name == "Pollo Listo"
|
||||
assert user.email == data["email"]
|
||||
|
||||
@@ -17,6 +17,7 @@ def test_create_team(client: TestClient, superuser_token_headers: dict[str, str]
|
||||
event = create_random_event(db)
|
||||
data = {
|
||||
"theme_name": "Foo",
|
||||
"short_name": "1",
|
||||
"event_id": str(event.id),
|
||||
}
|
||||
response = client.post(
|
||||
@@ -27,12 +28,14 @@ def test_create_team(client: TestClient, superuser_token_headers: dict[str, str]
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["theme_name"] == data["theme_name"]
|
||||
assert content["short_name"] == data["short_name"]
|
||||
assert content["event_id"] == str(event.id)
|
||||
assert "id" in content
|
||||
|
||||
def test_create_team_without_event(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
data = {
|
||||
"theme_name": "No Event Team",
|
||||
"short_name": "2",
|
||||
}
|
||||
response = client.post(
|
||||
f"{settings.API_V1_STR}/teams/",
|
||||
@@ -46,6 +49,7 @@ def test_create_team_without_event(client: TestClient, superuser_token_headers:
|
||||
def test_create_team_with_incorrect_event(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
data = {
|
||||
"theme_name": "No Event Team",
|
||||
"short_name": "3",
|
||||
"event_id": str(uuid.uuid4()), # Non-existent event
|
||||
}
|
||||
response = client.post(
|
||||
@@ -63,6 +67,7 @@ def test_create_team_with_normal_user(
|
||||
event = create_random_event(db)
|
||||
data = {
|
||||
"theme_name": "Normal user",
|
||||
"short_name": "4",
|
||||
"event_id": str(event.id),
|
||||
}
|
||||
response = client.post(
|
||||
@@ -80,6 +85,7 @@ def test_create_team_with_event_user(
|
||||
event = event_user_token_headers.event
|
||||
data = {
|
||||
"theme_name": "Event user",
|
||||
"short_name": "5",
|
||||
"event_id": str(event.id),
|
||||
}
|
||||
response = client.post(
|
||||
@@ -90,6 +96,7 @@ def test_create_team_with_event_user(
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
content = response.json()
|
||||
assert content["theme_name"] == data["theme_name"]
|
||||
assert content["short_name"] == data["short_name"]
|
||||
assert content["event_id"] == str(event.id)
|
||||
|
||||
|
||||
@@ -99,6 +106,7 @@ def test_create_team_for_event_user(
|
||||
event = create_random_event(db)
|
||||
data = {
|
||||
"theme_name": "Other event user",
|
||||
"short_name": "6",
|
||||
"event_id": str(event.id),
|
||||
}
|
||||
response = client.post(
|
||||
@@ -120,6 +128,7 @@ def test_read_team(client: TestClient, superuser_token_headers: dict[str, str],
|
||||
content = response.json()
|
||||
assert content["id"] == str(team.id)
|
||||
assert content["theme_name"] == team.theme_name
|
||||
assert content["short_name"] == team.short_name
|
||||
assert content["event_id"] == str(team.event_id)
|
||||
|
||||
def test_read_team_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
@@ -153,6 +162,7 @@ def test_read_team_with_event_user(client: TestClient, event_user_token_headers:
|
||||
content = response.json()
|
||||
assert content["id"] == str(team.id)
|
||||
assert content["theme_name"] == team.theme_name
|
||||
assert content["short_name"] == team.short_name
|
||||
assert content["event_id"] == str(event_user_token_headers.event.id)
|
||||
|
||||
|
||||
@@ -230,7 +240,10 @@ def test_read_teams_with_event_user_team_manager(client: TestClient, db: Session
|
||||
|
||||
def test_update_team_name(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
data = {"theme_name": "Updated Team Name"}
|
||||
data = {
|
||||
"theme_name": "Updated Team Name",
|
||||
"short_name": "7",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}",
|
||||
headers=superuser_token_headers,
|
||||
@@ -240,11 +253,15 @@ def test_update_team_name(client: TestClient, superuser_token_headers: dict[str,
|
||||
content = response.json()
|
||||
assert content["id"] == str(team.id)
|
||||
assert content["theme_name"] == data["theme_name"]
|
||||
assert content["short_name"] == data["short_name"]
|
||||
assert content["event_id"] == str(team.event_id)
|
||||
|
||||
|
||||
def test_update_team_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
|
||||
data = {"theme_name": "Non-existent team"}
|
||||
data = {
|
||||
"theme_name": "Non-existent team",
|
||||
"short_name": "8",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/teams/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
@@ -256,7 +273,10 @@ def test_update_team_not_found(client: TestClient, superuser_token_headers: dict
|
||||
|
||||
def test_update_team_not_enough_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
|
||||
team = create_random_team(db)
|
||||
data = {"theme_name": "Not enough permissions team"}
|
||||
data = {
|
||||
"theme_name": "Not enough permissions team",
|
||||
"short_name": "9",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}",
|
||||
headers=normal_user_token_headers,
|
||||
@@ -268,7 +288,10 @@ def test_update_team_not_enough_permissions(client: TestClient, normal_user_toke
|
||||
|
||||
def test_update_team_name_with_event_permissions(client: TestClient, event_user_token_headers: EventUserHeader, db: Session) -> None:
|
||||
team = create_random_team(db, event=event_user_token_headers.event)
|
||||
data = {"theme_name": "Updated Team Name with Event permissions"}
|
||||
data = {
|
||||
"theme_name": "Updated Team Name with Event permissions",
|
||||
"short_name": "10",
|
||||
}
|
||||
response = client.put(
|
||||
f"{settings.API_V1_STR}/teams/{team.id}",
|
||||
headers=event_user_token_headers.headers,
|
||||
@@ -278,6 +301,7 @@ def test_update_team_name_with_event_permissions(client: TestClient, event_user_
|
||||
content = response.json()
|
||||
assert content["id"] == str(team.id)
|
||||
assert content["theme_name"] == data["theme_name"]
|
||||
assert content["short_name"] == data["short_name"]
|
||||
assert content["event_id"] == str(event_user_token_headers.event.id)
|
||||
|
||||
|
||||
@@ -328,6 +352,7 @@ def test_update_team_event_with_event_user(client: TestClient, event_user_token_
|
||||
content = response.json()
|
||||
assert content["id"] == str(team.id)
|
||||
assert content["theme_name"] == team.theme_name
|
||||
assert content["short_name"] == team.short_name
|
||||
assert content["event_id"] == str(new_event.id)
|
||||
|
||||
|
||||
|
||||
@@ -172,9 +172,8 @@ def test_retrieve_users(
|
||||
def test_update_user_me(
|
||||
client: TestClient, normal_user_token_headers: dict[str, str], db: Session
|
||||
) -> None:
|
||||
full_name = "Updated Name"
|
||||
email = random_email()
|
||||
data = {"full_name": full_name, "email": email}
|
||||
data = {"email": email}
|
||||
r = client.patch(
|
||||
f"{settings.API_V1_STR}/users/me",
|
||||
headers=normal_user_token_headers,
|
||||
@@ -183,13 +182,11 @@ def test_update_user_me(
|
||||
assert r.status_code == status.HTTP_200_OK
|
||||
updated_user = r.json()
|
||||
assert updated_user["email"] == email
|
||||
assert updated_user["full_name"] == full_name
|
||||
|
||||
user_query = select(User).where(User.email == email)
|
||||
user_db = db.exec(user_query).first()
|
||||
assert user_db
|
||||
assert user_db.email == email
|
||||
assert user_db.full_name == full_name
|
||||
|
||||
|
||||
def test_update_password_me(
|
||||
@@ -311,8 +308,7 @@ def test_update_password_me_same_password_error(
|
||||
def test_register_user(client: TestClient, db: Session) -> None:
|
||||
username = random_email()
|
||||
password = random_lower_string()
|
||||
full_name = random_lower_string()
|
||||
data = {"email": username, "password": password, "full_name": full_name}
|
||||
data = {"email": username, "password": password}
|
||||
r = client.post(
|
||||
f"{settings.API_V1_STR}/users/signup",
|
||||
json=data,
|
||||
@@ -320,23 +316,19 @@ def test_register_user(client: TestClient, db: Session) -> None:
|
||||
assert r.status_code == status.HTTP_200_OK
|
||||
created_user = r.json()
|
||||
assert created_user["email"] == username
|
||||
assert created_user["full_name"] == full_name
|
||||
|
||||
user_query = select(User).where(User.email == username)
|
||||
user_db = db.exec(user_query).first()
|
||||
assert user_db
|
||||
assert user_db.email == username
|
||||
assert user_db.full_name == full_name
|
||||
assert verify_password(password, user_db.hashed_password)
|
||||
|
||||
|
||||
def test_register_user_already_exists_error(client: TestClient) -> None:
|
||||
password = random_lower_string()
|
||||
full_name = random_lower_string()
|
||||
data = {
|
||||
"email": settings.FIRST_SUPERUSER,
|
||||
"password": password,
|
||||
"full_name": full_name,
|
||||
}
|
||||
r = client.post(
|
||||
f"{settings.API_V1_STR}/users/signup",
|
||||
@@ -346,45 +338,6 @@ def test_register_user_already_exists_error(client: TestClient) -> None:
|
||||
assert r.json()["detail"] == "The user with this email already exists in the system"
|
||||
|
||||
|
||||
def test_update_user(
|
||||
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
||||
) -> None:
|
||||
username = random_email()
|
||||
password = random_lower_string()
|
||||
user_in = UserCreate(email=username, password=password)
|
||||
user = User.create(session=db, create_obj=user_in)
|
||||
|
||||
data = {"full_name": "Updated_full_name"}
|
||||
r = client.patch(
|
||||
f"{settings.API_V1_STR}/users/{user.id}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert r.status_code == status.HTTP_200_OK
|
||||
updated_user = r.json()
|
||||
|
||||
assert updated_user["full_name"] == "Updated_full_name"
|
||||
|
||||
user_query = select(User).where(User.email == username)
|
||||
user_db = db.exec(user_query).first()
|
||||
db.refresh(user_db)
|
||||
assert user_db
|
||||
assert user_db.full_name == "Updated_full_name"
|
||||
|
||||
|
||||
def test_update_user_not_exists(
|
||||
client: TestClient, superuser_token_headers: dict[str, str]
|
||||
) -> None:
|
||||
data = {"full_name": "Updated_full_name"}
|
||||
r = client.patch(
|
||||
f"{settings.API_V1_STR}/users/{uuid.uuid4()}",
|
||||
headers=superuser_token_headers,
|
||||
json=data,
|
||||
)
|
||||
assert r.status_code == status.HTTP_404_NOT_FOUND
|
||||
assert r.json()["detail"] == "The user with this id does not exist in the system"
|
||||
|
||||
|
||||
def test_update_user_email_exists(
|
||||
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
||||
) -> None:
|
||||
|
||||
@@ -19,8 +19,8 @@ def db() -> Generator[Session, None, None]:
|
||||
with Session(engine) as session:
|
||||
init_db(session)
|
||||
yield session
|
||||
statement = delete(User)
|
||||
session.execute(statement)
|
||||
# statement = delete(User)
|
||||
# session.execute(statement)
|
||||
session.commit()
|
||||
|
||||
|
||||
|
||||
12
backend/app/tests/utils/association.py
Normal file
12
backend/app/tests/utils/association.py
Normal file
@@ -0,0 +1,12 @@
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.models.association import Association, AssociationCreate
|
||||
from app.tests.utils.utils import random_lower_string
|
||||
|
||||
|
||||
def create_random_association(db: Session, name: str = None) -> Association:
|
||||
if not name:
|
||||
name = random_lower_string()
|
||||
|
||||
association_in = AssociationCreate(name=name)
|
||||
return Association.create(session=db, create_obj=association_in)
|
||||
17
backend/app/tests/utils/division.py
Normal file
17
backend/app/tests/utils/division.py
Normal file
@@ -0,0 +1,17 @@
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.models.association import Association
|
||||
from app.models.division import Division, DivisionCreate
|
||||
from app.tests.utils.association import create_random_association
|
||||
from app.tests.utils.utils import random_lower_string
|
||||
|
||||
|
||||
def create_random_division(db: Session, name: str = None, association: Association = None) -> Division:
|
||||
if not name:
|
||||
name = random_lower_string()
|
||||
|
||||
if not association:
|
||||
association = create_random_association(db)
|
||||
|
||||
division_in = DivisionCreate(name=name, association_id=association.id)
|
||||
return Division.create(session=db, create_obj=division_in)
|
||||
19
backend/app/tests/utils/member.py
Normal file
19
backend/app/tests/utils/member.py
Normal file
@@ -0,0 +1,19 @@
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.models.team import Team
|
||||
from app.models.member import Member, MemberCreate, MemberTeamLink, MemberTeamLinkCreate, MemberRank
|
||||
from app.tests.utils.utils import random_lower_string
|
||||
|
||||
|
||||
def create_random_member(db: Session) -> Member:
|
||||
member_in = MemberCreate(
|
||||
name=random_lower_string(),
|
||||
contact=random_lower_string(),
|
||||
scouting_id=random_lower_string(),
|
||||
comment=random_lower_string(),
|
||||
allergy=random_lower_string(),
|
||||
# birthday_at=random_datetime(),
|
||||
# canceled_at=random_datetime(),
|
||||
canceled_reason=random_lower_string(),
|
||||
)
|
||||
return Member.create(session=db, create_obj=member_in)
|
||||
@@ -1,3 +1,5 @@
|
||||
import random
|
||||
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.models.event import Event
|
||||
@@ -7,12 +9,16 @@ from app.tests.utils.event import create_random_event
|
||||
from app.tests.utils.utils import random_lower_string
|
||||
|
||||
|
||||
def random_short_name() -> str:
|
||||
return str(random.Random().randrange(1, 200))
|
||||
|
||||
def create_random_team(db: Session, event: Event | None = None) -> Team:
|
||||
name = random_lower_string()
|
||||
short_name = random_short_name()
|
||||
|
||||
if not event:
|
||||
event = create_random_event(db)
|
||||
|
||||
team_in = TeamCreate(theme_name=name, event_id=event.id)
|
||||
team_in = TeamCreate(theme_name=name, short_name=short_name, event_id=event.id)
|
||||
team = Team.create(session=db, create_obj=team_in)
|
||||
return team
|
||||
|
||||
Reference in New Issue
Block a user