add test data, fix bugs, refactor

Author: EarthenSky

src/alembic/versions/243190df5588_create_election_tables.py

@@ -22,8 +22,8 @@
 def upgrade() -> None:
     op.create_table(
         "election",
-        sa.Column("slug", sa.String(length=32), nullable=False),
-        sa.Column("name", sa.String(length=32), nullable=False),
+        sa.Column("slug", sa.String(length=64), nullable=False),
+        sa.Column("name", sa.String(length=64), nullable=False),
         sa.Column("type", sa.String(length=64), default="general_election"),
         sa.Column("datetime_start_nominations", sa.DateTime(), nullable=False),
         sa.Column("datetime_start_voting", sa.DateTime(), nullable=False),
@@ -35,8 +35,8 @@ def upgrade() -> None:
         "election_nominee",
         sa.Column("computing_id", sa.String(length=32), nullable=False),
         sa.Column("full_name", sa.String(length=64), nullable=False),
-        sa.Column("facebook", sa.String(length=64), nullable=True),
-        sa.Column("instagram", sa.String(length=64), nullable=True),
+        sa.Column("facebook", sa.String(length=128), nullable=True),
+        sa.Column("instagram", sa.String(length=128), nullable=True),
         sa.Column("email", sa.String(length=64), nullable=True),
         sa.Column("discord", sa.String(length=32), nullable=True),
         sa.Column("discord_id", sa.String(length=32), nullable=True),

src/elections/crud.py

@@ -46,6 +46,6 @@ async def update_election(db_session: AsyncSession, new_election: Election) -> b
             sqlalchemy
             .update(Election)
             .where(Election.slug == target_slug)
-            .values(new_election)
+            .values(new_election.to_update_dict())
         )
         return True

src/elections/tables.py

@@ -17,12 +17,15 @@
 
 election_types = ["general_election", "by_election", "council_rep_election"]
 
+MAX_ELECTION_NAME = 64
+MAX_ELECTION_SLUG = 64
+
 class Election(Base):
     __tablename__ = "election"
 
     # Slugs are unique identifiers
-    slug = Column(String(32), primary_key=True)
-    name = Column(String(32), nullable=False)
+    slug = Column(String(MAX_ELECTION_SLUG), primary_key=True)
+    name = Column(String(MAX_ELECTION_NAME), nullable=False)
     type = Column(String(64), default="general_election")
     datetime_start_nominations = Column(DateTime, nullable=False)
     datetime_start_voting = Column(DateTime, nullable=False)
@@ -53,15 +56,28 @@ def public_details(self) -> dict:
             "datetime_end_voting": self.datetime_end_voting.isoformat(),
         }
 
+    def to_update_dict(self) -> dict:
+        return {
+            "slug": self.slug,
+            "name": self.name,
+            "type": self.type,
+
+            "datetime_start_nominations": self.datetime_start_nominations,
+            "datetime_start_voting": self.datetime_start_voting,
+            "datetime_end_voting": self.datetime_end_voting,
+
+            "survey_link": self.survey_link,
+        }
+
 # Each row represents a nominee of a given election
 class Nominee(Base):
     __tablename__ = "election_nominee"
 
     # Previously named sfuid
     computing_id = Column(String(COMPUTING_ID_LEN), primary_key=True)
     full_name = Column(String(64), nullable=False)
-    facebook = Column(String(64))
-    instagram = Column(String(64))
+    facebook = Column(String(128))
+    instagram = Column(String(128))
     email = Column(String(64))
     discord = Column(String(DISCORD_NAME_LEN))
     discord_id = Column(String(DISCORD_ID_LEN))

src/elections/urls.py

@@ -10,7 +10,7 @@
 import elections
 from elections.tables import Election, election_types
 from permission.types import ElectionOfficer, WebsiteAdmin
-from utils.urls import logged_in_or_raise
+from utils.urls import is_logged_in
 
 _logger = logging.getLogger(__name__)
 
@@ -21,20 +21,50 @@
 
 def _slugify(text: str) -> str:
     """Creates a unique slug based on text passed in. Assumes non-unicode text."""
-    return re.sub(r"[\W_]+", "-", text)
+    return re.sub(r"[\W_]+", "-", text.replace("/", "").replace("&", ""))
 
 async def _validate_user(
     request: Request,
     db_session: database.DBSession,
 ) -> tuple[bool, str, str]:
-    session_id, computing_id = logged_in_or_raise(request, db_session)
+    logged_in, session_id, computing_id = await is_logged_in(request, db_session)
+    if not logged_in:
+        return False, None, None
+
     has_permission = await ElectionOfficer.has_permission(db_session, computing_id)
     if not has_permission:
         has_permission = await WebsiteAdmin.has_permission(db_session, computing_id)
+
     return has_permission, session_id, computing_id
 
 # elections ------------------------------------------------------------- #
 
