Add base tests for post

.idea/league-manager.iml

@@ -3,8 +3,9 @@
   <component name="NewModuleRootManager">
     <content url="file://$MODULE_DIR$">
       <sourceFolder url="file://$MODULE_DIR$/backend" isTestSource="false" />
+      <excludeFolder url="file://$MODULE_DIR$/.venv" />
     </content>
-    <orderEntry type="jdk" jdkName="Python 3.13" jdkType="Python SDK" />
+    <orderEntry type="jdk" jdkName="Python 3.14 (RSW_puntensysteem)" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
   <component name="PyDocumentationSettings">

.idea/misc.xml

@@ -3,5 +3,5 @@
   <component name="Black">
     <option name="sdkName" value="Python 3.13" />
   </component>
-  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.13" project-jdk-type="Python SDK" />
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.14 (RSW_puntensysteem)" project-jdk-type="Python SDK" />
 </project>

backend/app/models/base.py

@@ -62,6 +62,11 @@ class ApiTags(DocumentedStrEnum):
     SERIES = "Series"
 
 
+class VisitedCountType(DocumentedStrEnum):
+    NONE = auto_enum()
+    ONE_VISIT = auto_enum()
+    EACH_VISIT = auto_enum()
+
 # endregion
 
 

backend/app/models/mixin.py

@@ -6,7 +6,7 @@ from sqlmodel import (
     Field,
 )
 
-from .base import RowId as RowIdType
+from .base import RowId as RowIdType, VisitedCountType
 from ..core.config import settings
 
 
@@ -120,3 +120,28 @@ class Birthday(BaseModel):
 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")
+
+
+class QuestionInfo(BaseModel):
+    question_file: str | None = Field(default=None, nullable=True, max_length=255)
+    answer_file: str | None = Field(default=None, nullable=True, max_length=255)
+
+
+class TeamAmmounts(BaseModel):
+    min_teams: int = Field(default=None, nullable=True, ge=1)
+    max_teams: int = Field(default=None, nullable=True, ge=1)
+
+
+class MaxPoints(BaseModel):
+    max_points: int = Field(default=None, nullable=True, ge=0)
+
+
+class VisitedPoints(BaseModel):
+    visited_points: int | None = Field(
+        default=None,
+        ge=0,
+        description="Visited points for this place, None will disable this function",
+    )
+    visited_count_type: VisitedCountType = Field(
+        default=VisitedCountType.NONE,
+    )

backend/app/models/post.py

