Make the main tree pass the full quality contract

Register the enrollment accessibility and sorting test modules with the
CI adoption gate and name them in the strict typecheck override; fix the
tar-spec annotations and use the production limits dataclass in the
snapshot resource budgets tests; format the three files the contract's
format component flagged.

Author: Snapshot Budgets

content_sync/tests/test_snapshot_resource_budgets.py

@@ -18,7 +18,6 @@
 import subprocess
 import tarfile
 import time
-from dataclasses import dataclass
 from pathlib import Path
 from unittest.mock import patch
 
@@ -27,6 +26,7 @@
 from content_sync import snapshot
 from content_sync.course_repository_ingest import (
     CourseRepositoryFetchError,
+    CourseRepositoryLimits,
     fetch_course_repository_snapshot,
     read_course_repository_checkout,
 )
@@ -35,16 +35,7 @@
 SCRATCH_ROOT = PROJECT_ROOT / ".tmp" / "snapshot-resource-budgets"
 
 
-@dataclass(frozen=True)
-class _Limits:
-    """The tightest ceilings the budgets can be expressed against."""
-
-    max_files: int = 1
-    max_total_bytes: int = 1
-    max_file_bytes: int = 1
-
-
-def _build_tar(*specs: tuple[str, int, bytes, dict[str, str] | None]) -> bytes:
+def _build_tar(*specs: tuple[str, bytes, bytes, dict[str, str] | None]) -> bytes:
     """One tar session holding every spec: ``(name, type, content, pax)``."""
 
     buffer = io.BytesIO()
@@ -59,11 +50,11 @@ def _build_tar(*specs: tuple[str, int, bytes, dict[str, str] | None]) -> bytes:
     return buffer.getvalue()
 
 
-def _file(name: str, content: bytes = b"x") -> tuple[str, int, bytes, None]:
+def _file(name: str, content: bytes = b"x") -> tuple[str, bytes, bytes, None]:
     return (name, tarfile.REGTYPE, content, None)
 
 
-def _dir(name: str) -> tuple[str, int, bytes, None]:
+def _dir(name: str) -> tuple[str, bytes, bytes, None]:
     return (name, tarfile.DIRTYPE, b"", None)
 
 
@@ -147,7 +138,7 @@ def test_success_under_the_cap_returns_the_archive(self) -> None:
         self.assertEqual(
             snapshot.read_snapshot_archive(
                 archive,
-                limits=_Limits(max_files=5, max_total_bytes=100, max_file_bytes=100),
+                limits=CourseRepositoryLimits(max_files=5, max_total_bytes=100, max_file_bytes=100),
                 strip_root=False,
             ),
             {"a.txt": b"hello\n"},
@@ -217,7 +208,11 @@ def test_many_directory_entries_exhaust_the_structural_budget(self) -> None:
         archive = _build_tar(*[_dir(f"dir-{index:03d}/") for index in range(100)])
 
         with self.assertRaises(snapshot.SnapshotError) as raised:
-            snapshot.read_snapshot_archive(archive, limits=_Limits(), strip_root=False)
+            snapshot.read_snapshot_archive(
+                archive,
+                limits=CourseRepositoryLimits(max_files=1, max_total_bytes=1, max_file_bytes=1),
+                strip_root=False,
+            )
 
         self.assertEqual(raised.exception.code, "archive_members_exceeded")
 
@@ -227,7 +222,7 @@ def test_legitimate_directories_do_not_consume_the_file_count(self) -> None:
 
         result = snapshot.read_snapshot_archive(
             _build_tar(*specs),
-            limits=_Limits(max_files=10, max_total_bytes=100, max_file_bytes=10),
+            limits=CourseRepositoryLimits(max_files=10, max_total_bytes=100, max_file_bytes=10),
             strip_root=False,
         )
 
@@ -248,7 +243,9 @@ def test_a_compressed_metadata_bomb_is_bounded(self) -> None:
 
         with self.assertRaises(snapshot.SnapshotError) as raised:
             snapshot.read_snapshot_archive(
-                gzip.compress(_build_tar(*specs)), limits=_Limits(), strip_root=False
+                gzip.compress(_build_tar(*specs)),
+                limits=CourseRepositoryLimits(max_files=1, max_total_bytes=1, max_file_bytes=1),
+                strip_root=False,
             )
 
         self.assertEqual(raised.exception.code, "archive_expansion_exceeded")
@@ -258,7 +255,7 @@ def test_symlinks_and_duplicates_are_still_refused(self) -> None:
         with self.assertRaises(snapshot.SnapshotError) as raised:
             snapshot.read_snapshot_archive(
                 _build_tar(link),
-                limits=_Limits(max_files=5, max_total_bytes=100, max_file_bytes=10),
+                limits=CourseRepositoryLimits(max_files=5, max_total_bytes=100, max_file_bytes=10),
                 strip_root=False,
             )
         self.assertEqual(raised.exception.code, "archive_entry_invalid")
@@ -267,7 +264,7 @@ def test_symlinks_and_duplicates_are_still_refused(self) -> None:
         with self.assertRaises(snapshot.SnapshotError) as raised:
             snapshot.read_snapshot_archive(
                 duplicate,
-                limits=_Limits(max_files=5, max_total_bytes=100, max_file_bytes=10),
+                limits=CourseRepositoryLimits(max_files=5, max_total_bytes=100, max_file_bytes=10),
                 strip_root=False,
             )
         self.assertEqual(raised.exception.code, "duplicate_path")
@@ -325,7 +322,9 @@ def test_a_budget_spent_on_the_headers_refuses_before_reading(self) -> None:
                     owner="owner",
                     repository="repo",
                     commit_sha="a" * 40,
-                    limits=_Limits(max_files=5, max_total_bytes=100, max_file_bytes=10),
+                    limits=CourseRepositoryLimits(
+                        max_files=5, max_total_bytes=100, max_file_bytes=10
+                    ),
                 )
 
         self.assertEqual(raised.exception.code, "course_repository_fetch_timeout")
