Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _on_request method in responses

Best Python code snippet using responses

mixin.py

Source:mixin.py

...51        user = json.loads(txt)52    else:53        user = None54    callback(user)55def _on_request(self, callback, response):56    if response.error:57        logging.warning('Error response %s fetching %s', response.error,58                        response.request.url)59        callback(None)60        return61    callback(response.body)62class DoubanMixin(tornado.auth.OAuthMixin):63    _OAUTH_REQUEST_TOKEN_URL = 'http://www.douban.com/service/auth/request_token'64    _OAUTH_ACCESS_TOKEN_URL = 'http://www.douban.com/service/auth/access_token'65    _OAUTH_AUTHORIZE_URL = 'http://www.douban.com/service/auth/authorize'66    _OAUTH_VERSION = '1.0'67    _OAUTH_NO_CALLBACKS = False68    _on_request = _on_request69    callback_url = callback_url...

webserver.py

Source:webserver.py

...134    # Recognized by SocketServer.135    self._server.timeout = 10136    self._on_request = None137  def _OnRequest(self, request, responder):138    self._on_request(responder)139    self._on_request = None140  def Respond(self, on_request):141    """Blocks until request comes in, then calls given handler function.142    Args:143      on_request: Function that handles the request. Invoked with single144          parameter, an instance of Responder.145    """146    if self._on_request:147      raise RuntimeError('Must handle 1 request at a time.')148    self._on_request = on_request149    while self._on_request:150      # Don't use handle_one_request, because it won't work with the timeout.151      self._server.handle_request()152  def RespondWithContent(self, content):...

test_outgoing_http.py

Source:test_outgoing_http.py

...6from urllib3.util import Retry7from zerver.lib.outgoing_http import OutgoingSession8from zerver.lib.test_classes import ZulipTestCase9class RequestMockWithProxySupport(responses.RequestsMock):10    def _on_request(11        self,12        adapter: requests.adapters.HTTPAdapter,13        request: requests.PreparedRequest,14        **kwargs: Any,15    ) -> requests.Response:16        if "proxies" in kwargs and request.url:17            proxy_uri = requests.utils.select_proxy(request.url, kwargs["proxies"])18            if proxy_uri is not None:19                request = requests.Request(20                    method="GET",21                    url=f"{proxy_uri}/",22                    headers=adapter.proxy_headers(proxy_uri),23                ).prepare()24        return super()._on_request(  # type: ignore[misc]  # This is an undocumented internal API25            adapter,26            request,27            **kwargs,28        )29class RequestMockWithTimeoutAsHeader(responses.RequestsMock):30    def _on_request(31        self,32        adapter: requests.adapters.HTTPAdapter,33        request: requests.PreparedRequest,34        **kwargs: Any,35    ) -> requests.Response:36        if kwargs.get("timeout") is not None:37            request.headers["X-Timeout"] = kwargs["timeout"]38        return super()._on_request(  # type: ignore[misc]  # This is an undocumented internal API39            adapter,40            request,41            **kwargs,42        )43class TestOutgoingHttp(ZulipTestCase):44    def test_headers(self) -> None:45        with RequestMockWithProxySupport() as mock_requests:46            mock_requests.add(responses.GET, "http://example.com/")47            OutgoingSession(role="testing", timeout=1, headers={"X-Foo": "bar"}).get(48                "http://example.com/"49            )50            self.assert_length(mock_requests.calls, 1)51            headers = mock_requests.calls[0].request.headers52            # We don't see a proxy header with no proxy set...

test_webhook.py

Source:test_webhook.py

...44def test_handler_not_called_with_different_event(mocked_req, webhook):45    mocked_req.headers['X-Github-Event'] = "pull_request"46    handler = mock.Mock()47    webhook.hook()(handler)48    webhook._on_request()49    handler.assert_not_called()50def test_handler_called_with_valid_event(mocked_req, webhook):51    handler = mock.Mock()52    webhook.hook()(handler)53    webhook._on_request()54    handler.assert_called_once()55def test_multiple_handlers(mocked_req, webhook):56    handlers = []57    for _ in range(10):58        handler = mock.Mock()59        webhook.hook()(handler)60        handlers.append(handler)61    webhook._on_request()62    for handler in handlers:63        handler.assert_called_once()64def _assert_request_fails(hook, err_msg):65    with pytest.raises(werkzeug.exceptions.BadRequest) as excpt:66        hook._on_request()...

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.