diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index 194512a4..4791009c 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -28,6 +28,7 @@ ENV DOCUMENT_BASE_PATH="/workspaces/WebWebWeb/test-assets/documents" ENV ALBUM_BASE_PATH="/workspaces/WebWebWeb/test-assets/albums" ENV ASSETS_BASE_PATH="/workspaces/WebWebWeb/test-assets/assets" ENV MOOSE_GAME_TOKEN="sad_secret_key" +ENV ENCLOSE_MOOSE_TOKEN="happy_secret_key" RUN mkdir -p "$DOCUMENT_BASE_PATH" \ "$ALBUM_BASE_PATH" \ diff --git a/.github/workflows/custom-tests-v2.yml b/.github/workflows/custom-tests-v2.yml index 829b4e8f..b9e15637 100644 --- a/.github/workflows/custom-tests-v2.yml +++ b/.github/workflows/custom-tests-v2.yml @@ -81,4 +81,5 @@ jobs: TEST_REDIS_URL: redis://localhost:6379/0 ENVIRONMENT: "testing" USER_MANAGER_SECRET: "test_secret" + ENCLOSE_MOOSE_TOKEN: "happy_secret_key" run: pytest -v diff --git a/api_schemas/enclose_moose_level_schema.py b/api_schemas/enclose_moose_level_schema.py new file mode 100644 index 00000000..b6ddd366 --- /dev/null +++ b/api_schemas/enclose_moose_level_schema.py @@ -0,0 +1,55 @@ +from pydantic import Field, model_validator +from datetime import datetime, date +from zoneinfo import ZoneInfo +from api_schemas.base_schema import BaseSchema +from api_schemas.enclose_moose_submission_schema import EncloseMooseSubmissionRead + + +class EncloseMooseLevelRead(BaseSchema): + level_id: int + release_date: date + day_index: int | None + name_sv: str + name_en: str + + encoded_grid: str + wall_budget: int + + show_spoilers: bool = Field(default=False, exclude=True) + + optimal_score: int | None = None + optimal_solution: set[int] | None = None + optimal_is_unique: bool | None = None + + player_submission: EncloseMooseSubmissionRead | None = None + score_distribution: dict[int, int] | None = None + + @model_validator(mode="after") + def redact_spoilers(self): + if not self.show_spoilers: + self.optimal_score = None + self.optimal_solution = None + self.optimal_is_unique = None + self.score_distribution = None + + return self + + +class EncloseMooseLevelCreate(BaseSchema): + release_date: date = Field(default_factory=lambda: datetime.now(ZoneInfo("Europe/Stockholm")).date()) + day_index: int | None = None + name_sv: str + name_en: str + + encoded_grid: str + wall_budget: int + + +class EncloseMooseLevelUpdate(BaseSchema): + release_date: date | None = None + day_index: int | None = None + name_sv: str | None = None + name_en: str | None = None + + encoded_grid: str | None = None + wall_budget: int | None = None diff --git a/api_schemas/enclose_moose_submission_schema.py b/api_schemas/enclose_moose_submission_schema.py new file mode 100644 index 00000000..48e05e25 --- /dev/null +++ b/api_schemas/enclose_moose_submission_schema.py @@ -0,0 +1,15 @@ +from api_schemas.base_schema import BaseSchema +from helpers.types import datetime + + +class EncloseMooseSubmissionRead(BaseSchema): + level_id: int + submission_time: datetime + + player_id: int + player_score: int + player_solution: set[int] + + +class EncloseMooseSubmissionCreate(BaseSchema): + player_solution: set[int] diff --git a/db_models/enclose_moose_level_model.py b/db_models/enclose_moose_level_model.py new file mode 100644 index 00000000..beeb11ca --- /dev/null +++ b/db_models/enclose_moose_level_model.py @@ -0,0 +1,38 @@ +from datetime import date +from helpers.constants import MAX_ENCLOSE_LEVEL_NAME, MAX_ENCLOSE_GRID +from .base_model import BaseModel_DB +from sqlalchemy.orm import relationship, mapped_column, Mapped +from typing import TYPE_CHECKING, Optional +from sqlalchemy import String, Date, JSON + +if TYPE_CHECKING: + from db_models.enclose_moose_submission_model import EncloseMooseSubmission_DB + + +class EncloseMooseLevel_DB(BaseModel_DB): + __tablename__ = "enclose_moose_level_table" + + level_id: Mapped[int] = mapped_column(primary_key=True, init=False) + release_date: Mapped[Optional[date]] = mapped_column(Date) + day_index: Mapped[Optional[int]] = mapped_column() + name_sv: Mapped[str] = mapped_column(String(MAX_ENCLOSE_LEVEL_NAME)) + name_en: Mapped[str] = mapped_column(String(MAX_ENCLOSE_LEVEL_NAME)) + + encoded_grid: Mapped[str] = mapped_column(String(MAX_ENCLOSE_GRID)) + wall_budget: Mapped[int] = mapped_column() + + optimal_score: Mapped[int] = mapped_column() + optimal_solution: Mapped[list[int]] = mapped_column(JSON) + optimal_is_unique: Mapped[Optional[bool]] = mapped_column() + + submissions: Mapped[list["EncloseMooseSubmission_DB"]] = relationship( + back_populates="level", cascade="all, delete-orphan", init=False + ) + + @property + def score_distribution(self): + score_distribution: dict[int, int] = {} + for submission in self.submissions: + score_distribution[submission.player_score] = score_distribution.get(submission.player_score, 0) + 1 + + return score_distribution diff --git a/db_models/enclose_moose_submission_model.py b/db_models/enclose_moose_submission_model.py new file mode 100644 index 00000000..dae186bf --- /dev/null +++ b/db_models/enclose_moose_submission_model.py @@ -0,0 +1,27 @@ +from .base_model import BaseModel_DB +from sqlalchemy.orm import relationship, mapped_column, Mapped +from sqlalchemy import ForeignKey, JSON +from typing import TYPE_CHECKING +from helpers.types import datetime_utc + +if TYPE_CHECKING: + from db_models.enclose_moose_level_model import EncloseMooseLevel_DB + + +class EncloseMooseSubmission_DB(BaseModel_DB): + __tablename__ = "enclose_moose_submission_table" + + level_id: Mapped[int] = mapped_column( + ForeignKey("enclose_moose_level_table.level_id", ondelete="CASCADE"), + primary_key=True, + ) + submission_time: Mapped[datetime_utc] = mapped_column() + + player_id: Mapped[int] = mapped_column( + ForeignKey("user_table.id", ondelete="CASCADE"), + primary_key=True, + ) + player_score: Mapped[int] = mapped_column() + player_solution: Mapped[list[int]] = mapped_column(JSON) + + level: Mapped["EncloseMooseLevel_DB"] = relationship(back_populates="submissions", init=False, viewonly=True) diff --git a/helpers/constants.py b/helpers/constants.py index bc9191dc..78602b55 100644 --- a/helpers/constants.py +++ b/helpers/constants.py @@ -110,3 +110,8 @@ # Keyval, used for example for storing the links to different very important documents (e.g. reglementet) MAX_KEYVAL_KEY = 100 MAX_KEYVAL_VALUE = 1000 + + +# enclose.moose +MAX_ENCLOSE_LEVEL_NAME = 100 +MAX_ENCLOSE_GRID = 10000 diff --git a/helpers/types.py b/helpers/types.py index 34af1a1f..23aedf38 100644 --- a/helpers/types.py +++ b/helpers/types.py @@ -57,6 +57,7 @@ def force_utc(date: datetime): "Council", "RoomBookings", "Moosegame", + "EncloseMoose", "MailAlias", "GuildMeeting", "Keyvals", diff --git a/requirements.txt b/requirements.txt index 85941318..cd7149cd 100644 --- a/requirements.txt +++ b/requirements.txt @@ -36,6 +36,7 @@ Mako==1.3.8 MarkupSafe==3.0.3 numpy==2.2.4 oauthlib==3.3.1 +ortools==9.15.6755 packaging==23.2 pandas==2.2.3 passlib==1.7.4 diff --git a/routes/__init__.py b/routes/__init__.py index 5d02dc3d..d6ed3559 100644 --- a/routes/__init__.py +++ b/routes/__init__.py @@ -2,6 +2,7 @@ from .mail_alias_router import mail_alias_router from .moose_game_router import moose_game_router +from .enclose_moose_router import enclose_moose_router from .council_router import council_router from .user_router import user_router from .post_router import post_router @@ -87,6 +88,8 @@ main_router.include_router(moose_game_router, prefix="/moose-game", tags=["moose-game"]) +main_router.include_router(enclose_moose_router, prefix="/enclose-moose", tags=["enclose-moose"]) + main_router.include_router(mail_alias_router, prefix="/mail-alias", tags=["mail-alias"]) main_router.include_router(access_serve_router, prefix="/access-serve", tags=["access serve"]) diff --git a/routes/enclose_moose_router.py b/routes/enclose_moose_router.py new file mode 100644 index 00000000..9c668c6d --- /dev/null +++ b/routes/enclose_moose_router.py @@ -0,0 +1,231 @@ +import os +from datetime import datetime +from zoneinfo import ZoneInfo +from fastapi import APIRouter, HTTPException, Request +from typing import Annotated +from sqlalchemy.exc import DataError, IntegrityError +from api_schemas.enclose_moose_level_schema import ( + EncloseMooseLevelRead, + EncloseMooseLevelCreate, + EncloseMooseLevelUpdate, +) +from api_schemas.enclose_moose_submission_schema import EncloseMooseSubmissionRead, EncloseMooseSubmissionCreate +from user.permission import Permission +from database import DB_dependency +from db_models.user_model import User_DB +from db_models.enclose_moose_level_model import EncloseMooseLevel_DB +from db_models.enclose_moose_submission_model import EncloseMooseSubmission_DB +from services.enclose_moose_service import level_create, level_update, solution_submit + +ENCLOSE_MOOSE_SECRET = os.getenv("ENCLOSE_MOOSE_TOKEN") + +enclose_moose_router = APIRouter() + + +# Admin routes +@enclose_moose_router.post( + "/admin/levels", + response_model=EncloseMooseLevelRead, + dependencies=[Permission.require("manage", "EncloseMoose")], +) +def admin_create_level(data: EncloseMooseLevelCreate, db: DB_dependency): + level = level_create(data) + + db.add(level) + try: + db.commit() + except DataError: + db.rollback() + raise HTTPException(400, detail="Some string is too long") + + level.show_spoilers = True # pyright: ignore + + return level + + +@enclose_moose_router.get( + "/admin/levels/{level_id}", + response_model=EncloseMooseLevelRead, + dependencies=[Permission.require("manage", "EncloseMoose")], +) +def admin_get_level(level_id: int, db: DB_dependency): + level = db.get(EncloseMooseLevel_DB, level_id) + if level is None: + raise HTTPException(404, detail=f'No level with level_id "{level_id}" exists') + + level.show_spoilers = True # pyright: ignore + + return level + + +@enclose_moose_router.get( + "/admin/levels", + response_model=list[EncloseMooseLevelRead], + dependencies=[Permission.require("manage", "EncloseMoose")], +) +def admin_get_all_levels(db: DB_dependency): + levels = db.query(EncloseMooseLevel_DB).order_by(EncloseMooseLevel_DB.release_date).all() + + for level in levels: + level.show_spoilers = True # pyright: ignore + + return levels + + +@enclose_moose_router.patch( + "/admin/levels/{level_id}", + response_model=EncloseMooseLevelRead, + dependencies=[Permission.require("manage", "EncloseMoose")], +) +def admin_update_level(level_id: int, data: EncloseMooseLevelUpdate, db: DB_dependency): + level = db.get(EncloseMooseLevel_DB, level_id) + if level is None: + raise HTTPException(404, detail=f'No level with level_id "{level_id}" exists') + + updated_level = level_update(level, data) + try: + db.commit() + except DataError: + db.rollback() + raise HTTPException(400, detail="Some string is too long") + + updated_level.show_spoilers = True # pyright: ignore + + return updated_level + + +@enclose_moose_router.delete( + "/admin/levels/{level_id}", + response_model=EncloseMooseLevelRead, + dependencies=[Permission.require("manage", "EncloseMoose")], +) +def admin_delete_level(level_id: int, db: DB_dependency): + level = db.get(EncloseMooseLevel_DB, level_id) + if level is None: + raise HTTPException(404, detail=f'No level with level_id "{level_id}" exists') + + db.delete(level) + db.commit() + + level.show_spoilers = True # pyright: ignore + + return level + + +@enclose_moose_router.get( + "/admin/submissions/{level_id}", + response_model=list[EncloseMooseSubmissionRead], + dependencies=[Permission.require("manage", "EncloseMoose")], +) +def admin_get_all_level_submissions( + level_id: int, + db: DB_dependency, +): + submissions = db.query(EncloseMooseSubmission_DB).filter(EncloseMooseSubmission_DB.level_id == level_id).all() + + return submissions + + +# Non-admin routes +@enclose_moose_router.get("/levels/{level_id}", response_model=EncloseMooseLevelRead) +def get_level(level_id: int, me: Annotated[User_DB, Permission.member()], db: DB_dependency): + date_today = datetime.now(ZoneInfo("Europe/Stockholm")).date() + level = ( + db.query(EncloseMooseLevel_DB) + .filter(EncloseMooseLevel_DB.release_date <= date_today, EncloseMooseLevel_DB.level_id == level_id) + .one_or_none() + ) + if level is None: + raise HTTPException(404, detail=f'No level with level_id "{level_id}" exists') + + submission = db.get(EncloseMooseSubmission_DB, (level_id, me.id)) + level.player_submission = submission # pyright: ignore + level.show_spoilers = submission is not None # pyright: ignore + + return level + + +@enclose_moose_router.get("/levels", response_model=list[EncloseMooseLevelRead]) +def get_all_levels(me: Annotated[User_DB, Permission.member()], db: DB_dependency): + date_today = datetime.now(ZoneInfo("Europe/Stockholm")).date() + + results = ( + db.query(EncloseMooseLevel_DB, EncloseMooseSubmission_DB) + .outerjoin( + EncloseMooseSubmission_DB, + (EncloseMooseSubmission_DB.level_id == EncloseMooseLevel_DB.level_id) + & (EncloseMooseSubmission_DB.player_id == me.id), + ) + .filter(EncloseMooseLevel_DB.release_date <= date_today) + .order_by(EncloseMooseLevel_DB.release_date) + .all() + ) + + levels: list[EncloseMooseLevel_DB] = [] + for level, submission in results: + level.player_submission = submission + level.show_spoilers = submission is not None # pyright: ignore + levels.append(level) + + return levels + + +@enclose_moose_router.post("/submissions/{level_id}", response_model=EncloseMooseLevelRead) +def submit_solution( + level_id: int, + submission: EncloseMooseSubmissionCreate, + me: Annotated[User_DB, Permission.member()], + request: Request, + db: DB_dependency, +): + token = request.headers.get("enclose-moose-token") + if ENCLOSE_MOOSE_SECRET != token: + raise HTTPException(401, detail="Invalid enclose-moose-token") + + date_today = datetime.now(ZoneInfo("Europe/Stockholm")).date() + level = ( + db.query(EncloseMooseLevel_DB) + .filter(EncloseMooseLevel_DB.release_date <= date_today, EncloseMooseLevel_DB.level_id == level_id) + .one_or_none() + ) + if level is None: + raise HTTPException(404, detail=f'No level with level_id "{level_id}" exists') + + db_submission = solution_submit(level, submission.player_solution, me.id) + db.add(db_submission) + try: + db.commit() + except IntegrityError: + db.rollback() + raise HTTPException( + 409, + detail="The player has already submitted a solution to this level", + ) + + level.player_submission = db_submission # pyright: ignore + level.show_spoilers = True # pyright: ignore + + return level + + +@enclose_moose_router.get("/submissions/{level_id}", response_model=EncloseMooseSubmissionRead) +def get_submission( + level_id: int, + me: Annotated[User_DB, Permission.member()], + db: DB_dependency, +): + submission = db.get(EncloseMooseSubmission_DB, (level_id, me.id)) + if submission is None: + raise HTTPException(404, detail="No submission exists for this player and level") + + return submission + + +@enclose_moose_router.get("/submissions", response_model=list[EncloseMooseSubmissionRead]) +def get_all_submissions( + me: Annotated[User_DB, Permission.member()], + db: DB_dependency, +): + submissions = db.query(EncloseMooseSubmission_DB).filter(EncloseMooseSubmission_DB.player_id == me.id).all() + + return submissions diff --git a/seed.py b/seed.py index ca56f0d8..bdf61afd 100644 --- a/seed.py +++ b/seed.py @@ -218,6 +218,7 @@ def seed_permissions(db: Session, posts: list[Post_DB]): Permission(action="manage", target="Document", posts=["Buggmästare"]), Permission(action="view", target="Document", posts=["Buggmästare"]), Permission(action="manage", target="Moosegame", posts=["Buggmästare"]), + Permission(action="manage", target="EncloseMoose", posts=["Buggmästare"]), Permission(action="manage", target="UserPost", posts=["Buggmästare"]), Permission(action="view", target="GuildMeeting", posts=["Buggmästare"]), Permission(action="manage", target="GuildMeeting", posts=["Buggmästare"]), diff --git a/services/enclose_moose_service.py b/services/enclose_moose_service.py new file mode 100644 index 00000000..3b91d618 --- /dev/null +++ b/services/enclose_moose_service.py @@ -0,0 +1,301 @@ +from collections import deque +from datetime import datetime, timezone +from typing import Any +from ortools.sat.python import cp_model +from fastapi import HTTPException +from api_schemas.enclose_moose_level_schema import EncloseMooseLevelCreate, EncloseMooseLevelUpdate +from db_models.enclose_moose_level_model import EncloseMooseLevel_DB +from db_models.enclose_moose_submission_model import EncloseMooseSubmission_DB + + +class EncloseGrid: + def __init__(self, encoded_grid: str, wall_budget: int): + self.wall_budget = wall_budget + + if wall_budget < 0: + raise HTTPException(400, detail="Wall budget must not be negative") + + self.grid_string = encoded_grid.replace("\n", "") + self.N = len(self.grid_string) + + if self.N <= 1: + raise HTTPException(400, detail="Level must be bigger than one square") + + for tile in encoded_grid: + if tile not in ("\n", ".", "~", "H", "C", "G", "S") and not tile.isnumeric(): + raise HTTPException(400, detail=f'Level contained unknown encoding: "{tile}"') + + if "H" not in encoded_grid: + raise HTTPException(400, detail='Level must contain a moose ("H")') + + if "\n" not in encoded_grid: + raise HTTPException(400, detail="Level must be rectangular") + + if encoded_grid.strip("\n") != encoded_grid: + raise HTTPException(400, detail="Level must not have leading or trailing newlines") + + self.grid_width = encoded_grid.index("\n") + self.grid_height = self.N // self.grid_width + + if len(set(map(len, encoded_grid.split("\n")))) != 1: + raise HTTPException(400, detail="Level must be rectangular") + + self.moose_index = self.grid_string.index("H") + + self.portals: dict[str, set[int]] = {} + for flat_index, tile in enumerate(self.grid_string): + if tile.isnumeric(): + self.portals.setdefault(tile, set()).add(flat_index) + + def get_neighbors(self, flat_index: int): + row_index, column_index = divmod(flat_index, self.grid_width) + neighbors: set[int] = set() + if column_index != 0: + neighbors.add(flat_index - 1) + if column_index != self.grid_width - 1: + neighbors.add(flat_index + 1) + if row_index != 0: + neighbors.add(flat_index - self.grid_width) + if row_index != self.grid_height - 1: + neighbors.add(flat_index + self.grid_width) + + tile = self.grid_string[flat_index] + if tile in self.portals: + for flat_index_portal in self.portals[tile]: + if flat_index_portal != flat_index: + neighbors.add(flat_index_portal) + + return neighbors + + def is_boundary(self, flat_index: int): + row, col = divmod(flat_index, self.grid_width) + + return row == 0 or col == 0 or row == self.grid_height - 1 or col == self.grid_width - 1 + + def score_tile(self, tile: str): + bonus_score_dict = {"C": 3, "G": 10, "S": -5} + bonus_score = 1 + bonus_score_dict.get(tile, 0) + + return bonus_score + + def find_optimal_solution(self): + # Algorithm inspired by https://dynomight.substack.com/p/horse, https://blog.macuyiko.com/post/2026/solving-enclosehorse-with-cp-sat.html and Gemini + + model = cp_model.CpModel() + + w = [model.new_bool_var(f"w_{i}") for i in range(self.N)] # Whether wall is present + e = [model.new_bool_var(f"r_{i}") for i in range(self.N)] # Whether the tile is enclosed + + never_enclosed_indices: set[int] = set() + for flat_index, tile in enumerate(self.grid_string): + if tile == "~" or self.is_boundary(flat_index): + model.add(e[flat_index] == 0) # Water or boundary tiles cannot be enclosed + never_enclosed_indices.add(flat_index) + + if tile != ".": + model.add(w[flat_index] == 0) # Non-grass tiles cannot have walls + + model.add(e[flat_index] + w[flat_index] <= 1) # A tile cannot be both enclosed and have a wall + + model.add(e[self.moose_index] == 1) # Moose must be enclosed + + model.add(sum(w) <= self.wall_budget) # Enforce wall budget + + d = [model.new_int_var(0, self.N, f"d_{i}") for i in range(self.N)] # Distance from moose + model.add(d[self.moose_index] == 0) # Moose is at distance 0 from moose. + + for flat_index, tile in enumerate(self.grid_string): + if flat_index in never_enclosed_indices: + continue + + parents: list[cp_model.IntVar] = [] + for neighbor_index in self.get_neighbors(flat_index): + if self.grid_string[neighbor_index] == "~": + continue + + model.add( + e[flat_index] <= e[neighbor_index] + w[neighbor_index] + ) # If a tile is enclosed, its neighbor must either also be enclosed or have a wall + + if flat_index != self.moose_index: + p_var = model.new_bool_var( + f"p_{neighbor_index}_{flat_index}" + ) # Whether neighbor_index is parent of flat_index + parents.append(p_var) + + model.add_implication(p_var, e[neighbor_index]) # A parent must be an enclosed tile + model.add(d[flat_index] == d[neighbor_index] + 1).only_enforce_if( # pyright: ignore + p_var + ) # Distance increases to prevent cycles (isolated enclosed areas) + + if flat_index != self.moose_index: + if parents: + model.add(sum(parents) == e[flat_index]) # If a tile is enclosed, it has exactly one parent + else: + model.add(e[flat_index] == 0) # If completely surrounded by water, it cannot be enclosed + + enclosed_score = [self.score_tile(tile) * e[flat_index] for flat_index, tile in enumerate(self.grid_string)] + model.maximize(sum(enclosed_score)) + + solver = cp_model.CpSolver() + solver.parameters.max_time_in_seconds = 15 + solver.parameters.num_search_workers = 8 + + status = solver.solve(model) + # print(f"Solving took {solver.wall_time} seconds") + + if status not in (cp_model.INFEASIBLE, cp_model.OPTIMAL): + raise HTTPException( + 400, + detail=f"Level could not be solved, likely because it exceeded the time limit (solver status: {solver.status_name(status)})", + ) + + if status == cp_model.INFEASIBLE: + raise HTTPException(400, detail="Level is unsolvable") + + score = int(solver.objective_value) + wall_indices = set(i for i in range(self.N) if solver.value(w[i]) == 1) + + model.add_bool_or( + [w[i].Not() for i in wall_indices] + [w_i for i, w_i in enumerate(w) if i not in wall_indices] + ) + status2 = solver.solve(model) + if status2 == cp_model.OPTIMAL: + solution_is_unique = int(solver.objective_value) != score + elif status2 == cp_model.INFEASIBLE: + solution_is_unique = True + else: + solution_is_unique = None + + """ # Some debug visualisations + import numpy as np + + e_sol = np.array([solver.value(e[i]) for i in range(self.N)]) + + np.set_printoptions(linewidth=1000) + print("MAP") + grid: list[list[str]] = [] + for flat_index, tile in enumerate(self.grid_string): + if flat_index % self.grid_width == 0: + grid.append([]) + + if flat_index in wall_indices: + grid[-1].append("W") + else: + grid[-1].append(tile) + print(np.array(grid).reshape((self.grid_height, self.grid_width))) + + print("REGION") + print(e_sol.reshape((self.grid_height, self.grid_width))) + print("SCORES") + print( + (e_sol * np.array(list(map(self.score_tile, self.grid_string)))) + .astype(int) + .reshape((self.grid_height, self.grid_width)) + ) + """ + + return score, wall_indices, solution_is_unique + + def score_solution(self, solution: set[int]): + if len(solution) > self.wall_budget: + raise HTTPException(400, "The solution contained too many walls") + + for wall_index in solution: + if wall_index < 0 or wall_index > self.N - 1: + raise HTTPException(400, f"The solution contained a wall that was of out bounds ({wall_index})") + + if self.grid_string[wall_index] != ".": + raise HTTPException(400, f"The solution contained a wall placed on non-grass ({wall_index})") + + queue: deque[int] = deque() + visited: set[int] = set() + score = 0 + + queue.append(self.moose_index) + visited.add(self.moose_index) + + while queue: + current_index = queue.popleft() + score += self.score_tile(self.grid_string[current_index]) + + for neighbor_index in self.get_neighbors(current_index): + is_visited = neighbor_index in visited + if is_visited: + continue + + visited.add(neighbor_index) + + is_blocked = self.grid_string[neighbor_index] == "~" or neighbor_index in solution + if is_blocked: + continue + + is_escaped = self.is_boundary(neighbor_index) + if is_escaped: + raise HTTPException(400, detail="The solution does not enclose the moose") + + queue.append(neighbor_index) + + return score # len(visited) + bonus score + + +def level_create(data: EncloseMooseLevelCreate): + + grid = EncloseGrid(data.encoded_grid, data.wall_budget) + optimal_score, optimal_solution, optimal_is_unique = grid.find_optimal_solution() + + level = EncloseMooseLevel_DB( + release_date=data.release_date, + day_index=data.day_index, + name_sv=data.name_sv, + name_en=data.name_en, + encoded_grid=data.encoded_grid, + wall_budget=data.wall_budget, + optimal_score=optimal_score, + optimal_solution=list(optimal_solution), + optimal_is_unique=optimal_is_unique, + ) + + return level + + +def level_update(level: EncloseMooseLevel_DB, data: EncloseMooseLevelUpdate): + data_dump = data.model_dump(exclude_unset=True) + updates: dict[str, Any] = {} + for key, value in data_dump.items(): + if value is not None: + updates[key] = value + elif key == "day_index": # day_index is nullable + updates[key] = value + + used_encoded_grid = updates.get("encoded_grid", level.encoded_grid) + used_wall_budget = updates.get("wall_budget", level.wall_budget) + if used_encoded_grid != level.encoded_grid or used_wall_budget != level.wall_budget: + level.submissions.clear() + + grid = EncloseGrid(used_encoded_grid, used_wall_budget) + optimal_score, optimal_solution, optimal_is_unique = grid.find_optimal_solution() + + updates["optimal_score"] = optimal_score + updates["optimal_solution"] = list(optimal_solution) + updates["optimal_is_unique"] = optimal_is_unique + + for var, value in updates.items(): + setattr(level, var, value) + + return level + + +def solution_submit(level: EncloseMooseLevel_DB, solution: set[int], player_id: int): + grid = EncloseGrid(level.encoded_grid, level.wall_budget) + player_score = grid.score_solution(solution) + + db_submission = EncloseMooseSubmission_DB( + level_id=level.level_id, + submission_time=datetime.now(timezone.utc), + player_id=player_id, + player_score=player_score, + player_solution=list(solution), + ) + + return db_submission diff --git a/tests/basic_fixtures.py b/tests/basic_fixtures.py index 35a19176..e4324741 100644 --- a/tests/basic_fixtures.py +++ b/tests/basic_fixtures.py @@ -90,6 +90,7 @@ def admin_post(db_session): Permission_DB(action="view", target="Council"), Permission_DB(action="manage", target="User"), Permission_DB(action="manage", target="RoomBookings"), + Permission_DB(action="manage", target="EncloseMoose"), Permission_DB(action="view", target="UserPost"), Permission_DB(action="manage", target="UserPost"), Permission_DB(action="view", target="RoomBookings"), diff --git a/tests/conftest.py b/tests/conftest.py index 516e7c49..2d3fce2a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,6 +10,10 @@ # Ensure we're in testing mode before importing important modules os.environ["ENVIRONMENT"] = "testing" + +# Must match the value used in test_enclose_moose.py +os.environ["ENCLOSE_MOOSE_TOKEN"] = "happy_secret_key" + from main import app from database import get_db from .basic_fixtures import * diff --git a/tests/test_enclose_moose.py b/tests/test_enclose_moose.py new file mode 100644 index 00000000..327591be --- /dev/null +++ b/tests/test_enclose_moose.py @@ -0,0 +1,247 @@ +# type: ignore +import pytest +from .basic_factories import auth_headers +from datetime import datetime, UTC, timedelta + + +def create_level(client, token, **kwargs): + default_data = {"name_sv": "test", "name_en": "test", "encoded_grid": ".~.\n.H.\n~.~", "wall_budget": 4} + return client.post("/enclose-moose/admin/levels", json=default_data | kwargs, headers=auth_headers(token)) + + +def patch_level(client, token, level_id, **kwargs): + return client.patch(f"/enclose-moose/admin/levels/{level_id}", json=kwargs, headers=auth_headers(token)) + + +def delete_level(client, token, level_id): + return client.delete(f"/enclose-moose/admin/levels/{level_id}", headers=auth_headers(token)) + + +def admin_get_level(client, token, level_id): + return client.get(f"/enclose-moose/admin/levels/{level_id}", headers=auth_headers(token)) + + +def get_level(client, token, level_id): + return client.get(f"/enclose-moose/levels/{level_id}", headers=auth_headers(token)) + + +def admin_get_all_levels(client, token): + return client.get("/enclose-moose/admin/levels", headers=auth_headers(token)) + + +def get_all_levels(client, token): + return client.get("/enclose-moose/levels", headers=auth_headers(token)) + + +def submit_solution(client, token, level_id, player_solution, secret_header="happy_secret_key"): + headers = auth_headers(token) + if secret_header is not None: + headers["enclose-moose-token"] = secret_header + + body = {"player_solution": player_solution} + return client.post(f"/enclose-moose/submissions/{level_id}", json=body, headers=headers) + + +def get_submission(client, token, level_id): + return client.get(f"/enclose-moose/submissions/{level_id}", headers=auth_headers(token)) + + +def admin_get_all_level_submissions(client, token, level_id): + return client.get(f"/enclose-moose/admin/submissions/{level_id}", headers=auth_headers(token)) + + +def get_all_my_submissions(client, token): + return client.get("/enclose-moose/submissions", headers=auth_headers(token)) + + +def test_admin_manage_level(client, member_token, admin_token): + res_create_invalid = create_level(client, admin_token, encoded_grid=".~.") + assert res_create_invalid.status_code == 400 + + res_create_unsolvable = create_level(client, admin_token, wall_budget=1) + assert res_create_unsolvable.status_code == 400 + + res_create = create_level(client, admin_token) + assert res_create.status_code == 200 + released_level_id = res_create.json()["level_id"] + + res_get = admin_get_level(client, admin_token, released_level_id) + assert res_get.status_code == 200 + + res_get_all = admin_get_all_levels(client, admin_token) + assert res_get_all.status_code == 200 + assert len(res_get_all.json()) == 1 + + submit_solution(client, member_token, released_level_id, player_solution=[3, 5, 7]) + res_get_submissions = admin_get_all_level_submissions(client, admin_token, released_level_id) + assert res_get_submissions.status_code == 200 + assert len(res_get_submissions.json()) == 1 + + res_patch = patch_level(client, admin_token, released_level_id, name_sv="updated_name") + assert res_patch.status_code == 200 + assert res_patch.json()["name_sv"] == "updated_name" + + res_delete = delete_level(client, admin_token, released_level_id) + assert res_delete.status_code == 200 + + res_get = admin_get_level(client, admin_token, released_level_id) + assert res_get.status_code == 404 + + +def test_member_cannot_access_admin_routes(client, member_token, admin_token): + res_create = create_level(client, member_token) + assert res_create.status_code == 403 + + released_level_id = create_level(client, admin_token).json()["level_id"] + + res_admin_get = admin_get_level(client, member_token, released_level_id) + assert res_admin_get.status_code == 403 + + res_admin_get_all = admin_get_all_levels(client, member_token) + assert res_admin_get_all.status_code == 403 + + res_patch = patch_level(client, member_token, released_level_id, name_sv="updated_name") + assert res_patch.status_code == 403 + res_patch_get = get_level(client, member_token, released_level_id) + assert res_patch_get.json()["name_sv"] != "updated_name" + + res_del = delete_level(client, member_token, released_level_id) + assert res_del.status_code == 403 + + res_del_get = get_level(client, member_token, released_level_id) + assert res_del_get.status_code == 200 + + res_submissions = admin_get_all_level_submissions(client, member_token, released_level_id) + assert res_submissions.status_code == 403 + + +def test_levels(client, member_token, admin_token): + future_date = (datetime.now(UTC).date() + timedelta(days=2)).isoformat() + unreleased_level_id = create_level(client, admin_token, release_date=future_date).json()["level_id"] + released_level_id = create_level(client, admin_token).json()["level_id"] + + res_get_admin_unreleased = admin_get_level(client, admin_token, unreleased_level_id) + assert res_get_admin_unreleased.status_code == 200 + + res_get_admin_released = admin_get_level(client, admin_token, released_level_id) + assert res_get_admin_released.status_code == 200 + + res_get_member_unreleased = get_level(client, member_token, unreleased_level_id) + assert res_get_member_unreleased.status_code == 404 + + res_get_member_released = get_level(client, member_token, released_level_id) + assert res_get_member_released.status_code == 200 + + res_admin_get_all = admin_get_all_levels(client, admin_token) + assert res_admin_get_all.status_code == 200 + assert len(res_admin_get_all.json()) == 2 + + res_get_all_member = get_all_levels(client, member_token) + assert res_get_all_member.status_code == 200 + assert len(res_get_all_member.json()) == 1 + + +def test_submission(client, member_token, admin_token): + res_non_existent = submit_solution(client, member_token, 99999, player_solution=[3, 5, 7]) + assert res_non_existent.status_code == 404 + + res_get_non_existent = get_submission(client, member_token, 99999) + assert res_get_non_existent.status_code == 404 + + released_level_id = create_level(client, admin_token).json()["level_id"] + res_member = submit_solution(client, member_token, released_level_id, player_solution=[3, 5, 7]) + assert res_member.status_code == 200 + + res_member_invalid = submit_solution(client, member_token, released_level_id, player_solution=[3, 5]) + assert res_member_invalid.status_code == 400 + + res_invalid_token = submit_solution( + client, member_token, released_level_id, player_solution=[3, 5, 7], secret_header="I love tests!" + ) + assert res_invalid_token.status_code == 401 + + res_get = get_submission(client, member_token, released_level_id) + assert res_get.status_code == 200 + + future_date = (datetime.now(UTC).date() + timedelta(days=2)).isoformat() + unreleased_level_id = create_level(client, admin_token, release_date=future_date).json()["level_id"] + res_unreleased = submit_solution(client, member_token, unreleased_level_id, player_solution=[3, 5, 7]) + assert res_unreleased.status_code == 404 + + submit_solution(client, admin_token, released_level_id, player_solution=[3, 5, 7]) + res_get_all = get_all_my_submissions(client, member_token) + assert res_get_all.status_code == 200 + assert len(res_get_all.json()) == 1 + + +def test_submissions_clear(client, member_token, admin_token): + released_level_id = create_level(client, admin_token).json()["level_id"] + submit_solution(client, member_token, released_level_id, player_solution=[3, 5, 7]) + + patch_level(client, admin_token, released_level_id, name_sv="updated_name") + res_get_unchanged = get_submission(client, member_token, released_level_id) + assert res_get_unchanged.status_code == 200 + + patch_level(client, admin_token, released_level_id, wall_budget=10) + res_get_changed = get_submission(client, member_token, released_level_id) + assert res_get_changed.status_code == 404 + + submit_solution(client, member_token, released_level_id, player_solution=[3, 5, 7]) + delete_level(client, admin_token, released_level_id) + res_get_deleted = get_submission(client, member_token, released_level_id) + assert res_get_deleted.status_code == 404 + + +def test_non_member_cannot_access_member_routes(client, non_member_token, admin_token): + released_level_id = create_level(client, admin_token).json()["level_id"] + + res_get = get_level(client, non_member_token, released_level_id) + assert res_get.status_code == 403 + + res_get_all = get_all_levels(client, non_member_token) + assert res_get_all.status_code == 403 + + res_submit = submit_solution(client, non_member_token, released_level_id, player_solution=[3, 5, 7]) + assert res_submit.status_code == 403 + + res_get_submission = get_submission(client, non_member_token, released_level_id) + assert res_get_submission.status_code == 403 + + res_get_all_submissions = get_all_my_submissions(client, non_member_token) + assert res_get_all_submissions.status_code == 403 + + +def test_leading_newline_rejected(client, admin_token): + res_create = create_level(client, admin_token, encoded_grid="\n.~.\n.H.\n~.~") + assert res_create.status_code == 400 + + res_create_working = create_level(client, admin_token, encoded_grid=".~.\n.H.\n~.~") + assert res_create_working.status_code == 200 + res_patch = patch_level(client, admin_token, res_create_working.json()["level_id"], encoded_grid="\n.~.\n.H.\n~.~") + assert res_patch.status_code == 400 + + +def test_updates_with_none(client, admin_token): + res_create = create_level(client, admin_token, day_index=1) + assert res_create.status_code == 200 + level_id = res_create.json()["level_id"] + + # Clearable day_index + res_patch = patch_level(client, admin_token, level_id, day_index=None) + assert res_patch.status_code == 200 + assert res_patch.json()["day_index"] is None + + res_patch = patch_level(client, admin_token, level_id, day_index=14) + assert res_patch.status_code == 200 + assert res_patch.json()["day_index"] == 14 + + # Should not update + res_patch_again = patch_level( + client, admin_token, level_id, name_sv=None, name_en=None, encoded_grid=None, wall_budget=None, day_index=None + ) + assert res_patch_again.status_code == 200 + assert res_patch_again.json()["name_sv"] != None + assert res_patch_again.json()["name_en"] != None + assert res_patch_again.json()["encoded_grid"] != None + assert res_patch_again.json()["wall_budget"] != None + assert res_patch_again.json()["day_index"] is None