diff --git a/backend/src/flotilla/api/events_api.py b/backend/src/flotilla/api/events_api.py new file mode 100644 index 000000000..fa2a585ed --- /dev/null +++ b/backend/src/flotilla/api/events_api.py @@ -0,0 +1,133 @@ +from http import HTTPStatus +from logging import getLogger +from typing import List + +from fastapi import APIRouter, Body, Depends, Path, Response, Security +from flotilla_openapi.models.event import Event +from flotilla_openapi.models.event_request import EventRequest +from flotilla_openapi.models.problem_details import ProblemDetails +from pytest import Session + +from flotilla.api.authentication import authentication_scheme +from flotilla.database.crud import ( + DBException, + create_event, + read_event_by_id, + read_events, + remove_event, +) +from flotilla.database.db import get_db +from flotilla.database.models import EventDBModel + +logger = getLogger("api") + +router = APIRouter() + +NOT_FOUND_DESCRIPTION = "Not Found - No event with given id" +INTERNAL_SERVER_ERROR_DESCRIPTION = "Internal Server Error" + + +@router.get( + "/events", + responses={ + HTTPStatus.OK.value: { + "model": List[Event], + "description": "Request successful.", + }, + }, + tags=["Events"], + summary="Lookup events", + dependencies=[Security(authentication_scheme)], +) +async def get_events( + db: Session = Depends(get_db), +) -> List[Event]: + """Lookup events.""" + db_events: List[EventDBModel] = read_events(db) + events: List[Event] = [event.get_api_event() for event in db_events] + return events + + +@router.post( + "/events", + responses={ + HTTPStatus.CREATED.value: {"model": Event, "description": "Request successful"}, + }, + tags=["Events"], + summary="Create new event", + dependencies=[Security(authentication_scheme)], +) +async def post_event( + response: Response, + db: Session = Depends(get_db), + event_request: EventRequest = Body(None, description="Time entry update"), +) -> Event: + """Add a new event to the robot schedule""" + try: + event_id: int = create_event( + db, + event_request.robot_id, + event_request.mission_id, + event_request.start_time, + ) + db_event: EventDBModel = read_event_by_id(db, event_id) + event: EventDBModel = db_event.get_api_event() + except DBException: + logger.exception(f"An error occured while creating event.") + response.status_code = HTTPStatus.INTERNAL_SERVER_ERROR.value + return ProblemDetails(title=INTERNAL_SERVER_ERROR_DESCRIPTION) + response.status_code = HTTPStatus.CREATED.value + return event + + +@router.delete( + "/events/{event_id}", + responses={ + HTTPStatus.NO_CONTENT.value: {"description": "Event successfully deleted"}, + }, + tags=["Events"], + summary="Delete event with specified id", + dependencies=[Security(authentication_scheme)], +) +async def delete_event( + response: Response, + db: Session = Depends(get_db), + event_id: int = Path(None, description=""), +) -> None: + """Deletes an event from the robot schedule. Can only be used for events that have not started yet.""" + try: + remove_event(db, event_id) + except DBException: + logger.exception(f"Could not delete event with id {event_id}.") + response.status_code = HTTPStatus.NOT_FOUND.value + return ProblemDetails(title=NOT_FOUND_DESCRIPTION) + response.status_code = HTTPStatus.NO_CONTENT.value + return None + + +@router.get( + "/events/{event_id}", + responses={ + HTTPStatus.OK.value: { + "model": Event, + "description": "Request successful.", + }, + }, + tags=["Events"], + summary="Lookup event with specified id", + dependencies=[Security(authentication_scheme)], +) +async def get_event( + response: Response, + db: Session = Depends(get_db), + event_id: int = Path(None, description=""), +) -> Event: + """Lookup event with specified id. Can only be used for events that have not started yet.""" + try: + db_event: EventDBModel = read_event_by_id(db, event_id) + event: EventDBModel = db_event.get_api_event() + except DBException: + logger.exception(f"Could not get event with id {event_id}.") + response.status_code = HTTPStatus.NOT_FOUND.value + return ProblemDetails(title=NOT_FOUND_DESCRIPTION) + return event diff --git a/backend/src/flotilla/main.py b/backend/src/flotilla/main.py index 38c77d8c5..92c0ca5e8 100644 --- a/backend/src/flotilla/main.py +++ b/backend/src/flotilla/main.py @@ -5,6 +5,7 @@ from fastapi.middleware.cors import CORSMiddleware from flotilla.api.authentication import authenticator +from flotilla.api.events_api import router as events_router from flotilla.api.missions_api import router as missions_router from flotilla.api.reports_api import router as reports_router from flotilla.api.robots_api import router as robots_router @@ -46,6 +47,7 @@ def startup_event(): app.include_router(robots_router) app.include_router(missions_router) app.include_router(reports_router) +app.include_router(events_router) if __name__ == "__main__": diff --git a/backend/tests/test_api/test_events_api.py b/backend/tests/test_api/test_events_api.py new file mode 100644 index 000000000..7ca6c4ac0 --- /dev/null +++ b/backend/tests/test_api/test_events_api.py @@ -0,0 +1,86 @@ +import json +from datetime import datetime +from http import HTTPStatus + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from flotilla_openapi.models.event_request import EventRequest + + +def test_get_events(test_app: FastAPI): + + with TestClient(test_app) as client: + response = client.get("/events") + assert response.status_code == HTTPStatus.OK.value + + +@pytest.mark.parametrize( + "event_id, expected_status_code", + [ + ( + 2, + HTTPStatus.OK.value, + ), + ( + 4, + HTTPStatus.NOT_FOUND.value, + ), + ], +) +def test_get_event( + test_app: FastAPI, + event_id: int, + expected_status_code: int, +): + + with TestClient(test_app) as client: + response = client.get(f"/events/{event_id}") + assert response.status_code == expected_status_code + + +@pytest.mark.parametrize( + "event_request, expected_status_code", + [ + ( + EventRequest(robot_id=1, mission_id=234, start_time=datetime.now()), + HTTPStatus.CREATED.value, + ), + ], +) +def test_post_event( + test_app: FastAPI, + event_request: EventRequest, + expected_status_code: int, +): + + with TestClient(test_app) as client: + response = client.post( + f"/events", + data=json.dumps(event_request.dict(), default=str), + ) + assert response.status_code == expected_status_code + + +@pytest.mark.parametrize( + "event_id, expected_status_code", + [ + ( + 1, + HTTPStatus.NO_CONTENT.value, + ), + ( + 5, + HTTPStatus.NOT_FOUND.value, + ), + ], +) +def test_delete_event( + test_app: FastAPI, + event_id: int, + expected_status_code: int, +): + + with TestClient(test_app) as client: + response = client.delete(f"/events/{event_id}") + assert response.status_code == expected_status_code