@@ -0,0 +1,206 @@
+# app/models/post.py
+from typing import TYPE_CHECKING
+
+from sqlmodel import Field, Relationship, Session, select
+
+from . import mixin
+from .base import BaseSQLModel, RowId, DocumentedStrEnum, DocumentedIntFlag, auto_enum
+from .user import PermissionRight, User
+
+if TYPE_CHECKING:
+    from .team import Team
+
+# region # Post ##############################################################
+
+class PostType(DocumentedIntFlag):
+    POINTS = auto_enum() #allows for the ability to get points
+    OPTION = auto_enum() #allows groups to enter waiting list
+    POINTS_OPTIONS = POINTS | POINTS
+
+    VISITED = auto_enum() #only mark if a team has visited this post
+    CORRECT = auto_enum() #possible to fill in if correct (get max points)
+
+
+class ReplayType(DocumentedStrEnum):
+    NOT_POSSIBLE = auto_enum()
+    ONLY_WHEN_NO_POINTS = auto_enum()
+    YES_BUT_FIRST_POINTS_COUNT = auto_enum()
+    YES_BUT_LAST_POINTS_COUNT = auto_enum()
+    YES_BUT_HIGHEST_POINTS_COUNT = auto_enum()
+    ALWAYS_AS_NEW = auto_enum()  # add points together
+
+# ##############################################################################
+
+# Shared properties
+class PostUserLinkBase(BaseSQLModel):
+    rights: PermissionRight = Field(default=PermissionRight.READ, nullable=False)
+
+# Properties to receive via API on creation
+class PostUserLinkCreate(PostUserLinkBase):
+    user_id: RowId = Field(nullable=False)
+
+# Properties to receive via API on update, all are optional
+class PostUserLinkUpdate(PostUserLinkBase):
+    pass
+
+# Database model (link tussen post en user)
+class PostUserLink(PostUserLinkBase, table=True):
+    post_id: RowId = Field(
+        foreign_key="post.id",
+        primary_key=True,
+        nullable=False,
+        ondelete="CASCADE",
+    )
+    user_id: RowId = Field(
+        foreign_key="user.id",
+        primary_key=True,
+        nullable=False,
+        ondelete="CASCADE",
+    )
+
+    post: "Post" = Relationship(back_populates="user_links")
+    user: "User" = Relationship(back_populates="post_links")
+
+# Properties to return via API
+class PostUserLinkPublic(PostUserLinkBase):
+    user_id: RowId
+    post_id: RowId
+
+class PostUserLinksPublic(BaseSQLModel):
+    data: list[PostUserLinkPublic]
+    count: int
+
+# Shared properties
+class PostBase(
+    #common
+    mixin.Name, #post name
+    mixin.Contact, #name contact person
+    mixin.IsActive,  #scouting active (unused)
+
+    #post specific
+    mixin.ShortName, #shortname for post
+
+    mixin.MaxPoints, #max obtainable points
+    mixin.VisitedPoints, #points obtained for visiting post
+
+    mixin.TeamAmmounts, #teams, min&max teams required
+
+    mixin.QuestionInfo, #field for questions
+
+    BaseSQLModel,
+):
+    post_type: PostType = Field(default=PostType.POINTS, nullable=False)
+    replay: ReplayType = Field(default=ReplayType.NOT_POSSIBLE, nullable=False)
+
+# Properties to receive via API on creation
+class PostCreate(PostBase):
+    pass
+
+# Properties to receive via API on update, all are optional
+class PostUpdate(mixin.ShortNameUpdate, PostBase):
+    post_type: PostType | None = Field(default=None, nullable=True)  # type: ignore
+    replay: ReplayType | None = Field(default=None, nullable=True)  # type: ignore
+
+# Database model
+class Post(mixin.RowId, PostBase, table=True):
+
+    user_links: list["PostUserLink"] = Relationship(back_populates="post", cascade_delete=True)
+    #teams: list["Team"] = Relationship(back_populates="post", cascade_delete=True)
+
+    @classmethod
+    def create(cls, *, session: Session, create_obj: PostCreate) -> "Post":
+        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: "Post", in_obj: PostUpdate) -> "Post":
+        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
+
+    def get_user_link(self, user: User) -> "PostUserLink | None":
+        return next((link for link in self.user_links if link.user == user), None)
+
+    def add_user(
+        self,
+        user: User,
+        rights: PermissionRight = PermissionRight.READ,
+        *,
+        session: Session,
+    ) -> "PostUserLink | None":
+        existing = self.get_user_link(user=user)
+        if existing is None:
+            new_link = PostUserLink(post=self, user=user, rights=rights)
+            self.user_links.append(new_link)
+            session.add(new_link)
+            session.commit()
+            return new_link
+        return None
+
+    def update_user(
+        self,
+        user: User,
+        rights: PermissionRight = PermissionRight.READ,
+        *,
+        session: Session,
+    ) -> "PostUserLink | None":
+        link = self.get_user_link(user=user)
+        if link:
+            link.rights = rights
+            session.add(link)
+            session.commit()
+            return link
+        return None
+
+    def remove_user(self, user: User, *, session: Session) -> None:
+        link = self.get_user_link(user=user)
+        if link:
+            session.delete(link)
+            session.commit()
+
+    def user_has_rights(
+        self,
+        user: User,
+        rights: PermissionRight | None = None,
+    ) -> bool:
+        return any(
+            (
+                link.user == user
+                and link.rights
+                and (not rights or (link.rights & rights) == rights)
+            )
+            for link in self.user_links
+        )
+
+    def user_has_right(
+        self,
+        user: User,
+        rights: PermissionRight | None = None,
+    ) -> bool:
+        return any(
+            (
+                link.user == user
+                and link.rights
+                and (not rights or (link.rights & rights))
+            )
+            for link in self.user_links
+        )
+
+
+# API output models
+class PostPublic(mixin.RowIdPublic, PostBase):
+    pass
+
+
+class PostsPublic(BaseSQLModel):
+    data: list[PostPublic]
+    count: int
+
+# endregion

backend/app/models/serie.py

