Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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)
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')
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_config=endpoint)
agent = Agent(None, None, interpreter)
result = await agent.parse_message_using_nlu_interpreter("lunch?", tracker)
assert result["requested_language"] == "en"