bernsteinbear.com
this repo has no description
1#!/usr/bin/env python3
2# Copyright (c) Facebook, Inc. and its affiliates. (http://www.facebook.com)
3import asyncio
4import unittest
5
6from test_support import pyro_only
7
8
9class AsyncioTaskTest(unittest.TestCase):
10 class SpecialTask(asyncio.Task):
11 def __init__(self, *args, **kwargs):
12 self.step_counter = 0
13 super().__init__(*args, **kwargs)
14
15 def _step(self, exc=None):
16 self.step_counter += 1
17 super()._step(exc)
18
19 @pyro_only
20 def test_step_of_special_task_raises_exception(self):
21 async def run():
22 return "ok"
23
24 loop = asyncio.new_event_loop()
25 t = self.SpecialTask(run(), loop=loop, name="TestTask")
26 self.assertEqual(t.step_counter, 0)
27 loop.run_until_complete(t)
28 self.assertEqual(t.step_counter, 1)
29
30
31if __name__ == "__main__":
32 unittest.main()