@@ -344,7 +343,9 @@ def test_a_budget_spent_midstream_stops_the_reads(self) -> None:
                     owner="owner",
                     repository="repo",
                     commit_sha="a" * 40,
-                    limits=_Limits(max_files=5, max_total_bytes=100, max_file_bytes=10),
+                    limits=CourseRepositoryLimits(
+                        max_files=5, max_total_bytes=100, max_file_bytes=10
+                    ),
                 )
 
         self.assertEqual(raised.exception.code, "course_repository_fetch_timeout")
@@ -358,7 +359,7 @@ class TransportBudgetParityTests(SimpleTestCase):
     def test_a_structurally_exhausting_tree_refuses_identically(self) -> None:
         deep = "/".join(f"level-{index:02d}" for index in range(100))
         root, commit_sha = _git_repo("deep-tree", file_relative=f"{deep}/deepest.txt")
-        limits = _Limits(max_files=1, max_total_bytes=1_000_000, max_file_bytes=100)
+        limits = CourseRepositoryLimits(max_files=1, max_total_bytes=1_000_000, max_file_bytes=100)
 
         with self.assertRaises(CourseRepositoryFetchError) as pulled:
             read_course_repository_checkout(root, commit_sha=commit_sha, limits=limits)

events/tests/test_qna_lifecycle_contract.py

@@ -101,9 +101,7 @@ def _poll(self) -> dict:
     def test_open_poll_reports_capabilities(self) -> None:
         payload = self._poll()
         self.assertEqual(payload["state"], "open")
-        self.assertEqual(
-            payload["capabilities"], {"can_ask": True, "can_vote": True}
-        )
+        self.assertEqual(payload["capabilities"], {"can_ask": True, "can_vote": True})
 
     def test_closed_poll_answers_200_with_the_new_state(self) -> None:
         stale = self._poll()
@@ -115,9 +113,7 @@ def test_closed_poll_answers_200_with_the_new_state(self) -> None:
         self.assertEqual(response.status_code, 200)
         payload = response.json()
         self.assertEqual(payload["state"], "closed")
-        self.assertEqual(
-            payload["capabilities"], {"can_ask": False, "can_vote": False}
-        )
+        self.assertEqual(payload["capabilities"], {"can_ask": False, "can_vote": False})
 
     def test_participant_page_renders_lifecycle_from_the_session(self) -> None:
         participant, _token = security.new_participant()

playwright_tests/test_enrollment_sorting.py

