diff --git a/src/elections/crud.py b/src/elections/crud.py index 9541ebfc..84cb711c 100644 --- a/src/elections/crud.py +++ b/src/elections/crud.py @@ -3,7 +3,12 @@ import sqlalchemy from sqlalchemy.ext.asyncio import AsyncSession +import candidates.crud +import nominees.crud +from candidates.tables import CandidateDB +from elections.models import ElectionNomineeSummary from elections.tables import ElectionDB +from nominees.tables import NomineeInfoDB async def get_all_elections(db_session: AsyncSession) -> Sequence[ElectionDB]: @@ -37,3 +42,96 @@ async def delete_election(db_session: AsyncSession, slug: str) -> None: Deletes a given election by its slug. Does not validate if an election exists """ await db_session.execute(sqlalchemy.delete(ElectionDB).where(ElectionDB.slug == slug)) + + +async def get_all_nominees_by_election( + db_session: AsyncSession, + has_permission: bool, +) -> dict[str, list[ElectionNomineeSummary]]: + """ + Fetches all nominees across all elections in a JOIN query. + Returns a dict mapping election slug -> list of ElectionNomineeSummary. + Only fetches contact fields (computing_id, linked_in, etc.) when has_permission is True. + """ + if has_permission: + query = sqlalchemy.select( + CandidateDB.nominee_election, + CandidateDB.position, + CandidateDB.speech, + NomineeInfoDB.full_name, + NomineeInfoDB.computing_id, + NomineeInfoDB.linked_in, + NomineeInfoDB.instagram, + NomineeInfoDB.email, + NomineeInfoDB.discord_username, + ).join(NomineeInfoDB, CandidateDB.computing_id == NomineeInfoDB.computing_id) + else: + query = sqlalchemy.select( + CandidateDB.nominee_election, + CandidateDB.position, + CandidateDB.speech, + NomineeInfoDB.full_name, + ).join(NomineeInfoDB, CandidateDB.computing_id == NomineeInfoDB.computing_id) + + rows = (await db_session.execute(query)).all() + + nominees_by_election: dict[str, list[ElectionNomineeSummary]] = {} + for row in rows: + if has_permission: + nominee = ElectionNomineeSummary( + full_name=row.full_name, + position=row.position, + speech=row.speech or "No speech provided by this candidate", + computing_id=row.computing_id, + linked_in=row.linked_in, + instagram=row.instagram, + email=row.email, + discord_username=row.discord_username, + ) + else: + nominee = ElectionNomineeSummary( + full_name=row.full_name, + position=row.position, + speech=row.speech or "No speech provided by this candidate", + ) + if row.nominee_election not in nominees_by_election: + nominees_by_election[row.nominee_election] = [] + nominees_by_election[row.nominee_election].append(nominee) + + return nominees_by_election + + +async def _get_election_nominees( + db_session: AsyncSession, + election_row: ElectionDB, + has_permission: bool, +) -> list[ElectionNomineeSummary]: + candidates_list = [] + all_nominations = await candidates.crud.get_all_candidates_in_election(db_session, election_row.slug) + if not all_nominations: + return [] + for nomination in all_nominations: + # NOTE: if a nominee does not input their legal name, they are not considered a nominee + nominee_info = await nominees.crud.get_nominee_info(db_session, nomination.computing_id) + if nominee_info is None: + continue + + if has_permission: + candidate_entry = ElectionNomineeSummary( + full_name=nominee_info.full_name, + position=nomination.position, + speech=nomination.speech or "No speech provided by this candidate", + computing_id=nomination.computing_id, + linked_in=nominee_info.linked_in, + instagram=nominee_info.instagram, + email=nominee_info.email, + discord_username=nominee_info.discord_username, + ) + else: + candidate_entry = ElectionNomineeSummary( + full_name=nominee_info.full_name, + position=nomination.position, + speech=nomination.speech or "No speech provided by this candidate", + ) + candidates_list.append(candidate_entry) + return candidates_list diff --git a/src/elections/models.py b/src/elections/models.py index 01b97bcb..1514d000 100644 --- a/src/elections/models.py +++ b/src/elections/models.py @@ -3,7 +3,6 @@ from pydantic import BaseModel, Field -from candidates.models import Candidate from officers.constants import OfficerPositionEnum @@ -20,6 +19,17 @@ class ElectionStatusEnum(StrEnum): AFTER_VOTING = "after_voting" +class ElectionNomineeSummary(BaseModel): + full_name: str + position: OfficerPositionEnum + speech: str + computing_id: str | None = None + linked_in: str | None = None + instagram: str | None = None + email: str | None = None + discord_username: str | None = None + + class ElectionResponse(BaseModel): slug: str name: str @@ -32,7 +42,10 @@ class ElectionResponse(BaseModel): # Private fields survey_link: str | None = Field(None, description="Only available to admins") - candidates: list[Candidate] | None = Field(None, description="Only available to admins") + candidates: list[ElectionNomineeSummary] | None = Field( + None, + description="Included when with_nominees is true, contact fields only for election admins", + ) class ElectionParams(BaseModel): diff --git a/src/elections/urls.py b/src/elections/urls.py index 62d9a2aa..2b3c55f7 100644 --- a/src/elections/urls.py +++ b/src/elections/urls.py @@ -1,6 +1,6 @@ import datetime -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException, Query, status from fastapi.responses import JSONResponse from sqlalchemy.exc import IntegrityError @@ -74,7 +74,7 @@ def _raise_if_bad_election_data( @router.get( "", - description="Returns a list of all election & their status", + description="Return a list of all elections, their statuses and nominees (if requested)", response_model=list[ElectionResponse], responses={status.HTTP_404_NOT_FOUND: {"description": "No election found", "model": DetailModel}}, operation_id="get_all_elections", @@ -82,16 +82,32 @@ def _raise_if_bad_election_data( async def list_elections( computing_id: SessionUser, db_session: database.DBSession, + with_nominees: bool = Query(False), ): election_list = await elections.crud.get_all_elections(db_session) if election_list is None or len(election_list) == 0: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="no election found") current_time = datetime.datetime.now(datetime.UTC) - if await is_user_election_admin(computing_id, db_session): - election_metadata_list = [election.private_details(current_time) for election in election_list] - else: - election_metadata_list = [election.public_details(current_time) for election in election_list] + election_metadata_list = [] + has_permission = await is_user_election_admin(computing_id, db_session) + + nominees_by_election = {} + if with_nominees: + nominees_by_election = await elections.crud.get_all_nominees_by_election(db_session, has_permission) + + for election in election_list: + if has_permission: + election_data = election.private_details(current_time) + else: + election_data = election.public_details(current_time) + if with_nominees: + election_nominees = nominees_by_election.get(election.slug, []) + candidates_list = [] + for nominee in election_nominees: + candidates_list.append(nominee.model_dump(exclude_none=True)) + election_data["candidates"] = candidates_list + election_metadata_list.append(election_data) return JSONResponse(election_metadata_list) @@ -100,14 +116,18 @@ async def list_elections( "/{election_name}", description=""" Retrieves the election data for an election by name. - Returns private details when the time is allowed. - If user is an admin or election officer, returns computing ids for each candidate as well. + If user is an admin or election officer, returns complete nominee information (if requested). """, response_model=ElectionResponse, responses={404: {"description": "Election of that name doesn't exist", "model": DetailModel}}, operation_id="get_election_by_name", ) -async def get_election(db_session: database.DBSession, computing_id: SessionUser, election_name: str): +async def get_election( + db_session: database.DBSession, + computing_id: SessionUser, + election_name: str, + with_nominees: bool = Query(False), +): current_time = datetime.datetime.now(datetime.UTC) slugified_name = slugify(election_name) election = await elections.crud.get_election(db_session, slugified_name) @@ -117,41 +137,16 @@ async def get_election(db_session: database.DBSession, computing_id: SessionUser ) has_permission = await is_user_election_admin(computing_id, db_session) - if current_time >= election.datetime_start_voting or has_permission: + if has_permission: election_json = election.private_details(current_time) - all_nominations = await candidates.crud.get_all_candidates_in_election(db_session, slugified_name) - if not all_nominations: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="no candidates found") - election_json["candidates"] = [] - - available_positions_list = election.available_positions - for nomination in all_nominations: - if nomination.position not in available_positions_list: - # ignore any positions that are **no longer** active - continue - - # NOTE: if a nominee does not input their legal name, they are not considered a nominee - nominee_info = await nominees.crud.get_nominee_info(db_session, nomination.computing_id) - if nominee_info is None: - continue - - candidate_entry = { - "position": nomination.position, - "full_name": nominee_info.full_name, - "linked_in": nominee_info.linked_in, - "instagram": nominee_info.instagram, - "email": nominee_info.email, - "discord_username": nominee_info.discord_username, - "speech": ("No speech provided by this candidate" if nomination.speech is None else nomination.speech), - } - if has_permission: - candidate_entry["computing_id"] = nomination.computing_id - election_json["candidates"].append(candidate_entry) - - # after the voting period starts, all election data becomes public - return JSONResponse(election_json) else: election_json = election.public_details(current_time) + if with_nominees: + nominees = await elections.crud._get_election_nominees(db_session, election, has_permission) + candidates_list = [] + for nominee in nominees: + candidates_list.append(nominee.model_dump(exclude_none=True)) + election_json["candidates"] = candidates_list return JSONResponse(election_json) diff --git a/src/nominees/urls.py b/src/nominees/urls.py index da3bd2db..913a9877 100644 --- a/src/nominees/urls.py +++ b/src/nominees/urls.py @@ -84,7 +84,7 @@ async def get_nominee_info(db_session: database.DBSession, computing_id: str): operation_id="delete_nominee", dependencies=[Depends(perm_election)], ) -async def delete_nominee_info(db_session: database.tDBSession, computing_id: str): +async def delete_nominee_info(db_session: database.DBSession, computing_id: str): try: await nominees.crud.delete_nominee_info(db_session, computing_id) await db_session.commit() diff --git a/src/officers/crud.py b/src/officers/crud.py index f18cf622..92240b08 100644 --- a/src/officers/crud.py +++ b/src/officers/crud.py @@ -177,7 +177,7 @@ async def get_active_officer_terms( ) query = utils.is_active_officer(query) if positions: - query.where(OfficerTermDB.position.in_(positions)) + query = query.where(OfficerTermDB.position.in_(positions)) return list((await db_session.scalars(query)).all()) diff --git a/tests/integration/test_elections.py b/tests/integration/test_elections.py index 1d6d2cae..ed90a70e 100644 --- a/tests/integration/test_elections.py +++ b/tests/integration/test_elections.py @@ -21,6 +21,18 @@ pytestmark = pytest.mark.asyncio(loop_scope="session") +CANDIDATE_CONTACT_FIELDS = ["computing_id", "linked_in", "instagram", "email", "discord_username"] + + +def assert_public_candidate_fields(candidate: dict): + for field in CANDIDATE_CONTACT_FIELDS: + assert field not in candidate + + +def assert_private_candidate_fields(candidate: dict): + for field in CANDIDATE_CONTACT_FIELDS: + assert field in candidate + # TODO: Split the candidate tests into their own file # database testing------------------------------- @@ -57,21 +69,46 @@ async def test_read_elections(db_session: DBSession): # API endpoint testing (without AUTH)-------------------------------------- -async def test__get_all_elections(client): +async def test__get_all_elections(client: AsyncClient): response = await client.get("/election") assert response.status_code == 200 assert response.json() != {} + for election in response.json(): + assert "candidates" not in election -async def test__get_single_election(client): +async def test__get_all_elections_with_nominees_true(client: AsyncClient): + # Test on election 2, because it has candidates + response = await client.get("/election", params={"with_nominees": "true"}) + assert response.status_code == 200 + elections_list = {election["slug"]: election for election in response.json()} + election_2_response = elections_list[slugify(TEST_ELECTION_2)] + assert "survey_link" not in election_2_response + assert "candidates" in election_2_response + assert len(election_2_response["candidates"]) >= 1 + for candidate in election_2_response["candidates"]: + assert_public_candidate_fields(candidate) + + +async def test__get_single_election(client: AsyncClient): # Returns private details when the time is allowed. If user is an admin or election officer, returns computing ids for each candidate as well. response = await client.get(f"/election/{TEST_ELECTION_2}") assert response.status_code == 200 assert response.json() != {} - # if candidates filled, enure unauthorized values remain hidden - if "candidates" in response.json() and response.json()["candidates"]: - for cand in response.json()["candidates"]: - assert "computing_id" not in cand + assert "candidates" not in response.json() + assert "survey_link" not in response.json() + + +async def test__get_single_election_with_nominees_true(client: AsyncClient): + response = await client.get(f"/election/{TEST_ELECTION_2}", params={"with_nominees": "true"}) + assert response.status_code == 200 + election_2_response = response.json() + assert election_2_response != {} + assert "candidates" in election_2_response + assert "survey_link" not in election_2_response + assert len(election_2_response["candidates"]) >= 1 + for candidate in election_2_response["candidates"]: + assert_public_candidate_fields(candidate) async def test__get_single_candidates(client: AsyncClient): @@ -164,17 +201,24 @@ async def test__delete_candidate(client: AsyncClient): # Admin API testing (with AUTH)----------------------------------- async def test__admin_get_all_elections(admin_client: AsyncClient): - # Login in as the website admin - # session_id = "temp_id_" + load_test_db.SYSADMIN_COMPUTING_ID - # async with database_setup.session() as db_session: - # await create_user_session(db_session, session_id, load_test_db.SYSADMIN_COMPUTING_ID) - - # client.cookies = {"session_id": session_id} - - # test that more info is given if logged in & with access to it response = await admin_client.get("/election") assert response.status_code == 200 assert response.json() != {} + for election in response.json(): + assert "candidates" not in election + + +async def test__admin_get_all_elections_with_nominees_true(admin_client: AsyncClient): + # Test on election 2, because it has candidates + response = await admin_client.get("/election", params={"with_nominees": "true"}) + assert response.status_code == 200 + elections_list = {election["slug"]: election for election in response.json()} + election_2_response = elections_list[slugify(TEST_ELECTION_2)] + assert "survey_link" in election_2_response + assert "candidates" in election_2_response + assert len(election_2_response["candidates"]) >= 1 + for candidate in election_2_response["candidates"]: + assert_private_candidate_fields(candidate) async def test__admin_get_single_election(admin_client: AsyncClient): @@ -182,10 +226,20 @@ async def test__admin_get_single_election(admin_client: AsyncClient): response = await admin_client.get(f"/election/{TEST_ELECTION_2}") assert response.status_code == 200 assert response.json() != {} - # if candidates filled, enure unauthorized values remain hidden - if "candidates" in response.json() and response.json()["candidates"]: - for cand in response.json()["candidates"]: - assert "computing_id" in cand + assert "candidates" not in response.json() + assert "survey_link" in response.json() + + +async def test__admin_get_single_election_with_nominees_true(admin_client: AsyncClient): + response = await admin_client.get(f"/election/{TEST_ELECTION_2}", params={"with_nominees": "true"}) + assert response.status_code == 200 + election_2_response = response.json() + assert election_2_response != {} + assert "survey_link" in election_2_response + assert "candidates" in election_2_response + assert len(election_2_response["candidates"]) >= 1 + for candidate in election_2_response["candidates"]: + assert_private_candidate_fields(candidate) async def test__admin_create_election(admin_client: AsyncClient):