How to use _load_test method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

gui_2.py

Source:gui_2.py

1import asyncio2from concurrent.futures import Future3from asyncio import AbstractEventLoop4from typing import Callable, Optional5from aiohttp import ClientSession6from queue import Queue7from tkinter import Tk8from tkinter import Label9from tkinter import Entry10from tkinter import ttk11from threading import Thread12class StressTest:13    def __init__(self, loop: AbstractEventLoop, url: str, total_requests: int,14                 callback: Callable[[int, int], None]):15        self._completed_requests: int = 016        self._load_test_future: Optional[Future] = None17        self._loop = loop18        self._url = url19        self._total_requests = total_requests20        self._callback = callback21        self._refresh_rate = total_requests // 10022    def start(self):23        future = asyncio.run_coroutine_threadsafe(self._make_requests(), self._loop)24        self._load_test_future = future25    def cancel(self):26        if self._load_test_future:27            self._loop.call_soon_threadsafe(self._load_test_future.cancel)28    async def _get_url(self, session: ClientSession, url: str):29        try:30            await session.get(url)31        except Exception as e:32            print(e)33        self._completed_requests = self._completed_requests + 134        if self._completed_requests % self._refresh_rate == 0 \35           or self._completed_requests == self._total_requests:36            self._callback(self._completed_requests, self._total_requests)37    async def _make_requests(self):38        async with ClientSession() as session:39            reqs = [self._get_url(session, self._url) for _ in range(self._total_requests)]40            await asyncio.gather(*reqs)41class LoadTester(Tk):42    def __init__(self, loop, *args, **kwargs):43        Tk.__init__(self, *args, **kwargs)44        self._queue = Queue()45        self._refresh_ms = 2546        self._loop = loop47        self._load_test: Optional[StressTest] = None48        self.title('URL Requester')49        self._url_label = Label(self, text="URL:")50        self._url_label.grid(column=0, row=0)51        self._url_field = Entry(self, width=10)52        self._url_field.grid(column=1, row=0)53        self._request_label = Label(self, text="Number of requests:")54        self._request_label.grid(column=0, row=1)55        self._request_field = Entry(self, width=10)56        self._request_field.grid(column=1, row=1)57        self._submit = ttk.Button(self, text="Submit", command=self._start)58        self._submit.grid(column=2, row=1)59        self._pb_label = Label(self, text="Progress:")60        self._pb_label.grid(column=0, row=3)61        self._pb = ttk.Progressbar(self, orient="horizontal", length=200, mode="determinate")62        self._pb.grid(column=1, row=3, columnspan=2)63    def _update_bar(self, pct: int):64        if pct == 100:65            self._load_test = None66            self._submit['text'] = 'Submit'67        else:68            self._pb['value'] = pct69            self.after(self._refresh_ms, self._poll_queue)70    def _queue_update(self, completed_requests: int, total_requests: int):71        self._queue.put(int(completed_requests / total_requests * 100))72    def _poll_queue(self):73        if not self._queue.empty():74            percent_complete = self._queue.get()75            self._update_bar(percent_complete)76        else:77            if self._load_test:78                self.after(self._refresh_ms, self._poll_queue)79    def _start(self):80        if self._load_test is None:81            self._submit['text'] = 'Cancel'82            test = StressTest(self._loop,83                              self._url_field.get(),84                              int(self._request_field.get()),85                              self._queue_update)86            self.after(self._refresh_ms, self._poll_queue)87            test.start()88            self._load_test = test89        else:90            self._load_test.cancel()91            self._load_test = None92            self._submit['text'] = 'Submit'93class ThreadedEventLoop(Thread):94    def __init__(self, loop: AbstractEventLoop):95        super().__init__()96        self._loop = loop97        self.daemon = True98    def run(self):99        self._loop.run_forever()100loop = asyncio.new_event_loop()101asyncio_thread = ThreadedEventLoop(loop)102asyncio_thread.start()103app = LoadTester(loop)...

listing_7_14.py

Source:listing_7_14.py

