How to use wait_task method in avocado

Best Python code snippet using avocado_python

sound_alsa_test.py

Source:sound_alsa_test.py

1import asyncio2import os3import sys4import time5import unittest6import asynctest  # type: ignore7import pytest8from .utils import close_old_async_connections9@pytest.mark.skipif(10    sys.platform != "linux", reason="Alsa is only available on Linux"11)12@pytest.mark.django_db13class TestPlaySound(unittest.TestCase):14    def setUp(self):15        from nabd.nabio import NabIO16        from nabd.sound_alsa import SoundAlsa17        self.loop = asyncio.new_event_loop()18        asyncio.set_event_loop(self.loop)19        try:20            if SoundAlsa.sound_configuration()[1] == "sndrpihifiberry":21                model = NabIO.MODEL_201822            else:23                if SoundAlsa.sound_configuration()[1] == "tagtagtagsound":24                    model = NabIO.MODEL_2019_TAGTAG25                else:26                    raise unittest.SkipTest("No compatible sound card found")27        except RuntimeError as error:28            raise unittest.SkipTest(29                "Runtime error getting sound card %s" % str(error)30            )31        self.sound = SoundAlsa(model)32    def tearDown(self):33        close_old_async_connections()34    def test_mp3(self):35        start_task = self.loop.create_task(36            self.sound.start_playing("choreographies/3notesE5A5C6.mp3")37        )38        self.loop.run_until_complete(start_task)39        wait_task = self.loop.create_task(self.sound.wait_until_done())40        self.loop.run_until_complete(wait_task)41    def test_two_mp3s(self):42        start_task = self.loop.create_task(43            self.sound.start_playing("choreographies/3notesA4G5G5.mp3")44        )45        self.loop.run_until_complete(start_task)46        wait_task = self.loop.create_task(self.sound.wait_until_done())47        self.loop.run_until_complete(wait_task)48        start_task = self.loop.create_task(49            self.sound.start_playing("choreographies/2notesE5E4.mp3")50        )51        self.loop.run_until_complete(start_task)52        wait_task = self.loop.create_task(self.sound.wait_until_done())53        self.loop.run_until_complete(wait_task)54    def test_two_wavs(self):55        start_task = self.loop.create_task(56            self.sound.start_playing("nabmastodond/communion.wav")57        )58        self.loop.run_until_complete(start_task)59        wait_task = self.loop.create_task(self.sound.wait_until_done())60        self.loop.run_until_complete(wait_task)61        start_task = self.loop.create_task(62            self.sound.start_playing("nabmastodond/communion.wav")63        )64        self.loop.run_until_complete(start_task)65        wait_task = self.loop.create_task(self.sound.wait_until_done())66        self.loop.run_until_complete(wait_task)67    def test_wav(self):68        start_task = self.loop.create_task(69            self.sound.start_playing("nabmastodond/communion.wav")70        )71        self.loop.run_until_complete(start_task)72        wait_task = self.loop.create_task(self.sound.wait_until_done())73        self.loop.run_until_complete(wait_task)74    def test_stream_mp3(self):75        start_task = self.loop.create_task(76            self.sound.start_playing(77                "https://github.com/nabaztag2018/pynab/raw/master/nabd/"78                "sounds/choreographies/3notesE5A5C6.mp3"79            )80        )81        self.loop.run_until_complete(start_task)82        wait_task = self.loop.create_task(self.sound.wait_until_done())83        self.loop.run_until_complete(wait_task)84@pytest.mark.skipif(85    sys.platform != "linux", reason="Alsa is only available on Linux"86)87@pytest.mark.django_db88class TestWaitUntilComplete(asynctest.TestCase):89    def setUp(self):90        from nabd.nabio import NabIO91        from nabd.sound_alsa import SoundAlsa92        # Workaround for asynctest bug?93        os.environ["DJANGO_ALLOW_ASYNC_UNSAFE"] = "yes"94        try:95            if SoundAlsa.sound_configuration()[1] == "sndrpihifiberry":96                model = NabIO.MODEL_201897            else:98                if SoundAlsa.sound_configuration()[1] == "tagtagtagsound":99                    model = NabIO.MODEL_2019_TAGTAG100                else:101                    raise unittest.SkipTest("No compatible sound card found")102        except RuntimeError as error:103            raise unittest.SkipTest(104                "Runtime error getting sound card %s" % str(error)105            )106        self.sound = SoundAlsa(model)107    async def test_wait_until_done(self):108        before = time.time()109        await self.sound.start_playing("fr_FR/asr/failed/5.mp3")110        await self.sound.wait_until_done()111        after = time.time()112        self.assertGreater(after - before, 3.0)113    async def test_wait_until_done_with_event_not_set(self):114        before = time.time()115        await self.sound.start_playing("fr_FR/asr/failed/5.mp3")116        event = asyncio.Event()117        await self.sound.wait_until_done(event)118        after = time.time()119        self.assertGreater(after - before, 3.0)120    async def test_wait_until_done_with_event_set(self):121        before = time.time()122        await self.sound.start_playing("fr_FR/asr/failed/5.mp3")123        event = asyncio.Event()124        self.loop.call_later(1.5, lambda: event.set())125        await self.sound.wait_until_done(event)126        after = time.time()127        self.assertLess(after - before, 3.0)128@pytest.mark.skipif(129    sys.platform != "linux", reason="Alsa is only available on Linux"130)131@pytest.mark.django_db132class TestRecord(unittest.TestCase):133    def setUp(self):134        from nabd.nabio import NabIO135        from nabd.sound_alsa import SoundAlsa136        self.loop = asyncio.new_event_loop()137        asyncio.set_event_loop(self.loop)138        try:139            if SoundAlsa.sound_configuration()[1] != "tagtagtagsound":140                raise unittest.SkipTest(141                    "Test should be run on a 2019 card only"142                )143        except RuntimeError as error:144            raise unittest.SkipTest(145                "Runtime error getting sound card %s" % str(error)146            )147        self.sound = SoundAlsa(NabIO.MODEL_2019_TAGTAG)148        self.recorded_raw = open("test_recording.raw", "wb")149    def tearDown(self):150        self.recorded_raw.close()151        close_old_async_connections()152    def test_recording_playback(self):153        start_task = self.loop.create_task(154            self.sound.start_playing("asr/listen.mp3")155        )156        self.loop.run_until_complete(start_task)157        wait_task = self.loop.create_task(self.sound.wait_until_done())158        self.loop.run_until_complete(wait_task)159        start_task = self.loop.create_task(160            self.sound.start_recording(self.record_cb)161        )162        self.loop.run_until_complete(start_task)163        time.sleep(5)164        stop_task = self.loop.create_task(self.sound.stop_recording())165        self.loop.run_until_complete(stop_task)166        start_task = self.loop.create_task(167            self.sound.start_playing("asr/acquired.mp3")168        )169        self.loop.run_until_complete(start_task)170        wait_task = self.loop.create_task(self.sound.wait_until_done())171        self.loop.run_until_complete(wait_task)172    def record_cb(self, data, finalize):...

