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

How to use request_callback method in responses

Best Python code snippet using responses

wdn_api.py

Source:wdn_api.py

...28    ]29    return products_with_brand_id30def get_products(brand_id, page_size=DEFAULT_PAGE_SIZE, request_callback=None):31    resource_uri = "/brands/{}/products".format(brand_id)32    modified_callback = (lambda resp: request_callback(products_add_brand_id(resp, brand_id)))33    products = paginate_request(resource_uri, page_size, modified_callback)34    products_with_brand_id = [35        dict(product, **{"brand_id": brand_id}) for product in products36    ]37    return products_with_brand_id38def get_retailers(retailer_ids, request_callback=None):39    return iteration_request_multithread("/retailers/{}", retailer_ids, request_callback, request_callback=request_callback, processes=PARALLEL_PROCESS_COUNT)40def get_products_offers(product_ids, request_callback=None):41    offers = iteration_request_multithread("/products/{}/offers", product_ids, request_callback=request_callback, processes=PARALLEL_PROCESS_COUNT)42    flat_offers = [43        offer44        for offer_data in offers45        for offer in offer_data46        if offer["attributes"]["product_id"] in product_ids47    ]48    return flat_offers49@timeout_decorator.timeout(REQUEST_TIMEOUT)50@retry(delay=1, backoff=2, max_delay=60, tries=10)51def request(api_uri, ignore_404=False):52    request = requests.get("{}{}".format(API_URL, api_uri))53    if request.status_code == 404 and ignore_404:54        LOGGER.error("request failed 404 - {}".format(api_uri))55        return56    if not request.ok:57        raise Exception(58            "request failed with status: {} for url: {}".format(59                request.status_code, request.url60            )61        )62    return request.json()63def iteration_request(id, uri_templete, ignore_404, request_callback):64    resource_uri = uri_templete.format(id)65    response = request(resource_uri, ignore_404=ignore_404)66    if response and response['data']:67        if request_callback:68            request_callback(response['data'])69        LOGGER.info("{} - done".format(resource_uri))70        return response['data']71    LOGGER.info("{} - response empty".format(resource_uri))72    return {}73def chunks(lst, chunk_size):74    for i in range(0, len(lst), chunk_size):75        chunk = lst[i:i + chunk_size]76        yield chunk, i+chunk_size77def iteration_request_multithread(uri_templete, ids, ignore_404=True, request_callback=None, processes=1):78    iteration_request_=partial(iteration_request, uri_templete=uri_templete, ignore_404=ignore_404, request_callback=request_callback)79    records = []80    for ids_chunk, i in chunks(ids, chunk_size=ITERATION_CHUNK_SIZE):81        with get_context("spawn").Pool(processes=processes) as pool:82            records += pool.map(iteration_request_, ids_chunk)83            records = [record for record in records if record]84            LOGGER.info(85                "Processed: {}/{}, records with values: {}".format(86                    i, len(ids), len(records)87                )88            )89    return records90def paginate_request(resource_uri, page_size, request_callback=None):91    page = 192    data = []93    while True:94        param_prefix = "&" if "?" in resource_uri else "?"95        pagination_params = "{}page={}&page_size={}".format(96            param_prefix, page, page_size97        )98        api_uri = "{resource_uri}{pagination_params}".format(99            resource_uri=resource_uri, pagination_params=pagination_params100        )101        response = request(api_uri)102        response_data = response["data"]103        if request_callback:104            request_callback(response_data)105        data += response_data106        if not data:107            break108        total_count = response["meta"]["page"]["total_count"]109        page = response["meta"]["page"]["current_page"]110        records_fetched_count = page_size * page111        LOGGER.info(112            "paginate_request: {}\t records fetched: {}/{}".format(113                resource_uri, records_fetched_count, total_count114            )115        )116        if records_fetched_count >= total_count:117            break118        page += 1...

test_auth_client.py

Source:test_auth_client.py

