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

How to use queue_work method in autotest

Best Python code snippet using autotest_python

movies_favorites.py

Source:movies_favorites.py

1import os2import sys3import csv45from threading import Thread6from queue import Queue7from tqdm import tqdm8from imdb import IMDb910class IMDbThreadWorker(Thread):11    """12    IMDbThreadWorker class13    Main class for IMDb(Internet Movie Database) API requests using threads14    __init__:15    @param queue_work: Queue with information from the given list.16    @param queue_data: Queue with movie data obtained from the imdb api.17    @return: IMDbThreadWorker object18    """19    def __init__(self, queue_work, queue_data):20        Thread.__init__(self)21        self.queue_work = queue_work22        self.queue_data = queue_data23        self.movies = []2425    def get_movie(self, name):26        """27        Get information of each movie using the imdb api.28        @param name: movie name.29        @return: movie data information30        """31        imdb_api = IMDb()32        search_movie = imdb_api.search_movie(name)33        for number, search in enumerate(search_movie):34            if search.data["kind"] == "movie":35                movie_id = search_movie[number].movieID36                movie = imdb_api.get_movie(movie_id)37                break38        return movie3940    def run(self):41        """42        Thread queue work.43        @return: Completed work queue44        """45        while True:46            number, name = self.queue_work.get()47            try:48                self.queue_data.put((number, self.get_movie(name)))49            finally:50                self.queue_work.task_done()5152class MoviesFavorites(object):53    """54    MoviesFavorites class55    Main class MoviesFavorites to initialize56    __init__:57    @param filename: list name in text file.58    @return: MoviesFavorites object59    """60    def __init__(self, filename=""):61        self.filename = filename62        self.threads = 1663        self.fieldnames = [64            "original title", "localized title(Brasil)", "year",65            "rating", "votes", "top 250 rank", "run time", 66            "genres", "kind", "languages"67        ]6869    def print_help(self):70        """71        Print usage.72        @return: code 073        """74        print("usage: python movies_favorites.py [list.txt]")75        return 07677    def get_readlines(self):78        """79        Open text file and get the lines.80        @return: file lines81        """82        with open(os.path.join(self.filename), "r", encoding="utf-8") as file:83            file_readlines = file.readlines()84            file.close()85        return file_readlines8687    def get_runtime_formatted(self, runtime):88        """89        Simple formatted runtime90        @param runtime: movie runtime91        @return: runtime formatted92        """93        return "{}h {}m".format(int(runtime / 60), int(runtime % 60))9495    def generate_queue_and_movies(self):96        """97        Create a queue using class IMDbThreadWorker and get the movies data.98        @return: movies data information99        """100        queue_work = Queue()101        queue_data = Queue()102        lines = self.get_readlines()103        for number, line in enumerate(lines):104            movie_name = line.replace("\n", "")105            queue_work.put((number, movie_name))106        for i in range(self.threads):107            worker = IMDbThreadWorker(queue_work, queue_data)108            worker.daemon = True109            worker.start()110        movies = []111        for i in tqdm(range(len(lines)), desc="movies"):112            movies.append(queue_data.get())113        return movies114115    def generate_movies_data(self, movies_queue):116        """117        Create array with movie information.118        @param movies_queue: Queue with all movies.119        @return: movies information120        """121        data = []122        for movie in sorted(movies_queue, key=lambda m: m[0]):123            movie = movie[1]124            data.append([{125                "original title": movie.data["original title"] if movie.data.get("original title") else movie.data["title"],126                "localized title(Brasil)": movie.data["localized title"] if movie.data.get("localized title") else original_title,127                "year": movie.data["year"] if movie.data.get("year") else "None",128                "rating": movie.data["rating"] if movie.data.get("rating") else "None",129                "votes": movie.data["votes"] if movie.data.get("votes") else "None",130                "top 250 rank": movie.data["top 250 rank"] if movie.data.get("top 250 rank") else "None",131                "run time": self.get_runtime_formatted(int(movie.data["runtimes"][0])) if movie.data.get("runtimes") else "None",132                "genres": ", ".join(movie.data["genres"]) if movie.data.get("genres") else "None",133                "kind": movie.data["kind"] if movie.data.get("kind") else "None",134                "languages": movie.data["languages"][0] if movie.data.get("languages") else "None"135            }])136        return data137138    def generate_csv(self, rows):139        """140        Create cvs file.141        @param rows: Array with all movies data information.142        @return: True143        """144        with open(os.path.join(self.filename.replace(".txt", ".csv")), "w", encoding="utf-8", newline="") as file:145            writer = csv.DictWriter(file, fieldnames=self.fieldnames)146            writer.writeheader()147            for row in rows:148                writer.writerows(row)149            file.close()150        return True151152if __name__ == "__main__":153    if len(sys.argv) < 2:154        sys.exit(MoviesFavorites().print_help())155    movies_favorites = MoviesFavorites(sys.argv[1])156    movies_queue = movies_favorites.generate_queue_and_movies()157    movies_data = movies_favorites.generate_movies_data(movies_queue)158    movies_csv = movies_favorites.generate_csv(movies_data)
...

