How to use queue_work method in autotest

Best Python code snippet using autotest_python

movies_favorites.py

Source:movies_favorites.py Github

copy

Full Screen

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) ...

Full Screen

Full Screen

test_threadpool.py

Source:test_threadpool.py Github

copy

Full Screen

...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__':...

Full Screen

Full Screen

threads_unittest.py

Source:threads_unittest.py Github

copy

Full Screen

...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__':...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful