Coverage for test/end2end/test_stop.py: 100%

1import time 

2from unittest import TestCase 

3 

4from ovos_bus_client.message import Message 

5from ovos_bus_client.session import Session 

6from ovos_utils import create_daemon 

7from ovos_utils.log import LOG 

8 

9from ovoscope import End2EndTest, get_minicroft 

10 

11 

12class TestStopNoSkills(TestCase): 

13 

14 def setUp(self): 

15 LOG.set_level("DEBUG") 

16 self.minicroft = get_minicroft([]) # reuse for speed, but beware if skills keeping internal state # to make tests easier to grok 

17 self.ignore_messages = ["speak", 

18 "ovos.common_play.stop.response", 

19 "common_query.openvoiceos.stop.response", 

20 "persona.openvoiceos.stop.response" 

21 ] 

22 

23 def tearDown(self): 

24 if self.minicroft: 

25 self.minicroft.stop() 

26 LOG.set_level("CRITICAL") 

27 

28 def test_exact(self): 

29 session = Session("123") 

30 session.lang = "en-US" 

31 session.pipeline = ['ovos-stop-pipeline-plugin-high'] 

32 message = Message("recognizer_loop:utterance", 

33 {"utterances": ["stop"], "lang": session.lang}, 

34 {"session": session.serialize()}) 

35 

36 test = End2EndTest( 

37 minicroft=self.minicroft, 

38 skill_ids=[], 

39 eof_msgs=["ovos.utterance.handled"], 

40 flip_points=["recognizer_loop:utterance"], 

41 ignore_messages=self.ignore_messages, 

42 source_message=message, 

43 # keep_original_src=["stop.openvoiceos.activate"], # TODO 

44 expected_messages=[ 

45 message, 

46 Message("stop.openvoiceos.activate", {}), # stop pipeline counts as active_skill 

47 

48 Message("stop:global", {}), # global stop, no active skill 

49 Message("mycroft.stop", {}), 

50 

51 Message("ovos.utterance.handled", {}) 

52 ] 

53 ) 

54 

55 test.execute() 

56 

57 def test_not_exact_high(self): 

58 session = Session("123") 

59 session.lang = "en-US" 

60 session.pipeline = ['ovos-stop-pipeline-plugin-high'] 

61 message = Message("recognizer_loop:utterance", 

62 {"utterances": ["could you stop that"], "lang": session.lang}, 

63 {"session": session.serialize()}) 

64 

65 test = End2EndTest( 

66 minicroft=self.minicroft, 

67 skill_ids=[], 

68 eof_msgs=["ovos.utterance.handled"], 

69 flip_points=["recognizer_loop:utterance"], 

70 ignore_messages=self.ignore_messages, 

71 source_message=message, 

72 expected_messages=[ 

73 message, 

74 Message("mycroft.audio.play_sound", {"uri": "snd/error.mp3"}), 

75 Message("complete_intent_failure", {}), 

76 Message("ovos.utterance.handled", {}), 

77 ] 

78 ) 

79 

80 test.execute() 

81 

82 def test_not_exact_med(self): 

83 session = Session("123") 

84 session.lang = "en-US" 

85 session.pipeline = ['ovos-stop-pipeline-plugin-medium'] 

86 message = Message("recognizer_loop:utterance", 

87 {"utterances": ["could you stop that"], "lang": session.lang}, 

88 {"session": session.serialize()}) 

89 

90 test = End2EndTest( 

91 minicroft=self.minicroft, 

92 skill_ids=[], 

93 eof_msgs=["ovos.utterance.handled"], 

94 flip_points=["recognizer_loop:utterance"], 

95 source_message=message, 

96 ignore_messages=self.ignore_messages, 

97 # keep_original_src=["stop.openvoiceos.activate"], # TODO 

98 expected_messages=[ 

99 message, 

100 Message("stop.openvoiceos.activate", {}), # stop pipeline counts as active_skill 

101 

102 Message("stop:global", {}), # global stop, no active skill 

103 Message("mycroft.stop", {}), 

104 

105 Message("ovos.utterance.handled", {}) 

106 ] 

107 ) 

108 

109 test.execute() 

110 

111 

112class TestCountSkills(TestCase): 

113 

114 def setUp(self): 

115 LOG.set_level("DEBUG") 

116 self.skill_id = "ovos-skill-count.openvoiceos" 

117 self.minicroft = get_minicroft([self.skill_id]) # reuse for speed, but beware if skills keeping internal state 

118 # to make tests easier to grok 

119 self.ignore_messages = ["speak", 

120 "ovos.common_play.stop.response", 

121 "common_query.openvoiceos.stop.response", 

122 "persona.openvoiceos.stop.response" 

123 ] 

124 

125 def tearDown(self): 

126 if self.minicroft: 

127 self.minicroft.stop() 

128 LOG.set_level("CRITICAL") 

129 

130 def test_count(self): 

131 session = Session("123") 

132 session.lang = "en-US" 

133 session.pipeline = ['ovos-stop-pipeline-plugin-high', "ovos-padatious-pipeline-plugin-high"] 

134 

135 message = Message("recognizer_loop:utterance", 

136 {"utterances": ["count to 3"], "lang": session.lang}, 

137 {"session": session.serialize()}) 

138 

139 # first count to 10 to validate skill is working 

140 activate_skill = [ 

141 message, 

142 Message(f"{self.skill_id}.activate", {}), # skill is activated 

143 Message(f"{self.skill_id}:count_to_N.intent", {}), # intent triggers 

144 

145 Message("mycroft.skill.handler.start", { 

146 "name": "CountSkill.handle_how_are_you_intent" 

147 }), 

148 # here would be N speak messages, but we ignore them in this test 

149 Message("mycroft.skill.handler.complete", { 

150 "name": "CountSkill.handle_how_are_you_intent" 

151 }), 

152 

153 Message("ovos.utterance.handled", {}) 

154 ] 

155 test = End2EndTest( 

156 minicroft=self.minicroft, 

157 skill_ids=[], 

158 eof_msgs=["ovos.utterance.handled"], 

159 flip_points=["recognizer_loop:utterance"], 

160 ignore_messages=self.ignore_messages, 

161 source_message=message, 

162 # keep_original_src=[f"{self.skill_id}.activate"], # TODO 

163 expected_messages=activate_skill 

164 ) 

165 test.execute() 

166 

167 def test_count_infinity_active(self): 

168 session = Session("123") 

169 session.lang = "en-US" 

170 session.pipeline = ['ovos-stop-pipeline-plugin-high', 

171 "ovos-padatious-pipeline-plugin-high"] 

172 

173 def make_it_count(): 

174 nonlocal session 

175 message = Message("recognizer_loop:utterance", 

176 {"utterances": ["count to infinity"], "lang": session.lang}, 

177 {"session": session.serialize(), "source": "A", "destination": "B"}) 

178 session.activate_skill(self.skill_id) # ensure in active skill list 

179 self.minicroft.bus.emit(message) 

180 

181 # count to infinity, the skill will keep running in the background 

182 create_daemon(make_it_count) 

183 

184 time.sleep(2) 

185 

186 message = Message("recognizer_loop:utterance", 

187 {"utterances": ["stop"], "lang": session.lang}, 

188 {"session": session.serialize(), "source": "A", "destination": "B"}) 

189 

190 stop_skill_active = [ 

191 message, 

192 Message(f"{self.skill_id}.stop.ping", 

193 {"skill_id":self.skill_id}), 

194 Message("skill.stop.pong", 

195 {"skill_id": self.skill_id, "can_handle": True}, 

196 {"skill_id": self.skill_id}), 

197 

198 Message("stop.openvoiceos.activate", 

199 context={"skill_id": "stop.openvoiceos"}), 

200 Message("stop:skill", 

201 context={"skill_id": "stop.openvoiceos"}), 

202 Message(f"{self.skill_id}.stop", 

203 context={"skill_id": "stop.openvoiceos"}), 

204 Message(f"{self.skill_id}.stop.response", 

205 {"skill_id": self.skill_id, "result": True}, 

206 {"skill_id": self.skill_id}), 

207 

208 # async stop pipeline callback emits these messages 

209 # but we cant guarantee where in the test they will be emitted 

210 

211 # if skill is in middle of get_response 

212 #Message("mycroft.skills.abort_question", 

213 # {"skill_id": self.skill_id}, 

214 # {"skill_id": self.skill_id}), 

215 

216 # if skill is in active_list 

217 #Message("ovos.skills.converse.force_timeout", 

218 # {"skill_id": self.skill_id}, 

219 # {"skill_id": self.skill_id}), 

220 

221 # if skill is executing TTS 

222 #Message("mycroft.audio.speech.stop", 

223 # {"skill_id": self.skill_id}, 

224 # {"skill_id": self.skill_id}), 

225 

226 # the intent running in the daemon thread exits cleanly 

227 Message("mycroft.skill.handler.complete", 

228 {"name": "CountSkill.handle_how_are_you_intent"}, 

229 {"skill_id": self.skill_id}), 

230 Message("ovos.utterance.handled", 

231 {"name": "CountSkill.handle_how_are_you_intent"}, 

232 {"skill_id": self.skill_id}) 

233 ] 

234 test = End2EndTest( 

235 minicroft=self.minicroft, 

236 skill_ids=[], 

237 eof_msgs=[], 

238 flip_points=["recognizer_loop:utterance"], 

239 # messages in 'keep_original_src' would not be sent to hivemind clients 

240 # i.e. they are directed towards ovos-core 

241 keep_original_src=[f"{self.skill_id}.stop.ping", 

242 f"{self.skill_id}.stop", 

243 "mycroft.skills.abort_question", 

244 "ovos.skills.converse.force_timeout", 

245 # "stop.openvoiceos.activate" # TODO 

246 ], 

247 async_messages=[ 

248 "ovos.skills.converse.force_timeout" 

249 ], # order that it wil be received unknown 

250 ignore_messages=self.ignore_messages, 

251 source_message=message, 

252 expected_messages=stop_skill_active 

253 ) 

254 test.execute() 

255 

256 def test_count_infinity_global(self): 

257 session = Session("123") 

258 session.lang = "en-US" 

259 session.pipeline = ['ovos-stop-pipeline-plugin-high', 

260 "ovos-padatious-pipeline-plugin-high"] 

261 

262 def make_it_count(): 

263 message = Message("recognizer_loop:utterance", 

264 {"utterances": ["count to infinity"], "lang": session.lang}, 

265 {"session": session.serialize()}) 

266 self.minicroft.bus.emit(message) 

267 

268 # count to infinity, the skill will keep running in the background 

269 create_daemon(make_it_count) 

270 

271 time.sleep(3) 

272 

273 # NOTE: skill not in active skill list for this Session, global stop will match instead 

274 # this doesnt typically happen at runtime, but possible since clients send whatever Session they want 

275 message = Message("recognizer_loop:utterance", 

276 {"utterances": ["stop"], "lang": session.lang}, 

277 {"session": session.serialize()}) 

278 stop_skill_from_global = [ 

279 message, 

280 Message("stop.openvoiceos.activate", {}), # stop pipeline counts as active_skill 

281 

282 Message("stop:global", {}), # global stop, no active skill 

283 Message("mycroft.stop", {}), 

284 

285 Message(f"{self.skill_id}.stop.response", 

286 {"skill_id": self.skill_id, "result": True}), 

287 Message("ovos.utterance.handled", {}) 

288 ] 

289 test = End2EndTest( 

290 minicroft=self.minicroft, 

291 skill_ids=[], 

292 eof_msgs=["ovos.utterance.handled"], 

293 flip_points=["recognizer_loop:utterance"], 

294 ignore_messages=self.ignore_messages, 

295 source_message=message, 

296 expected_messages=stop_skill_from_global, 

297 #keep_original_src=["stop.openvoiceos.activate"], # TODO 

298 ) 