mouse.py

Source:mouse.py

...23        }24        self._buttonStateChanges: Final[List[Tuple[Button, int]]] = []25    @property26    def visible(self) -> bool:27        return wait_task(28            lambda: glfw.get_input_mode(Window.handle, glfw.CURSOR)29            == glfw.CURSOR_NORMAL30        )31    def _show(self) -> None:32        if glfw.get_input_mode(Window.handle, glfw.CURSOR) == glfw.CURSOR_DISABLED:33            new_pos = Window.bounds.size * 0.534            self._posChanges[:] = new_pos35            self._pos[:] = new_pos36        glfw.set_input_mode(Window.handle, glfw.CURSOR, glfw.CURSOR_NORMAL)37    def show(self) -> None:38        run_task(self._show)39class ButtonInput(Input):40    def __init__(self):41        super().__init__()42        self.down_pos: Vector2 = Vector2()43@unique44class Button(Enum):45    UNKNOWN = -146    LEFT = glfw.MOUSE_BUTTON_LEFT47    RIGHT = glfw.MOUSE_BUTTON_RIGHT48    MIDDLE = glfw.MOUSE_BUTTON_MIDDLE49    FOUR = glfw.MOUSE_BUTTON_450    FIVE = glfw.MOUSE_BUTTON_551    SIX = glfw.MOUSE_BUTTON_652    SEVEN = glfw.MOUSE_BUTTON_753    EIGHT = glfw.MOUSE_BUTTON_854# noinspection PyFinal55class Shape:56    ARROW_CURSOR: Final[Shape]57    IBEAM_CURSOR: Final[Shape]58    CROSSHAIR_CURSOR: Final[Shape]59    HAND_CURSOR: Final[Shape]60    HRESIZE_CURSOR: Final[Shape]61    VRESIZE_CURSOR: Final[Shape]62    @classmethod63    def create(cls, name: str, image: Image, x_hot: int, y_hot: int) -> Shape:64        return cls(name, wait_task(lambda: glfw.create_cursor(image, x_hot, y_hot)))65    def __init__(self, name: str, handle: int):66        self._name = name67        self._handle = handle68    def __repr__(self) -> str:69        return f"Shape(name={self._name})"70    def destroy(self) -> None:71        run_task(lambda: glfw.destroy_cursor(self._handle))72# noinspection PyFinal73Shape.ARROW_CURSOR = Shape(74    "ARROW", wait_task(lambda: glfw.create_standard_cursor(glfw.ARROW_CURSOR))75)76# noinspection PyFinal77Shape.IBEAM_CURSOR = Shape(78    "IBEAM", wait_task(lambda: glfw.create_standard_cursor(glfw.IBEAM_CURSOR))79)80# noinspection PyFinal81Shape.CROSSHAIR_CURSOR = Shape(82    "CROSSHAIR", wait_task(lambda: glfw.create_standard_cursor(glfw.CROSSHAIR_CURSOR))83)84# noinspection PyFinal85Shape.HAND_CURSOR = Shape(86    "HAND", wait_task(lambda: glfw.create_standard_cursor(glfw.HAND_CURSOR))87)88# noinspection PyFinal89Shape.HRESIZE_CURSOR = Shape(90    "HRESIZE", wait_task(lambda: glfw.create_standard_cursor(glfw.HRESIZE_CURSOR))91)92# noinspection PyFinal93Shape.VRESIZE_CURSOR = Shape(94    "VRESIZE", wait_task(lambda: glfw.create_standard_cursor(glfw.VRESIZE_CURSOR))...

