diff --git a/test/integrationtests/test_common_query_skill.py b/test/integrationtests/test_common_query_skill.py deleted file mode 100644 index 0d70ff50eae..00000000000 --- a/test/integrationtests/test_common_query_skill.py +++ /dev/null @@ -1,62 +0,0 @@ -from unittest import TestCase, mock - -from ovos_bus_client.message import Message - -from ovos_workshop.skills.common_query_skill import CommonQuerySkill - -# TODO - move test to ovos-workshop - -class AnyCallable: - """Class matching any callable. - - Useful for assert_called_with arguments. - """ - def __eq__(self, other): - return callable(other) - - - -class TestCommonQuerySkill(TestCase): - def setUp(self): - self.skill = CQSTest() - self.bus = mock.Mock(name='bus') - self.skill.bind(self.bus) - self.skill.config_core = {'enclosure': {'platform': 'mycroft_mark_1'}} - - def test_lifecycle(self): - """Test startup and shutdown.""" - skill = CQSTest() - bus = mock.Mock(name='bus') - skill.bind(bus) - bus.on.assert_any_call('question:query', AnyCallable()) - bus.on.assert_any_call('question:action', AnyCallable()) - skill.shutdown() - - def test_common_test_skill_action(self): - """Test that the optional action is triggered.""" - query_action = self.bus.on.call_args_list[-2][0][1] - query_action(Message('query:action', data={ - 'phrase': 'What\'s the meaning of life', - 'skill_id': 'asdf'})) - self.skill.CQS_action.assert_not_called() - query_action(Message('query:action', data={ - 'phrase': 'What\'s the meaning of life', - 'skill_id': 'CQSTest'})) - self.skill.CQS_action.assert_called_once_with( - 'What\'s the meaning of life', {}) - - -class CQSTest(CommonQuerySkill): - """Simple skill for testing the CommonQuerySkill""" - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.CQS_match_query_phrase = mock.Mock(name='match_phrase') - self.CQS_action = mock.Mock(name='selected_action') - self.skill_id = 'CQSTest' - - def CQS_match_query_phrase(self, phrase): - pass - - def CQS_action(self, phrase, data): - pass diff --git a/test/integrationtests/test_context.py b/test/integrationtests/test_context.py deleted file mode 100644 index 549fce6b840..00000000000 --- a/test/integrationtests/test_context.py +++ /dev/null @@ -1,40 +0,0 @@ -from unittest import TestCase, mock - -# TODO - move to ovos-workshop -from ovos_workshop.decorators import adds_context, removes_context - - -class ContextSkillMock(mock.Mock): - """Mock class to apply decorators on.""" - @adds_context('DestroyContext') - def handler_adding_context(self): - pass - - @adds_context('DestroyContext', 'exterminate') - def handler_adding_context_with_words(self): - pass - - @removes_context('DestroyContext') - def handler_removing_context(self): - pass - - -class TestContextDecorators(TestCase): - def test_adding_context(self): - """Check that calling handler adds the correct Keyword.""" - skill = ContextSkillMock() - skill.handler_adding_context() - skill.set_context.assert_called_once_with('DestroyContext', '') - - def test_adding_context_with_words(self): - """Ensure that decorated handler adds Keyword and content.""" - skill = ContextSkillMock() - skill.handler_adding_context_with_words() - skill.set_context.assert_called_once_with('DestroyContext', - 'exterminate') - - def test_removing_context(self): - """Make sure the decorated handler removes the specified context.""" - skill = ContextSkillMock() - skill.handler_removing_context() - skill.remove_context.assert_called_once_with('DestroyContext') diff --git a/test/integrationtests/test_fallback_skill.py b/test/integrationtests/test_fallback_skill.py deleted file mode 100644 index b86974d23b1..00000000000 --- a/test/integrationtests/test_fallback_skill.py +++ /dev/null @@ -1,54 +0,0 @@ -from unittest import TestCase, mock - -from ovos_workshop.skills.fallback import FallbackSkillV1 as FallbackSkill - - -def setup_fallback(fb_class): - fb_skill = fb_class() - fb_skill.bind(mock.Mock(name='bus')) - fb_skill.initialize() - return fb_skill - - -# TODO - move to ovos-workshop -class TestFallbackSkill(TestCase): - def test_life_cycle(self): - """Test startup and shutdown of a fallback skill. - - Ensure that an added handler is removed as part of default shutdown. - """ - self.assertEqual(len(FallbackSkill.fallback_handlers), 0) - fb_skill = setup_fallback(SimpleFallback) - self.assertEqual(len(FallbackSkill.fallback_handlers), 1) - self.assertEqual(FallbackSkill.wrapper_map[0][0], - fb_skill.fallback_handler) - self.assertEqual(len(FallbackSkill.wrapper_map), 1) - - fb_skill.default_shutdown() - self.assertEqual(len(FallbackSkill.fallback_handlers), 0) - self.assertEqual(len(FallbackSkill.wrapper_map), 0) - - def test_manual_removal(self): - """Test that the call to remove_fallback() removes the handler""" - self.assertEqual(len(FallbackSkill.fallback_handlers), 0) - - # Create skill adding a single handler - fb_skill = setup_fallback(SimpleFallback) - self.assertEqual(len(FallbackSkill.fallback_handlers), 1) - - self.assertTrue(fb_skill.remove_fallback(fb_skill.fallback_handler)) - # Both internal trackers of handlers should be cleared now - self.assertEqual(len(FallbackSkill.fallback_handlers), 0) - self.assertEqual(len(FallbackSkill.wrapper_map), 0) - - # Removing after it's already been removed should fail - self.assertFalse(fb_skill.remove_fallback(fb_skill.fallback_handler)) - - -class SimpleFallback(FallbackSkill): - """Simple fallback skill used for test.""" - def initialize(self): - self.register_fallback(self.fallback_handler, 42) - - def fallback_handler(self): - pass diff --git a/test/integrationtests/test_intent_service_interface.py b/test/integrationtests/test_intent_service_interface.py deleted file mode 100644 index 61fea26ed5d..00000000000 --- a/test/integrationtests/test_intent_service_interface.py +++ /dev/null @@ -1,145 +0,0 @@ -import unittest -# TODO - move test to ovos-workshop -from ovos_workshop.intents import IntentBuilder, IntentServiceInterface - - -class MockEmitter: - def __init__(self): - self.reset() - - def emit(self, message): - self.types.append(message.msg_type) - self.results.append(message.data) - - def get_types(self): - return self.types - - def get_results(self): - return self.results - - def on(self, event, f): - pass - - def reset(self): - self.types = [] - self.results = [] - - -class KeywordRegistrationTest(unittest.TestCase): - def check_emitter(self, expected_message_data): - """Verify that the registration messages matches the expected.""" - for msg_type in self.emitter.get_types(): - self.assertEqual(msg_type, 'register_vocab') - self.assertEqual( - sorted(self.emitter.get_results(), - key=lambda d: sorted(d.items())), - sorted(expected_message_data, key=lambda d: sorted(d.items()))) - self.emitter.reset() - - def setUp(self): - self.emitter = MockEmitter() - - def test_register_keyword(self): - intent_service = IntentServiceInterface(self.emitter) - intent_service.register_adapt_keyword('test_intent', 'test', lang='en-US') - entity_data = {'entity_value': 'test', 'entity_type': 'test_intent', 'lang': 'en-US'} - compatibility_data = {'start': 'test', 'end': 'test_intent'} - expected_data = {**entity_data, **compatibility_data} - self.check_emitter([expected_data]) - - def test_register_keyword_with_aliases(self): - # TODO 22.02: Remove compatibility data - intent_service = IntentServiceInterface(self.emitter) - intent_service.register_adapt_keyword('test_intent', 'test', - ['test2', 'test3'], - lang='en-US') - - entity_data = {'entity_value': 'test', 'entity_type': 'test_intent', 'lang': 'en-US'} - compatibility_data = {'start': 'test', 'end': 'test_intent'} - expected_initial_vocab = {**entity_data, **compatibility_data} - - alias_data = { - 'entity_value': 'test2', - 'entity_type': 'test_intent', - 'alias_of': 'test', - 'lang': 'en-US' - } - alias_compatibility = {'start': 'test2', 'end': 'test_intent'} - expected_alias1 = {**alias_data, **alias_compatibility} - - alias_data2 = { - 'entity_value': 'test3', - 'entity_type': 'test_intent', - 'alias_of': 'test', - 'lang': 'en-US' - } - alias_compatibility2 = {'start': 'test3', 'end': 'test_intent'} - expected_alias2 = {**alias_data2, **alias_compatibility2} - - self.check_emitter([expected_initial_vocab, - expected_alias1, - expected_alias2]) - - def test_register_regex(self): - intent_service = IntentServiceInterface(self.emitter) - intent_service.register_adapt_regex('.*', lang="en-us") - self.check_emitter([{'regex': '.*', 'lang': 'en-us'}]) - - -class KeywordIntentRegistrationTest(unittest.TestCase): - def check_emitter(self, expected_message_data): - """Verify that the registration messages matches the expected.""" - for msg_type in self.emitter.get_types(): - self.assertEqual(msg_type, 'register_intent') - self.assertEqual( - sorted(self.emitter.get_results(), - key=lambda d: sorted(d.items())), - sorted(expected_message_data, key=lambda d: sorted(d.items()))) - self.emitter.reset() - - def setUp(self): - self.emitter = MockEmitter() - - def test_register_intent(self): - intent_service = IntentServiceInterface(self.emitter) - intent_service.register_adapt_keyword('testA', 'testA', lang='en-US') - intent_service.register_adapt_keyword('testB', 'testB', lang='en-US') - self.emitter.reset() - - intent = IntentBuilder("test").require("testA").optionally("testB") - intent_service.register_adapt_intent("test", intent) - expected_data = {'at_least_one': [], - 'name': 'test', - 'excludes': [], - 'optional': [('testB', 'testB')], - 'requires': [('testA', 'testA')]} - self.check_emitter([expected_data]) - - - -class UtteranceIntentRegistrationTest(unittest.TestCase): - def check_emitter(self, expected_message_data): - """Verify that the registration messages matches the expected.""" - for msg_type in self.emitter.get_types(): - self.assertEqual(msg_type, 'padatious:register_intent') - - self.assertEqual( - sorted(self.emitter.get_results(), - key=lambda d: sorted(d.items())), - sorted(expected_message_data, key=lambda d: sorted(d.items()))) - self.emitter.reset() - - def setUp(self): - self.emitter = MockEmitter() - - def test_register_intent(self): - intent_service = IntentServiceInterface(self.emitter) - filename = "/tmp/test.intent" - with open(filename, "w") as f: - f.write("this is a test\ntest the intent") - - intent_service.register_padatious_intent('test', filename, lang='en-US') - expected_data = {'file_name': '/tmp/test.intent', 'lang': 'en-US', 'name': 'test', - 'samples': ['this is a test', 'test the intent']} - self.check_emitter([expected_data]) - diff --git a/test/integrationtests/test_mycroft_skill.py b/test/integrationtests/test_mycroft_skill.py deleted file mode 100644 index 09059179b94..00000000000 --- a/test/integrationtests/test_mycroft_skill.py +++ /dev/null @@ -1,641 +0,0 @@ -# -# 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. -# - -import json -import sys -import unittest -from datetime import datetime -from os.path import join, dirname, abspath -from unittest.mock import MagicMock, patch - -from ovos_bus_client.message import Message -from ovos_config import Configuration -from ovos_workshop.decorators import resting_screen_handler, intent_handler -from ovos_workshop.intents import IntentBuilder, open_intent_envelope -from ovos_workshop.skills.mycroft_skill import MycroftSkill -from ovos_workshop.skills.ovos import OVOSSkill -from copy import deepcopy -from ovos_config import LocalConf, DEFAULT_CONFIG - -# TODO - move test to ovos-workshop - -BASE_CONF = deepcopy(LocalConf(DEFAULT_CONFIG)) - - -class MockEmitter(object): - def __init__(self): - self.types = [] - self.results = [] - self.reset() - - def emit(self, message): - self.types.append(message.msg_type) - self.results.append(message.data) - - def get_types(self): - return self.types - - def get_results(self): - return self.results - - def on(self, event, f): - pass - - def reset(self): - self.types = [] - self.results = [] - - -def vocab_base_path(): - return join(dirname(__file__), '..', 'vocab_test') - - -class TestFunction(unittest.TestCase): - def test_resting_screen_handler(self): - class T(OVOSSkill): - @resting_screen_handler('humbug') - def f(self): - pass - - test_class = T() - self.assertTrue('resting_handler' in dir(test_class.f)) - self.assertEqual(test_class.f.resting_handler, 'humbug') - - -class TestOVOSSkill(unittest.TestCase): - emitter = MockEmitter() - regex_path = abspath(join(dirname(__file__), '../regex_test')) - vocab_path = abspath(join(dirname(__file__), '../vocab_test')) - - def setUp(self): - self.emitter.reset() - - def compare_dicts(self, d1, d2): - self.assertEqual(json.dumps(d1, sort_keys=True), - json.dumps(d2, sort_keys=True)) - - def check_emitter(self, result_list): - for msg_type in self.emitter.get_types(): - self.assertEqual(msg_type, 'register_vocab') - self.assertEqual(sorted(self.emitter.get_results(), - key=lambda d: sorted(d.items())), - sorted(result_list, key=lambda d: sorted(d.items()))) - self.emitter.reset() - - def test_open_envelope(self): - name = 'Jerome' - intent = IntentBuilder(name).require('Keyword') - intent.name = name - m = Message("register_intent", intent.__dict__) - unpacked_intent = open_intent_envelope(m) - self.assertEqual(intent.__dict__, unpacked_intent.__dict__) - - def check_detach_intent(self): - self.assertTrue(len(self.emitter.get_types()) > 0) - for msg_type in self.emitter.get_types(): - self.assertEqual(msg_type, 'detach_intent') - self.emitter.reset() - - def check_register_intent(self, result_list): - for msg_type in self.emitter.get_types(): - self.assertEqual(msg_type, 'register_intent') - self.assertEqual(sorted(self.emitter.get_results()), - sorted(result_list)) - self.emitter.reset() - - def check_register_vocabulary(self, result_list): - for msg_type in self.emitter.get_types(): - self.assertEqual(msg_type, 'register_vocab') - self.assertEqual(sorted(self.emitter.get_results()), - sorted(result_list)) - self.emitter.reset() - - def test_register_intent(self): - # Test register Intent object - s = SimpleSkill1(bus=self.emitter, skill_id="A") - expected = [{'at_least_one': [], - 'name': 'A:a', - 'excludes': [], - 'optional': [], - 'requires': [('AKeyword', 'AKeyword')]}] - msg_data = self.emitter.get_results() - self.assertTrue(expected[0] in msg_data) - self.emitter.reset() - - # Test register IntentBuilder object - s = SimpleSkill2(bus=self.emitter, skill_id="A") - expected = [{'at_least_one': [], - 'name': 'A:a', - 'excludes': [], - 'optional': [], - 'requires': [('AKeyword', 'AKeyword')]}] - - msg_data = self.emitter.get_results() - self.assertTrue(expected[0] in msg_data) - self.emitter.reset() - - # Test register IntentBuilder object - with self.assertRaises(ValueError): - s = SimpleSkill3(bus=self.emitter, skill_id="A") - - def test_enable_disable_intent(self): - """Test disable/enable intent.""" - # Setup basic test - s = SimpleSkill1(bus=self.emitter, skill_id="A") - - # check that intent was registered - expected = [{'at_least_one': [], - 'name': 'A:a', - 'excludes': [], - 'optional': [], - 'requires': [('AKeyword', 'AKeyword')]}] - msg_data = self.emitter.get_results() - self.assertTrue(expected[0] in msg_data) - self.emitter.reset() - - # Test disable/enable cycle - s.disable_intent('a') - self.check_detach_intent() - s.enable_intent('a') - self.check_register_intent(expected) - - def test_enable_disable_intent_handlers(self): - """Test disable/enable intent.""" - # Setup basic test - s = SimpleSkill1(bus=self.emitter, skill_id="A") - expected = [{'at_least_one': [], - 'name': 'A:a', - 'excludes': [], - 'optional': [], - 'requires': [('AKeyword', 'AKeyword')]}] - msg_data = self.emitter.get_results() - self.assertTrue(expected[0] in msg_data) - self.emitter.reset() - - # Test disable/enable cycle - msg = Message('test.msg', data={'intent_name': 'a'}) - intent_disabled = s.handle_disable_intent(msg) - self.assertTrue(intent_disabled) - self.check_detach_intent() - intent_enabled = s.handle_enable_intent(msg) - self.assertTrue(intent_enabled) - self.check_register_intent(expected) - - def test_register_vocab(self): - """Test disable/enable intent.""" - # Setup basic test - s = SimpleSkill1(bus=self.emitter, skill_id="A") - - # Normal vocaubulary - self.emitter.reset() - expected = [ - { - 'start': 'hello', - 'end': 'AHelloKeyword', - 'entity_value': 'hello', - 'lang': 'en-us', - 'entity_type': 'AHelloKeyword' - } - ] - s.register_vocabulary('hello', 'HelloKeyword') - self.check_register_vocabulary(expected) - # Regex - s.register_regex('weird (?P.+) stuff') - expected = [{'lang': 'en-us', 'regex': 'weird (?P.+) stuff'}] - self.check_register_vocabulary(expected) - - def check_register_object_file(self, types_list, result_list): - self.assertEqual(sorted(self.emitter.get_types()), - sorted(types_list)) - self.assertEqual(sorted(self.emitter.get_results(), - key=lambda d: sorted(d.items())), - sorted(result_list, key=lambda d: sorted(d.items()))) - self.emitter.reset() - - def test_register_intent_file(self): - self._test_intent_file(SimpleSkill4()) - - def test_register_intent_intent_file(self): - """Test register intent files using register_intent.""" - self._test_intent_file(SimpleSkill6()) - - def _test_intent_file(self, s): - s.res_dir = abspath(join(dirname(__file__), 'intent_file')) - s._startup(bus=self.emitter, skill_id="A") - - expected_types = [ - 'padatious:register_intent', - 'padatious:register_entity' - ] - - expected_results = [ - { - 'file_name': join(dirname(__file__), 'intent_file', - 'vocab', 'en-us', 'test.intent'), - 'lang': 'en-us', - 'name': str(s.skill_id) + ':test.intent', - 'samples': [] - }, - { - 'file_name': join(dirname(__file__), 'intent_file', - 'vocab', 'en-us', 'test_ent.entity'), - 'lang': 'en-us', - 'name': str(s.skill_id) + ':test_ent_87af9db6c8402bcfaa8ebc719ae4427c', - 'samples': [] - } - ] - self.check_register_object_file(expected_types, expected_results) - - def check_register_decorators(self, result_list): - self.assertEqual(sorted(self.emitter.get_results(), - key=lambda d: sorted(d.items())), - sorted(result_list, key=lambda d: sorted(d.items()))) - self.emitter.reset() - - def test_register_decorators(self): - """ Test decorated intents """ - path_orig = sys.path - sys.path.append(abspath(dirname(__file__))) - SimpleSkill5 = __import__('decorator_test_skill').TestSkill - s = SimpleSkill5() - s.res_dir = abspath(join(dirname(__file__), 'intent_file')) - s._startup(bus=self.emitter, skill_id="A") - - expected = [{'at_least_one': [], - 'name': 'A:a', - 'optional': [], - 'excludes': [], - 'requires': [('AKeyword', 'AKeyword')]}, - { - 'file_name': join(dirname(__file__), 'intent_file', - 'vocab', 'en-us', 'test.intent'), - 'lang': 'en-us', - 'samples': [], - 'name': str(s.skill_id) + ':test.intent'} - ] - - self.check_register_decorators(expected) - # Restore sys.path - sys.path = path_orig - - def test_failing_set_context(self): - s = SimpleSkill1(bus=self.emitter, skill_id="A") - with self.assertRaises(ValueError): - s.set_context(1) - with self.assertRaises(ValueError): - s.set_context(1, 1) - with self.assertRaises(ValueError): - s.set_context('Kowabunga', 1) - - def test_set_context(self): - def check_set_context(result_list): - for msg_type in self.emitter.get_types(): - self.assertEqual(msg_type, 'add_context') - self.assertEqual(sorted(self.emitter.get_results()), - sorted(result_list)) - self.emitter.reset() - - s = SimpleSkill1(bus=self.emitter, skill_id="A") - self.emitter.reset() - # No context content - s.set_context('TurtlePower') - expected = [{'context': 'ATurtlePower', 'origin': '', 'word': ''}] - check_set_context(expected) - - # context with content - s.set_context('Technodrome', 'Shredder') - expected = [{'context': 'ATechnodrome', 'origin': '', - 'word': 'Shredder'}] - check_set_context(expected) - - # UTF-8 context - s.set_context('Smörgåsbord€15') - expected = [{'context': 'ASmörgåsbord€15', 'origin': '', 'word': ''}] - check_set_context(expected) - - self.emitter.reset() - - def test_failing_remove_context(self): - s = SimpleSkill1(bus=self.emitter, skill_id="A") - with self.assertRaises(ValueError): - s.remove_context(1) - - def test_remove_context(self): - def check_remove_context(result_list): - for type in self.emitter.get_types(): - self.assertEqual(type, 'remove_context') - self.assertEqual(sorted(self.emitter.get_results()), - sorted(result_list)) - self.emitter.reset() - - s = SimpleSkill1(bus=self.emitter, skill_id="A") - self.emitter.reset() - s.remove_context('Donatello') - expected = [{'context': 'ADonatello'}] - check_remove_context(expected) - - @patch.dict(Configuration._Configuration__patch, BASE_CONF) - def test_skill_location(self): - s = SimpleSkill1() - self.assertEqual(s.location, BASE_CONF.get('location')) - self.assertEqual(s.location_pretty, - BASE_CONF['location']['city']['name']) - self.assertEqual(s.location_timezone, - BASE_CONF['location']['timezone']['code']) - - @patch.dict(Configuration._Configuration__patch, BASE_CONF) - def test_add_event(self): - emitter = MagicMock() - s = SimpleSkill1(bus=emitter, skill_id="A") - s.add_event('handler1', s.handler) - # Check that the handler was registered with the emitter - self.assertEqual(emitter.on.call_args[0][0], 'handler1') - # Check that the handler was stored in the skill - self.assertTrue('handler1' in [e[0] for e in s.events]) - - @patch.dict(Configuration._Configuration__patch, BASE_CONF) - def test_remove_event(self): - emitter = MagicMock() - s = SimpleSkill1(bus=emitter, skill_id="A") - s.add_event('handler1', s.handler) - self.assertTrue('handler1' in [e[0] for e in s.events]) - # Remove event handler - s.remove_event('handler1') - # make sure it's not in the event list anymore - self.assertTrue('handler1' not in [e[0] for e in s.events]) - # Check that the handler was registered with the emitter - self.assertEqual(emitter.remove_all_listeners.call_args[0][0], - 'handler1') - - @patch.dict(Configuration._Configuration__patch, BASE_CONF) - def test_add_scheduled_event(self): - emitter = MagicMock() - s = SimpleSkill1(bus=emitter, skill_id="A") - - s.schedule_event(s.handler, datetime.now(), name='datetime_handler') - # Check that the handler was registered with the emitter - self.assertEqual(emitter.once.call_args[0][0], 'A:datetime_handler') - sched_events = [e[0] for e in s.event_scheduler.events] - self.assertTrue('A:datetime_handler' in sched_events) - - s.schedule_event(s.handler, 1, name='int_handler') - # Check that the handler was registered with the emitter - self.assertEqual(emitter.once.call_args[0][0], 'A:int_handler') - sched_events = [e[0] for e in s.event_scheduler.events] - self.assertTrue('A:int_handler' in sched_events) - - s.schedule_event(s.handler, .5, name='float_handler') - # Check that the handler was registered with the emitter - self.assertEqual(emitter.once.call_args[0][0], 'A:float_handler') - sched_events = [e[0] for e in s.event_scheduler.events] - self.assertTrue('A:float_handler' in sched_events) - - @patch.dict(Configuration._Configuration__patch, BASE_CONF) - def test_remove_scheduled_event(self): - emitter = MagicMock() - s = SimpleSkill1(bus=emitter, skill_id="A") - s.schedule_event(s.handler, datetime.now(), name='sched_handler1') - # Check that the handler was registered with the emitter - events = [e[0] for e in s.event_scheduler.events] - print(events) - self.assertTrue('A:sched_handler1' in events) - s.cancel_scheduled_event('sched_handler1') - # Check that the handler was removed - self.assertEqual(emitter.remove_all_listeners.call_args[0][0], - 'A:sched_handler1') - events = [e[0] for e in s.event_scheduler.events] - self.assertTrue('A:sched_handler1' not in events) - - @patch.dict(Configuration._Configuration__patch, BASE_CONF) - def test_run_scheduled_event(self): - emitter = MagicMock() - s = SimpleSkill1(bus=emitter, skill_id="A") - with patch.object(s, '_settings', - create=True, value=MagicMock()): - s.schedule_event(s.handler, datetime.now(), name='sched_handler1') - # Check that the handler was registered with the emitter - emitter.once.call_args[0][1](Message('message')) - # Check that the handler was run - self.assertTrue(s.handler_run) - # Check that the handler was removed from the list of registred - # handler - self.assertTrue('A:sched_handler1' not in [e[0] for e in s.events]) - - def test_voc_match(self): - s = SimpleSkill1() - s.root_dir = abspath(dirname(__file__)) - - self.assertTrue(s.voc_match("turn off the lights", "turn_off_test")) - self.assertTrue(s.voc_match("would you please turn off the lights", - "turn_off_test")) - self.assertFalse(s.voc_match("return office", "turn_off_test")) - self.assertTrue(s.voc_match("switch off the lights", "turn_off_test")) - self.assertFalse(s.voc_match("", "turn_off_test")) - self.assertFalse(s.voc_match("switch", "turn_off_test")) - self.assertFalse(s.voc_match("My hovercraft is full of eels", - "turn_off_test")) - - self.assertTrue(s.voc_match("turn off the lights", "turn_off2_test")) - self.assertFalse(s.voc_match("return office", "turn_off2_test")) - self.assertTrue(s.voc_match("switch off the lights", "turn_off2_test")) - self.assertFalse(s.voc_match("", "turn_off_test")) - self.assertFalse(s.voc_match("switch", "turn_off_test")) - self.assertFalse(s.voc_match("My hovercraft is full of eels", - "turn_off_test")) - - def test_voc_match_exact(self): - s = SimpleSkill1() - s.root_dir = abspath(dirname(__file__)) - - self.assertTrue(s.voc_match("yes", "yes", exact=True)) - self.assertFalse(s.voc_match("yes please", "yes", exact=True)) - self.assertTrue(s.voc_match("switch off", "turn_off_test", - exact=True)) - self.assertFalse(s.voc_match("would you please turn off the lights", - "turn_off_test", exact=True)) - - def test_translate_locations(self): - """Assert that the a translatable list can be loaded from dialog and - locale. - """ - # Check that translatables can be loaded from the dialog directory - s = SimpleSkill1() - s.res_dir = abspath(join(dirname(__file__), - 'translate', 'in-dialog/')) - self.assertEqual(s.lang, "en-us") - lst = s.translate_list('good_things') - self.assertIsInstance(lst, list) - vals = s.translate_namedvalues('named_things') - self.assertIsInstance(vals, dict) - template = s.translate_template('test', - data={'thing': 'test framework'}) - self.assertEqual(template, - ['Oh look it\'s my favourite test framework']) - # Check that translatables can be loaded from locale folder - s = SimpleSkill1() - s.res_dir = abspath(join(dirname(__file__), - 'translate', 'in-locale')) - lst = s.translate_list('good_things') - self.assertIsInstance(lst, list) - vals = s.translate_namedvalues('named_things') - self.assertIsInstance(vals, dict) - template = s.translate_template('test', - data={'thing': 'test framework'}) - self.assertEqual(template, - ['Oh look it\'s my favourite test framework']) - - # Check loading in a non-en-us language - s = SimpleSkill1() - s.config_core['lang'] = 'de-de' - s.res_dir = abspath(join(dirname(__file__), - 'translate', 'in-locale')) - lst = s.translate_list('good_things') - self.assertEqual(lst, ['sonne', 'mycroft', 'zahne']) - vals = s.translate_namedvalues('named_things') - self.assertEqual(vals['blau'], '2') - template = s.translate_template('test', - data={'thing': 'test framework'}) - self.assertEqual(template, - ['Aber setzen sie sich herr test framework']) - - # Restore lang to en-us - s.config_core['lang'] = 'en-us' - - def test_native_langs(self): - s = _TestSkill() - lang = s.config_core['lang'] - secondary = s.config_core['secondary_langs'] - - s.config_core['lang'] = 'en-US' - s.config_core['secondary_langs'] = ['en', 'en-us', 'en-AU', - 'es', 'pt-PT'] - self.assertEqual(s.lang, 'en-us') - self.assertEqual(s.secondary_langs, ['en', 'en-au', 'es', - 'pt-pt']) - self.assertEqual(len(s.native_langs), len(set(s.native_langs))) - self.assertEqual(set(s.native_langs), {'en-us', 'en-au', 'pt-pt'}) - s.config_core['lang'] = lang - s.config_core['secondary_langs'] = secondary - - -class TestIntentCollisions(unittest.TestCase): - def test_two_intents_with_same_name(self): - emitter = MockEmitter() - skill = SameIntentNameSkill() - skill.bind(emitter) - with self.assertRaises(ValueError): - skill.initialize() - - def test_two_anonymous_intent_decorators(self): - """Two anonymous intent handlers should be ok.""" - emitter = MockEmitter() - skill = SameAnonymousIntentDecoratorsSkill() - skill.bind(emitter) - skill._register_decorated() - self.assertEqual(len(skill.intent_service.registered_intents), 2) - - -class _TestSkill(OVOSSkill): - pass - - -class SimpleSkill1(MycroftSkill): - """ Test skill for normal intent builder syntax """ - - def initialize(self): - self.handler_run = False - i = IntentBuilder('a').require('Keyword').build() - self.register_intent(i, self.handler) - - def handler(self, message): - self.handler_run = True - - def stop(self): - pass - - -class SimpleSkill2(_TestSkill): - """ Test skill for intent builder without .build() """ - - def initialize(self): - i = IntentBuilder('a').require('Keyword') - self.register_intent(i, self.handler) - - def handler(self, message): - pass - - def stop(self): - pass - - -class SimpleSkill3(_TestSkill): - """ Test skill for invalid Intent for register_intent """ - - def initialize(self): - self.register_intent('string', self.handler) - - def handler(self, message): - pass - - def stop(self): - pass - - -class SimpleSkill4(_TestSkill): - """ Test skill for padatious intent """ - - def initialize(self): - self.register_intent_file('test.intent', self.handler) - self.register_entity_file('test_ent.entity') - - def handler(self, message): - pass - - def stop(self): - pass - - -class SimpleSkill6(_TestSkill): - """ Test skill for padatious intent """ - - def initialize(self): - self.register_intent('test.intent', self.handler) - self.register_entity_file('test_ent.entity') - - def handler(self, message): - pass - - -class SameIntentNameSkill(_TestSkill): - """Test skill for duplicate intent namesr.""" - - def initialize(self): - intent = IntentBuilder('TheName').require('Keyword') - intent2 = IntentBuilder('TheName').require('Keyword') - self.register_intent(intent, self.handler) - self.register_intent(intent2, self.handler) - - def handler(self, message): - pass - - -class SameAnonymousIntentDecoratorsSkill(_TestSkill): - """Test skill for duplicate anonymous intent handlers.""" - - @intent_handler(IntentBuilder('').require('Keyword')) - @intent_handler(IntentBuilder('').require('OtherKeyword')) - def handler(self, message): - pass diff --git a/test/integrationtests/test_skill_api.py b/test/integrationtests/test_skill_api.py deleted file mode 100644 index fba8878ae11..00000000000 --- a/test/integrationtests/test_skill_api.py +++ /dev/null @@ -1,153 +0,0 @@ -from unittest import TestCase, mock - -# TODO - move test to ovos-workshop -from ovos_workshop.skills.ovos import OVOSSkill -from ovos_bus_client.message import Message -from ovos_workshop.decorators import skill_api_method -from ovos_workshop.skills.api import SkillApi - - -class Skill(OVOSSkill): - """Test skill with registered API methods.""" - def __init__(self, *args, **kwargs): - self.registered_methods = {} - super().__init__(*args, **kwargs) - - def add_event(self, event_type, func, **kwargs): - """Mock handler of add_event, simply storing type and method. - - Used in testing to verify the wrapped methods - """ - self.registered_methods[event_type] = func - - @skill_api_method - def test_method(self): - """Documentation.""" - return True - - @skill_api_method - def test_method2(self, arg): - """Documentation.""" - return 'TestResult' - - -def load_test_skill(): - """Helper for setting up the test skill. - - Returns: - (MycroftSkill): created test skill - """ - bus = mock.Mock() - test_skill = Skill(skill_id = 'test_skill', bus=bus) - return test_skill - - -def create_skill_api_from_skill(skill): - """Helper creating an api from a skill. - - Args: - skill (MycroftSkill): Skill to create api from. - - Returns: - (SkillApi): API for the skill. - """ - SkillApi.connect_bus(skill.bus) - return SkillApi(skill.public_api) - - -class testSkillMethod(TestCase): - """Tests for the MycroftSkill class API setup.""" - def test_public_api_event(self): - """Test that public api event handler is created.""" - test_skill = load_test_skill() - self.assertTrue( - 'test_skill.public_api' in test_skill.registered_methods - ) - - def test_public_api(self): - """Test that the public_api structure matches the decorators.""" - test_skill = load_test_skill() - # Check that methods has been added - self.assertTrue('test_method' in test_skill.public_api) - self.assertTrue('test_method2' in test_skill.public_api) - # Test docstring - self.assertEqual(test_skill.public_api['test_method']['help'], - 'Documentation.') - # Test type - self.assertEqual(test_skill.public_api['test_method']['type'], - '{}.{}'.format(test_skill.skill_id, 'test_method')) - - def test_public_api_method(self): - """Verify message from wrapped api method.""" - test_skill = load_test_skill() - api_method = test_skill.registered_methods['test_skill.test_method'] - - # Call method - call_msg = Message('test_skill.test_method', - data={'args': [], 'kwargs': {}}) - api_method(call_msg) - # Check response sent on the bus is the same as the method's return - # value - response = test_skill.bus.emit.call_args[0][0] - self.assertEqual(response.data['result'], test_skill.test_method()) - - def test_public_api_request(self): - """Test public api request handling. - - Ensures that a request for the skill's available public api returns - expected content. - """ - test_skill = load_test_skill() - sent_message = None - - def capture_sent_message(message): - """Capture sent message.""" - nonlocal sent_message - sent_message = message - - test_skill.bus.emit.side_effect = capture_sent_message - get_api_method = test_skill.registered_methods['test_skill.public_api'] - request_api_msg = Message('test_skill.public_api') - - # Ensure that the sent public api contains the correct items - get_api_method(request_api_msg) - public_api = sent_message.data - self.assertTrue('test_method' in public_api) - self.assertTrue('test_method2' in public_api) - self.assertEqual(len(public_api), 2) - - -class TestApiObject(TestCase): - """Tests for the generated SkillApi objects.""" - def test_create_api_object(self): - """Check that expected methods are available.""" - test_skill = load_test_skill() - test_api = create_skill_api_from_skill(test_skill) - - hasattr(test_api, 'test_method') - hasattr(test_api, 'test_method2') - - def test_call_api_method(self): - """Ensure that calling the methods works as expected.""" - test_skill = load_test_skill() - test_api = create_skill_api_from_skill(test_skill) - - expected_response = 'all is good' - sent_message = None - - def capture_sent_message(message, timeout=3): - """Capture sent message and return expected response message.""" - nonlocal sent_message - sent_message = message - return Message('', data={'result': expected_response}) - - test_api.bus.wait_for_response.side_effect = capture_sent_message - - response = test_api.test_method('hello', person='you') - - # Verify response - self.assertEqual(response, expected_response) - # Verify sent message - self.assertEqual(sent_message.msg_type, 'test_skill.test_method') - self.assertEqual(sent_message.data['args'], ('hello',)) - self.assertEqual(sent_message.data['kwargs'], {'person': 'you'}) diff --git a/test/integrationtests/test_skill_loader.py b/test/integrationtests/test_skill_loader.py deleted file mode 100644 index 7f81420f268..00000000000 --- a/test/integrationtests/test_skill_loader.py +++ /dev/null @@ -1,192 +0,0 @@ -# Copyright 2019 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. -# -"""Unit tests for the SkillLoader class.""" -import json -import unittest -from pathlib import Path -from unittest.mock import Mock - -from ovos_utils import classproperty -from ovos_utils.messagebus import FakeBus -from ovos_utils.process_utils import RuntimeRequirements -from ovos_workshop.skill_launcher import SkillLoader -# TODO - move test to ovos-workshop -from ovos_workshop.skills.mycroft_skill import MycroftSkill - -ONE_MINUTE = 60 - - -class OfflineSkill(MycroftSkill): - @classproperty - def runtime_requirements(self): - return RuntimeRequirements(internet_before_load=False, - network_before_load=False, - requires_internet=False, - requires_network=False, - no_internet_fallback=True, - no_network_fallback=True) - - -class LANSkill(MycroftSkill): - @classproperty - def runtime_requirements(self): - scans_on_init = True - return RuntimeRequirements(internet_before_load=False, - network_before_load=scans_on_init, - requires_internet=False, - requires_network=True, - no_internet_fallback=True, - no_network_fallback=False) - - -class TestSkillNetwork(unittest.TestCase): - - def test_class_property(self): - self.assertEqual(OfflineSkill.runtime_requirements, - RuntimeRequirements(internet_before_load=False, - network_before_load=False, - requires_internet=False, - requires_network=False, - no_internet_fallback=True, - no_network_fallback=True) - ) - self.assertEqual(LANSkill.runtime_requirements, - RuntimeRequirements(internet_before_load=False, - network_before_load=True, - requires_internet=False, - requires_network=True, - no_internet_fallback=True, - no_network_fallback=False) - ) - self.assertEqual(MycroftSkill.runtime_requirements, - RuntimeRequirements() - ) - - -msgs = [] -bus = FakeBus() -bus.msgs = [] - - -def _handle(msg): - global bus - bus.msgs.append(json.loads(msg)) - - -bus.on("message", _handle) - - -class TestSkillLoader(unittest.TestCase): - skill_directory = Path('/tmp/test_skill') - skill_directory.mkdir(exist_ok=True) - for file_name in ('__init__.py', 'bar.py', '.foobar', 'bar.pyc'): - skill_directory.joinpath(file_name).touch() - - def test_skill_reload(self): - """Test reloading a skill that was modified.""" - bus.msgs = [] - loader = SkillLoader(bus, str(self.skill_directory)) - loader.instance = Mock() - loader.loaded = True - loader.load_attempted = False - loader.last_loaded = 10 - loader.instance.reload_skill = True - loader.instance.name = "MySkill" - loader.skill_id = 'test_skill' - - # Mock to return a known (Mock) skill instance - real_create_skill_instance = loader._create_skill_instance - - def _update_skill_instance(*args, **kwargs): - loader.instance = Mock() - loader.loaded = True - loader.load_attempted = True - loader.last_loaded = 100 - loader.skill_id = 'test_skill' - loader.instance.name = "MySkill" - return True - - loader._create_skill_instance = _update_skill_instance - - loader.reload() - - self.assertTrue(loader.load_attempted) - self.assertTrue(loader.loaded) - - self.assertListEqual( - ['mycroft.skills.shutdown', 'mycroft.skills.loaded'], - [m["type"] for m in bus.msgs] - ) - loader._create_skill_instance = real_create_skill_instance - - def test_skill_load(self): - loader = SkillLoader(bus, str(self.skill_directory)) - bus.msgs = [] - loader.instance = None - loader.loaded = False - loader.last_loaded = 0 - - # Mock to return a known (Mock) skill instance - real_create_skill_instance = loader._create_skill_instance - - def _update_skill_instance(*args, **kwargs): - loader.instance = Mock() - loader.loaded = True - loader.last_loaded = 100 - loader.skill_id = 'test_skill' - loader.instance.name = "MySkill" - return True - - loader._create_skill_instance = _update_skill_instance - - loader.load() - - self.assertTrue(loader.load_attempted) - self.assertTrue(loader.loaded) - - self.assertListEqual( - ['mycroft.skills.loaded'], - [m["type"] for m in bus.msgs] - ) - loader._create_skill_instance = real_create_skill_instance - - def test_skill_load_blacklisted(self): - """Skill should not be loaded if it is blacklisted""" - loader = SkillLoader(bus, str(self.skill_directory)) - loader.instance = Mock() - loader.loaded = False - loader.last_loaded = 0 - loader.skill_id = 'test_skill' - loader.name = "MySkill" - bus.msgs = [] - - config = dict(loader.config) - config['skills']['blacklisted_skills'] = ['test_skill'] - loader.config = config - self.assertEqual(loader.config['skills']['blacklisted_skills'], - ['test_skill']) - loader.skill_id = 'test_skill' - - loader.load() - - self.assertTrue(loader.load_attempted) - self.assertFalse(loader.loaded) - - self.assertListEqual( - ['mycroft.skills.loading_failure'], - [m["type"] for m in bus.msgs] - ) - - loader.config['skills']['blacklisted_skills'].remove('test_skill')