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

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() 

178 

179 def test_load_plugin_skill_success(self): 

180 """Test successful plugin skill loading emits the correct message.""" 

181 skill_id = 'test.plugin.skill' 

182 mock_plugin = Mock() 

183 

184 # Setup mock loader following existing patterns 

185 mock_loader = Mock(spec=SkillLoader) 

186 mock_loader.skill_id = skill_id 

187 mock_loader.load.return_value = True 

188 

189 # Mock _get_plugin_skill_loader to return our mock 

190 self.skill_manager._get_plugin_skill_loader = Mock(return_value=mock_loader) 

191 

192 # Reset message tracking 

193 self.message_bus_mock.message_types = [] 

194 self.message_bus_mock.message_data = [] 

195 self.skill_manager.plugin_skills = {} 

196 

197 # Call the method 

198 result = self.skill_manager._load_plugin_skill(skill_id, mock_plugin) 

199 

200 # Verify message was emitted 

201 self.assertIn('mycroft.skill.loaded', self.message_bus_mock.message_types) 

202 loaded_msg_idx = self.message_bus_mock.message_types.index('mycroft.skill.loaded') 

203 self.assertEqual( 

204 {'skill_id': skill_id}, 

205 self.message_bus_mock.message_data[loaded_msg_idx] 

206 ) 

207 

208 # Verify loader was called 

209 mock_loader.load.assert_called_once_with(mock_plugin) 

210 

211 # Verify skill was added to plugin_skills 

212 self.assertIn(skill_id, self.skill_manager.plugin_skills) 

213 self.assertEqual(mock_loader, self.skill_manager.plugin_skills[skill_id]) 

214 

215 # Verify return value 

216 self.assertEqual(result, mock_loader) 

217 

218 def test_load_plugin_skill_failure(self): 

219 """Test failed plugin skill loading is handled gracefully.""" 

220 skill_id = 'test.failing.skill' 

221 mock_plugin = Mock() 

222 

223 # Setup mock loader to raise exception 

224 mock_loader = Mock(spec=SkillLoader) 

225 mock_loader.skill_id = skill_id 

226 mock_loader.load.side_effect = Exception("Skill load failed!") 

227 

228 # Mock _get_plugin_skill_loader to return our mock 

229 self.skill_manager._get_plugin_skill_loader = Mock(return_value=mock_loader) 

230 

231 # Reset message tracking 

232 self.message_bus_mock.message_types = [] 

233 self.message_bus_mock.message_data = [] 

234 self.skill_manager.plugin_skills = {} 

235 

236 # Call the method 

237 result = self.skill_manager._load_plugin_skill(skill_id, mock_plugin) 

238 

239 # Verify NO success message was emitted 

240 self.assertNotIn('mycroft.skill.loaded', self.message_bus_mock.message_types) 

241 

242 # Verify exception was logged 

243 self.log_mock.exception.assert_called_once() 

244 

245 # Verify skill was still added to plugin_skills (even on failure) 

246 self.assertIn(skill_id, self.skill_manager.plugin_skills) 

247 self.assertEqual(mock_loader, self.skill_manager.plugin_skills[skill_id]) 

248 

249 # Verify return value is None on failure 

250 self.assertIsNone(result) 

251 

252 def test_load_plugin_skill_returns_false(self): 

253 """Test plugin skill loading that returns False (load failed gracefully).""" 

254 skill_id = 'test.false.skill' 

255 mock_plugin = Mock() 

256 

257 # Setup mock loader to return False (failed but no exception) 

258 mock_loader = Mock(spec=SkillLoader) 

259 mock_loader.skill_id = skill_id 

260 mock_loader.load.return_value = False 

261 

262 # Mock _get_plugin_skill_loader to return our mock 

263 self.skill_manager._get_plugin_skill_loader = Mock(return_value=mock_loader) 

264 

265 # Reset message tracking 

266 self.message_bus_mock.message_types = [] 

267 self.skill_manager.plugin_skills = {} 

268 

269 # Call the method 

270 result = self.skill_manager._load_plugin_skill(skill_id, mock_plugin) 

271 

272 # Verify NO success message was emitted (load returned False) 

273 self.assertNotIn('mycroft.skill.loaded', self.message_bus_mock.message_types) 

274 

275 # Verify skill was added to plugin_skills 

276 self.assertIn(skill_id, self.skill_manager.plugin_skills) 

277 

278 # Verify return value is None when load returns False 

279 self.assertIsNone(result)