299 test.execute() 

300 

301 def test_count_infinity_stop_low(self): 

302 session = Session("123") 

303 session.lang = "en-US" 

304 session.pipeline = ["ovos-padatious-pipeline-plugin-high", 

305 'ovos-stop-pipeline-plugin-low'] 

306 

307 def make_it_count(): 

308 nonlocal session 

309 message = Message("recognizer_loop:utterance", 

310 {"utterances": ["count to infinity"], "lang": session.lang}, 

311 {"session": session.serialize(), "source": "A", "destination": "B"}) 

312 session.activate_skill(self.skill_id) # ensure in active skill list 

313 self.minicroft.bus.emit(message) 

314 

315 # count to infinity, the skill will keep running in the background 

316 create_daemon(make_it_count) 

317 

318 time.sleep(2) 

319 

320 message = Message("recognizer_loop:utterance", 

321 {"utterances": ["full stop"], "lang": session.lang}, 

322 {"session": session.serialize(), "source": "A", "destination": "B"}) 

323 

324 stop_skill_active = [ 

325 message, 

326 Message(f"{self.skill_id}.stop.ping", 

327 {"skill_id":self.skill_id}), 

328 Message("skill.stop.pong", 

329 {"skill_id": self.skill_id, "can_handle": True}, 

330 {"skill_id": self.skill_id}), 

331 

332 Message("stop.openvoiceos.activate", 

333 context={"skill_id": "stop.openvoiceos"}), 

334 Message("stop:skill", 

335 context={"skill_id": "stop.openvoiceos"}), 

336 Message(f"{self.skill_id}.stop", 

337 context={"skill_id": "stop.openvoiceos"}), 

338 Message(f"{self.skill_id}.stop.response", 

339 {"skill_id": self.skill_id, "result": True}, 

340 {"skill_id": self.skill_id}), 

341 

342 # async stop pipeline callback emits these messages 

343 # but we cant guarantee where in the test they will be emitted 

344 

345 # if skill is in middle of get_response 

346 #Message("mycroft.skills.abort_question", 

347 # {"skill_id": self.skill_id}, 

348 # {"skill_id": self.skill_id}), 

349 

350 # if skill is in active_list 

351 #Message("ovos.skills.converse.force_timeout", 

352 # {"skill_id": self.skill_id}, 

353 # {"skill_id": self.skill_id}), 

354 

355 # if skill is executing TTS 

356 #Message("mycroft.audio.speech.stop", 

357 # {"skill_id": self.skill_id}, 

358 # {"skill_id": self.skill_id}), 

359 

360 # the intent running in the daemon thread exits cleanly 

361 Message("mycroft.skill.handler.complete", 

362 {"name": "CountSkill.handle_how_are_you_intent"}, 

363 {"skill_id": self.skill_id}), 

364 Message("ovos.utterance.handled", 

365 {"name": "CountSkill.handle_how_are_you_intent"}, 

366 {"skill_id": self.skill_id}) 

367 ] 

368 test = End2EndTest( 

369 minicroft=self.minicroft, 

370 skill_ids=[], 

371 eof_msgs=[], 

372 flip_points=["recognizer_loop:utterance"], 

373 # messages in 'keep_original_src' would not be sent to hivemind clients 

374 # i.e. they are directed towards ovos-core 

375 keep_original_src=[f"{self.skill_id}.stop.ping", 

376 f"{self.skill_id}.stop", 

377 "mycroft.skills.abort_question", 

378 # "stop.openvoiceos.activate", # TODO 

379 "ovos.skills.converse.force_timeout"], 

380 ignore_messages=self.ignore_messages, 

381 async_messages=[ 

382 "ovos.skills.converse.force_timeout" 

383 ], # order that it wil be received unknown 

384 source_message=message, 

385 expected_messages=stop_skill_active 

386 ) 

387 test.execute()