@@ -0,0 +1,173 @@
+# app/models/serie.py
+from typing import TYPE_CHECKING
+
+from sqlmodel import Field, Relationship, Session, select
+
+from . import mixin
+from .base import BaseSQLModel, RowId
+from .user import PermissionRight, User
+
+if TYPE_CHECKING:
+    from .team import Team
+
+# region # Serie ##############################################################
+
+# Shared properties
+class SerieUserLinkBase(BaseSQLModel):
+    rights: PermissionRight = Field(default=PermissionRight.READ, nullable=False)
+
+# Properties to receive via API on creation
+class SerieUserLinkCreate(SerieUserLinkBase):
+    user_id: RowId = Field(nullable=False)
+
+# Properties to receive via API on update, all are optional
+class SerieUserLinkUpdate(SerieUserLinkBase):
+    pass
+
+# Database model (link tussen serie en user)
+class SerieUserLink(SerieUserLinkBase, table=True):
+    serie_id: RowId = Field(
+        foreign_key="serie.id",
+        primary_key=True,
+        nullable=False,
+        ondelete="CASCADE",
+    )
+    user_id: RowId = Field(
+        foreign_key="user.id",
+        primary_key=True,
+        nullable=False,
+        ondelete="CASCADE",
+    )
+
+    serie: "Serie" = Relationship(back_populates="user_links")
+    user: "User" = Relationship(back_populates="serie_links")
+
+# Properties to return via API
+class SerieUserLinkPublic(SerieUserLinkBase):
+    user_id: RowId
+    serie_id: RowId
+
+class SerieUserLinksPublic(BaseSQLModel):
+    data: list[SerieUserLinkPublic]
+    count: int
+
+# Shared properties
+class SerieBase(
+    mixin.Name,
+    mixin.Contact,
+    mixin.IsActive,
+    BaseSQLModel,
+):
+    pass
+
+# Properties to receive via API on creation
+class SerieCreate(SerieBase):
+    pass
+
+# Properties to receive via API on update, all are optional
+class SerieUpdate(SerieBase):
+    pass
+
+# Database model
+class Serie(mixin.RowId, SerieBase, table=True):
+
+    user_links: list["SerieUserLink"] = Relationship(back_populates="serie", cascade_delete=True)
+    #teams: list["Team"] = Relationship(back_populates="serie", cascade_delete=True)
+
+    @classmethod
+    def create(cls, *, session: Session, create_obj: SerieCreate) -> "Serie":
+        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: "Serie", in_obj: SerieUpdate) -> "Serie":
+        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
+
+    def get_user_link(self, user: User) -> "SerieUserLink | None":
+        return next((link for link in self.user_links if link.user == user), None)
+
+    def add_user(
+        self,
+        user: User,
+        rights: PermissionRight = PermissionRight.READ,
+        *,
+        session: Session,
+    ) -> "SerieUserLink | None":
+        existing = self.get_user_link(user=user)
+        if existing is None:
+            new_link = SerieUserLink(serie=self, user=user, rights=rights)
+            self.user_links.append(new_link)
+            session.add(new_link)
+            session.commit()
+            return new_link
+        return None #TODO aanpassen??
+
+    def update_user(
+        self,
+        user: User,
+        rights: PermissionRight = PermissionRight.READ,
+        *,
+        session: Session,
+    ) -> "SerieUserLink | None":
+        link = self.get_user_link(user=user)
+        if link:
+            link.rights = rights
+            session.add(link)
+            session.commit()
+            return link
+        return None
+
+    def remove_user(self, user: User, *, session: Session) -> None:
+        link = self.get_user_link(user=user)
+        if link:
+            session.delete(link)
+            session.commit()
+
+    def user_has_rights(
+        self,
+        user: User,
+        rights: PermissionRight | None = None,
+    ) -> bool:
+        return any(
+            (
+                link.user == user
+                and link.rights
+                and (not rights or (link.rights & rights) == rights)
+            )
+            for link in self.user_links
+        )
+
+    def user_has_right(
+        self,
+        user: User,
+        rights: PermissionRight | None = None,
+    ) -> bool:
+        return any(
+            (
+                link.user == user
+                and link.rights
+                and (not rights or (link.rights & rights))
+            )
+            for link in self.user_links
+        )
+
+
+# API output models
+class SeriePublic(mixin.RowIdPublic, SerieBase):
+    pass
+
+
+class SeriesPublic(BaseSQLModel):
+    data: list[SeriePublic]
+    count: int
+
+# endregion

backend/app/models/user.py

@@ -33,6 +33,7 @@ class PermissionModule(DocumentedStrEnum):
     DIVISION = auto_enum()
     MEMBER = auto_enum()
     SERIE = auto_enum()
+    POST = auto_enum()
 
 
 class PermissionPart(DocumentedStrEnum):

backend/app/tests/api/routes/test_post.py

