fix: unit tests working with only one set up

Author: jbriones1

tests/conftest.py

@@ -9,7 +9,7 @@
 from httpx import ASGITransport, AsyncClient
 
 from src import load_test_db
-from src.auth.crud import create_user_session
+from src.auth.crud import create_user_session, remove_user_session
 from src.database import SQLALCHEMY_TEST_DATABASE_URL, DatabaseSessionManager
 from src.main import app
 
@@ -21,22 +21,7 @@ def suppress_sqlalchemy_logs():
     yield
     logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
 
-@pytest.fixture(scope="session")
-def event_loop(request):
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield loop
-    loop.close()
-
-
-@pytest.fixture(scope="function")
-async def client() -> AsyncGenerator[Any, None]:
-    # base_url is just a random placeholder url
-    # ASGITransport is just telling the async client to pass all requests to app
-    # `async with` syntax used so that the connecton will automatically be closed once done
-    async with AsyncClient(transport=ASGITransport(app), base_url="http://test") as client:
-        yield client
-
-@pytest_asyncio.fixture(scope="module")
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
 async def database_setup():
     # reset the database again, just in case
     print("Resetting DB...")
@@ -47,24 +32,25 @@ async def database_setup():
     yield sessionmanager
     await sessionmanager.close()
 
-@pytest_asyncio.fixture(scope="function")
-async def db_transaction(database_setup):
-    async with database_setup.session() as session:
-        try:
-            await session.begin()
-            yield session
-        finally:
-            await session.rollback()
+@pytest_asyncio.fixture(scope="function", loop_scope="session")
+async def client() -> AsyncGenerator[Any, None]:
+    # base_url is just a random placeholder url
+    # ASGITransport is just telling the async client to pass all requests to app
+    # `async with` syntax used so that the connecton will automatically be closed once done
+    async with AsyncClient(transport=ASGITransport(app), base_url="http://test") as client:
+        yield client
 
-@pytest_asyncio.fixture(scope="function")
-async def db_session(db_transaction):
-    yield db_transaction
+@pytest_asyncio.fixture(scope="function", loop_scope="session")
+async def db_session(database_setup):
+    async with database_setup.session() as session:
+        yield session
 
-@pytest_asyncio.fixture(scope="function")
+@pytest_asyncio.fixture(scope="function", loop_scope="session")
 async def admin_session(database_setup):
     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)
-    yield
+    async with database_setup.session() as session:
+        await create_user_session(session, session_id, load_test_db.SYSADMIN_COMPUTING_ID)
+        yield
+        await remove_user_session(session, session_id)
 
 

tests/integration/test_officers.py

@@ -3,6 +3,7 @@
 from datetime import date, timedelta
 
 import pytest
+from httpx import AsyncClient
 
 from src import load_test_db
 from src.officers.constants import OfficerPositionEnum
@@ -11,76 +12,108 @@
 # TODO: setup a database on the CI machine & run this as a unit test then (since
 # this isn't really an integration test)
 
