KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_uvloop method in Molotov

Best Python code snippet using molotov_python

test_libraries.py

Source:test_libraries.py

...181        import pinyin182        """)183@importorskip('uvloop')184@skipif_win185def test_uvloop(pyi_builder):186    pyi_builder.test_source("import uvloop")187@importorskip('web3')188def test_web3(pyi_builder):189    pyi_builder.test_source("import web3")190@importorskip('phonenumbers')191def test_phonenumbers(pyi_builder):192    pyi_builder.test_source("""193        import phonenumbers194        number = '+17034820623'195        parsed_number = phonenumbers.parse(number)196        assert(parsed_number.country_code == 1)197        assert(parsed_number.national_number == 7034820623)198        """)199@importorskip('pendulum')...

test_serial.py

Source:test_serial.py

1"""2    A USB Serial converter is needed for testing, simply short RxD and TxD3    and run the test suite with tox4"""5import random6import asyncio7import unittest8import pytest9from asyncserial import Serial10try:11    import uvloop12    test_uvloop = True13except ImportError:14    pass15PORT = "/dev/ttyUSB0"16# Max baudrate of the FTDI UART converter, greatly decreases test time17BAUD = 921600 18test_bytes = bytearray(range(256))19def generate_random_input(n):20    """21    Helper to generate random input22    """23    return bytes(random.sample(test_bytes, k=n))24class Test_serial_asyncio(unittest.TestCase):25    """26    Tests with standard asyncio loop27    """28    29    def setUp(self):30        """31        Initializes eventloop and serial32        """33        self.setup_loop()34        self._serial = Serial(self._loop, PORT, baudrate=BAUD)35    36    def setup_loop(self):37        """38        Setup asyncio eventloop39        """40        self._loop = asyncio.get_event_loop()41    42    def run_async(self, test_async):43        """44        Helper to test async code45        46        :param test_async: async coroutine to run47        """48        self._loop.run_until_complete(test_async)49    def test_await_write(self):50        """51        Test if flush is working correctly52        """53        async def await_write(n):54            tmp = generate_random_input(n)55            # Waits until the output queue is flushed56            await self._serial.write(tmp, await_blocking=True)57            assert self._serial.out_waiting == 058            # Cleanup for th next tests59            await self._serial.read()60        61        62        for n in [100, 200, 256]:63            self.run_async(await_write(n))64    def test_n_write_n_out_async(self):65        """66        Test if every a random bytesequence of length from 1 to 25667        gets written and received correctly68        """69        70        async def n_write_n_out(n):71            tmp = generate_random_input(n)72            await self._serial.write(tmp)73            assert tmp == await self._serial.read(n)74            75        for n in [100, 200, 256]:76            self.run_async(n_write_n_out(n))77    78    def test_read_all(self):79        """80        Test if read automatically detemines how many bytes are in the input81        queue82        """83        async def read_auto(n):84            tmp = generate_random_input(n)85            await self._serial.write(tmp, await_blocking=True)86            # Giving the input buffer some time to refresh87            await asyncio.sleep(0.1)88            # Automaticall determines how many bytes to read89            assert tmp == await self._serial.read()90        for n in [100, 200, 256]:91            self.run_async(read_auto(n))92    93    def test_flush(self):94        """95        Test if flush is working96        """97        async def flush_2048():98            # write 4096 bytes99            await self._serial.write(("a" * 2048).encode("ASCII"), await_blocking=False)100            await self._serial.flush()101            # Output queue should be cleared by now102            assert self._serial.out_waiting == 0103            # Give the input queue some time to keep up104            await asyncio.sleep(0.2)105            assert self._serial.in_waiting == 2048106            # Clear input queue for the next test 107            await self._serial.read()108        self.run_async(flush_2048())109    110    def test_readline(self):111        """112        Test if readline is working113        """114        async def readline_async(line):115            line += b"\r\n"116            await self._serial.write(line, await_blocking=True)117            # Readline should be equal to the line118            assert line == await self._serial.readline()119        120        lines = [121            "Hello World!",122            "This is ASYNCSERIAL",123            "\x13\x17 LEET \xbe\xef"124        ]125        for line in lines:126            self.run_async(readline_async(line.encode()))127@pytest.mark.skipif(not test_uvloop, reason="UVLOOP is not available, skipping test")128class Test_serial_uvloop(Test_serial_asyncio):129    """130    Tests with uvloop131    """132    def setup_loop(self):133        """134        Setup uvloop eventloop135        """...

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.