@@ -0,0 +1,213 @@
+import uuid
+
+from fastapi import status
+from fastapi.testclient import TestClient
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.models.base import VisitedCountType
+from app.tests.utils.post import create_random_post
+
+from app.models.post import PostType, ReplayType
+
+
+def test_create_post(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
+    data = {
+        "post_type": PostType.POINTS | PostType.VISITED | PostType.OPTION,
+        "replay": ReplayType.NOT_POSSIBLE,
+
+        "name": "Post 1",
+        "short_name": "1",
+
+        "contact": "Rick",
+        "description": "Post met renspel",
+
+        "question_file": None,
+        "answer_file": None,
+
+        "max_points": 5,
+        "visited_points": 5,
+        "visited_count_type": VisitedCountType.ONE_VISIT,
+
+        "min_teams": 1,
+        "max_teams": 2,
+    }
+    response = client.post(
+        f"{settings.API_V1_STR}/posts/",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert response.status_code == status.HTTP_200_OK
+    content = response.json()
+    assert content["post_type"] == data["post_type"]
+    assert content["replay"] == data["replay"]
+    assert content["name"] == data["name"]
+    assert content["short_name"] == data["short_name"]
+    assert content["contact"] == data["contact"]
+    assert content["description"] == data["description"]
+    assert content["question_file"] == data["question_file"]
+    assert content["answer_file"] == data["answer_file"]
+    assert content["max_points"] == data["max_points"]
+    assert content["visited_points"] == data["visited_points"]
+    assert content["visited_count_type"] == data["visited_count_type"]
+    assert content["min_teams"] == data["min_teams"]
+    assert content["max_teams"] == data["max_teams"]
+    assert "id" in content
+
+
+def test_create_post_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
+    data = {
+        "post_type": PostType.POINTS | PostType.VISITED | PostType.OPTION,
+        "replay": ReplayType.NOT_POSSIBLE,
+        "name": "No permissions",
+        "short_name": "No perm",
+    }
+    response = client.post(
+        f"{settings.API_V1_STR}/posts/",
+        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_post(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    post = create_random_post(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/posts/{post.id}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == status.HTTP_200_OK
+    content = response.json()
+    assert content["id"] == str(post.id)
+    assert content["name"] == post.name
+    assert content["contact"] == post.contact
+
+
+def test_read_post_not_found(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
+    response = client.get(
+        f"{settings.API_V1_STR}/posts/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.json()["detail"] == "Post not found"
+
+
+def test_read_post_no_permission(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
+    post = create_random_post(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/posts/{post.id}",
+        headers=normal_user_token_headers,
+    )
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert response.json()["detail"] == "Not enough permissions"
+
+
+def test_read_posts(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
+    create_random_post(db)
+    create_random_post(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/posts/",
+        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_posts_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
+    create_random_post(db)
+    create_random_post(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/posts/",
+        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_post(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
+    post = create_random_post(db)
+    data = {
+        "name": "Updated name",
+        "short_name": "4",
+    }
+    response = client.put(
+        f"{settings.API_V1_STR}/posts/{post.id}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert response.status_code == status.HTTP_200_OK
+    content = response.json()
+    assert content["id"] == str(post.id)
+    assert content["name"] == data["name"]
+    assert content["short_name"] == data["short_name"]
+
+
+def test_update_post_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
+    data = {
+        "name": "Not found",
+        "short_name": "5",
+    }
+    response = client.put(
+        f"{settings.API_V1_STR}/posts/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.json()["detail"] == "Post not found"
+
+
+def test_update_post_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
+    post = create_random_post(db)
+    data = {
+        "name": "No permissions",
+        "short_name": "6",
+    }
+    response = client.put(
+        f"{settings.API_V1_STR}/posts/{post.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_post(client: TestClient, superuser_token_headers: dict[str, str], db: Session) -> None:
+    post = create_random_post(db)
+    response = client.delete(
+        f"{settings.API_V1_STR}/posts/{post.id}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["message"] == "Post deleted successfully"
+
+
+def test_delete_post_not_found(client: TestClient, superuser_token_headers: dict[str, str]) -> None:
+    response = client.delete(
+        f"{settings.API_V1_STR}/posts/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.json()["detail"] == "Post not found"
+
+
+def test_delete_post_no_permissions(client: TestClient, normal_user_token_headers: dict[str, str], db: Session) -> None:
+    post = create_random_post(db)
+    response = client.delete(
+        f"{settings.API_V1_STR}/posts/{post.id}",
+        headers=normal_user_token_headers,
+    )
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert response.json()["detail"] == "Not enough permissions"

backend/app/tests/utils/post.py

@@ -0,0 +1,12 @@
+from sqlmodel import Session
+
+from app.models.post import Post, PostCreate
+from app.tests.utils.utils import random_lower_string
+
+
+def create_random_post(db: Session, name: str = None) -> Post:
+    if not name:
+        name = random_lower_string()
+
+    post_in = PostCreate(name=name)
+    return Post.create(session=db, create_obj=post_in)