Add a util for extracting the chosen archives from the database

Author: PeterJCLaw

code_submitter/utils.py

@@ -0,0 +1,31 @@
+from typing import Dict
+
+import databases
+from sqlalchemy.sql import select
+
+from .tables import Archive, ChoiceHistory
+
+
+async def get_chosen_submissions(database: databases.Database) -> Dict[str, bytes]:
+    """
+    Return a mapping of teams to their the chosen archive.
+    """
+
+    # Note: Ideally we'd group by team in SQL, however that doesn't seem to work
+    # properly -- we don't get the ordering applied before the grouping.
+
+    rows = await database.fetch_all(
+        select([
+            Archive.c.team,
+            Archive.c.content,
+            ChoiceHistory.c.created,
+        ]).select_from(
+            Archive.join(ChoiceHistory),
+        ).order_by(
+            Archive.c.team,
+            ChoiceHistory.c.created.asc(),
+        ),
+    )
+
+    # Rely on later keys replacing earlier occurrences of the same key.
+    return {x['team']: x['content'] for x in rows}

tests/tests_utils.py

@@ -0,0 +1,74 @@
+import datetime
+
+import test_utils
+from code_submitter import utils
+from code_submitter.tables import Archive, ChoiceHistory
+
+
+class UtilsTests(test_utils.InTransactionTestCase):
+    def setUp(self) -> None:
+        super().setUp()
+
+        self.await_(self.database.execute(
+            Archive.insert().values(
+                id=8888888888,
+                content=b'8888888888',
+                username='someone_else',
+                team='ABC',
+                created=datetime.datetime(2020, 8, 8, 12, 0),
+            ),
+        ))
+        self.await_(self.database.execute(
+            Archive.insert().values(
+                id=2222222222,
+                content=b'2222222222',
+                username='a_colleague',
+                team='SRZ2',
+                created=datetime.datetime(2020, 2, 2, 12, 0),
+            ),
+        ))
+        self.await_(self.database.execute(
+            Archive.insert().values(
+                id=1111111111,
+                content=b'1111111111',
+                username='test_user',
+                team='SRZ2',
+                created=datetime.datetime(2020, 1, 1, 12, 0),
+            ),
+        ))
+
+    def test_get_chosen_submissions_nothing_chosen(self) -> None:
+        result = self.await_(utils.get_chosen_submissions(self.database))
+        self.assertEqual({}, result)
+
+    def test_get_chosen_submissions_multiple_chosen(self) -> None:
+        self.await_(self.database.execute(
+            ChoiceHistory.insert().values(
+                archive_id=8888888888,
+                username='someone_else',
+                created=datetime.datetime(2020, 8, 8, 12, 0),
+            ),
+        ))
+        self.await_(self.database.execute(
+            ChoiceHistory.insert().values(
+                archive_id=1111111111,
+                username='test_user',
+                created=datetime.datetime(2020, 3, 3, 12, 0),
+            ),
+        ))
+        self.await_(self.database.execute(
+            ChoiceHistory.insert().values(
+                archive_id=2222222222,
+                username='test_user',
+                created=datetime.datetime(2020, 2, 2, 12, 0),
+            ),
+        ))
+
+        result = self.await_(utils.get_chosen_submissions(self.database))
+        self.assertEqual(
+            {
+                'SRZ2': b'1111111111',
+                'ABC': b'8888888888',
+            },
+            result,
+        )