325 lines
11 KiB
Python
325 lines
11 KiB
Python
# Copyright 2017 Mycroft AI Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
#
|
|
from copy import copy
|
|
from http import HTTPStatus
|
|
from unittest import TestCase
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from requests import HTTPError
|
|
|
|
from mycroft.api import (
|
|
Api, DeviceApi, STTApi, has_been_paired, is_paired, BackendDown
|
|
)
|
|
from ..mocks import base_config
|
|
|
|
API_DOMAIN = 'https://api-test.mycroft.ai'
|
|
|
|
|
|
def mock_identity(paired=True):
|
|
identity_mock = MagicMock()
|
|
identity_mock.is_expired.return_value = False
|
|
if paired:
|
|
identity_mock.uuid = "1234"
|
|
else:
|
|
identity_mock.uuid = ""
|
|
|
|
return identity_mock
|
|
|
|
|
|
def mock_http_response(status, json=None, url=''):
|
|
response = MagicMock()
|
|
response.status_code = status
|
|
response.json.return_value = json or {}
|
|
response.url = url
|
|
|
|
return response
|
|
|
|
|
|
class ApiTestBase(TestCase):
|
|
def setUp(self):
|
|
self._patch_config()
|
|
self._patch_requests()
|
|
self._patch_identity()
|
|
super().setUp()
|
|
|
|
def _patch_config(self):
|
|
config = base_config()
|
|
server_config = dict(
|
|
url=API_DOMAIN,
|
|
version='v1',
|
|
update=True,
|
|
metrics=False
|
|
)
|
|
enclosure_config = dict(packaging_type='pantacor')
|
|
config.update(
|
|
data_dir='/opt/mycroft',
|
|
server=server_config,
|
|
enclosure=enclosure_config
|
|
)
|
|
patcher = patch(
|
|
'mycroft.configuration.Configuration.get', return_value=config
|
|
)
|
|
self.config_mock = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
def _patch_requests(self):
|
|
patcher = patch('mycroft.api.requests.request')
|
|
self.http_request_mock = patcher.start()
|
|
self.http_response_ok = mock_http_response(HTTPStatus.OK)
|
|
self.http_request_mock.return_value = self.http_response_ok
|
|
self.addCleanup(patcher.stop)
|
|
patcher = patch('mycroft.api.requests.post')
|
|
self.mock_post = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
def _patch_identity(self):
|
|
patcher = patch('mycroft.api.IdentityManager')
|
|
self.identity_mock = patcher.start()
|
|
self.identity_mock.get = MagicMock(return_value=mock_identity())
|
|
self.addCleanup(patcher.stop)
|
|
|
|
def _check_api_request(self, url, method):
|
|
request_url = self.http_request_mock.call_args[0][1]
|
|
self.assertEqual(request_url, API_DOMAIN + url)
|
|
request_method = self.http_request_mock.call_args[0][0]
|
|
self.assertEqual(request_method, method)
|
|
|
|
|
|
class TestApi(ApiTestBase):
|
|
def test_init(self):
|
|
self.identity_mock.get = MagicMock(return_value=mock_identity())
|
|
api = Api('test-path')
|
|
self.assertEqual(api.url, API_DOMAIN)
|
|
self.assertEqual(api.version, 'v1')
|
|
self.assertEqual(api.identity.uuid, '1234')
|
|
|
|
def test_send(self):
|
|
mock_response_301 = mock_http_response(HTTPStatus.MOVED_PERMANENTLY)
|
|
mock_response_401 = mock_http_response(
|
|
HTTPStatus.UNAUTHORIZED, url='auth/token'
|
|
)
|
|
mock_response_refresh = mock_http_response(
|
|
HTTPStatus.UNAUTHORIZED, url=''
|
|
)
|
|
|
|
self.http_request_mock.return_value = self.http_response_ok
|
|
api = Api('test-path')
|
|
req = {'path': 'something', 'headers': {}}
|
|
|
|
# Check successful
|
|
self.assertEqual(api.send(req), self.http_response_ok.json())
|
|
|
|
# check that a 300+ status code generates Exception
|
|
self.http_request_mock.return_value = mock_response_301
|
|
with self.assertRaises(HTTPError):
|
|
api.send(req)
|
|
|
|
# Check 401
|
|
self.http_request_mock.return_value = mock_response_401
|
|
req = {'path': '', 'headers': {}}
|
|
with self.assertRaises(HTTPError):
|
|
api.send(req)
|
|
|
|
# Check refresh token
|
|
api.identity.is_expired = MagicMock(return_value=True)
|
|
api.old_params = copy(req)
|
|
self.http_request_mock.side_effect = [
|
|
mock_response_refresh, self.http_response_ok, self.http_response_ok
|
|
]
|
|
req = {'path': 'something', 'headers': {}}
|
|
api.send(req)
|
|
self.assertTrue(self.identity_mock.save.called)
|
|
|
|
|
|
class TestDeviceApi(ApiTestBase):
|
|
def test_init(self):
|
|
device = DeviceApi()
|
|
self.assertEqual(device.identity.uuid, '1234')
|
|
self.assertEqual(device.path, 'device')
|
|
|
|
def test_device_activate(self):
|
|
device = DeviceApi()
|
|
device.activate('state', 'token')
|
|
json = self.http_request_mock.call_args[1]['json']
|
|
self.assertEqual(json['state'], 'state')
|
|
self.assertEqual(json['token'], 'token')
|
|
|
|
def test_device_get(self):
|
|
device = DeviceApi()
|
|
device.get()
|
|
self._check_api_request('/v1/device/1234', 'GET')
|
|
|
|
def test_device_get_code(self):
|
|
self.http_request_mock.return_value = mock_http_response(200, '123ABC')
|
|
device = DeviceApi()
|
|
pairing_code = device.get_code('state')
|
|
self.assertEqual(pairing_code, '123ABC')
|
|
self._check_api_request(
|
|
'/v1/device/code?state=state', 'GET'
|
|
)
|
|
|
|
def test_device_get_settings(self):
|
|
device = DeviceApi()
|
|
device.get_settings()
|
|
self._check_api_request('/v1/device/1234/setting', 'GET')
|
|
|
|
def test_device_report_metric(self):
|
|
device = DeviceApi()
|
|
device.report_metric('my_metric', {'data': 'my_data'})
|
|
params = self.http_request_mock.call_args[1]
|
|
|
|
content_type = params['headers']['Content-Type']
|
|
correct_json = {'data': 'my_data'}
|
|
self.assertEqual(content_type, 'application/json')
|
|
self.assertEqual(params['json'], correct_json)
|
|
self._check_api_request('/v1/device/1234/metric/my_metric', 'POST')
|
|
|
|
def test_device_send_email(self):
|
|
device = DeviceApi()
|
|
device.send_email('title', 'body', 'sender')
|
|
params = self.http_request_mock.call_args[1]
|
|
|
|
content_type = params['headers']['Content-Type']
|
|
correct_json = {'body': 'body', 'sender': 'sender', 'title': 'title'}
|
|
self.assertEqual(content_type, 'application/json')
|
|
self.assertEqual(params['json'], correct_json)
|
|
self._check_api_request('/v1/device/1234/message', 'PUT')
|
|
|
|
def test_device_get_oauth_token(self):
|
|
device = DeviceApi()
|
|
device.get_oauth_token(1)
|
|
self._check_api_request('/v1/device/1234/token/1', 'GET')
|
|
|
|
def test_device_get_location(self):
|
|
device = DeviceApi()
|
|
device.get_location()
|
|
self._check_api_request('/v1/device/1234/location', 'GET')
|
|
|
|
def test_device_get_subscription(self):
|
|
device = DeviceApi()
|
|
device.get_subscription()
|
|
self._check_api_request('/v1/device/1234/subscription', 'GET')
|
|
|
|
request_json = {'@type': 'free'}
|
|
self.http_request_mock.return_value = mock_http_response(200, request_json)
|
|
self.assertFalse(device.is_subscriber)
|
|
|
|
request_json = {'@type': 'monthly'}
|
|
self.http_request_mock.return_value = mock_http_response(200, request_json)
|
|
self.assertTrue(device.is_subscriber)
|
|
|
|
request_json = {'@type': 'yearly'}
|
|
self.http_request_mock.return_value = mock_http_response(200, request_json)
|
|
self.assertTrue(device.is_subscriber)
|
|
|
|
def test_device_upload_skills_data(self):
|
|
device = DeviceApi()
|
|
device.upload_skills_data({})
|
|
data = self.http_request_mock.call_args[1]['json']
|
|
|
|
# Check that the correct url is called
|
|
self._check_api_request('/v1/device/1234/skillJson', 'PUT')
|
|
|
|
# Check that the correct data is sent as json
|
|
self.assertTrue('blacklist' in data)
|
|
self.assertTrue('skills' in data)
|
|
|
|
with self.assertRaises(ValueError):
|
|
device.upload_skills_data('This isn\'t right at all')
|
|
|
|
def test_has_been_paired(self):
|
|
identity_load_mock = MagicMock()
|
|
self.identity_mock.load = MagicMock(return_value=identity_load_mock)
|
|
# Test None
|
|
identity_load_mock.uuid = None
|
|
self.assertFalse(has_been_paired())
|
|
# Test empty string
|
|
identity_load_mock.uuid = ""
|
|
self.assertFalse(has_been_paired())
|
|
# Test actual id number
|
|
identity_load_mock.uuid = "1234"
|
|
self.assertTrue(has_been_paired())
|
|
|
|
def test_upload_meta(self):
|
|
device = DeviceApi()
|
|
|
|
settings_section = dict(
|
|
name='Settings',
|
|
fields=[dict(name='Set me', type='number', value=4)]
|
|
)
|
|
settings_meta = dict(
|
|
name='TestMeta',
|
|
skill_gid='test_skill|19.02',
|
|
skillMetadata=dict(sections=[settings_section])
|
|
)
|
|
device.upload_skill_metadata(settings_meta)
|
|
params = self.http_request_mock.call_args[1]
|
|
|
|
content_type = params['headers']['Content-Type']
|
|
self.assertEqual(content_type, 'application/json')
|
|
self.assertEqual(params['json'], settings_meta)
|
|
self._check_api_request('/v1/device/1234/settingsMeta', 'PUT')
|
|
|
|
def test_get_skill_settings(self):
|
|
device = DeviceApi()
|
|
device.get_skill_settings()
|
|
|
|
self._check_api_request('/v1/device/1234/skill/settings', 'GET')
|
|
|
|
|
|
class TestSttApi(ApiTestBase):
|
|
def test_stt(self):
|
|
stt = STTApi('stt')
|
|
self.assertEqual(stt.path, 'stt')
|
|
|
|
def test_stt_stt(self):
|
|
stt = STTApi('stt')
|
|
stt.stt('La la la', 'en-US', 1)
|
|
self._check_api_request('/v1/stt', 'POST')
|
|
data = self.http_request_mock.call_args[1].get('data')
|
|
self.assertEqual(data, 'La la la')
|
|
params = self.http_request_mock.call_args[1].get('params')
|
|
self.assertEqual(params['lang'], 'en-US')
|
|
|
|
|
|
@patch('mycroft.api._paired_cache', False)
|
|
class TestIsPaired(ApiTestBase):
|
|
def test_is_paired_true(self):
|
|
num_calls = self.identity_mock.get.num_calls
|
|
self.assertTrue(is_paired())
|
|
self.assertEqual(num_calls, self.identity_mock.get.num_calls)
|
|
self._check_api_request('/v1/device/1234', 'GET')
|
|
|
|
def test_is_paired_false_local(self):
|
|
self.identity_mock.get.return_value = mock_identity(paired=False)
|
|
self.assertFalse(is_paired())
|
|
self.identity_mock.uuid = None
|
|
self.assertFalse(is_paired())
|
|
|
|
def test_is_paired_false_remote(self):
|
|
self.http_request_mock.return_value = mock_http_response(
|
|
HTTPStatus.UNAUTHORIZED
|
|
)
|
|
self.assertFalse(is_paired())
|
|
|
|
def test_is_paired_error_remote(self):
|
|
self.http_request_mock.return_value = mock_http_response(
|
|
HTTPStatus.INTERNAL_SERVER_ERROR
|
|
)
|
|
self.assertFalse(is_paired())
|
|
|
|
with self.assertRaises(BackendDown):
|
|
is_paired(ignore_errors=False)
|