diff options
Diffstat (limited to 't')
| -rw-r--r-- | t/test_functionality.py | 83 | ||||
| -rw-r--r-- | t/utils.py | 43 |
2 files changed, 69 insertions, 57 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() @@ -10,26 +10,35 @@ class AergiaUnitTestCase(unittest.TestCase): def setUp(self): self.Aergia.clear() - def assertFuncContains(self, func_name, samples_expected, samples): - samples_actual = self.extract_values_by_func(samples, - func_name) - self.assertTrue(len(samples_expected) == len(samples_actual), - f'{samples_expected} (expected) not length of ' - f'{samples_actual} (actual)') - s1 = sorted(samples_expected) - s2 = sorted(samples_actual) - for v1, v2 in zip(s1, s2): - self.assertRoughlyEqual(v1, v2) - - def assertRoughlyEqual(self, v1, v2): + def assert_reasonable_delay(self, func_name, time_expected, samples): + '''Compares the results reported by Aergia for FUNC_NAME + with time_expected.''' + samples_expected = self.aergia_expected_samples(time_expected) + samples_actual = self.aergia_extract_values_by_func(samples, func_name) + self.assert_roughly_equal(samples_actual, samples_expected) + + def assert_roughly_equal(self, v1, v2): + '''Throws an exception if values V1 and V2 are not within 2.5 + samples of each other.''' a = abs(v1 - v2) - self.assertTrue(a <= 2, f'{v1} (expected) not roughly {v2} (actual)') + self.assertTrue(a <= 2.5, f'{v1} (expected) not roughly {v2} (actual)') - def expected_samples(self, total_seconds): + def aergia_expected_samples(self, total_seconds): + '''Given TOTAL_SECONDS, returns the expected number of samples, using + the sampling interval.''' return (total_seconds / self.interval) - def extract_values_by_func(self, samples, func_name): - return [ + def aergia_extract_values_by_func(self, samples, func_name): + '''Returns the total number of samples for function name, given an + Aergia SAMPLES object.''' + return sum( value for key, value in samples.items() if key.func == func_name - ] + ) + + def yappi_extract_values_by_func(self, stats, func_name): + '''Returns the total number of samples for function name, given a + yappi stats object.''' + for s in stats: + if s.name == func_name: + return s.ttot |