test_graceful_shutdown.py

Source:test_graceful_shutdown.py

1import asyncio2import os3import signal4from unittest import IsolatedAsyncioTestCase, mock5from ustack_tornado_shutdown import graceful_shutdown6class TestGracefulShutdown(IsolatedAsyncioTestCase):7    async def send_sigterm(self, wait_task: asyncio.Task[None]) -> bool:8        done = wait_task.done()9        os.kill(os.getpid(), signal.SIGTERM)10        return done11    @mock.patch("asyncio.sleep")12    async def test_on_sigterm_returns_after_sigterm_is_trapped(13        self,14        mock_sleep: mock.Mock15    ) -> None:16        wait_task = asyncio.create_task(graceful_shutdown.on_sigterm())17        finish_task = asyncio.create_task(self.send_sigterm(wait_task))18        await wait_task19        assert not await finish_task20        mock_sleep.assert_not_called()21    @mock.patch("asyncio.sleep")22    async def test_on_sigterm_calls_function_before_returning_when_specified(23        self,24        mock_sleep: mock.Mock25    ) -> None:26        fn = mock.Mock()27        wait_task = asyncio.create_task(graceful_shutdown.on_sigterm(fn))28        finish_task = asyncio.create_task(self.send_sigterm(wait_task))29        await wait_task30        await finish_task31        fn.assert_called_once_with()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.