+@router.get(
+    "/by_name/{name:str}",
+    description="Retrieves the election data for an election by name"
+)
+async def get_election(
+    request: Request,
+    db_session: database.DBSession,
+    name: str,
+):
+    election = await elections.crud.get_election(db_session, _slugify(name))
+    if election is None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"election with slug {_slugify(name)} does not exist"
+        )
+    elif datetime.now() >= election.datetime_start_voting:
+        # after the voting period starts, all election data becomes public
+        return JSONResponse(election.serializable_dict())
+
+    is_valid_user, _, _ = await _validate_user(request, db_session)
+    return JSONResponse(
+        election.serializable_dict()
+        if is_valid_user
+        else election.public_details()
+    )
+
 @router.post(
     "/by_name/{name:str}",
     description="Creates an election and places it in the database. Returns election json on success",
@@ -60,59 +90,57 @@ async def create_election(
             # TODO: is this header actually required?
             headers={"WWW-Authenticate": "Basic"},
         )
-    elif elections.crud.get_election(db_session, _slugify(name)) is not None:
+    elif len(name) <= elections.tables.MAX_ELECTION_NAME:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"election name {name} is too long",
+        )
+    elif len(_slugify(name)) <= elections.tables.MAX_SLUG_NAME:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"election slug {_slugify(name)} is too long",
+        )
+    elif await elections.crud.get_election(db_session, _slugify(name)) is not None:
         # don't overwrite a previous election
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail="would overwrite previous election",
         )
+    elif not (
+        (datetime_start_nominations <= datetime_start_voting)
+        and (datetime_start_voting <= datetime_end_voting)
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="dates must be in order from earliest to latest",
+        )
+
+    # TODO: force dates to be in order; here & on the update election endpoint
 
     await elections.crud.create_election(
         Election(
-            _slugify(name),
-            name,
-            election_type,
-            datetime_start_nominations,
-            datetime_start_voting,
-            datetime_end_voting,
-            survey_link
+            slug = _slugify(name),
+            name = name,
+            type = election_type,
+            datetime_start_nominations = datetime_start_nominations,
+            datetime_start_voting = datetime_start_voting,
+            datetime_end_voting = datetime_end_voting,
+            survey_link = survey_link
         ),
         db_session
     )
     await db_session.commit()
 
-    election = elections.crud.get_election(db_session, _slugify(name))
+    election = await elections.crud.get_election(db_session, _slugify(name))
     return JSONResponse(election.serializable_dict())
 
-@router.delete(
-    "/by_name/{name:str}",
-    description="Deletes an election from the database. Returns whether the election exists after deletion."
-)
-async def delete_election(
-    request: Request,
-    db_session: database.DBSession,
-    name: str
-):
-    is_valid_user, _, _ = await _validate_user(request, db_session)
-    if not is_valid_user:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="must have election officer permission",
-            # TODO: is this header actually required?
-            headers={"WWW-Authenticate": "Basic"},
-        )
-
-    await elections.crud.delete_election(_slugify(name), db_session)
-    await db_session.commit()
-
-    old_election = elections.crud.get_election(db_session, _slugify(name))
-    return JSONResponse({"exists": old_election is None})
-
 @router.patch(
     "/by_name/{name:str}",
     description="""
         Updates an election in the database.
-        Note that this don't let you to change the name of an election as it would generate a new slug!
+
+        Note that this doesn't let you change the name of an election, unless the new
+        name produces the same slug.
 
         Returns election json on success.
     """
@@ -135,15 +163,23 @@ async def update_election(
             detail="must have election officer or admin permission",
             headers={"WWW-Authenticate": "Basic"},
         )
+    elif not (
+        (datetime_start_nominations <= datetime_start_voting)
+        and (datetime_start_voting <= datetime_end_voting)
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="dates must be in order from earliest to latest",
+        )
 
     new_election = Election(
-        _slugify(name),
-        name,
-        election_type,
-        datetime_start_nominations,
-        datetime_start_voting,
-        datetime_end_voting,
-        survey_link
+        slug = _slugify(name),
+        name = name,
+        type = election_type,
+        datetime_start_nominations = datetime_start_nominations,
+        datetime_start_voting = datetime_start_voting,
+        datetime_end_voting = datetime_end_voting,
+        survey_link = survey_link
     )
     success = await elections.crud.update_election(db_session, new_election)
     if not success:
@@ -154,31 +190,32 @@ async def update_election(
     else:
         await db_session.commit()
 
-        election = elections.crud.get_election(db_session, _slugify(name))
+        election = await elections.crud.get_election(db_session, _slugify(name))
         return JSONResponse(election.serializable_dict())
 
-@router.get(
+@router.delete(
     "/by_name/{name:str}",
-    description="Retrieves the election data for an election by name"
+    description="Deletes an election from the database. Returns whether the election exists after deletion."
 )
-async def get_election(
+async def delete_election(
     request: Request,
     db_session: database.DBSession,
-    name: str,
+    name: str
 ):
-    election = elections.crud.get_election(db_session, _slugify(name))
-    if election is None:
+    is_valid_user, _, _ = await _validate_user(request, db_session)
+    if not is_valid_user:
         raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"election with slug {_slugify(name)} does not exist"
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="must have election officer permission",
+            # TODO: is this header actually required?
+            headers={"WWW-Authenticate": "Basic"},
         )
 
-    is_valid_user, _, _ = await _validate_user(request, db_session)
-    return JSONResponse(
-        election.serializable_dict()
-        if is_valid_user
-        else election.public_details()
-    )
+    await elections.crud.delete_election(_slugify(name), db_session)
+    await db_session.commit()
+
+    old_election = await elections.crud.get_election(db_session, _slugify(name))
+    return JSONResponse({"exists": old_election is None})
 
 # registration ------------------------------------------------------------- #