KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use add_req method in lisa
Best Python code snippet using lisa_python
request_test.py
Source:request_test.py 
...54 self.assertEqual(self.session.get_torrents(), [self.handle])55 # TODO: can we test the download-after-graceful-pause case?56class TestAddRemove(request_test_utils.RequestServiceTestCase):57 def test_add_remove(self) -> None:58 req = self.add_req()59 self.wait_for_torrent()60 self.assertEqual(61 [str(h.info_hash()) for h in self.session.get_torrents()],62 [self.torrent.info_hash],63 )64 self.service.discard_request(req)65 with self.assertRaises(request_lib.CanceledError):66 req.read(timeout=5)67 def test_fetch_error(self) -> None:68 req = self.add_req(configure_atp=lambda atp: _raise_dummy())69 with self.assertRaises(request_lib.FetchError):70 req.read(timeout=5)71 def test_shutdown(self) -> None:72 req = self.add_req()73 self.service.terminate()74 with self.assertRaises(request_lib.CanceledError):75 req.read(timeout=5)76 def test_already_shutdown(self) -> None:77 self.service.terminate()78 req = self.add_req()79 with self.assertRaises(request_lib.CanceledError):80 req.read(timeout=5)81class TestRead(request_test_utils.RequestServiceTestCase):82 def test_all(self) -> None:83 req = self.add_req()84 self.feed_pieces()85 data = request_test_utils.read_all(req)86 self.assertEqual(data, self.torrent.data)87 def test_unaligned_multi_pieces(self) -> None:88 start = self.torrent.piece_length // 289 stop = min(start + self.torrent.piece_length, self.torrent.length)90 req = self.add_req(start=start, stop=stop)91 self.feed_pieces()92 data = request_test_utils.read_all(req)93 self.assertEqual(data, self.torrent.data[start:stop])94 def test_unaligned_single_piece(self) -> None:95 start = self.torrent.piece_length // 496 stop = 3 * self.torrent.piece_length // 497 req = self.add_req(start=start, stop=stop)98 self.feed_pieces()99 data = request_test_utils.read_all(req)100 self.assertEqual(data, self.torrent.data[start:stop])101 def test_existing_torrent(self) -> None:102 req = self.add_req()103 self.feed_pieces()104 request_test_utils.read_all(req)105 req = self.add_req()106 data = request_test_utils.read_all(req, msg="second read")107 self.assertEqual(data, self.torrent.data)108 def test_simultaneous(self) -> None:109 req1 = self.add_req()110 req2 = self.add_req()111 executor = concurrent.futures.ThreadPoolExecutor()112 future1 = executor.submit(request_test_utils.read_all, req1)113 future2 = executor.submit(request_test_utils.read_all, req2)114 self.feed_pieces()115 self.assertEqual(future1.result(), self.torrent.data)116 self.assertEqual(future2.result(), self.torrent.data)117 def test_two_readers(self) -> None:118 req1 = self.add_req()119 req2 = self.add_req()120 self.feed_pieces()121 data1 = request_test_utils.read_all(req1)122 data2 = request_test_utils.read_all(req2)123 self.assertEqual(data1, self.torrent.data)124 self.assertEqual(data2, self.torrent.data)125 def test_download(self) -> None:126 seed = lib.create_isolated_session_service().session127 seed_dir = tempfile.TemporaryDirectory()128 atp = self.torrent.atp()129 atp.save_path = seed_dir.name130 atp.flags &= ~lt.torrent_flags.paused131 handle = seed.add_torrent(atp)132 # https://github.com/arvidn/libtorrent/issues/4980: add_piece() while133 # checking silently fails in libtorrent 1.2.8.134 request_test_utils.wait_done_checking_or_error(handle)135 for i, piece in enumerate(self.torrent.pieces):136 # NB: bug in libtorrent where add_piece accepts str but not bytes137 handle.add_piece(i, piece.decode(), 0)138 req = self.add_req()139 self.wait_for_torrent().connect_peer(("127.0.0.1", seed.listen_port()))140 # The peer connection takes a long time, not sure why141 data = request_test_utils.read_all(req, timeout=60)142 self.assertEqual(data, self.torrent.data)143 def test_file_error(self) -> None:144 # Create a file in tempdir, try to use it as the save_path145 path = os.path.join(self.tempdir.name, "file.txt")146 with open(path, mode="w"):147 pass148 self.config["torrent_default_save_path"] = path149 self.service.set_config(self.config)150 req = self.add_req()151 self.feed_pieces()152 with self.assertRaises(NotADirectoryError):153 request_test_utils.read_all(req)154 def test_read_checked_pieces(self) -> None:155 # Download a torrent156 req = self.add_req()157 self.feed_pieces()158 data = request_test_utils.read_all(req)159 self.assertEqual(data, self.torrent.data)160 # query_save_path not bound in python161 save_path = self.wait_for_torrent().status(flags=128).save_path162 # Wait for the file to be written to disk163 for _ in lib.loop_until_timeout(5, msg="write file"):164 path = os.path.join(save_path, self.torrent.files[0].path.decode())165 if os.path.exists(path):166 data = open(path, mode="rb").read()167 if data == self.torrent.data:168 break169 # Create a new session170 self.teardown_session()171 self.init_session()172 req = self.add_req()173 # We should be able to read the data without feeding pieces174 data = request_test_utils.read_all(req)175 self.assertEqual(data, self.torrent.data)176 def test_read_after_cancelled_read(self) -> None:177 # Start reading178 req = self.add_req()179 # Feed one piece, so the torrent stays in the session180 self.feed_pieces(piece_indexes=(0,))181 # Wait for pieces to be prioritized182 for _ in lib.loop_until_timeout(5, msg="prioritize"):183 if all(self.wait_for_torrent().get_piece_priorities()):184 break185 # Cancel the request -- resets piece deadlines186 self.service.discard_request(req)187 # Wait until deadlines have been reset188 for _ in lib.loop_until_timeout(5, msg="deprioritize"):189 if not any(self.wait_for_torrent().get_piece_priorities()):190 break191 # Recreate the request -- listens for read_piece_alert192 req = self.add_req()193 # Feed all pieces and check that we can read the data194 self.feed_pieces()195 data = request_test_utils.read_all(req)196 self.assertEqual(data, self.torrent.data)197class TestRemoveTorrent(request_test_utils.RequestServiceTestCase):198 def test_with_active_requests(self) -> None:199 req = self.add_req()200 self.session.remove_torrent(self.wait_for_torrent())201 with self.assertRaises(request_lib.TorrentRemovedError):202 req.read(timeout=5)203class TestConfig(request_test_utils.RequestServiceTestCase):204 def test_config_defaults(self) -> None:205 save_path = str(self.config_dir.joinpath("downloads"))206 self.assertEqual(207 self.config, config_lib.Config(torrent_default_save_path=save_path)208 )209 atp = lt.add_torrent_params()210 self.service.configure_atp(atp)211 self.assertEqual(atp.save_path, save_path)212 def test_set_config(self) -> None:213 # Set all non-default configs...
favorites_endpoint.py
Source:favorites_endpoint.py
1from .endpoint import Endpoint, api2from .exceptions import MissingRequiredFieldError3from .. import RequestFactory4from ...models import FavoriteItem5from ..pager import Pager6import xml.etree.ElementTree as ET7import logging8import copy9logger = logging.getLogger('tableau.endpoint.favorites')10class Favorites(Endpoint):11 @property12 def baseurl(self):13 return "{0}/sites/{1}/favorites".format(self.parent_srv.baseurl, self.parent_srv.site_id)14 # Gets all favorites15 @api(version="2.5")16 def get(self, user_item, req_options=None):17 logger.info('Querying all favorites for user {0}'.format(user_item.name))18 url = '{0}/{1}'.format(self.baseurl, user_item.id)19 server_response = self.get_request(url, req_options)20 user_item._favorites = FavoriteItem.from_response(server_response.content, self.parent_srv.namespace)21 @api(version="2.0")22 def add_favorite_workbook(self, user_item, workbook_item):23 url = '{0}/{1}'.format(self.baseurl, user_item.id)24 add_req = RequestFactory.Favorite.add_workbook_req(workbook_item.id, workbook_item.name)25 server_response = self.put_request(url, add_req)26 logger.info('Favorited {0} for user (ID: {1})'.format(workbook_item.name, user_item.id))27 @api(version="2.0")28 def add_favorite_view(self, user_item, view_item):29 url = '{0}/{1}'.format(self.baseurl, user_item.id)30 add_req = RequestFactory.Favorite.add_view_req(view_item.id, view_item.name)31 server_response = self.put_request(url, add_req)32 logger.info('Favorited {0} for user (ID: {1})'.format(view_item.name, user_item.id))33 @api(version="2.3")34 def add_favorite_datasource(self, user_item, datasource_item):35 url = '{0}/{1}'.format(self.baseurl, user_item.id)36 add_req = RequestFactory.Favorite.add_datasource_req(datasource_item.id, datasource_item.name)37 server_response = self.put_request(url, add_req)38 logger.info('Favorited {0} for user (ID: {1})'.format(datasource_item.name, user_item.id))39 @api(version="3.1")40 def add_favorite_project(self, user_item, project_item):41 url = '{0}/{1}'.format(self.baseurl, user_item.id)42 add_req = RequestFactory.Favorite.add_project_req(project_item.id, project_item.name)43 server_response = self.put_request(url, add_req)44 logger.info('Favorited {0} for user (ID: {1})'.format(project_item.name, user_item.id))45 @api(version="2.0")46 def delete_favorite_workbook(self, user_item, workbook_item):47 url = '{0}/{1}/workbooks/{2}'.format(self.baseurl, user_item.id, workbook_item.id)48 logger.info('Removing favorite {0} for user (ID: {1})'.format(workbook_item.id, user_item.id))49 self.delete_request(url)50 @api(version="2.0")51 def delete_favorite_view(self, user_item, view_item):52 url = '{0}/{1}/views/{2}'.format(self.baseurl, user_item.id, view_item.id)53 logger.info('Removing favorite {0} for user (ID: {1})'.format(view_item.id, user_item.id))54 self.delete_request(url)55 @api(version="2.3")56 def delete_favorite_datasource(self, user_item, datasource_item):57 url = '{0}/{1}/datasources/{2}'.format(self.baseurl, user_item.id, datasource_item.id)58 logger.info('Removing favorite {0} for user (ID: {1})'.format(datasource_item.id, user_item.id))59 self.delete_request(url)60 @api(version="3.1")61 def delete_favorite_project(self, user_item, project_item):62 url = '{0}/{1}/projects/{2}'.format(self.baseurl, user_item.id, project_item.id)63 logger.info('Removing favorite {0} for user (ID: {1})'.format(project_item.id, user_item.id))...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run lisa 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!!
Was this article helpful?
