update sched to 3.13.3

arihant2math · arihant2math · commit 1c64bde0ee11 · 2025-04-21T21:11:42.000-07:00
diff --git a/Lib/sched.py b/Lib/sched.py
@@ -11,7 +11,7 @@
 implement simulated time by writing your own functions.  This can
 also be used to integrate scheduling with STDWIN events; the delay
 function is allowed to modify the queue.  Time can be expressed as
-integers or floating point numbers, as long as it is consistent.
+integers or floating-point numbers, as long as it is consistent.
 
 Events are specified by tuples (time, priority, action, argument, kwargs).
 As in UNIX, lower priority numbers mean higher priority; in this
diff --git a/Lib/test/test_sched.py b/Lib/test/test_sched.py
@@ -58,6 +58,7 @@ def test_enterabs(self):
         scheduler.run()
         self.assertEqual(l, [0.01, 0.02, 0.03, 0.04, 0.05])
 
+    @threading_helper.requires_working_threading()
     def test_enter_concurrent(self):
         q = queue.Queue()
         fun = q.put
@@ -91,10 +92,23 @@ def test_priority(self):
         l = []
         fun = lambda x: l.append(x)
         scheduler = sched.scheduler(time.time, time.sleep)
-        for priority in [1, 2, 3, 4, 5]:
-            z = scheduler.enterabs(0.01, priority, fun, (priority,))
-        scheduler.run()
-        self.assertEqual(l, [1, 2, 3, 4, 5])
+
+        cases = [
+            ([1, 2, 3, 4, 5], [1, 2, 3, 4, 5]),
+            ([5, 4, 3, 2, 1], [1, 2, 3, 4, 5]),
+            ([2, 5, 3, 1, 4], [1, 2, 3, 4, 5]),
+            ([1, 2, 3, 2, 1], [1, 1, 2, 2, 3]),
+        ]
+        for priorities, expected in cases:
+            with self.subTest(priorities=priorities, expected=expected):
+                for priority in priorities:
+                    scheduler.enterabs(0.01, priority, fun, (priority,))
+                scheduler.run()
+                self.assertEqual(l, expected)
+
+                # Cleanup:
+                self.assertTrue(scheduler.empty())
+                l.clear()
 
     def test_cancel(self):
         l = []
@@ -111,6 +125,7 @@ def test_cancel(self):
         scheduler.run()
         self.assertEqual(l, [0.02, 0.03, 0.04])
 
+    @threading_helper.requires_working_threading()
     def test_cancel_concurrent(self):
         q = queue.Queue()
         fun = q.put
