Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.