...12    def setUpClass(cls):13        os.environ["SPYLIB_AUTH_BASE_URL"] = "https://auth.localhost:8000"14    @responses.activate15    def test_create_entity_makes_call(self):16        def request_callback(request):17            resp_body = {}18            headers = {"set-cookie": "refresh_token=1234;"}19            return (201, headers, json.dumps(resp_body))20        responses.add_callback(21            responses.POST,22            "https://auth.localhost:8000/api/v1/entities",23            callback=request_callback,24            content_type="application/json",25        )26        auth_client = AuthClient()27        auth_client.create_entity()28        assert responses.calls.__len__() == 129    @responses.activate30    def test_create_entity_api_key_makes_call(self):31        def request_callback(request):32            resp_body = {}33            headers = {"set-cookie": "refresh_token=1234;"}34            return (201, headers, json.dumps(resp_body))35        responses.add_callback(36            responses.POST,37            "https://auth.localhost:8000/api/v1/entity-api-keys",38            callback=request_callback,39            content_type="application/json",40        )41        auth_client = AuthClient()42        auth_client.create_entity_api_key(uuid.uuid4())43        assert responses.calls.__len__() == 144    @responses.activate45    def test_create_auth_perm_makes_call(self):46        def request_callback(request):47            resp_body = {}48            headers = {"set-cookie": "refresh_token=1234;"}49            return (201, headers, json.dumps(resp_body))50        responses.add_callback(51            responses.POST,52            "https://auth.localhost:8000/api/v1/permissions",53            callback=request_callback,54            content_type="application/json",55        )56        auth_client = AuthClient()57        auth_client.create_auth_perm(58            uuid.uuid4(), "test_perm", "a description used for testing"59        )60        assert responses.calls.__len__() == 161    @responses.activate62    def test_associate_entity_with_perm_makes_call(self):63        def request_callback(request):64            resp_body = {"status_code": 201}65            headers = {"set-cookie": "refresh_token=1234;"}66            return (201, headers, json.dumps(resp_body))67        responses.add_callback(68            responses.POST,69            "https://auth.localhost:8000/api/v1/entity-permissions",70            callback=request_callback,71            content_type="application/json",72        )73        auth_client = AuthClient()74        auth_client.associate_entity_with_permission(75            uuid.uuid4(), "test_perm", "a description used for testing"76        )77        assert responses.calls.__len__() == 178    @responses.activate79    def test_deassociate_entity_with_perm_makes_call(self):80        def request_callback(request):81            resp_body = {"status_code": 200}82            headers = {"set-cookie": "refresh_token=1234;"}83            return (200, headers, json.dumps(resp_body))84        responses.add_callback(85            responses.DELETE,86            "https://auth.localhost:8000/api/v1/entity-permissions",87            callback=request_callback,88            content_type="application/json",89        )90        auth_client = AuthClient()91        auth_client.deassociate_entity_with_permission(92            uuid.uuid4(), "test_perm", "a description used for testing"93        )94        assert responses.calls.__len__() == 1

httpretty_mock.py

Source:httpretty_mock.py

1import json2from http import HTTPStatus3from typing import Callable, Dict, List, Optional, Pattern, Union4import httpretty5from httpretty.core import HTTPrettyRequest6RequestCallback = Callable[7    [HTTPrettyRequest, str, Dict[str, str]],8    List[object],  # noqa: WPS2219]10class RequestCallbackFactory:11    """Create request callback."""12    def __init__(13        self,14        body: Optional[object] = None,15        status_code: int = HTTPStatus.OK,16    ) -> None:17        """Initializing."""18        self._body = {} if body is None else body19        self._status_code = status_code20    def __call__(21        self,22        request: HTTPrettyRequest,23        uri: str,24        response_headers: Dict[str, str],25    ) -> List[object]:26        """Call request callback."""27        response_headers["Content-Type"] = "application/json"28        return [self._status_code, response_headers, json.dumps(self._body)]29class HttprettyMock:30    """Httpretty mocker."""31    base_api_url = ""32    def __init__(self) -> None:33        """Initializing."""34        assert httpretty.is_enabled()35    def register_get(36        self,37        path: str,38        body: Optional[object] = None,39        status_code: int = HTTPStatus.OK,40    ) -> None:41        """Registry url for mock get-query."""42        request_callback = body43        if not callable(body):44            request_callback = RequestCallbackFactory(body, status_code)45        self.register_url(46            method=httpretty.GET,47            uri=self._prepare_uri(path),48            request_callback=request_callback,49            priority=1,50        )51    def register_post(52        self,53        path: str,54        body: Optional[object] = None,55        status_code: int = HTTPStatus.OK,56    ) -> None:57        """Registry url for mock post-query."""58        request_callback = body59        if not callable(body):60            request_callback = RequestCallbackFactory(body, status_code)61        self.register_url(62            method=httpretty.POST,63            uri=self._prepare_uri(path),64            request_callback=request_callback,65            priority=1,66        )67    def register_delete(68        self,69        path: str,70        body: Optional[object] = None,71        status_code: int = HTTPStatus.OK,72    ) -> None:73        """Registry url for mock delete-query."""74        request_callback = body75        if not callable(body):76            request_callback = RequestCallbackFactory(body, status_code)77        self.register_url(78            method=httpretty.DELETE,79            uri=self._prepare_uri(path),80            request_callback=request_callback,81            priority=1,82        )83    def register_url(84        self,85        method: str,86        uri: Union[Pattern[str], str],87        request_callback: RequestCallback,88        priority: int = 0,89    ) -> None:90        """Register url for mock."""91        httpretty.register_uri(92            method=method,93            uri=uri,94            body=request_callback,95            priority=priority,96        )97    def _prepare_uri(self, path: str) -> str:98        """99        Prepare uri.100        :param path:101        :type path: str102        :rtype: str103        """...

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.