526 lines
17 KiB
Python
526 lines
17 KiB
Python
"""The tests for the Dialogflow component."""
|
|
# pylint: disable=protected-access
|
|
import json
|
|
import unittest
|
|
|
|
import requests
|
|
from aiohttp.hdrs import CONTENT_TYPE
|
|
|
|
from homeassistant.core import callback
|
|
from homeassistant import setup, const
|
|
from homeassistant.components import dialogflow, http
|
|
|
|
from tests.common import get_test_instance_port, get_test_home_assistant
|
|
|
|
API_PASSWORD = 'test1234'
|
|
SERVER_PORT = get_test_instance_port()
|
|
BASE_API_URL = "http://127.0.0.1:{}".format(SERVER_PORT)
|
|
INTENTS_API_URL = "{}{}".format(BASE_API_URL, dialogflow.INTENTS_API_ENDPOINT)
|
|
|
|
HA_HEADERS = {
|
|
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
|
|
CONTENT_TYPE: const.CONTENT_TYPE_JSON,
|
|
}
|
|
|
|
SESSION_ID = "a9b84cec-46b6-484e-8f31-f65dba03ae6d"
|
|
INTENT_ID = "c6a74079-a8f0-46cd-b372-5a934d23591c"
|
|
INTENT_NAME = "tests"
|
|
REQUEST_ID = "19ef7e78-fe15-4e94-99dd-0c0b1e8753c3"
|
|
REQUEST_TIMESTAMP = "2017-01-21T17:54:18.952Z"
|
|
CONTEXT_NAME = "78a5db95-b7d6-4d50-9c9b-2fc73a5e34c3_id_dialog_context"
|
|
MAX_RESPONSE_TIME = 5 # https://dialogflow.com/docs/fulfillment
|
|
|
|
# An unknown action takes 8 s to return. Request timeout should be bigger to
|
|
# allow the test to finish
|
|
REQUEST_TIMEOUT = 15
|
|
|
|
# pylint: disable=invalid-name
|
|
hass = None
|
|
calls = []
|
|
|
|
|
|
# pylint: disable=invalid-name
|
|
def setUpModule():
|
|
"""Initialize a Home Assistant server for testing this module."""
|
|
global hass
|
|
|
|
hass = get_test_home_assistant()
|
|
|
|
setup.setup_component(
|
|
hass, http.DOMAIN, {
|
|
http.DOMAIN: {
|
|
http.CONF_API_PASSWORD: API_PASSWORD,
|
|
http.CONF_SERVER_PORT: SERVER_PORT,
|
|
}
|
|
}
|
|
)
|
|
|
|
@callback
|
|
def mock_service(call):
|
|
"""Mock action call."""
|
|
calls.append(call)
|
|
|
|
hass.services.register('test', 'dialogflow', mock_service)
|
|
|
|
assert setup.setup_component(hass, dialogflow.DOMAIN, {
|
|
"dialogflow": {},
|
|
})
|
|
assert setup.setup_component(hass, "intent_script", {
|
|
"intent_script": {
|
|
"WhereAreWeIntent": {
|
|
"speech": {
|
|
"type": "plain",
|
|
"text": """
|
|
{%- if is_state("device_tracker.paulus", "home")
|
|
and is_state("device_tracker.anne_therese",
|
|
"home") -%}
|
|
You are both home, you silly
|
|
{%- else -%}
|
|
Anne Therese is at {{
|
|
states("device_tracker.anne_therese")
|
|
}} and Paulus is at {{
|
|
states("device_tracker.paulus")
|
|
}}
|
|
{% endif %}
|
|
""",
|
|
}
|
|
},
|
|
"GetZodiacHoroscopeIntent": {
|
|
"speech": {
|
|
"type": "plain",
|
|
"text": "You told us your sign is {{ ZodiacSign }}.",
|
|
}
|
|
},
|
|
"CallServiceIntent": {
|
|
"speech": {
|
|
"type": "plain",
|
|
"text": "Service called",
|
|
},
|
|
"action": {
|
|
"service": "test.dialogflow",
|
|
"data_template": {
|
|
"hello": "{{ ZodiacSign }}"
|
|
},
|
|
"entity_id": "switch.test",
|
|
}
|
|
}
|
|
}
|
|
})
|
|
|
|
hass.start()
|
|
|
|
|
|
# pylint: disable=invalid-name
|
|
def tearDownModule():
|
|
"""Stop the Home Assistant server."""
|
|
hass.stop()
|
|
|
|
|
|
def _intent_req(data):
|
|
return requests.post(
|
|
INTENTS_API_URL, data=json.dumps(data), timeout=REQUEST_TIMEOUT,
|
|
headers=HA_HEADERS)
|
|
|
|
|
|
class TestDialogflow(unittest.TestCase):
|
|
"""Test Dialogflow."""
|
|
|
|
def tearDown(self):
|
|
"""Stop everything that was started."""
|
|
hass.block_till_done()
|
|
|
|
def test_intent_action_incomplete(self):
|
|
"""Test when action is not completed."""
|
|
data = {
|
|
"id": REQUEST_ID,
|
|
"timestamp": REQUEST_TIMESTAMP,
|
|
"result": {
|
|
"source": "agent",
|
|
"resolvedQuery": "my zodiac sign is virgo",
|
|
"speech": "",
|
|
"action": "GetZodiacHoroscopeIntent",
|
|
"actionIncomplete": True,
|
|
"parameters": {
|
|
"ZodiacSign": "virgo"
|
|
},
|
|
"metadata": {
|
|
"intentId": INTENT_ID,
|
|
"webhookUsed": "true",
|
|
"webhookForSlotFillingUsed": "false",
|
|
"intentName": INTENT_NAME
|
|
},
|
|
"fulfillment": {
|
|
"speech": "",
|
|
"messages": [
|
|
{
|
|
"type": 0,
|
|
"speech": ""
|
|
}
|
|
]
|
|
},
|
|
"score": 1
|
|
},
|
|
"status": {
|
|
"code": 200,
|
|
"errorType": "success"
|
|
},
|
|
"sessionId": SESSION_ID,
|
|
"originalRequest": None
|
|
}
|
|
|
|
req = _intent_req(data)
|
|
self.assertEqual(200, req.status_code)
|
|
self.assertEqual("", req.text)
|
|
|
|
def test_intent_slot_filling(self):
|
|
"""Test when Dialogflow asks for slot-filling return none."""
|
|
data = {
|
|
"id": REQUEST_ID,
|
|
"timestamp": REQUEST_TIMESTAMP,
|
|
"result": {
|
|
"source": "agent",
|
|
"resolvedQuery": "my zodiac sign is",
|
|
"speech": "",
|
|
"action": "GetZodiacHoroscopeIntent",
|
|
"actionIncomplete": True,
|
|
"parameters": {
|
|
"ZodiacSign": ""
|
|
},
|
|
"contexts": [
|
|
{
|
|
"name": CONTEXT_NAME,
|
|
"parameters": {
|
|
"ZodiacSign.original": "",
|
|
"ZodiacSign": ""
|
|
},
|
|
"lifespan": 2
|
|
},
|
|
{
|
|
"name": "tests_ha_dialog_context",
|
|
"parameters": {
|
|
"ZodiacSign.original": "",
|
|
"ZodiacSign": ""
|
|
},
|
|
"lifespan": 2
|
|
},
|
|
{
|
|
"name": "tests_ha_dialog_params_zodiacsign",
|
|
"parameters": {
|
|
"ZodiacSign.original": "",
|
|
"ZodiacSign": ""
|
|
},
|
|
"lifespan": 1
|
|
}
|
|
],
|
|
"metadata": {
|
|
"intentId": INTENT_ID,
|
|
"webhookUsed": "true",
|
|
"webhookForSlotFillingUsed": "true",
|
|
"intentName": INTENT_NAME
|
|
},
|
|
"fulfillment": {
|
|
"speech": "What is the ZodiacSign?",
|
|
"messages": [
|
|
{
|
|
"type": 0,
|
|
"speech": "What is the ZodiacSign?"
|
|
}
|
|
]
|
|
},
|
|
"score": 0.77
|
|
},
|
|
"status": {
|
|
"code": 200,
|
|
"errorType": "success"
|
|
},
|
|
"sessionId": SESSION_ID,
|
|
"originalRequest": None
|
|
}
|
|
|
|
req = _intent_req(data)
|
|
self.assertEqual(200, req.status_code)
|
|
self.assertEqual("", req.text)
|
|
|
|
def test_intent_request_with_parameters(self):
|
|
"""Test a request with parameters."""
|
|
data = {
|
|
"id": REQUEST_ID,
|
|
"timestamp": REQUEST_TIMESTAMP,
|
|
"result": {
|
|
"source": "agent",
|
|
"resolvedQuery": "my zodiac sign is virgo",
|
|
"speech": "",
|
|
"action": "GetZodiacHoroscopeIntent",
|
|
"actionIncomplete": False,
|
|
"parameters": {
|
|
"ZodiacSign": "virgo"
|
|
},
|
|
"contexts": [],
|
|
"metadata": {
|
|
"intentId": INTENT_ID,
|
|
"webhookUsed": "true",
|
|
"webhookForSlotFillingUsed": "false",
|
|
"intentName": INTENT_NAME
|
|
},
|
|
"fulfillment": {
|
|
"speech": "",
|
|
"messages": [
|
|
{
|
|
"type": 0,
|
|
"speech": ""
|
|
}
|
|
]
|
|
},
|
|
"score": 1
|
|
},
|
|
"status": {
|
|
"code": 200,
|
|
"errorType": "success"
|
|
},
|
|
"sessionId": SESSION_ID,
|
|
"originalRequest": None
|
|
}
|
|
req = _intent_req(data)
|
|
self.assertEqual(200, req.status_code)
|
|
text = req.json().get("speech")
|
|
self.assertEqual("You told us your sign is virgo.", text)
|
|
|
|
def test_intent_request_with_parameters_but_empty(self):
|
|
"""Test a request with parameters but empty value."""
|
|
data = {
|
|
"id": REQUEST_ID,
|
|
"timestamp": REQUEST_TIMESTAMP,
|
|
"result": {
|
|
"source": "agent",
|
|
"resolvedQuery": "my zodiac sign is virgo",
|
|
"speech": "",
|
|
"action": "GetZodiacHoroscopeIntent",
|
|
"actionIncomplete": False,
|
|
"parameters": {
|
|
"ZodiacSign": ""
|
|
},
|
|
"contexts": [],
|
|
"metadata": {
|
|
"intentId": INTENT_ID,
|
|
"webhookUsed": "true",
|
|
"webhookForSlotFillingUsed": "false",
|
|
"intentName": INTENT_NAME
|
|
},
|
|
"fulfillment": {
|
|
"speech": "",
|
|
"messages": [
|
|
{
|
|
"type": 0,
|
|
"speech": ""
|
|
}
|
|
]
|
|
},
|
|
"score": 1
|
|
},
|
|
"status": {
|
|
"code": 200,
|
|
"errorType": "success"
|
|
},
|
|
"sessionId": SESSION_ID,
|
|
"originalRequest": None
|
|
}
|
|
req = _intent_req(data)
|
|
self.assertEqual(200, req.status_code)
|
|
text = req.json().get("speech")
|
|
self.assertEqual("You told us your sign is .", text)
|
|
|
|
def test_intent_request_without_slots(self):
|
|
"""Test a request without slots."""
|
|
data = {
|
|
"id": REQUEST_ID,
|
|
"timestamp": REQUEST_TIMESTAMP,
|
|
"result": {
|
|
"source": "agent",
|
|
"resolvedQuery": "where are we",
|
|
"speech": "",
|
|
"action": "WhereAreWeIntent",
|
|
"actionIncomplete": False,
|
|
"parameters": {},
|
|
"contexts": [],
|
|
"metadata": {
|
|
"intentId": INTENT_ID,
|
|
"webhookUsed": "true",
|
|
"webhookForSlotFillingUsed": "false",
|
|
"intentName": INTENT_NAME
|
|
},
|
|
"fulfillment": {
|
|
"speech": "",
|
|
"messages": [
|
|
{
|
|
"type": 0,
|
|
"speech": ""
|
|
}
|
|
]
|
|
},
|
|
"score": 1
|
|
},
|
|
"status": {
|
|
"code": 200,
|
|
"errorType": "success"
|
|
},
|
|
"sessionId": SESSION_ID,
|
|
"originalRequest": None
|
|
}
|
|
req = _intent_req(data)
|
|
self.assertEqual(200, req.status_code)
|
|
text = req.json().get("speech")
|
|
|
|
self.assertEqual("Anne Therese is at unknown and Paulus is at unknown",
|
|
text)
|
|
|
|
hass.states.set("device_tracker.paulus", "home")
|
|
hass.states.set("device_tracker.anne_therese", "home")
|
|
|
|
req = _intent_req(data)
|
|
self.assertEqual(200, req.status_code)
|
|
text = req.json().get("speech")
|
|
self.assertEqual("You are both home, you silly", text)
|
|
|
|
def test_intent_request_calling_service(self):
|
|
"""Test a request for calling a service.
|
|
|
|
If this request is done async the test could finish before the action
|
|
has been executed. Hard to test because it will be a race condition.
|
|
"""
|
|
data = {
|
|
"id": REQUEST_ID,
|
|
"timestamp": REQUEST_TIMESTAMP,
|
|
"result": {
|
|
"source": "agent",
|
|
"resolvedQuery": "my zodiac sign is virgo",
|
|
"speech": "",
|
|
"action": "CallServiceIntent",
|
|
"actionIncomplete": False,
|
|
"parameters": {
|
|
"ZodiacSign": "virgo"
|
|
},
|
|
"contexts": [],
|
|
"metadata": {
|
|
"intentId": INTENT_ID,
|
|
"webhookUsed": "true",
|
|
"webhookForSlotFillingUsed": "false",
|
|
"intentName": INTENT_NAME
|
|
},
|
|
"fulfillment": {
|
|
"speech": "",
|
|
"messages": [
|
|
{
|
|
"type": 0,
|
|
"speech": ""
|
|
}
|
|
]
|
|
},
|
|
"score": 1
|
|
},
|
|
"status": {
|
|
"code": 200,
|
|
"errorType": "success"
|
|
},
|
|
"sessionId": SESSION_ID,
|
|
"originalRequest": None
|
|
}
|
|
call_count = len(calls)
|
|
req = _intent_req(data)
|
|
self.assertEqual(200, req.status_code)
|
|
self.assertEqual(call_count + 1, len(calls))
|
|
call = calls[-1]
|
|
self.assertEqual("test", call.domain)
|
|
self.assertEqual("dialogflow", call.service)
|
|
self.assertEqual(["switch.test"], call.data.get("entity_id"))
|
|
self.assertEqual("virgo", call.data.get("hello"))
|
|
|
|
def test_intent_with_no_action(self):
|
|
"""Test an intent with no defined action."""
|
|
data = {
|
|
"id": REQUEST_ID,
|
|
"timestamp": REQUEST_TIMESTAMP,
|
|
"result": {
|
|
"source": "agent",
|
|
"resolvedQuery": "my zodiac sign is virgo",
|
|
"speech": "",
|
|
"action": "",
|
|
"actionIncomplete": False,
|
|
"parameters": {
|
|
"ZodiacSign": ""
|
|
},
|
|
"contexts": [],
|
|
"metadata": {
|
|
"intentId": INTENT_ID,
|
|
"webhookUsed": "true",
|
|
"webhookForSlotFillingUsed": "false",
|
|
"intentName": INTENT_NAME
|
|
},
|
|
"fulfillment": {
|
|
"speech": "",
|
|
"messages": [
|
|
{
|
|
"type": 0,
|
|
"speech": ""
|
|
}
|
|
]
|
|
},
|
|
"score": 1
|
|
},
|
|
"status": {
|
|
"code": 200,
|
|
"errorType": "success"
|
|
},
|
|
"sessionId": SESSION_ID,
|
|
"originalRequest": None
|
|
}
|
|
req = _intent_req(data)
|
|
self.assertEqual(200, req.status_code)
|
|
text = req.json().get("speech")
|
|
self.assertEqual(
|
|
"You have not defined an action in your Dialogflow intent.", text)
|
|
|
|
def test_intent_with_unknown_action(self):
|
|
"""Test an intent with an action not defined in the conf."""
|
|
data = {
|
|
"id": REQUEST_ID,
|
|
"timestamp": REQUEST_TIMESTAMP,
|
|
"result": {
|
|
"source": "agent",
|
|
"resolvedQuery": "my zodiac sign is virgo",
|
|
"speech": "",
|
|
"action": "unknown",
|
|
"actionIncomplete": False,
|
|
"parameters": {
|
|
"ZodiacSign": ""
|
|
},
|
|
"contexts": [],
|
|
"metadata": {
|
|
"intentId": INTENT_ID,
|
|
"webhookUsed": "true",
|
|
"webhookForSlotFillingUsed": "false",
|
|
"intentName": INTENT_NAME
|
|
},
|
|
"fulfillment": {
|
|
"speech": "",
|
|
"messages": [
|
|
{
|
|
"type": 0,
|
|
"speech": ""
|
|
}
|
|
]
|
|
},
|
|
"score": 1
|
|
},
|
|
"status": {
|
|
"code": 200,
|
|
"errorType": "success"
|
|
},
|
|
"sessionId": SESSION_ID,
|
|
"originalRequest": None
|
|
}
|
|
req = _intent_req(data)
|
|
self.assertEqual(200, req.status_code)
|
|
text = req.json().get("speech")
|
|
self.assertEqual(
|
|
"This intent is not yet configured within Home Assistant.", text)
|