@@ -163,9 +163,7 @@ def test_sort_is_global_and_survives_pagination_without_javascript(
         assert "page=" not in page.url
         settle(page, PAGE_SIZE)
         expect(first_row(page)).to_contain_text("page-student-01")
-        expect(page.locator("th", has_text="Pos").first).to_have_attribute(
-            "aria-sort", "ascending"
-        )
+        expect(page.locator("th", has_text="Pos").first).to_have_attribute("aria-sort", "ascending")
         page.screenshot(path=str(EVIDENCE / "position-asc-reset.png"), full_page=True)
     finally:
         context.close()
@@ -174,9 +172,7 @@ def test_sort_is_global_and_survives_pagination_without_javascript(
 def test_sort_is_keyboard_operable_and_announces_state(
     browser: Browser, live_server, sort_course: Cohort
 ) -> None:
-    page = studio_page(
-        browser, live_server, sort_course, viewport={"width": 1440, "height": 900}
-    )
+    page = studio_page(browser, live_server, sort_course, viewport={"width": 1440, "height": 900})
     context = page.context
     try:
         settle(page, PAGE_SIZE)

playwright_tests/test_issue_237_qna_review.py

@@ -84,9 +84,7 @@ def test_qna_participant_cohost_and_error_shells(
         assert detail.status == 200
         # UX-08 added the CSRF token as a hidden input: not an interactive
         # target, so the tap-target contract skips it.
-        for target in page.locator(
-            "button, textarea, input:not([type='hidden']), select"
-        ).all():
+        for target in page.locator("button, textarea, input:not([type='hidden']), select").all():
             box = target.bounding_box()
             assert box is not None and box["height"] >= 44
         sort = page.locator("#qna-sort")

pyproject.toml

@@ -158,6 +158,7 @@ module = [
     "scripts.projection_build.public_projection_source",
     "accounts.services.email_verification",
     "courses.services.mailchimp_course_tag_import",
+    "courses.tests.test_enrollment_error_accessibility",
     "courses.tests.test_family_page_content",
     "courses.tests.test_homework_submission_learning_public_markup",
     "courses.tests.test_mailchimp_course_tag_import",
@@ -169,6 +170,7 @@ module = [
     "scripts.tests.test_ml_zoomcamp_2021_identity_merge",
     "scripts.tests.test_registrant_import",
     "scripts.tests.test_scoring_import_submitted_at",
+    "studio_courses.tests.test_enrollment_sorting",
 ]
 ignore_errors = false
 

scripts/ci.py

@@ -103,6 +103,8 @@
     "scripts/tests/test_identity_manifest.py",
     "scripts/tests/test_ml_zoomcamp_2021_identity_merge.py",
     "scripts/tests/test_registrant_import.py",
+    "courses/tests/test_enrollment_error_accessibility.py",
+    "studio_courses/tests/test_enrollment_sorting.py",
 )
 PRODUCTION_IMPORT_PYTHON: Final = (
     "scripts/prod",

studio_courses/tests/test_enrollment_sorting.py

@@ -107,43 +107,35 @@ def test_score_ties_keep_the_queryset_order(self):
         # Positions 2, 4 and 6 collapse onto one score (student-16 already
         # holds it): a stable sort must present the whole tied group in the
         # queryset's (position, id) order, after every higher score.
-        Enrollment.objects.filter(
-            course=self.course, position_on_leaderboard__in=[2, 4, 6]
-        ).update(total_score=15)
+        Enrollment.objects.filter(course=self.course, position_on_leaderboard__in=[2, 4, 6]).update(
+            total_score=15
+        )
         ranked = usernames(self.ranked("total_score", "desc"))
         tied = [
             name
             for name in ranked
             if name in {"student-02", "student-04", "student-06", "student-16"}
         ]
-        self.assertEqual(
-            tied, ["student-02", "student-04", "student-06", "student-16"]
-        )
+        self.assertEqual(tied, ["student-02", "student-04", "student-06", "student-16"])
 
     def test_student_sort_is_case_insensitive(self):
-        uppercase = Enrollment.objects.get(
-            student__username="student-02", course=self.course
-        )
+        uppercase = Enrollment.objects.get(student__username="student-02", course=self.course)
         uppercase.student.username = "AAA-STUDENT"
         uppercase.student.save()
         ranked_enrollments = self.ranked("student", "asc")
         self.assertEqual(usernames(ranked_enrollments)[0], "AAA-STUDENT")
 
     def test_no_sort_arguments_match_the_legacy_ordering(self):
         legacy = list(
-            Enrollment.objects.filter(course=self.course).order_by(
-                "position_on_leaderboard", "id"
-            )
+            Enrollment.objects.filter(course=self.course).order_by("position_on_leaderboard", "id")
         )
         self.assertEqual(
             usernames(self.ranked("position", "asc")),
             [enrollment.student.username for enrollment in legacy],
         )
 
     def test_status_filter_and_sort_compose(self):
-        hidden = Enrollment.objects.get(
-            student__username="student-03", course=self.course
-        )
+        hidden = Enrollment.objects.get(student__username="student-03", course=self.course)
         hidden.display_on_leaderboard = False
         hidden.save()
         ranked_enrollments, counts = enrollment_list_data(
@@ -188,9 +180,7 @@ def test_descending_score_url_brings_the_global_top_to_page_one(self):
         self.assertEqual(self.first_row_username(response), "view-01")
 
     def test_unknown_sort_param_is_refused_by_the_allowlist(self):
-        response = self.client.get(
-            self.url, {"sort": "student__password", "dir": "desc"}
-        )
+        response = self.client.get(self.url, {"sort": "student__password", "dir": "desc"})
         self.assertEqual(response.status_code, 200)
         # The column falls back to the default; an explicit direction is
         # still honored, so this is position descending.
@@ -239,9 +229,7 @@ def test_sorted_state_is_announced_in_the_results_line(self):
         self.assertContains(response, "sorted by total_score (desc)")
 
     def test_second_page_of_a_sorted_view_is_the_sorted_sequence(self):
-        response = self.client.get(
-            self.url, {"sort": "total_score", "dir": "desc", "page": "2"}
-        )
+        response = self.client.get(self.url, {"sort": "total_score", "dir": "desc", "page": "2"})
         self.assertEqual(response.status_code, 200)
         page_two = usernames(response.context["enrollments"])
         self.assertEqual(len(page_two), 5)