1from queue import Queue2from tkinter import Tk3from tkinter import Label4from tkinter import Entry5from tkinter import ttk6from typing import Optional7from chapter_07.listing_7_13 import StressTest8class LoadTester(Tk):9    def __init__(self, loop, *args, **kwargs):  # A10        Tk.__init__(self, *args, **kwargs)11        self._queue = Queue()12        self._refresh_ms = 2513        self._loop = loop14        self._load_test: Optional[StressTest] = None15        self.title('URL Requester')16        self._url_label = Label(self, text="URL:")17        self._url_label.grid(column=0, row=0)18        self._url_field = Entry(self, width=10)19        self._url_field.grid(column=1, row=0)20        self._request_label = Label(self, text="Number of requests:")21        self._request_label.grid(column=0, row=1)22        self._request_field = Entry(self, width=10)23        self._request_field.grid(column=1, row=1)24        self._submit = ttk.Button(self, text="Submit", command=self._start)  # B25        self._submit.grid(column=2, row=1)26        self._pb_label = Label(self, text="Progress:")27        self._pb_label.grid(column=0, row=3)28        self._pb = ttk.Progressbar(self, orient="horizontal", length=200, mode="determinate")29        self._pb.grid(column=1, row=3, columnspan=2)30    def _update_bar(self, pct: int):  # C31        if pct == 100:32            self._load_test = None33            self._submit['text'] = 'Submit'34        else:35            self._pb['value'] = pct36            self.after(self._refresh_ms, self._poll_queue)37    def _queue_update(self, completed_requests: int, total_requests: int):  # D38        self._queue.put(int(completed_requests / total_requests * 100))39    def _poll_queue(self):  # E40        if not self._queue.empty():41            percent_complete = self._queue.get()42            self._update_bar(percent_complete)43        else:44            if self._load_test:45                self.after(self._refresh_ms, self._poll_queue)46    def _start(self):  # F47        if self._load_test is None:48            self._submit['text'] = 'Cancel'49            test = StressTest(self._loop,50                              self._url_field.get(),51                              int(self._request_field.get()),52                              self._queue_update)53            self.after(self._refresh_ms, self._poll_queue)54            test.start()55            self._load_test = test56        else:57            self._load_test.cancel()58            self._load_test = None...

dataset_warehouse.py

Source:dataset_warehouse.py

...4PATH = os.path.join(os.getcwd(), 'nlu_datasets/intent_classification/')5# ATIS dataset6atis = Dataset('atis')7atis._load_train(PATH + 'atis_train.json')8atis._load_test(PATH + 'atis_test.json')9atis._form_all()10# Snips dataset11snips = Dataset('snips')12snips._load_train(PATH + 'snips_train.json')13snips._load_test(PATH + 'snips_test.json')14snips._form_all()15# Banking77 dataset16banking77 = Dataset('banking77')17banking77._load_train(PATH + 'banking77_train.json')18banking77._load_test(PATH + 'banking77_test.json')19banking77._form_all()20# Clinc-150 dataset21clinc150 = Dataset('clinc-150')22clinc150._load_train(PATH + 'clinc150_train.json')23clinc150._load_test(PATH + 'clinc150_test.json')24clinc150._form_all()25# HWU-64 Dataset26hwu64 = Dataset('hwu-64')27hwu64._load_all(PATH + 'hwu64.json')28# AskUbuntu Dataset29ask_ubuntu = Dataset('ask_ubuntu')30ask_ubuntu._load_train(PATH + 'ask_ubuntu_train.json')31ask_ubuntu._load_test(PATH + 'ask_ubuntu_test.json')32ask_ubuntu._form_all()33# Chatbot Dataset34chatbot = Dataset('chatbot')35chatbot._load_train(PATH + 'chatbot_train.json')36chatbot._load_test(PATH + 'chatbot_test.json')37chatbot._form_all()38# WebApplications Dataset39web_applications = Dataset('web_applications')40web_applications._load_train(PATH + 'web_applications_train.json')41web_applications._load_test(PATH + 'web_applications_test.json')...

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.