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

How to use queue_log_record method in autotest

Best Python code snippet using autotest_python

subprocess_wrapper_windows.py

Source:subprocess_wrapper_windows.py

1"""Subprocess wrapper for Windows."""2import asyncio3import logging4import os5import queue6import re7import sys8import time9from logging import DEBUG, LogRecord, basicConfig, getLogger, handlers10from multiprocessing import Manager11import psutil12from tests.testlibraries import SECOND_SLEEP_FOR_TEST_LONG13from tests.testlibraries.example_use_case import example_use_case_interrupt14from tests.testlibraries.instance_resource import InstanceResource15from tests.testlibraries.keyboardinterrupter.keyboard_interrupter import KeyboardInterrupter16from tests.testlibraries.keyboardinterrupter.local_socket import LocalSocket17async def get_process_id() -> int:18    """The process to get process id to kill."""19    print("Await sleep")20    logger = getLogger(__name__)21    logger.debug("Get process id")22    # Wait for starting subprocess23    # otherwise, time.sleep() will block starting subprocess.24    current_process = psutil.Process(os.getpid())25    while len(current_process.children()) < 2:26        print(len(current_process.children()))27        await asyncio.sleep(0.01)28    logger.debug("Start sleep")29    time.sleep(SECOND_SLEEP_FOR_TEST_LONG)30    print("Kill all processes in this window.")31    return 032def worker_configurer() -> None:33    logger = getLogger()34    logger.setLevel(DEBUG)35def listener_configurer(queue_log_record: queue.Queue[LogRecord]) -> handlers.QueueListener:36    console_handler = logging.StreamHandler()37    console_handler.setFormatter(logging.Formatter("[%(levelname)s/%(processName)s] %(message)s"))38    return handlers.QueueListener(queue_log_record, console_handler)39class Checker:40    """Checks logger output from queue."""41    def __init__(self) -> None:42        self.ffmpeg_process_quit_finish = False43        self.ffmpeg_closed_log = False44        self.logger = getLogger(__name__)45    def check(self, queue_log_record: queue.Queue[LogRecord]) -> None:46        """Checks logger output from queue."""47        log_record: LogRecord = queue_log_record.get()48        self.logger.handle(log_record)49        if "FFmpeg process quit finish" in log_record.message:50            self.ffmpeg_process_quit_finish = True51        if re.search(InstanceResource.REGEX_STDERR_FFMPEG_LASTLINE, log_record.message,) is not None:52            self.ffmpeg_closed_log = True53def check_log(queue_log_record: "queue.Queue[LogRecord]") -> None:54    """Checks log."""55    logger = getLogger(__name__)56    checker = Checker()57    try:58        while not queue_log_record.empty():59            checker.check(queue_log_record)60        assert checker.ffmpeg_process_quit_finish61        assert checker.ffmpeg_closed_log62    except BaseException:63        logger.exception("Error!")64        time.sleep(10)65        raise66def main() -> None:67    """Tests CTRL + C."""68    manager = Manager()69    queue_log_record = manager.Queue(-1)70    path_file_input = LocalSocket.receive()71    LocalSocket.send("Next")72    path_file_output = LocalSocket.receive()73    basicConfig(stream=sys.stdout, level=DEBUG)74    logger = getLogger(__name__)75    try:76        KeyboardInterrupter(77            example_use_case_interrupt(path_file_input, path_file_output, queue_log_record, worker_configurer),78            get_process_id(),79        ).test_keyboard_interrupt()80    except KeyboardInterrupt:81        logger.debug("__main__ KeyboardInterrupt")82        check_log(queue_log_record)83        LocalSocket.send("Test succeed")84        logger.debug("__main__ sleep")85    finally:86        time.sleep(10)87if __name__ == "__main__":...

keyboard_interrupter.py

Source:keyboard_interrupter.py

