Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
update sched to 3.13.3
  • Loading branch information
arihant2math committed Apr 22, 2025
commit 1c64bde0ee11e3f457425565d87c88b9f048197d
2 changes: 1 addition & 1 deletion Lib/sched.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
23 changes: 19 additions & 4 deletions Lib/test/test_sched.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 = []
Expand All @@ -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
Expand Down