How to use aioresponses - 10 common examples

def test_address_as_instance_of_url_combined_with_pass_through(self):
        external_api = 'http://httpbin.org/status/201'

        @asyncio.coroutine
        def doit():
            api_resp = yield from self.session.get(self.url)
            # we have to hit actual url,
            # otherwise we do not test pass through option properly
            ext_rep = yield from self.session.get(URL(external_api))
            return api_resp, ext_rep

        with aioresponses(passthrough=[external_api]) as m:
            m.get(self.url, status=200)
            api, ext = yield from doit()

            self.assertEqual(api.status, 200)
            self.assertEqual(ext.status, 201)

def callback(url, **kwargs):
            yield from event.wait()
            self.assertEqual(str(url), self.url)
            self.assertEqual(kwargs, {'allow_redirects': True})
            return CallbackResult(body=body)

def doit():
            api_resp = yield from self.session.get(self.url)
            # we have to hit actual url,
            # otherwise we do not test pass through option properly
            ext_rep = yield from self.session.get(URL(external_api))
            return api_resp, ext_rep

async def test_parsing_with_tracker():
    tracker = DialogueStateTracker.from_dict("1", [], [Slot("requested_language")])

    # we'll expect this value 'en' to be part of the result from the interpreter
    tracker._set_slot("requested_language", "en")

    endpoint = EndpointConfig("https://interpreter.com")
    with aioresponses() as mocked:
        mocked.post("https://interpreter.com/parse", repeat=True, status=200)

        # mock the parse function with the one defined for this test
        with patch.object(RasaNLUHttpInterpreter, "parse", mocked_parse):
            interpreter = RasaNLUHttpInterpreter(endpoint=endpoint)
            agent = Agent(None, None, interpreter)
            result = await agent.parse_message_using_nlu_interpreter("lunch?", tracker)

            assert result["requested_language"] == "en"

def test_shutdown(self):
        """
        Make sure we call unsubscribe callbacks when shutting down.
        """
        mock = aioresponses()
        conf = TEST_CONFIG['SERVICES']['ws_test']

        async def task():
            # Wait until backend is ready.
            await asyncio.sleep(0.1)

            aiosession = aiohttp.ClientSession()

            async with aiosession.ws_connect(self.ws_url) as ws:
                await ws.send_str(json.dumps({
                    'event': 'subscribe',
                    'subscription': 'ws_test.hello',
                }))
                msg = await ws.receive()
                assert msg.type == aiohttp.WSMsgType.TEXT
                data = json.loads(msg.data)

async def setUp(self):
        self.session = aiohttp.ClientSession(loop=self.loop)
        self.addCleanup(self.session.close)

        self.mocked = aioresponses()
        self.mocked.start()
        self.addCleanup(self.mocked.stop)

        # Just to make absolutely sure no tests leak to a real server
        telemetry.TELEMETRY_SERVER = "https://sql.telemetry.example.com"

async def get_async_data():
        darksky = DarkSkyAsync("api_key")
        with aioresponses.aioresponses() as resp:
            resp.get(re.compile(".+"), status=200, payload=copy.deepcopy(DATA))

            result = await darksky.get_forecast(
                DATA["latitude"],
                DATA["longitude"],
                client_session=aiohttp.ClientSession()
            )

        return result

async def test_remote_action_runs(
    default_channel, default_nlg, default_tracker, default_domain
):

    endpoint = EndpointConfig("https://example.com/webhooks/actions")
    remote_action = action.RemoteAction("my_action", endpoint)

    with aioresponses() as mocked:
        mocked.post(
            "https://example.com/webhooks/actions",
            payload={"events": [], "responses": []},
        )

        await remote_action.run(
            default_channel, default_nlg, default_tracker, default_domain
        )

        r = latest_request(mocked, "post", "https://example.com/webhooks/actions")

        assert r

        assert json_of_latest_request(r) == {
            "domain": default_domain.as_dict(),
            "next_action": "my_action",

    @aioresponses()
    def test_method_dont_match(self, m):
        m.get(self.url)
        with self.assertRaises(ClientConnectionError):
            self.run_async(self.session.post(self.url))

    @aioresponses()
    @asyncio.coroutine
    def test_streaming(self, m):
        m.get(self.url, body='Test')
        resp = yield from self.session.get(self.url)
        content = yield from resp.content.read()
        self.assertEqual(content, b'Test')