Best Python code snippet using responses
test_gitlab_http_methods.py
Source:test_gitlab_http_methods.py  
...23    )24    http_r = gl.http_request("get", "/projects")25    http_r.json()26    assert http_r.status_code == 20027    assert responses.assert_call_count(url, 1) is True28@responses.activate29def test_http_request_404(gl):30    url = "http://localhost/api/v4/not_there"31    responses.add(32        method=responses.GET,33        url=url,34        json={},35        status=400,36        match=MATCH_EMPTY_QUERY_PARAMS,37    )38    with pytest.raises(GitlabHttpError):39        gl.http_request("get", "/not_there")40    assert responses.assert_call_count(url, 1) is True41@responses.activate42@pytest.mark.parametrize("status_code", [500, 502, 503, 504])43def test_http_request_with_only_failures(gl, status_code):44    url = "http://localhost/api/v4/projects"45    responses.add(46        method=responses.GET,47        url=url,48        json={},49        status=status_code,50        match=MATCH_EMPTY_QUERY_PARAMS,51    )52    with pytest.raises(GitlabHttpError):53        gl.http_request("get", "/projects")54    assert responses.assert_call_count(url, 1) is True55@responses.activate56def test_http_request_with_retry_on_method_for_transient_failures(gl):57    call_count = 058    calls_before_success = 359    url = "http://localhost/api/v4/projects"60    def request_callback(request):61        nonlocal call_count62        call_count += 163        status_code = 200 if call_count >= calls_before_success else 50064        headers = {}65        body = "[]"66        return (status_code, headers, body)67    responses.add_callback(68        method=responses.GET,69        url=url,70        callback=request_callback,71        content_type="application/json",72    )73    http_r = gl.http_request("get", "/projects", retry_transient_errors=True)74    assert http_r.status_code == 20075    assert len(responses.calls) == calls_before_success76@responses.activate77def test_http_request_with_retry_on_class_for_transient_failures(gl_retry):78    call_count = 079    calls_before_success = 380    url = "http://localhost/api/v4/projects"81    def request_callback(request: requests.models.PreparedRequest):82        nonlocal call_count83        call_count += 184        status_code = 200 if call_count >= calls_before_success else 50085        headers = {}86        body = "[]"87        return (status_code, headers, body)88    responses.add_callback(89        method=responses.GET,90        url=url,91        callback=request_callback,92        content_type="application/json",93    )94    http_r = gl_retry.http_request("get", "/projects", retry_transient_errors=True)95    assert http_r.status_code == 20096    assert len(responses.calls) == calls_before_success97@responses.activate98def test_http_request_with_retry_on_class_and_method_for_transient_failures(gl_retry):99    call_count = 0100    calls_before_success = 3101    url = "http://localhost/api/v4/projects"102    def request_callback(request):103        nonlocal call_count104        call_count += 1105        status_code = 200 if call_count >= calls_before_success else 500106        headers = {}107        body = "[]"108        return (status_code, headers, body)109    responses.add_callback(110        method=responses.GET,111        url=url,112        callback=request_callback,113        content_type="application/json",114    )115    with pytest.raises(GitlabHttpError):116        gl_retry.http_request("get", "/projects", retry_transient_errors=False)117    assert len(responses.calls) == 1118def create_redirect_response(119    *, response: requests.models.Response, http_method: str, api_path: str120) -> requests.models.Response:121    """Create a Requests response object that has a redirect in it"""122    assert api_path.startswith("/")123    http_method = http_method.upper()124    # Create a history which contains our original request which is redirected125    history = [126        helpers.httmock_response(127            status_code=302,128            content="",129            headers={"Location": f"http://example.com/api/v4{api_path}"},130            reason="Moved Temporarily",131            request=response.request,132        )133    ]134    # Create a "prepped" Request object to be the final redirect. The redirect135    # will be a "GET" method as Requests changes the method to "GET" when there136    # is a 301/302 redirect code.137    req = requests.Request(138        method="GET",139        url=f"http://example.com/api/v4{api_path}",140    )141    prepped = req.prepare()142    resp_obj = helpers.httmock_response(143        status_code=200,144        content="",145        headers={},146        reason="OK",147        elapsed=5,148        request=prepped,149    )150    resp_obj.history = history151    return resp_obj152def test_http_request_302_get_does_not_raise(gl):153    """Test to show that a redirect of a GET will not cause an error"""154    method = "get"155    api_path = "/user/status"156    url = f"http://localhost/api/v4{api_path}"157    def response_callback(158        response: requests.models.Response,159    ) -> requests.models.Response:160        return create_redirect_response(161            response=response, http_method=method, api_path=api_path162        )163    with responses.RequestsMock(response_callback=response_callback) as req_mock:164        req_mock.add(165            method=responses.GET,166            url=url,167            status=302,168            match=MATCH_EMPTY_QUERY_PARAMS,169        )170        gl.http_request(verb=method, path=api_path)171def test_http_request_302_put_raises_redirect_error(gl):172    """Test to show that a redirect of a PUT will cause an error"""173    method = "put"174    api_path = "/user/status"175    url = f"http://localhost/api/v4{api_path}"176    def response_callback(177        response: requests.models.Response,178    ) -> requests.models.Response:179        return create_redirect_response(180            response=response, http_method=method, api_path=api_path181        )182    with responses.RequestsMock(response_callback=response_callback) as req_mock:183        req_mock.add(184            method=responses.PUT,185            url=url,186            status=302,187            match=MATCH_EMPTY_QUERY_PARAMS,188        )189        with pytest.raises(RedirectError) as exc:190            gl.http_request(verb=method, path=api_path)191    error_message = exc.value.error_message192    assert "Moved Temporarily" in error_message193    assert "http://localhost/api/v4/user/status" in error_message194    assert "http://example.com/api/v4/user/status" in error_message195@responses.activate196def test_get_request(gl):197    url = "http://localhost/api/v4/projects"198    responses.add(199        method=responses.GET,200        url=url,201        json={"name": "project1"},202        status=200,203        match=MATCH_EMPTY_QUERY_PARAMS,204    )205    result = gl.http_get("/projects")206    assert isinstance(result, dict)207    assert result["name"] == "project1"208    assert responses.assert_call_count(url, 1) is True209@responses.activate210def test_get_request_raw(gl):211    url = "http://localhost/api/v4/projects"212    responses.add(213        method=responses.GET,214        url=url,215        content_type="application/octet-stream",216        body="content",217        status=200,218        match=MATCH_EMPTY_QUERY_PARAMS,219    )220    result = gl.http_get("/projects")221    assert result.content.decode("utf-8") == "content"222    assert responses.assert_call_count(url, 1) is True223@responses.activate224def test_get_request_404(gl):225    url = "http://localhost/api/v4/not_there"226    responses.add(227        method=responses.GET,228        url=url,229        json=[],230        status=404,231        match=MATCH_EMPTY_QUERY_PARAMS,232    )233    with pytest.raises(GitlabHttpError):234        gl.http_get("/not_there")235    assert responses.assert_call_count(url, 1) is True236@responses.activate237def test_get_request_invalid_data(gl):238    url = "http://localhost/api/v4/projects"239    responses.add(240        method=responses.GET,241        url=url,242        body='["name": "project1"]',243        content_type="application/json",244        status=200,245        match=MATCH_EMPTY_QUERY_PARAMS,246    )247    with pytest.raises(GitlabParsingError):248        result = gl.http_get("/projects")249        print(type(result))250        print(result.content)251    assert responses.assert_call_count(url, 1) is True252@responses.activate253def test_list_request(gl):254    url = "http://localhost/api/v4/projects"255    responses.add(256        method=responses.GET,257        url=url,258        json=[{"name": "project1"}],259        headers={"X-Total": "1"},260        status=200,261        match=MATCH_EMPTY_QUERY_PARAMS,262    )263    result = gl.http_list("/projects", as_list=True)264    assert isinstance(result, list)265    assert len(result) == 1266    result = gl.http_list("/projects", as_list=False)267    assert isinstance(result, GitlabList)268    assert len(result) == 1269    result = gl.http_list("/projects", all=True)270    assert isinstance(result, list)271    assert len(result) == 1272    assert responses.assert_call_count(url, 3) is True273@responses.activate274def test_list_request_404(gl):275    url = "http://localhost/api/v4/not_there"276    responses.add(277        method=responses.GET,278        url=url,279        json=[],280        status=404,281        match=MATCH_EMPTY_QUERY_PARAMS,282    )283    with pytest.raises(GitlabHttpError):284        gl.http_list("/not_there")285    assert responses.assert_call_count(url, 1) is True286@responses.activate287def test_list_request_invalid_data(gl):288    url = "http://localhost/api/v4/projects"289    responses.add(290        method=responses.GET,291        url=url,292        body='["name": "project1"]',293        content_type="application/json",294        status=200,295        match=MATCH_EMPTY_QUERY_PARAMS,296    )297    with pytest.raises(GitlabParsingError):298        gl.http_list("/projects")299    assert responses.assert_call_count(url, 1) is True300@responses.activate301def test_post_request(gl):302    url = "http://localhost/api/v4/projects"303    responses.add(304        method=responses.POST,305        url=url,306        json={"name": "project1"},307        status=200,308        match=MATCH_EMPTY_QUERY_PARAMS,309    )310    result = gl.http_post("/projects")311    assert isinstance(result, dict)312    assert result["name"] == "project1"313    assert responses.assert_call_count(url, 1) is True314@responses.activate315def test_post_request_404(gl):316    url = "http://localhost/api/v4/not_there"317    responses.add(318        method=responses.POST,319        url=url,320        json=[],321        status=404,322        match=MATCH_EMPTY_QUERY_PARAMS,323    )324    with pytest.raises(GitlabHttpError):325        gl.http_post("/not_there")326    assert responses.assert_call_count(url, 1) is True327@responses.activate328def test_post_request_invalid_data(gl):329    url = "http://localhost/api/v4/projects"330    responses.add(331        method=responses.POST,332        url=url,333        content_type="application/json",334        body='["name": "project1"]',335        status=200,336        match=MATCH_EMPTY_QUERY_PARAMS,337    )338    with pytest.raises(GitlabParsingError):339        gl.http_post("/projects")340    assert responses.assert_call_count(url, 1) is True341@responses.activate342def test_put_request(gl):343    url = "http://localhost/api/v4/projects"344    responses.add(345        method=responses.PUT,346        url=url,347        json={"name": "project1"},348        status=200,349        match=MATCH_EMPTY_QUERY_PARAMS,350    )351    result = gl.http_put("/projects")352    assert isinstance(result, dict)353    assert result["name"] == "project1"354    assert responses.assert_call_count(url, 1) is True355@responses.activate356def test_put_request_404(gl):357    url = "http://localhost/api/v4/not_there"358    responses.add(359        method=responses.PUT,360        url=url,361        json=[],362        status=404,363        match=MATCH_EMPTY_QUERY_PARAMS,364    )365    with pytest.raises(GitlabHttpError):366        gl.http_put("/not_there")367    assert responses.assert_call_count(url, 1) is True368@responses.activate369def test_put_request_invalid_data(gl):370    url = "http://localhost/api/v4/projects"371    responses.add(372        method=responses.PUT,373        url=url,374        body='["name": "project1"]',375        content_type="application/json",376        status=200,377        match=MATCH_EMPTY_QUERY_PARAMS,378    )379    with pytest.raises(GitlabParsingError):380        gl.http_put("/projects")381    assert responses.assert_call_count(url, 1) is True382@responses.activate383def test_delete_request(gl):384    url = "http://localhost/api/v4/projects"385    responses.add(386        method=responses.DELETE,387        url=url,388        json=True,389        status=200,390        match=MATCH_EMPTY_QUERY_PARAMS,391    )392    result = gl.http_delete("/projects")393    assert isinstance(result, requests.Response)394    assert result.json() is True395    assert responses.assert_call_count(url, 1) is True396@responses.activate397def test_delete_request_404(gl):398    url = "http://localhost/api/v4/not_there"399    responses.add(400        method=responses.DELETE,401        url=url,402        json=[],403        status=404,404        match=MATCH_EMPTY_QUERY_PARAMS,405    )406    with pytest.raises(GitlabHttpError):407        gl.http_delete("/not_there")...test_slack_driver.py
Source:test_slack_driver.py  
...38    @responses.activate39    def test_sending_to_anonymous(self):40        responses.add(responses.POST, webhook_url, body=b"ok")41        self.notification.route("slack", webhook_url).notify(WelcomeNotification())42        self.assertTrue(responses.assert_call_count(webhook_url, 1))43    @responses.activate44    def test_sending_to_notifiable(self):45        responses.add(responses.POST, webhook_url, body=b"ok")46        User.route_notification_for_slack = lambda notifiable: webhook_url47        user = User.find(1)48        user.notify(WelcomeNotification())49        self.assertTrue(responses.assert_call_count(webhook_url, 1))50        User.route_notification_for_slack = route_for_slack51    @responses.activate52    def test_sending_to_multiple_webhooks(self):53        responses.add(responses.POST, webhook_url, body=b"ok")54        responses.add(responses.POST, webhook_url_2, body=b"ok")55        User.route_notification_for_slack = lambda notifiable: [56            webhook_url,57            webhook_url_2,58        ]59        user = User.find(1)60        user.notify(WelcomeNotification())61        self.assertTrue(responses.assert_call_count(webhook_url, 1))62        self.assertTrue(responses.assert_call_count(webhook_url_2, 1))63        User.route_notification_for_slack = route_for_slack64class TestSlackAPIDriver(TestCase):65    url = "https://slack.com/api/chat.postMessage"66    channel_url = "https://slack.com/api/conversations.list"67    def setUp(self):68        super().setUp()69        self.notification = self.application.make("notification")70    def test_sending_without_credentials(self):71        with self.assertRaises(NotificationException) as e:72            self.notification.route("slack", "123456").notify(WelcomeNotification())73        self.assertIn("not_authed", str(e.exception))74    @responses.activate75    def test_sending_to_anonymous(self):76        responses.add(77            responses.POST,78            self.url,79            body=b'{"ok": "True"}',80        )81        responses.add(82            responses.POST,83            self.channel_url,84            body=b'{"channels": [{"name": "bot", "id": "123"}]}',85        )86        self.notification.route("slack", "#bot").notify(WelcomeNotification())87        # to convert #bot to Channel ID88        self.assertTrue(responses.assert_call_count(self.channel_url, 1))89        self.assertTrue(responses.assert_call_count(self.url, 1))90    @responses.activate91    def test_sending_to_notifiable(self):92        user = User.find(1)93        responses.add(94            responses.POST,95            self.url,96            body=b'{"ok": "True"}',97        )98        responses.add(99            responses.POST,100            self.channel_url,101            body=b'{"channels": [{"name": "bot", "id": "123"}]}',102        )103        user.notify(WelcomeUserNotification())104        self.assertTrue(responses.assert_call_count(self.url, 1))105    @responses.activate106    @pytest.mark.skip(107        reason="Failing because user defined routing takes precedence. What should be the behaviour ?"108    )109    def test_sending_to_multiple_channels(self):110        user = User.find(1)111        responses.add(112            responses.POST,113            self.url,114            body=b'{"ok": "True"}',115        )116        responses.add(117            responses.POST,118            self.channel_url,119            body=b'{"channels": [{"name": "bot", "id": "123"}, {"name": "general", "id": "456"}]}',120        )121        user.notify(OtherNotification())122        self.assertTrue(responses.assert_call_count(self.channel_url, 2))123        self.assertTrue(responses.assert_call_count(self.url, 2))124    @responses.activate125    def test_convert_channel(self):126        channel_id = self.notification.get_driver("slack").convert_channel(127            "123456", "token"128        )129        self.assertEqual(channel_id, "123456")130        responses.add(131            responses.POST,132            self.channel_url,133            body=b'{"channels": [{"name": "general", "id": "654321"}]}',134        )135        channel_id = self.notification.get_driver("slack").convert_channel(136            "#general", "token"137        )...test_adapters.py
Source:test_adapters.py  
1import logging2from unittest import mock3import uuid4from driftwood.adapters import StatusUpdateAdapter5class TestStatusAdapter:6    def test_1(self):7        update_func = mock.MagicMock()8        log = logging.getLogger(uuid.uuid4().hex)9        adapter = StatusUpdateAdapter(update_func, log)10        update_func.assert_call_count == 011        adapter.info("test")12        update_func.assert_called_with(20, "INFO")13        update_func.assert_call_count == 114        adapter.debug("test")15        update_func.assert_call_count == 116        adapter.error("test")17        update_func.assert_called_with(40, "ERROR")18        update_func.assert_call_count == 219        adapter.warning("test")20        adapter.info("test")21        adapter.warning("test")22        update_func.assert_call_count == 223        ...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