-# TODO: switch to mark.anyio
-@pytest.mark.asyncio
-async def test__read_execs(database_setup):
-    async with database_setup.session() as db_session:
-        # test that reads from the database succeeded as expected
-        print(type(db_session))
-        assert (await get_active_officer_terms(db_session, "blarg")) == []
-        assert (await get_active_officer_terms(db_session, "abc22")) != []
-
-        abc11_officer_terms = await get_active_officer_terms(db_session, "abc11")
-        assert len(abc11_officer_terms) == 1
-        assert abc11_officer_terms[0].computing_id == "abc11"
-        assert abc11_officer_terms[0].position == OfficerPositionEnum.EXECUTIVE_AT_LARGE
-        assert abc11_officer_terms[0].start_date is not None
-        assert abc11_officer_terms[0].nickname == "the holy A"
-        assert abc11_officer_terms[0].favourite_course_0 == "CMPT 361"
-        assert abc11_officer_terms[0].biography == "Hi! I'm person A and I want school to be over ; _ ;"
-
-        current_exec_team = await current_officers(db_session)
-        assert current_exec_team is not None
-        assert len(current_exec_team) == 3
-        # assert next(iter(current_exec_team)) == OfficerPositionEnum.EXECUTIVE_AT_LARGE
-        # assert next(iter(current_exec_team))["favourite_course_0"] == "CMPT 361"
-        # assert next(iter(current_exec_team.values()))[0].csss_email == OfficerPosition.to_email(OfficerPositionEnum.EXECUTIVE_AT_LARGE)
-        # assert next(iter(current_exec_team.values()))[0].private_data is None
-
-        current_exec_team = await current_officers(db_session)
-        assert current_exec_team is not None
-        assert len(current_exec_team) == 3
-        # assert next(iter(current_exec_team.keys())) == OfficerPositionEnum.EXECUTIVE_AT_LARGE
-        # assert next(iter(current_exec_team.values()))[0].favourite_course_0 == "CMPT 361"
-        # assert next(iter(current_exec_team.values()))[0].csss_email == OfficerPosition.to_email(OfficerPositionEnum.EXECUTIVE_AT_LARGE)
-        # assert next(iter(current_exec_team.values()))[0].private_data is not None
-        # assert next(iter(current_exec_team.values()))[0].private_data.computing_id == "abc11"
-
-        all_terms = await all_officers(db_session, include_future_terms=False)
-        assert len(all_terms) == 8
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+async def test__read_execs(db_session):
+    # test that reads from the database succeeded as expected
+    print(type(db_session))
+    assert (await get_active_officer_terms(db_session, "blarg")) == []
+    assert (await get_active_officer_terms(db_session, "abc22")) != []
+
+    abc11_officer_terms = await get_active_officer_terms(db_session, "abc11")
+    assert len(abc11_officer_terms) == 1
+    assert abc11_officer_terms[0].computing_id == "abc11"
+    assert abc11_officer_terms[0].position == OfficerPositionEnum.EXECUTIVE_AT_LARGE
+    assert abc11_officer_terms[0].start_date is not None
+    assert abc11_officer_terms[0].nickname == "the holy A"
+    assert abc11_officer_terms[0].favourite_course_0 == "CMPT 361"
+    assert abc11_officer_terms[0].biography == "Hi! I'm person A and I want school to be over ; _ ;"
+
+    current_exec_team = await current_officers(db_session)
+    assert current_exec_team is not None
+    assert len(current_exec_team) == 3
+    # assert next(iter(current_exec_team)) == OfficerPositionEnum.EXECUTIVE_AT_LARGE
+    # assert next(iter(current_exec_team))["favourite_course_0"] == "CMPT 361"
+    # assert next(iter(current_exec_team.values()))[0].csss_email == OfficerPosition.to_email(OfficerPositionEnum.EXECUTIVE_AT_LARGE)
+    # assert next(iter(current_exec_team.values()))[0].private_data is None
+
+    current_exec_team = await current_officers(db_session)
+    assert current_exec_team is not None
+    assert len(current_exec_team) == 3
+    # assert next(iter(current_exec_team.keys())) == OfficerPositionEnum.EXECUTIVE_AT_LARGE
+    # assert next(iter(current_exec_team.values()))[0].favourite_course_0 == "CMPT 361"
+    # assert next(iter(current_exec_team.values()))[0].csss_email == OfficerPosition.to_email(OfficerPositionEnum.EXECUTIVE_AT_LARGE)
+    # assert next(iter(current_exec_team.values()))[0].private_data is not None
+    # assert next(iter(current_exec_team.values()))[0].private_data.computing_id == "abc11"
+
+    all_terms = await all_officers(db_session, include_future_terms=False)
+    assert len(all_terms) == 8
 
 
 #async def test__update_execs(database_setup):
 #    # TODO: the second time an update_officer_info call occurs, the user should be updated with info
 #    pass
 
-@pytest.mark.asyncio
-async def test__endpoints(client):
+async def test__get_officers(client):
+    # private data shoudn't be leaked
+    print(f"[DEBUG] Loop ID in {__name__}: {id(asyncio.get_running_loop())}")
     response = await client.get("/officers/current")
     assert response.status_code == 200
     assert response.json() != {}
     assert len(response.json().values()) == 3
-    assert not response.json()["executive at large"]["computing_id"]
+    assert "computing_id" not in response.json()[OfficerPositionEnum.EXECUTIVE_AT_LARGE]
+    assert "discord_id" not in response.json()[OfficerPositionEnum.EXECUTIVE_AT_LARGE]
+    assert "discord_name" not in response.json()[OfficerPositionEnum.EXECUTIVE_AT_LARGE]
+    assert "discord_nickname" not in response.json()[OfficerPositionEnum.EXECUTIVE_AT_LARGE]
+    assert "phone_number" not in response.json()[OfficerPositionEnum.EXECUTIVE_AT_LARGE]
+    assert "github_username" not in response.json()[OfficerPositionEnum.EXECUTIVE_AT_LARGE]
+    assert "google_drive_email" not in response.json()[OfficerPositionEnum.EXECUTIVE_AT_LARGE]
+    assert "photo_url" not in response.json()[OfficerPositionEnum.EXECUTIVE_AT_LARGE]
+
+    assert "computing_id" not in response.json()[OfficerPositionEnum.DIRECTOR_OF_ARCHIVES]
+    assert "discord_id" not in response.json()[OfficerPositionEnum.DIRECTOR_OF_ARCHIVES]
+    assert "discord_name" not in response.json()[OfficerPositionEnum.DIRECTOR_OF_ARCHIVES]
+    assert "discord_nickname" not in response.json()[OfficerPositionEnum.DIRECTOR_OF_ARCHIVES]
+    assert "phone_number" not in response.json()[OfficerPositionEnum.DIRECTOR_OF_ARCHIVES]
+    assert "github_username" not in response.json()[OfficerPositionEnum.DIRECTOR_OF_ARCHIVES]
+    assert "google_drive_email" not in response.json()[OfficerPositionEnum.DIRECTOR_OF_ARCHIVES]
+    assert "photo_url" not in response.json()[OfficerPositionEnum.DIRECTOR_OF_ARCHIVES]
+
+    assert "computing_id" not in response.json()[OfficerPositionEnum.PRESIDENT]
+    assert "discord_id" not in response.json()[OfficerPositionEnum.PRESIDENT]
+    assert "discord_name" not in response.json()[OfficerPositionEnum.PRESIDENT]
+    assert "discord_nickname" not in response.json()[OfficerPositionEnum.PRESIDENT]
+    assert "phone_number" not in response.json()[OfficerPositionEnum.PRESIDENT]
+    assert "github_username" not in response.json()[OfficerPositionEnum.PRESIDENT]
+    assert "google_drive_email" not in response.json()[OfficerPositionEnum.PRESIDENT]
+    assert "photo_url" not in response.json()[OfficerPositionEnum.PRESIDENT]
 
     response = await client.get("/officers/all?include_future_terms=false")
     assert response.status_code == 200
     assert response.json() != []
