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

How to use lookup_test_task method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

runner.py

Source:runner.py

...376    def skip(self, context, _):377        self.run(context)378def build_test_session_teardown_task(test_session_setup_task, dependencies):379    return TestSessionTeardownTask(test_session_setup_task, dependencies) if test_session_setup_task else None380def lookup_test_task(tasks, test_path):381    try:382        return next(task for task in tasks if isinstance(task, TestTask) and task.test.path == test_path)383    except StopIteration:384        raise LookupError("Cannot find test '%s' in tasks" % test_path)385def build_tasks(suites, fixture_registry, session_scheduled_fixtures, force_disabled):386    ###387    # Build test session setup task388    ###389    test_session_setup_task = build_test_session_setup_task(session_scheduled_fixtures)390    ###391    # Build suite tasks392    ###393    suite_tasks = []394    for suite in suites:395        suite_tasks.extend(396            build_suite_tasks(397                suite, fixture_registry, session_scheduled_fixtures, test_session_setup_task,398                force_disabled=force_disabled399            )400        )401    ###402    # Build test session teardown task403    ###404    if test_session_setup_task:405        test_session_teardown_dependencies = [406            task for task in suite_tasks if isinstance(task, SuiteEndingTask) and task.suite in suites407        ]408        test_session_teardown_task = build_test_session_teardown_task(409            test_session_setup_task, test_session_teardown_dependencies410        )411    else:412        test_session_teardown_task = None413    ###414    # Get all effective tasks (task != None)415    ###416    task_iter = itertools.chain((test_session_setup_task,), suite_tasks, (test_session_teardown_task,))417    tasks = list(filter(bool, task_iter))418    ###419    # Add extra dependencies in tasks for tests that depend on other tests420    ###421    for test in flatten_tests(suites):422        if not test.dependencies:423            continue424        test_task = lookup_test_task(tasks, test.path)425        for dep_test_path in test.dependencies:426            try:427                dep_test = lookup_test_task(tasks, dep_test_path)428            except LookupError:429                raise UserError(430                    "Cannot find dependency test '%s' for '%s', "431                    "either the test does not exist or is not going to be run" % (dep_test_path, test.path)432                )433            test_task.dependencies.append(dep_test)434    ###435    # Return tasks436    ###437    return tasks438def _run_suites(suites, fixture_registry, pre_run_scheduled_fixtures, session,439                force_disabled=False, stop_on_failure=False, nb_threads=1):440    # build tasks and run context441    session_scheduled_fixtures = fixture_registry.get_fixtures_scheduled_for_session(...

threaded_driver.py

Source:threaded_driver.py

...34# future for request tracing in case we'll ever need more accurate, per-request,35# individual measurements. In that case, both server and client would label36# durations using the request number so then when analysing the data we can37# tell exactly, for each request, how much time was spent where.38def lookup_test_task(test_id: str) -> TestTask:39    tasks = {40        VERSION_TEST: run_version_test,41        NOTIFY_TEST: run_notify_test42    }43    return tasks[test_id]44class ThreadedDriver(Driver):45    def __init__(self,46                 max_workers: int = MAX_THREAD_WORKERS,47                 requests_number: int = REQUESTS_N,48                 monitoring_dir: str = MONITORING_DIR):49        super().__init__(monitoring_dir=monitoring_dir)50        self._max_workers = max_workers51        self._request_n = requests_number52    def _do_run(self, test_id: str) -> TestRunResults:53        test_task = lookup_test_task(test_id)54        with ThreadPoolExecutor(max_workers=self._max_workers) as executor:55            return executor.map(test_task, range(self._request_n))56if __name__ == "__main__":...

asyncio_driver.py

Source:asyncio_driver.py

...23# in the event loop. While there's no accurate way of timing coroutines that24# I know, in the specific case of aiohttp, we could provide some half-meaningful25# measurements:26# - https://stackoverflow.com/questions/4600474527def lookup_test_task(test_id: str) -> TestTask:28    tasks = {29        VERSION_TEST: run_version_test,30        NOTIFY_TEST: run_notify_test31    }32    return tasks[test_id]33async def do_many(task: TestTask, how_many: int) -> TestRunResults:34    async with aiohttp.ClientSession() as session:35        tasks = [task(session) for _ in range(how_many)]36        return await asyncio.gather(*tasks, return_exceptions=True)37class AsyncioDriver(Driver):38    def _do_run(self, test_id: str) -> TestRunResults:39        test_task = lookup_test_task(test_id)40        return asyncio.run(do_many(test_task, REQUESTS_N))41if __name__ == "__main__":...

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.