Coverage for test/unittests/test_skill_manager.py: 98%

1# Copyright 2019 Mycroft AI Inc. 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# 

15import tempfile 

16from copy import deepcopy 

17from pathlib import Path 

18from shutil import rmtree 

19from unittest import TestCase 

20from unittest.mock import Mock, patch 

21 

22from ovos_bus_client.message import Message 

23from ovos_config import Configuration 

24from ovos_config import LocalConf, DEFAULT_CONFIG 

25from ovos_core.skill_manager import SkillManager 

26from ovos_workshop.skill_launcher import SkillLoader 

27 

28 

29class MessageBusMock: 

30 """Replaces actual message bus calls in unit tests. 

31 

32 The message bus should not be running during unit tests so mock it 

33 out in a way that makes it easy to test code that calls it. 

34 """ 

35 

36 def __init__(self): 

37 self.message_types = [] 

38 self.message_data = [] 

39 self.event_handlers = [] 

40 

41 def emit(self, message): 

42 self.message_types.append(message.msg_type) 

43 self.message_data.append(message.data) 

44 

45 def on(self, event, _): 

46 self.event_handlers.append(event) 

47 

48 def once(self, event, _): 

49 self.event_handlers.append(event) 

50 

51 def wait_for_response(self, message): 

52 self.emit(message) 

53 

54 

55def mock_config(): 

56 """Supply a reliable return value for the Configuration.get() method.""" 

57 config = deepcopy(LocalConf(DEFAULT_CONFIG)) 

58 config['skills']['priority_skills'] = ['foobar'] 

59 config['data_dir'] = str(tempfile.mkdtemp()) 

60 config['enclosure'] = {} 

61 return config 

62 

63 

64@patch.dict(Configuration._Configuration__patch, mock_config()) 

65class TestSkillManager(TestCase): 

66 mock_package = 'ovos_core.skill_manager.' 

67 

68 def setUp(self): 

69 temp_dir = tempfile.mkdtemp() 

70 self.temp_dir = Path(temp_dir) 

71 self.message_bus_mock = MessageBusMock() 

72 self._mock_log() 

73 self.skill_manager = SkillManager(self.message_bus_mock) 

74 self._mock_skill_loader_instance() 

75 

76 def _mock_log(self): 

77 log_patch = patch(self.mock_package + 'LOG') 

78 self.addCleanup(log_patch.stop) 

79 self.log_mock = log_patch.start() 

80 

81 def tearDown(self): 

82 rmtree(str(self.temp_dir)) 

83 

84 def _mock_skill_loader_instance(self): 

85 self.skill_dir = self.temp_dir.joinpath('test_skill') 

86 self.skill_loader_mock = Mock(spec=SkillLoader) 

87 self.skill_loader_mock.instance = Mock() 

88 self.skill_loader_mock.instance.default_shutdown = Mock() 

89 self.skill_loader_mock.instance.converse = Mock() 

90 self.skill_loader_mock.instance.converse.return_value = True 

91 self.skill_loader_mock.skill_id = 'test_skill' 

92 self.skill_manager.plugin_skills = { 

93 str(self.skill_dir): self.skill_loader_mock 

94 } 

95 

96 def test_instantiate(self): 

97 expected_result = [ 

98 'skillmanager.list', 

99 'skillmanager.deactivate', 

100 'skillmanager.keep', 

101 'skillmanager.activate', 

102 #'mycroft.skills.initialized', 

103 'mycroft.network.connected', 

104 'mycroft.internet.connected', 

105 'mycroft.gui.available', 

106 'mycroft.network.disconnected', 

107 'mycroft.internet.disconnected', 

108 'mycroft.gui.unavailable', 

109 'mycroft.skills.is_alive', 

110 'mycroft.skills.is_ready', 

111 'mycroft.skills.all_loaded' 

112 ] 

113 

114 self.assertListEqual(expected_result, 

115 self.message_bus_mock.event_handlers) 

116 

117 

118 def test_send_skill_list(self): 

119 self.skill_loader_mock.active = True 

120 self.skill_loader_mock.loaded = True 

121 self.skill_manager.send_skill_list(None) 

122 

123 self.assertListEqual( 

124 ['mycroft.skills.list'], 

125 self.message_bus_mock.message_types 

126 ) 

127 message_data = self.message_bus_mock.message_data[-1] 

128 self.assertIn('test_skill', message_data.keys()) 

129 skill_data = message_data['test_skill'] 

130 self.assertDictEqual(dict(active=True, id='test_skill'), skill_data) 

131 

132 def test_stop(self): 

133 self.skill_manager.stop() 

134 

135 self.assertTrue(self.skill_manager._stop_event.is_set()) 

136 instance = self.skill_loader_mock.instance 

137 instance.default_shutdown.assert_called_once_with() 

138 

139 def test_deactivate_skill(self): 

140 message = Message("test.message", {'skill': 'test_skill'}) 

141 message.response = Mock() 

142 self.skill_manager.deactivate_skill(message) 

143 self.skill_loader_mock.deactivate.assert_called_once() 

144 message.response.assert_called_once() 

145 

146 def test_deactivate_except(self): 

147 message = Message("test.message", {'skill': 'test_skill'}) 

148 message.response = Mock() 

149 self.skill_loader_mock.active = True 

150 foo_skill_loader = Mock(spec=SkillLoader) 

151 foo_skill_loader.skill_id = 'foo' 

152 foo2_skill_loader = Mock(spec=SkillLoader) 

153 foo2_skill_loader.skill_id = 'foo2' 

154 test_skill_loader = Mock(spec=SkillLoader) 

155 test_skill_loader.skill_id = 'test_skill' 

156 self.skill_manager.plugin_skills['foo'] = foo_skill_loader 

157 self.skill_manager.plugin_skills['foo2'] = foo2_skill_loader 

158 self.skill_manager.plugin_skills['test_skill'] = test_skill_loader 

159 

160 self.skill_manager.deactivate_except(message) 

161 foo_skill_loader.deactivate.assert_called_once() 

162 foo2_skill_loader.deactivate.assert_called_once() 

163 self.assertFalse(test_skill_loader.deactivate.called) 

164 

165 def test_activate_skill(self): 

166 message = Message("test.message", {'skill': 'test_skill'}) 

167 message.response = Mock() 

168 test_skill_loader = Mock(spec=SkillLoader) 

169 test_skill_loader.skill_id = 'test_skill' 

170 test_skill_loader.active = False 

171 

172 self.skill_manager.plugin_skills = {} 

173 self.skill_manager.plugin_skills['test_skill'] = test_skill_loader 

174 

175 self.skill_manager.activate_skill(message) 

176 test_skill_loader.activate.assert_called_once() 

177 message.response.assert_called_once()