From ac55d9ff0b588b91202ccad72ee71e508e33ad08 Mon Sep 17 00:00:00 2001 From: bd Date: Sat, 19 Jul 2025 22:21:10 -0600 Subject: Reformat repository to allow for new unit tests --- t/test_functionality.py | 57 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) create mode 100644 t/test_functionality.py (limited to 't/test_functionality.py') diff --git a/t/test_functionality.py b/t/test_functionality.py new file mode 100644 index 0000000..5253534 --- /dev/null +++ b/t/test_functionality.py @@ -0,0 +1,57 @@ +import utils +import aiohttp +import asyncio + + +class BasicUsage(utils.AergiaUnitTestCase): + + def test_asyncless(self): + def a(): + x = 100 + while x > 0: + x -= 1 + + self.Aergia.start() + a() + self.Aergia.stop() + + samples = self.Aergia.get_samples() + self.assertFalse(samples) + + def test_sequential_tasks(self): + delay = 0.2 + num_times = 5 + + async def b(tot, num): + await asyncio.sleep(delay) + return tot + num + + async def a(): + tot = 0 + for i in range(num_times): + tot = await b(tot, i) + assert tot == 10 + + self.Aergia.start() + asyncio.run(a()) + self.Aergia.stop() + + samples = self.Aergia.get_samples() + + self.assertFuncContains('b', [self.expected_samples(delay * num_times)], + samples) + + def test_simultaneous_tasks(self): + delay = 0.2 + async def b(): await asyncio.sleep(delay) + async def a(): await asyncio.gather(b(), b(), b()) + + self.Aergia.start() + asyncio.run(a()) + self.Aergia.stop() + + samples = self.Aergia.get_samples() + + print(self.expected_samples(delay * 3)) + self.assertFuncContains('b', [self.expected_samples(delay * 3)], + samples) -- cgit v1.2.3