1"""To keep task property even raise KeyboardInterrupt."""2import asyncio3import os4# Reason: For type hint. pylint: disable=unused-import5import queue6import signal7import traceback8from asyncio.tasks import Task9# Reason: For type hint. pylint: disable=unused-import10from logging import DEBUG, LogRecord, getLogger11from multiprocessing import Manager12from typing import Any, Awaitable, Callable, Optional13from tests.testlibraries.example_use_case import example_use_case14from tests.testlibraries.local_socket import LocalSocket15class KeyboardInterrupter:16    """To keep task property even raise KeyboardInterrupt."""17    def __init__(self, get_process_id: Awaitable[int]) -> None:18        self.get_process_id = get_process_id19        # Reason: pytest bug. pylint: disable=unsubscriptable-object20        self.task: Optional[Task[Any]] = None21        self.manager = Manager()22        self.queue_log_record: "queue.Queue[LogRecord]" = self.manager.Queue(-1)23        self.logger = getLogger(__name__)24    def test_keyboard_interrupt(self) -> None:25        """Tests keyboard interrupt and send response to pytest by socket when succeed."""26        try:27            self.test()28        except BaseException as error:29            self.logger.exception(error)30            traceback.print_exc()31            LocalSocket.send("Test failed")32            raise33        else:34            LocalSocket.send("Test succeed")35        finally:36            asyncio.run(asyncio.sleep(10))37    async def keyboard_interrupt(self) -> None:38        """Simulates keyboard interrupt by CTRL_C_EVENT."""39        print("Create task")40        coroutine = self.run_example_use_case_and_raise(self.queue_log_record, set_log_level_as_debug)41        self.task = asyncio.create_task(coroutine)42        process_id = await self.get_process_id43        # Reason: only for Windows. pylint: disable=no-member44        os.kill(process_id, signal.CTRL_C_EVENT)  # type: ignore45        print("Await task")46        await self.task47    @staticmethod48    async def run_example_use_case_and_raise(49        queue_logger: "queue.Queue[LogRecord]", worker_configurer: Optional[Callable[[], Any]]50    ) -> None:51        """The example use case of ProcessTaskPoolExecutor for E2E testing in case of cancel."""52        await example_use_case(queue_logger=queue_logger, configurer=worker_configurer)53        raise Exception("Keyboard interrupt isn't received.")54    def test(self) -> None:55        try:56            asyncio.run(self.keyboard_interrupt())57        except KeyboardInterrupt:58            pass59        assert self.expected_log_exists(), "Expected log not found"60    def expected_log_exists(self) -> bool:61        """62        True: Expected log exists.63        False: Expected log does not exist.64        """65        while not self.queue_log_record.empty():66            message = self.queue_log_record.get().message67            print(message)68            if message == "CPU-bound: KeyboardInterupt":69                return True70        return False71def set_log_level_as_debug() -> None:72    root_logger = getLogger()...

example_use_case.py

Source:example_use_case.py

1"""The example use case of FFmpegCroutine for E2E testing in case of interrupt."""2import queue3from logging import LogRecord, getLogger4from pathlib import Path5from typing import Any, Callable6# Reason: mypy issue: https://github.com/python/mypy/issues/101987from asynccpu import ProcessTaskPoolExecutor  # type: ignore8# Reason: Following export method in __init__.py from Effective Python 2nd Edition item 859from asyncffmpeg import FFmpegCoroutineFactory  # type: ignore10from tests.testlibraries.create_stream_spec_croutine import CreateStreamSpecCoroutineFilter11from tests.testlibraries.keyboardinterrupter.local_socket import LocalSocket12async def example_use_case_interrupt(13    path_file_input: str,14    path_file_output: str,15    queue_log_record: queue.Queue[LogRecord],16    configurer: Callable[..., Any],17) -> None:18    """The example use case of FFmpegCroutine for E2E testing in case of interrupt."""19    logger = getLogger(__name__)20    logger.info("Example use case interrupt start")21    with ProcessTaskPoolExecutor(22        max_workers=3, cancel_tasks_when_shutdown=True, queue=queue_log_record, configurer=configurer23    ) as executor:24        ffmpeg_coroutine = FFmpegCoroutineFactory.create()25        coroutine_create_stream_spec_filter = CreateStreamSpecCoroutineFilter(path_file_input, path_file_output)26        future = executor.create_process_task(ffmpeg_coroutine.execute, coroutine_create_stream_spec_filter.create)27        await future28        raise Exception("Failed")29async def example_use_case(path_file_input: Path, path_file_output: Path) -> None:30    """The example use case of FFmpegCroutine for E2E testing in case of interrupt."""31    with ProcessTaskPoolExecutor(max_workers=1, cancel_tasks_when_shutdown=True) as executor:32        task = executor.create_process_task(33            FFmpegCoroutineFactory.create().execute,34            CreateStreamSpecCoroutineFilter(path_file_input, path_file_output).create,35        )36        LocalSocket.send("Ready")...

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.