test_threadpool.py

Source:test_threadpool.py

...28    def after_work_cb(self, error):29        self.assertEqual(error, None)30        self.pool_after_work_cb_called += 131    def test_threadpool1(self):32        self.loop.queue_work(self.run_in_pool, self.after_work_cb)33        args = (1, 2, 3)34        kw = {}35        self.loop.queue_work(functools.partial(self.run_in_pool, *args, **kw), self.after_work_cb)36        args = ()37        kw = {'test': 1}38        self.loop.queue_work(functools.partial(self.run_in_pool, *args, **kw), self.after_work_cb)39        self.loop.run()40        self.assertEqual(self.pool_cb_called, 3)41        self.assertEqual(self.pool_after_work_cb_called, 3)42    def raise_in_pool(self, *args, **kw):43        1/044    def after_work_cb2(self, error):45        self.assertEqual(error, None)46        self.assertEqual(self.work_item.exc_info[0], ZeroDivisionError)47    def test_threadpool_exc(self):48        self.work_item = WorkItem(self.raise_in_pool)49        self.loop.queue_work(self.work_item, self.after_work_cb2)50        self.loop.run()51class ThreadPoolMultiLoopTest(unittest.TestCase):52    def setUp(self):53        self.pool_cb_called = 054        self.lock = threading.Lock()55    def run_in_pool(self):56        with self.lock:57            self.pool_cb_called += 158            time.sleep(0.2)59    def run_loop(self):60        loop = pyuv.Loop()61        loop.queue_work(self.run_in_pool)62        loop.run()63    def test_threadpool_multiple_loops(self):64        t1 = threading.Thread(target=self.run_loop)65        t2 = threading.Thread(target=self.run_loop)66        t3 = threading.Thread(target=self.run_loop)67        [t.start() for t in (t1, t2, t3)]68        [t.join(5) for t in (t1, t2, t3)]69        self.assertEqual(self.pool_cb_called, 3)70if __name__ == '__main__':...

threads_unittest.py

Source:threads_unittest.py

...24        th = threads.ThreadPool(self._workload, numthreads=5)25        self.god.check_playback()26    def test_one_thread(self):27        th = threads.ThreadPool(self._workload, numthreads=1)28        th.queue_work(range(10))29        th.wait()30        res = []31        while not self.results.empty():32            res.append(self.results.get())33        self.assertEqualNoOrder([0, 1, 4, 9, 16, 25, 36, 49, 64, 81], res)34    def _threading(self, numthreads, count):35        th = threads.ThreadPool(self._workload, numthreads=numthreads)36        th.queue_work(range(count))37        th.wait()38        res = []39        while not self.results.empty():40            res.append(self.results.get())41        self.assertEqualNoOrder([i * i for i in xrange(count)], res)42    def test_threading(self):43        self._threading(10, 10)44    def test_threading_lots(self):45        self._threading(100, 100)46    def test_threading_multi_queueing(self):47        th = threads.ThreadPool(self._workload, numthreads=5)48        th.queue_work(range(5))49        th.queue_work(range(5, 10))50        th.wait()51        res = []52        while not self.results.empty():53            res.append(self.results.get())54        self.assertEqualNoOrder([i * i for i in xrange(10)], res)55if __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.