xiio

commit

d39fb2b7a1d88a6447fe62d7206fea21e836dabe

parent

6b9965a49151195a363b7eb06f53a6379e1102ea

Author

Tobias Bengfort <tobias.bengfort@posteo.de>

Date

2026-02-26 18:34

allow to pause execution

The first step is to move the blocking call out of the function. To do
that, we pause execution by using the `yield` keyword. This turns our
function into a generator function. The separate `run()` function will
do the blocking call and resume the generator function once that is
done. The special `StopIteration` exception is raised when the generator
function returns.

Diffstat

2 files changed, 23 insertions, 6 deletions

diff --git a/tests.py b/tests.py

@@ -3,6 +3,8 @@ import time
    3     3 import unittest
    4     4 from unittest import mock
    5     5 
   -1     6 import xiio
   -1     7 
    6     8 
    7     9 class XiioTestCase(unittest.TestCase):
    8    10     @contextlib.contextmanager
@@ -18,11 +20,11 @@ class XiioTestCase(unittest.TestCase):
   18    20 class TestRun(XiioTestCase):
   19    21     def test_sleep(self):
   20    22         def foo():
   21    -1             time.sleep(0.1)
   -1    23             yield 0.1
   22    24             return 'Hello World'
   23    25 
   24    26         with self.assert_duration(0.1):
   25    -1             result = foo()
   -1    27             result = xiio.run(foo())
   26    28         self.assertEqual(result, 'Hello World')
   27    29 
   28    30     def test_runs_cleanup_on_error_while_paused(self):
@@ -30,13 +32,13 @@ class TestRun(XiioTestCase):
   30    32 
   31    33         def foo():
   32    34             try:
   33    -1                 time.sleep(0.1)
   -1    35                 yield 0.1
   34    36             finally:
   35    37                 stack.append(1)
   36    38 
   37    39         with mock.patch('time.sleep', side_effect=KeyboardInterrupt):
   38    40             with self.assertRaises(KeyboardInterrupt):
   39    -1                 foo()
   -1    41                 xiio.run(foo())
   40    42         self.assertEqual(stack, [1])
   41    43 
   42    44     def test_waits_for_cleanup(self):
@@ -44,8 +46,8 @@ class TestRun(XiioTestCase):
   44    46             try:
   45    47                 raise ValueError
   46    48             finally:
   47    -1                 time.sleep(0.1)
   -1    49                 yield 0.1
   48    50 
   49    51         with self.assertRaises(ValueError):
   50    52             with self.assert_duration(0.1):
   51    -1                 foo()
   -1    53                 xiio.run(foo())

diff --git a/xiio.py b/xiio.py

@@ -0,0 +1,15 @@
   -1     1 import time
   -1     2 
   -1     3 
   -1     4 def run(gen):
   -1     5     try:
   -1     6         timeout = next(gen)
   -1     7         while True:
   -1     8             try:
   -1     9                 time.sleep(timeout)
   -1    10             except BaseException as e:
   -1    11                 timeout = gen.throw(e)
   -1    12             else:
   -1    13                 timeout = next(gen)
   -1    14     except StopIteration as e:
   -1    15         return e.value