-    assert len(response.json()) == 6
-    assert response.json()[0]["private_data"] is None
+    assert len(response.json()) == 8
+    assert "computing_id" not in response.json()[0]
+    assert "discord_id" not in response.json()[0]
+    assert "discord_name" not in response.json()[0]
+    assert "discord_nickname" not in response.json()[0]
+    assert "phone_number" not in response.json()[0]
+    assert "github_username" not in response.json()[0]
+    assert "google_drive_email" not in response.json()[0]
+    assert "photo_url" not in response.json()[0]
 
     response = await client.get("/officers/all?include_future_terms=true")
     assert response.status_code == 401
 
+async def test__get_officer_terms(client: AsyncClient):
     response = await client.get(f"/officers/terms/{load_test_db.SYSADMIN_COMPUTING_ID}?include_future_terms=false")
     assert response.status_code == 200
-    assert response.json() != []
     assert len(response.json()) == 2
     assert response.json()[0]["nickname"] == "G2"
     assert response.json()[1]["nickname"] == "G1"
 
     response = await client.get("/officers/terms/balargho?include_future_terms=false")
     assert response.status_code == 200
-    assert response.json() == []
+    assert len(response.json()) == 0
 
     response = await client.get("/officers/terms/abc11?include_future_terms=true")
     assert response.status_code == 401
@@ -90,28 +123,35 @@ async def test__endpoints(client):
     response = await client.get(f"/officers/info/{load_test_db.SYSADMIN_COMPUTING_ID}")
     assert response.status_code == 401
 
-    response = await client.post("officers/term", content=json.dumps([{
+async def test__post_officer_terms(client: AsyncClient):
+    # Only admins can create new terms
+    response = await client.post("officers/term", json=[{
         "computing_id": "ehbc12",
         "position": OfficerPositionEnum.DIRECTOR_OF_MULTIMEDIA,
-        "start_date": "2025-12-29"
-    }]))
+        "start_date": "2025-12-29",
+        "legal_name": "Eh Bc"
+    }])
     assert response.status_code == 401
 
-    response = await client.post("officers/term", content=json.dumps([{
+    # Position must be one of the enum positions
+    response = await client.post("officers/term", json=[{
         "computing_id": "ehbc12",
         "position": "balargho",
-        "start_date": "2025-12-29"
-    }]))
-    assert response.status_code == 400
+        "start_date": "2025-12-29",
+        "legal_name": "Eh Bc"
+    }])
+    assert response.status_code == 422
 
-    response = await client.patch("officers/info/abc11", content=json.dumps({
+async def test__patch_officer_terms(client: AsyncClient):
+    # Only admins can update new terms
+    response = await client.patch("officers/info/abc11", json={
         "legal_name": "fancy name",
         "phone_number": None,
         "discord_name": None,
         "github_username": None,
         "google_drive_email": None,
-    }))
-    assert response.status_code == 401
+    })
+    assert response.status_code == 403
 
     response = await client.patch("officers/term/1", content=json.dumps({
         "computing_id": "abc11",
@@ -127,12 +167,12 @@ async def test__endpoints(client):
         "favourite_pl_1": "5",
         "biography": "hello"
     }))
-    assert response.status_code == 401
+    assert response.status_code == 403
 
     response = await client.delete("officers/term/1")
     assert response.status_code == 401
 
-@pytest.mark.asyncio
+@pytest.mark.skip
 async def test__endpoints_admin(client, database_setup, admin_session):
     # login as website admin
     session_id = "temp_id_" + load_test_db.SYSADMIN_COMPUTING_ID