diff --git a/app/crud.py b/app/crud.py index b93edca..b19c8dd 100644 --- a/app/crud.py +++ b/app/crud.py @@ -46,7 +46,8 @@ def get_progress(db: Session, election_ref: str, token: str) -> schemas.Progress raise errors.UnauthorizedError("Wrong election ref") # Check we can update the election - if not payload["admin"]: + # Use .get() for a safe check. If "admin" key is missing, it returns None (which is falsy). + if not payload.get("admin"): raise errors.ForbiddenError("You are not allowed to manage the election") # Votes are provided for each candidate and each voter @@ -284,11 +285,12 @@ def update_election( election_ref = payload["election"] # Check we can update the election - if not payload["admin"]: + # Use .get() for a safe check. If "admin" key is missing, it returns None (which is falsy). + if not payload.get("admin"): raise errors.ForbiddenError("You are not allowed to manage the election") if election_ref != election.ref: - raise errors.ForbiddenError("Wrong election ref") + raise errors.WrongElectionError("The provided admin token does not match this election.") db_election = get_election(db, election_ref) @@ -297,12 +299,12 @@ def update_election( if election.date_start is not None and election.date_end is None and db_election.date_end is not None: if schemas.parse_date(election.date_start) > schemas.parse_date(db_election.date_end): - raise errors.ForbiddenError( + raise errors.InvalidDateError( "The start date must be before the end date of the election" ) elif election.date_end is not None and election.date_start is None: if schemas.parse_date(election.date_end) < schemas.parse_date(db_election.date_start): - raise errors.ForbiddenError( + raise errors.InvalidDateError( "The end date must be after the start date of the election" ) @@ -332,13 +334,13 @@ def update_election( candidate_ids = {c.id for c in election.candidates} db_candidate_ids = {c.id for c in db_election.candidates} if candidate_ids != db_candidate_ids: - raise errors.ForbiddenError("You must have the same candidate ids") + raise errors.ImmutableIdsError("The set of candidate IDs cannot be changed during an update.") if election.grades is not None: grade_ids = {c.id for c in election.grades} db_grade_ids = {c.id for c in db_election.grades} if grade_ids != db_grade_ids: - raise errors.ForbiddenError("You must have the same grade ids") + raise errors.ImmutableIdsError("The set of grade IDs cannot be changed during an update.") # Update the candidates and grades if election.candidates is not None: @@ -346,6 +348,16 @@ def update_election( if election.grades is not None: update_grades(db, election.grades, db_election.grades) + # Check if start_date is being changed + if election.date_start is not None and str(db_election.date_start) != election.date_start: + # If so, check if any votes have been cast + num_votes_cast = db.query(models.Vote).filter( + models.Vote.election_ref == election_ref, + models.Vote.grade_id.is_not(None) + ).count() + if num_votes_cast > 0: + raise errors.ElectionIsActiveError("Cannot change the start date of an election that already has votes.") + for key in [ "name", "description", @@ -381,7 +393,7 @@ def _check_ballot_is_consistent( for c in election.candidates } if not all(len(votes) == 1 for votes in votes_by_candidate.values()): - raise errors.ForbiddenError("Unconsistent ballot") + raise errors.InconsistentBallotError("Inconsistent ballot: each candidate must have exactly one vote.") def create_ballot(db: Session, ballot: schemas.BallotCreate) -> schemas.BallotGet: @@ -426,7 +438,7 @@ def _check_public_election(db: Session, election_ref: str): if db_election is None: raise errors.NotFoundError("elections") if db_election.restricted: - raise errors.ForbiddenError( + raise errors.ElectionRestrictedError( "The election is restricted. You can not create new votes" ) return db_election @@ -437,7 +449,7 @@ def _check_election_is_started(election: models.Election): If it is not, raise an error. """ if election.date_start is not None and election.date_start > datetime.now(): - raise errors.ForbiddenError("The election has not started yet. You can not create votes") + raise errors.ElectionNotStartedError("The election has not started yet. You can not create votes") def _check_election_is_not_ended(election: models.Election): """ @@ -445,9 +457,9 @@ def _check_election_is_not_ended(election: models.Election): If it is, raise an error. """ if election.date_end is not None and election.date_end < datetime.now(): - raise errors.ForbiddenError("The election has ended. You can not create new votes") + raise errors.ElectionFinishedError("The election has ended. You can not create new votes") if election.force_close: - raise errors.ForbiddenError("The election is closed. You can not create or update votes") + raise errors.ElectionFinishedError("The election is closed. You can not create or update votes") def _check_items_in_election( db: Session, @@ -566,7 +578,7 @@ def get_results(db: Session, election_ref: str, token: t.Optional[str]) -> schem and (db_election.date_end is not None and db_election.date_end > datetime.now()) and not db_election.force_close ): - raise errors.ForbiddenError("The election is not closed") + raise errors.ResultsHiddenError("Results are hidden until the election is closed.") query = db.query( models.Vote.candidate_id, models.Grade.value, func.count(models.Vote.id) diff --git a/app/errors.py b/app/errors.py index 7a1e876..a097802 100644 --- a/app/errors.py +++ b/app/errors.py @@ -2,54 +2,100 @@ Utility to handle exceptions """ - -class NotFoundError(Exception): - """ - An item can not be found - """ - - def __init__(self, name: str): - self.name = name - - -class InconsistentDatabaseError(Exception): - """ - An inconsistent value was detected on the database - """ - - def __init__(self, name: str, details: str | None = None): - self.name = name - self.details = details - - -class BadRequestError(Exception): - """ - The request is made inconsistent - """ - - def __init__(self, details: str): - self.details = details - - -class ForbiddenError(Exception): +class CustomError(Exception): """ - The request is made inconsistent + Base class for custom application errors. """ + status_code: int = 500 + error_code: str = "UNEXPECTED_ERROR" + message: str = "An unexpected error occurred." - def __init__(self, details: str = "Forbidden"): - self.details = details + def __init__(self, message: str | None = None, status_code: int | None = None, error_code: str | None = None): + super().__init__(message or self.message) + if status_code is not None: + self.status_code = status_code + if error_code is not None: + self.error_code = error_code -class UnauthorizedError(Exception): - """ - The verification could not be verified - """ +class NotFoundError(CustomError): + status_code = 404 + error_code = "NOT_FOUND" + message = "The requested item could not be found." def __init__(self, name: str): - self.name = name + super().__init__(message=f"Oops! No {name} were found.") +class InconsistentDatabaseError(CustomError): + status_code = 500 + error_code = "INCONSISTENT_DATABASE" + message = "A serious internal error has occurred." + + def __init__(self, name: str, details: str | None = None): + super().__init__(message=f"A serious error has occurred with {name}. {details or ''}") + +class BadRequestError(CustomError): + status_code = 400 + error_code = "BAD_REQUEST" + message = "The request is invalid." + +class ForbiddenError(CustomError): + status_code = 403 + error_code = "FORBIDDEN" + message = "You are not authorized to perform this action." + +class UnauthorizedError(CustomError): + status_code = 401 + error_code = "UNAUTHORIZED" + message = "Authentication is required and has failed or has not yet been provided." + +class NoRecordedVotes(CustomError): + status_code = 403 + error_code = "NO_RECORDED_VOTES" + message = "No votes have been recorded yet." + +class ElectionFinishedError(CustomError): + status_code = 403 + error_code = "ELECTION_FINISHED" + message = "The election has finished and cannot be voted on." + +class InvalidDateError(CustomError): + status_code = 409 + error_code = "INVALID_DATE_CONFIGURATION" + message = "The provided date configuration is invalid." + +class ElectionNotStartedError(CustomError): + status_code = 403 + error_code = "ELECTION_NOT_STARTED" + message = "The election has not started yet." + +class ElectionRestrictedError(CustomError): + status_code = 403 + error_code = "ELECTION_RESTRICTED" + message = "This is a restricted election." + +class InconsistentBallotError(CustomError): + status_code = 403 + error_code = "INCONSISTENT_BALLOT" + message = "This ballot is inconsistent." + +class ResultsHiddenError(CustomError): + status_code = 403 + error_code = "RESULTS_HIDDEN" + message = "Results are hidden." + +class WrongElectionError(CustomError): + status_code = 403 + error_code = "WRONG_ELECTION" + message = "Wrong election." + +class ImmutableIdsError(CustomError): + status_code = 403 + error_code = "IMMUTABLE_IDS" + message = "The set of IDs is immutable." + +class ElectionIsActiveError(CustomError): + status_code = 403 + error_code = "ELECTION_IS_ACTIVE" + message = "This election is already active and cannot be modified." -class NoRecordedVotes(Exception): - """ - We can't display results if we don't have resutls - """ diff --git a/app/main.py b/app/main.py index 08795ed..8bbc6bb 100644 --- a/app/main.py +++ b/app/main.py @@ -3,8 +3,8 @@ from fastapi import Depends, FastAPI, HTTPException, Request, Body, Header from fastapi.responses import JSONResponse, PlainTextResponse from fastapi.middleware.cors import CORSMiddleware +from fastapi.exceptions import RequestValidationError from sqlalchemy.orm import Session -from jose import jwe, jws from jose.exceptions import JWEError, JWSError from . import crud, models, schemas, errors @@ -23,75 +23,36 @@ allow_headers=["*"], ) - -@app.get("/") -async def main(): - return {"message": "Hello World"} - -@app.exception_handler(schemas.ArgumentsSchemaError) -async def invalid_schema_exception_handler( - request: Request, exc: schemas.ArgumentsSchemaError -): +@app.exception_handler(RequestValidationError) +async def validation_exception_handler(request: Request, exc: RequestValidationError): + # This overrides FastAPI's default 422 validation error handler + # to produce our standardized error format. return JSONResponse( status_code=422, - content={ - "message": f"Validation Error. {exc}", - }, + content={"error": "VALIDATION_ERROR", "message": str(exc)}, ) -@app.exception_handler(errors.NotFoundError) -async def not_found_exception_handler(request: Request, exc: errors.NotFoundError): - return JSONResponse( - status_code=404, - content={"message": f"Oops! No {exc.name} were found."}, - ) - - -@app.exception_handler(errors.UnauthorizedError) -async def unauthorized_exception_handler(request: Request, exc: errors.NotFoundError): - return JSONResponse( - status_code=401, content={"message": "Unautorized", "details": exc.name} - ) - - -@app.exception_handler(errors.ForbiddenError) -async def forbidden_exception_handler(request: Request, exc: errors.ForbiddenError): - return JSONResponse( - status_code=403, - content={"message": f"Forbidden", "details": exc.details}, - ) - - -@app.exception_handler(errors.BadRequestError) -async def bad_request_exception_handler(request: Request, exc: errors.BadRequestError): - return JSONResponse( - status_code=400, - content={"message": f"Bad Request", "details": exc.details}, - ) +@app.get("/") +async def main(): + return {"message": "Hello World"} -@app.exception_handler(errors.NoRecordedVotes) -async def no_recorded_votes_exception_handler( - request: Request, exc: errors.NoRecordedVotes -): +@app.exception_handler(errors.CustomError) +async def custom_error_exception_handler(request: Request, exc: errors.CustomError): return JSONResponse( - status_code=403, - content={"message": f"No votes have been recorded yet"}, + status_code=exc.status_code, + content={"error": exc.error_code, "message": str(exc)}, ) - -@app.exception_handler(errors.InconsistentDatabaseError) -async def inconsistent_database_exception_handler( - request: Request, exc: errors.InconsistentDatabaseError +@app.exception_handler(schemas.ArgumentsSchemaError) +async def invalid_schema_exception_handler( + request: Request, exc: schemas.ArgumentsSchemaError ): return JSONResponse( - status_code=500, - content={ - "message": f"A serious error has occured with {exc.name}. {exc.details or ''}" - }, + status_code=422, + content={"error": "SCHEMA_VALIDATION_ERROR", "message": str(exc)}, ) - @app.get("/liveness") def read_root(): return "OK" diff --git a/app/tests/test_api.py b/app/tests/test_api.py index 90c2420..3c68172 100644 --- a/app/tests/test_api.py +++ b/app/tests/test_api.py @@ -30,12 +30,28 @@ def override_get_db(): finally: db.close() - app.dependency_overrides[get_db] = override_get_db client = TestClient(app) +def check_error_response(response, expected_status_code: int, expected_error_code: str): + """ + Helper function to assert a standardized error response from the API. + It checks the status code and the specific error code in the JSON body. + """ + assert response.status_code == expected_status_code, \ + f"Expected status {expected_status_code}, but got {response.status_code}. Body: {response.text}" + + data = response.json() + + assert "error" in data, f"Key 'error' not found in response body: {data}" + assert data["error"] == expected_error_code, \ + f"Expected error code '{expected_error_code}', but got '{data['error']}'" + + return data # Return the parsed data in case a test needs to check the message + + def test_liveness(): response = client.get("/liveness") assert response.status_code == 200, response.status_code @@ -44,7 +60,7 @@ def test_liveness(): def test_read_a_missing_election(): response = client.get("/elections/foo") - assert response.status_code == 404 + check_error_response(response, 404, "NOT_FOUND") def _random_string(length: int) -> str: @@ -111,13 +127,14 @@ def test_create_election(): data = response.json() assert data["name"] == body["name"] + def test_start_end_date_are_valid(): # cannot create an election where the start date is after the end date body = _random_election(2, 2) body["date_start"] = (datetime.now() + timedelta(days=1)).isoformat() body["date_end"] = (datetime.now()).isoformat() response = client.post("/elections", json=body) - assert response.status_code == 422, response.text + check_error_response(response, 422, "SCHEMA_VALIDATION_ERROR") body["date_start"] = (datetime.now()).isoformat() body["date_end"] = (datetime.now() + timedelta(days=1)).isoformat() @@ -133,24 +150,23 @@ def test_start_end_date_are_valid(): admin_token = election_data["admin"] election_ref = election_data["ref"] - # update election should not be allowed if the start date is after the end date + # update election should not be allowed if the new start date is after the new end date election_data["date_start"] = (datetime.now() + timedelta(days=1)).isoformat() election_data["date_end"] = (datetime.now()).isoformat() response = client.put("/elections", json=election_data, headers={"Authorization": f"Bearer {admin_token}"}) - assert response.status_code == 422, response.text + check_error_response(response, 422, "SCHEMA_VALIDATION_ERROR") - # update election should be allowed if the start date is before the end date + # update election should be rejected if the new end date is before the existing start date del election_data["date_start"] election_data["date_end"] = (datetime.now() - timedelta(days=1)).isoformat() response = client.put("/elections", json=election_data, headers={"Authorization": f"Bearer {admin_token}"}) - assert response.status_code == 403, response.text + check_error_response(response, 409, "INVALID_DATE_CONFIGURATION") - # update election should be allowed if the start date is before the end date + # update election should be rejected if the new start date is after the existing end date del election_data["date_end"] election_data["date_start"] = (datetime.now() + timedelta(days=2)).isoformat() response = client.put("/elections", json=election_data, headers={"Authorization": f"Bearer {admin_token}"}) - assert response.status_code == 403, response.text - + check_error_response(response, 409, "INVALID_DATE_CONFIGURATION") def test_get_election(): body = _random_election(3, 4) @@ -225,7 +241,7 @@ def test_create_ballot(): response = client.get( f"/ballots/", headers={"Authorization": f"Bearer {ballot_token}WRONG"} ) - assert response.status_code == 401, response.text + check_error_response(response, 401, "UNAUTHORIZED") response = client.get(f"/ballots/", headers={"Authorization": f"Bearer {ballot_token}"}) assert response.status_code == 200, response.text @@ -263,7 +279,7 @@ def test_reject_wrong_ballots_restricted_election(): json={"votes": votes[-1]}, headers={"Authorization": f"Bearer {ballot_token}"}, ) - assert response.status_code == 422, response.json() + check_error_response(response, 422, "VALIDATION_ERROR") # Check that a ballot with an empty grade_id is rejected grade_id = data["grades"][0]["id"] @@ -274,7 +290,7 @@ def test_reject_wrong_ballots_restricted_election(): json={"votes": votes2}, headers={"Authorization": f"Bearer {ballot_token}"}, ) - assert response.status_code == 422, response.json() + check_error_response(response, 422, "VALIDATION_ERROR") # Check that a ballot with an empty candidate is rejected votes2 = copy.deepcopy(votes) @@ -284,7 +300,7 @@ def test_reject_wrong_ballots_restricted_election(): json={"votes": votes2}, headers={"Authorization": f"Bearer {ballot_token}"}, ) - assert response.status_code == 422, response.json() + check_error_response(response, 422, "VALIDATION_ERROR") # But it should work with the whole ballot response = client.put( @@ -301,6 +317,26 @@ def test_reject_wrong_ballots_restricted_election(): ) assert response.status_code == 200, response.json() +def test_rejects_update_with_empty_ballot(): + """ + Tests that updating a ballot with an empty list of votes is rejected. + """ + # Create a restricted election to get a valid ballot token + body = _random_election(5, 3) + body["restricted"] = True + body["num_voters"] = 1 + response = client.post("/elections", json=body) + assert response.status_code == 200 + election_data = response.json() + ballot_token = election_data["invites"][0] + + # Attempt to update the ballot with an empty votes array + response = client.put( + "/ballots", + json={"votes": []}, + headers={"Authorization": f"Bearer {ballot_token}"}, + ) + check_error_response(response, 400, "BAD_REQUEST") def test_reject_wrong_ballots_unrestricted_election(): """ @@ -317,7 +353,7 @@ def test_reject_wrong_ballots_unrestricted_election(): response = client.post( f"/ballots", json={"votes": votes[:-1], "election_ref": data["ref"]} ) - assert response.status_code == 403, response.text + check_error_response(response, 403, "INCONSISTENT_BALLOT") # Check that a ballot with an empty grade_id is rejected votes = _generate_votes_from_response("id", data) @@ -325,7 +361,7 @@ def test_reject_wrong_ballots_unrestricted_election(): response = client.post( f"/ballots", json={"votes": votes, "election_ref": data["ref"]} ) - assert response.status_code == 422, response.text + check_error_response(response, 422, "VALIDATION_ERROR") # Check that a ballot with an empty candidate is rejected votes = _generate_votes_from_response("id", data) @@ -333,7 +369,7 @@ def test_reject_wrong_ballots_unrestricted_election(): response = client.post( f"/ballots", json={"votes": votes, "election_ref": data["ref"]} ) - assert response.status_code == 422, response.text + check_error_response(response, 422, "VALIDATION_ERROR") # But it should work with the whole ballot votes = _generate_votes_from_response("id", data) @@ -364,8 +400,7 @@ def test_cannot_create_vote_on_ended_election(): f"/ballots", json={"votes": votes, "election_ref": election_ref}, ) - data = response.json() - assert response.status_code == 403, data + check_error_response(response, 403, "ELECTION_FINISHED") # Try to close the election with force_close response = client.put( @@ -380,8 +415,7 @@ def test_cannot_create_vote_on_ended_election(): f"/ballots", json={"votes": votes, "election_ref": election_ref}, ) - data = response.json() - assert response.status_code == 403, data + check_error_response(response, 403, "ELECTION_FINISHED") def test_cannot_update_vote_on_ended_election(): """ @@ -422,7 +456,7 @@ def test_cannot_update_vote_on_ended_election(): json={"votes": votes}, headers={"Authorization": f"Bearer {ballot_token}"}, ) - assert response.status_code == 403, response.json() + check_error_response(response, 403, "ELECTION_FINISHED") # Test for date_end in the past response = client.put( @@ -440,11 +474,45 @@ def test_cannot_update_vote_on_ended_election(): json={"votes": votes}, headers={"Authorization": f"Bearer {ballot_token}"}, ) + check_error_response(response, 403, "ELECTION_FINISHED") + +def test_cannot_change_start_date_if_vote_is_cast(): + """ + Tests that the start_date of an election cannot be updated + once at least one vote has been cast. + """ + # Create a restricted election that is already open + body = _random_election(5, 3) + body["restricted"] = True + body["num_voters"] = 1 + body["date_start"] = (datetime.now() - timedelta(days=1)).isoformat() + body["date_end"] = (datetime.now() + timedelta(days=1)).isoformat() - assert response.status_code == 403, response.json() + response = client.post("/elections", json=body) + assert response.status_code == 200 + election_data = response.json() + election_ref = election_data["ref"] + admin_token = election_data["admin"] + ballot_token = election_data["invites"][0] + + # Cast a vote to make the election "active" + grade_id = election_data["grades"][0]["id"] + votes = [ + {"candidate_id": c["id"], "grade_id": grade_id} + for c in election_data["candidates"] + ] + response = client.put( + "/ballots", + json={"votes": votes}, + headers={"Authorization": f"Bearer {ballot_token}"} + ) + assert response.status_code == 200, "Setup failed: Could not cast the initial vote." + + # Attempt to change the start_date, which should be forbidden + update_payload = {"ref": election_ref, "date_start": (datetime.now() - timedelta(days=2)).isoformat()} + response = client.put("/elections", json=update_payload, headers={"Authorization": f"Bearer {admin_token}"}) + check_error_response(response, 403, "ELECTION_IS_ACTIVE") -## TODO: cannot change start_date if a people vote; -## def test_cannot_create_vote_on_unstarted_election(): """ On an unstarted election, we are not allowed to create new votes @@ -464,8 +532,7 @@ def test_cannot_create_vote_on_unstarted_election(): f"/ballots", json={"votes": votes, "election_ref": election_ref}, ) - data = response.json() - assert response.status_code == 403, data + check_error_response(response, 403, "ELECTION_NOT_STARTED") def test_cannot_update_vote_on_unstarted_election(): """ @@ -491,8 +558,7 @@ def test_cannot_update_vote_on_unstarted_election(): json={"votes": votes, "election_ref": election_ref}, headers={"Authorization": f"Bearer {tokens[0]}"}, ) - data = response.json() - assert response.status_code == 403, data + check_error_response(response, 403, "ELECTION_NOT_STARTED") def test_cannot_create_vote_on_restricted_election(): """ @@ -514,9 +580,7 @@ def test_cannot_create_vote_on_restricted_election(): f"/ballots", json={"votes": votes, "election_ref": election_ref}, ) - data = response.json() - assert response.status_code == 403, data - + check_error_response(response, 403, "ELECTION_RESTRICTED") def test_can_vote_on_restricted_election(): """ @@ -556,7 +620,7 @@ def test_can_vote_on_restricted_election(): assert payload2 == payload -def test_cannot_ballot_box_stuffing(): +def test_reject_ballot_box_stuffing(): # Create a random election body = _random_election(10, 5) response = client.post("/elections", json=body) @@ -570,9 +634,25 @@ def test_cannot_ballot_box_stuffing(): response = client.post( f"/ballots", json={"votes": votes + votes, "election_ref": election_ref} ) - data = response.json() - assert response.status_code == 403, data + check_error_response(response, 403, "INCONSISTENT_BALLOT") + +def test_get_results_for_election_with_no_votes(): + """ + Tests that requesting results for an election with zero votes + returns the correct specific error. + """ + # Create a new, open election. Do not cast any votes. + body = _random_election(10, 5) + # Ensure the election is considered "closed" so we don't get a RESULTS_HIDDEN error + body["date_start"] = (datetime.now() - timedelta(days=2)).isoformat() + body["date_end"] = (datetime.now() - timedelta(days=1)).isoformat() + response = client.post("/elections", json=body) + assert response.status_code == 200 + election_ref = response.json()["ref"] + # Request the results, which should fail predictably. + response = client.get(f"/results/{election_ref}") + check_error_response(response, 403, "NO_RECORDED_VOTES") def test_get_results(): # Create a random election @@ -600,7 +680,7 @@ def test_get_results(): assert len(profile) == len(data["candidates"]) -def test_get_results_with_hide_results(): +def test_get_results_with_hidden_results(): # Create a random election body = _random_election(10, 5) body["hide_results"] = True @@ -620,13 +700,15 @@ def test_get_results_with_hide_results(): # But, we can't get the results response = client.get(f"/results/{election_ref}") - assert response.status_code == 403, data + check_error_response(response, 403, "RESULTS_HIDDEN") # So, we close the election - print("UPDATE", data["force_close"]) - data["force_close"] = True + update_payload = { + "ref": election_ref, + "force_close": True + } response = client.put( - f"/elections", json=data, headers={"Authorization": f"Bearer {admin_token}"} + f"/elections", json=update_payload, headers={"Authorization": f"Bearer {admin_token}"} ) assert response.status_code == 200, response.text data = response.json() @@ -654,15 +736,17 @@ def test_get_results_with_auth_for_result(): ) assert response.status_code == 200, data + # Send a minimal update to ensure the election state is processed without triggering date change errors. + update_payload = {"ref": election_ref, "name": data["name"]} response = client.put( - f"/elections", json=data, headers={"Authorization": f"Bearer {admin_token}"} + f"/elections", json=update_payload, headers={"Authorization": f"Bearer {admin_token}"} ) assert response.status_code == 200, response.text # But, we can't get the results response = client.get(f"/results/{election_ref}") - assert response.status_code == 401, data + check_error_response(response, 401, "UNAUTHORIZED") # Now, we can access to the results response = client.get(f"/results/{election_ref}", headers={"Authorization": f"Bearer {admin_token}"}) @@ -675,7 +759,7 @@ def test_get_results_with_auth_for_result(): admin_token2 = data2["admin"] response = client.get(f"/results/{election_ref}", headers={"Authorization": f"Bearer {admin_token2}"}) - assert response.status_code == 401, data + check_error_response(response, 401, "UNAUTHORIZED") def test_update_election(): # Create a random election @@ -689,13 +773,13 @@ def test_update_election(): # Check we can not update without the ballot_token response = client.put("/elections", json=data) - assert response.status_code == 422, response.content + check_error_response(response, 422, "VALIDATION_ERROR") # Check that the request fails with a wrong ballot_token response = client.put( f"/elections", json=data, headers={"Authorization": f"Bearer {admin_token}WRONG"} ) - assert response.status_code == 401, response.text + check_error_response(response, 401, "UNAUTHORIZED") # Check that the request fails with a admnin token of other election response2 = client.post("/elections", json=body) @@ -704,7 +788,7 @@ def test_update_election(): response = client.put( f"/elections", json=data, headers={"Authorization": f"Bearer {admin_token2}"} ) - assert response.status_code == 403, response.text + check_error_response(response, 403, "WRONG_ELECTION") # But it works with the right ballot_token response = client.put( @@ -737,15 +821,32 @@ def test_update_election(): response = client.put( f"/elections", json=data2, headers={"Authorization": f"Bearer {admin_token}"} ) - assert response.status_code == 403, response.text + check_error_response(response, 403, "IMMUTABLE_IDS") data2 = copy.deepcopy(data) data2["grades"][0]["id"] += 100 response = client.put( f"/elections", json=data2, headers={"Authorization": f"Bearer {admin_token}"} ) - assert response.status_code == 403, response.text + check_error_response(response, 403, "IMMUTABLE_IDS") + +def test_update_election_as_non_admin(): + """ + Tests that a non-admin user cannot update an election. + """ + # Create a restricted election to get both an admin and a non-admin (ballot) token. + body = _random_election(5, 3) + body["restricted"] = True + body["num_voters"] = 1 + response = client.post("/elections", json=body) + assert response.status_code == 200 + election_data = response.json() + ballot_token = election_data["invites"][0] # This is a non-admin token + # Attempt to update the election using the ballot token. + update_payload = {"ref": election_data["ref"], "name": "New Name"} + response = client.put("/elections", json=update_payload, headers={"Authorization": f"Bearer {ballot_token}"}) + check_error_response(response, 403, "FORBIDDEN") def test_close_election2(): """ @@ -784,7 +885,7 @@ def test_close_election(): response = client.put( f"/elections", json=data, headers={"Authorization": f"Bearer {ballot_token}WRONG"} ) - assert response.status_code == 401, response.text + check_error_response(response, 401, "UNAUTHORIZED") # But it works with the right ballot_token response = client.put(