diff options
Diffstat (limited to 't/test_functionality.py')
| -rw-r--r-- | t/test_functionality.py | 83 |
1 files changed, 43 insertions, 40 deletions
diff --git a/t/test_functionality.py b/t/test_functionality.py index 75f0137..f0fa6ed 100644 --- a/t/test_functionality.py +++ b/t/test_functionality.py @@ -42,9 +42,7 @@ class BasicUsage(utils.AergiaUnitTestCase): samples = self.Aergia.get_samples() - self.assertFuncContains('b', - [self.expected_samples(delay * num_times)], - samples) + self.assert_reasonable_delay('b', delay * num_times, samples) def test_simultaneous_tasks(self): delay = 0.2 @@ -57,13 +55,12 @@ class BasicUsage(utils.AergiaUnitTestCase): samples = self.Aergia.get_samples() - self.assertFuncContains('b', [self.expected_samples(delay * 3)], - samples) + self.assert_reasonable_delay('b', delay * 3, samples) # the gather function is technically waiting for all tasks to finish. # This might be seen as unintuitive though I don't want to bias the # results by adding logic to add artificial consistency. # profiling does not mean obscuring implementation details. - self.assertFuncContains('a', [self.expected_samples(delay)], samples) + self.assert_reasonable_delay('a', delay, samples) def test_subthread_task(self): delay = 0.2 @@ -79,11 +76,10 @@ class BasicUsage(utils.AergiaUnitTestCase): self.Aergia.stop() samples = self.Aergia.get_samples() - self.assertFuncContains('c', [], samples) - self.assertFuncContains('b', [self.expected_samples(delay * 3)], - samples) + self.assert_reasonable_delay('c', 0, samples) + self.assert_reasonable_delay('b', delay * 3, samples) # see comment on `test_simultaneous_tasks'. - self.assertFuncContains('a', [self.expected_samples(delay)], samples) + self.assert_reasonable_delay('a', delay, samples) def test_eager_task(self): delay = 0.2 @@ -97,7 +93,7 @@ class BasicUsage(utils.AergiaUnitTestCase): self.Aergia.stop() samples = self.Aergia.get_samples() - self.assertFuncContains('a', [self.expected_samples(delay)], samples) + self.assert_reasonable_delay('a', delay, samples) def test_async_eager_scheduled_woven(self): d1 = 0.2 @@ -118,11 +114,7 @@ class BasicUsage(utils.AergiaUnitTestCase): self.Aergia.stop() samples = self.Aergia.get_samples() - self.assertFuncContains('a', [self.expected_samples(d1), - self.expected_samples(d2), - self.expected_samples(d3), - self.expected_samples(d4), - ], samples) + self.assert_reasonable_delay('a', d1 + d2 + d3 + d4, samples) def test_async_generator(self): delay = 0.2 @@ -147,15 +139,11 @@ class BasicUsage(utils.AergiaUnitTestCase): # also does not work at all with yappi. # doing so would be unique to this profiler. - # self.assertFuncContains('b', - # [self.expected_samples(delay * num_times)], - # samples) - # self.assertFuncContains('a', [], samples) + # self.assert_reasonable_delay('b', delay * num_times, samples) + # self.assert_reasonable_delay('a', [], samples) - self.assertFuncContains('a', - [self.expected_samples(delay * num_times)], - samples) - self.assertFuncContains('b', [], samples) + self.assert_reasonable_delay('a', delay * num_times, samples) + self.assert_reasonable_delay('b', 0, samples) def test_async_gen_and_comp(self): delay = 0.2 @@ -178,11 +166,9 @@ class BasicUsage(utils.AergiaUnitTestCase): # also does not work at all with yappi. # doing so would be unique to this profiler. - self.assertFuncContains('b', [], samples) - self.assertFuncContains('a', [], samples) - self.assertFuncContains('<listcomp>', - [self.expected_samples(delay * num_times)], - samples) + self.assert_reasonable_delay('b', 0, samples) + self.assert_reasonable_delay('a', 0, samples) + self.assert_reasonable_delay('<listcomp>', delay * num_times, samples) def test_deep_await(self): delay = 0.2 @@ -197,9 +183,10 @@ class BasicUsage(utils.AergiaUnitTestCase): samples = self.Aergia.get_samples() - self.assertFuncContains('c', [self.expected_samples(delay)], samples) - self.assertFuncContains('b', [], samples) - self.assertFuncContains('a', [], samples) + self.assert_reasonable_delay( + 'c', delay, samples) + self.assert_reasonable_delay('b', 0, samples) + self.assert_reasonable_delay('a', 0, samples) def test_task_groups(self): d1 = 0.2 @@ -221,16 +208,16 @@ class BasicUsage(utils.AergiaUnitTestCase): samples = self.Aergia.get_samples() - self.assertFuncContains('d', [self.expected_samples(d1)], samples) - self.assertFuncContains('c', [self.expected_samples(d2)], samples) - self.assertFuncContains('b', [self.expected_samples(d3)], samples) + self.assert_reasonable_delay('d', d1, samples) + self.assert_reasonable_delay('c', d2, samples) + self.assert_reasonable_delay('b', d3, samples) # the task group is technically waiting for all tasks to finish. # same as task.gather # This might be seen as unintuitive, (especially considering how # the next test works), though I don't want to bias the results # by adding logic to add artificial consistency. # Both are reporting correctly. - self.assertFuncContains('a', [self.expected_samples(d2)], samples) + self.assert_reasonable_delay('a', d2, samples) def test_task_groups_cancel(self): d1 = 0.1 @@ -258,9 +245,25 @@ class BasicUsage(utils.AergiaUnitTestCase): samples = self.Aergia.get_samples() - self.assertFuncContains('d', [], samples) - self.assertFuncContains('c', [self.expected_samples(d1)], samples) - self.assertFuncContains('b', [self.expected_samples(d3)], samples) + self.assert_reasonable_delay('d', 0, samples) + self.assert_reasonable_delay('c', d1, samples) + self.assert_reasonable_delay('b', d3, samples) # this time is attached to the sleep call itself. Aergia's print # function would confirm this! - self.assertFuncContains('a', [self.expected_samples(d3)], samples) + self.assert_reasonable_delay('a', d3, samples) + + def test_asyncio_recursion(self): + delay = 0.1 + + async def a(n): + if n <= 0: + return + await asyncio.sleep(delay) + await a(n - 1) + await a(n - 2) + + self.Aergia.start() + asyncio.run(a(3)) + self.Aergia.stop() + + samples = self.Aergia.get_samples() |