files: # 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.
  #
  """The module execute a test of one skill intent.
  
  Using a mocked message bus this module is responsible for sending utterences
  and testing that the intent is called.
  
  The module runner can test:
      That the expected intent in the skill is activated
      That the expected parameters are extracted from the utterance
      That Mycroft contexts are set or removed
      That the skill speak the intended answer
      The content of any message exchanged between the skill and the mycroft core
  
  To set up a test the test runner can
      Send an utterance, as the user would normally speak
      Set up and remove context
      Set up a custom timeout for the test runner, to allow for skills that runs
      for a very long time
  
  """
  from queue import Queue, Empty
  import json
  import time
  import os
  import re
  import ast
  from os.path import join, isdir, basename
  from pyee import EventEmitter
  from numbers import Number
  from mycroft.messagebus.message import Message
  from mycroft.skills.core import MycroftSkill, FallbackSkill
  from mycroft.skills.settings import SkillSettings
  from mycroft.skills.skill_loader import SkillLoader
  from mycroft.configuration import Configuration
  from mycroft.util.format import expand_options
  
  from logging import StreamHandler
  from io import StringIO
  from contextlib import contextmanager
  
  MainModule = '__init__'
  
  DEFAULT_EVALUAITON_TIMEOUT = 30
  
  # Set a configuration value to allow skills to check if they're in a test
  Configuration.get()['test_env'] = True
  
  
  class SkillTestError(Exception):
      pass
  
  
  # Easy way to show colors on terminals
  class clr:
      PINK = '\033[95m'
      BLUE = '\033[94m'
      CYAN = '\033[96m'
      GREEN = '\033[92m'
      YELLOW = '\033[93m'
      RED = '\033[91m'
      DKGRAY = '\033[90m'
      # Classes
      USER_UTT = '\033[96m'  # cyan
      MYCROFT = '\033[33m'   # bright yellow
      HEADER = '\033[94m'    # blue
      WARNING = '\033[93m'   # yellow
      FAIL = '\033[91m'      # red
      RESET = '\033[0m'
  
  
  class no_clr:
      PINK = ''
      BLUE = ''
      CYAN = ''
      GREEN = ''
      YELLOW = ''
      RED = ''
      DKGRAY = ''
      USER_UTT = ''
      MYCROFT = ''
      HEADER = ''
      WARNING = ''
      FAIL = ''
      RESET = ''
  
  
  # MST as in Mycroft Skill Tester
  if 'MST_NO_COLOR' not in os.environ:
      color = clr
  else:
      color = no_clr
  
  
  @contextmanager
  def temporary_handler(log, handler):
      """Context manager to replace the default logger with a temporary logger.
  
      Args:
          log (LOG): mycroft LOG object
          handler (logging.Handler): Handler object to use
      """
      old_handler = log.handler
      log.handler = handler
      yield
      log.handler = old_handler
  
  
  def create_skill_descriptor(skill_path):
      return {"path": skill_path}
  
  
  def get_skills(skills_folder):
      """Find skills in the skill folder or sub folders.
  
          Recursive traversal into subfolders stop when a __init__.py file
          is discovered
  
          Args:
              skills_folder:  Folder to start a search for skills __init__.py
                              files
  
          Returns:
              list: the skills
      """
  
      skills = []
  
      def _get_skill_descriptor(skills_folder):
          if not isdir(skills_folder):
              return
          if MainModule + ".py" in os.listdir(skills_folder):
              skills.append(create_skill_descriptor(skills_folder))
              return
  
          possible_skills = os.listdir(skills_folder)
          for i in possible_skills:
              _get_skill_descriptor(join(skills_folder, i))
  
      _get_skill_descriptor(skills_folder)
  
      skills = sorted(skills, key=lambda p: basename(p['path']))
      return skills
  
  
  def load_skills(emitter, skills_root):
      """Load all skills and set up emitter
  
          Args:
              emitter: The emmitter to use
              skills_root: Directory of the skills __init__.py
  
          Returns:
              tuple: (list of loaded skills, dict with logs for each skill)
  
      """
      skill_list = []
      log = {}
      for skill in get_skills(skills_root):
          path = skill["path"]
          skill_id = 'test-' + basename(path)
  
          # Catch the logs during skill loading
          from mycroft.skills.core import LOG as skills_log
          buf = StringIO()
          with temporary_handler(skills_log, StreamHandler(buf)):
              skill_loader = SkillLoader(emitter, path)
              skill_loader.skill_id = skill_id
              skill_loader.load()
              skill_list.append(skill_loader.instance)
          log[path] = buf.getvalue()
  
      return skill_list, log
  
  
  def unload_skills(skills):
      for s in skills:
          s.default_shutdown()
  
  
  class InterceptEmitter(object):
      """
      This class intercepts and allows emitting events between the
      skill_tester and the skill being tested.
      When a test is running emitted communication is intercepted for analysis
      """
  
      def __init__(self):
          self.emitter = EventEmitter()
          self.q = None
  
      def on(self, event, f):
          # run all events
          print("Event: ", event)
          self.emitter.on(event, f)
  
      def emit(self, event, *args, **kwargs):
          event_name = event.type
          if self.q:
              self.q.put(event)
          self.emitter.emit(event_name, event, *args, **kwargs)
  
      def wait_for_response(self, event, *args, **kwargs):
          return None
  
      def once(self, event, f):
          self.emitter.once(event, f)
  
      def remove(self, event_name, func):
          pass
  
      def remove_all_listeners(self, event_name):
          pass
  
  
  class TestSettings(SkillSettings):
      """ SkillSettings instance without saving/loading capability.
      """
      def save_skill_settings(self, skill_settings):
          pass
  
      def _poll_skill_settings(self):
          pass
  
      def _load_settings_meta(self):
          return None
  
      def update(self, settings_meta):
          pass
  
      def load_skill_settings_from_file(self):
          pass
  
      def store(self, force=False):
          pass
  
  
  class MockSkillsLoader(object):
      """Load a skill and set up emitter
      """
  
      def __init__(self, skills_root):
          self.load_log = None
  
          self.skills_root = skills_root
          self.emitter = InterceptEmitter()
          from mycroft.skills.intent_service import IntentService
          from mycroft.skills.padatious_service import PadatiousService
          self.ih = IntentService(self.emitter)
          self.ps = PadatiousService(self.emitter, self.ih)
          self.skills = None
          self.emitter.on(
              'intent_failure',
              FallbackSkill.make_intent_failure_handler(self.emitter))
  
          def make_response(message):
              skill_id = message.data.get('skill_id', '')
              data = dict(result=False, skill_id=skill_id)
              self.emitter.emit(Message('skill.converse.response', data))
          self.emitter.on('skill.converse.request', make_response)
  
      def load_skills(self):
          skills, self.load_log = load_skills(self.emitter, self.skills_root)
          self.skills = [s for s in skills if s]
          self.ps.train(Message('', data=dict(single_thread=True)))
          return self.emitter.emitter  # kick out the underlying emitter
  
      def unload_skills(self):
          unload_skills(self.skills)
  
  
  def load_test_case_file(test_case_file):
      """Load a test case to run."""
      print("")
      print(color.HEADER + "="*20 + " RUNNING TEST " + "="*20 + color.RESET)
      print('Test file: ', test_case_file)
      with open(test_case_file, 'r') as f:
          test_case = json.load(f)
      print('Test:', json.dumps(test_case, indent=4, sort_keys=False))
      return test_case
  
  
  class SkillTest(object):
      """
          This class is instantiated for each skill being tested. It holds the
          data needed for the test, and contains the methods doing the test
  
      """
  
      def __init__(self, skill, test_case_file, emitter, test_status=None):
          self.skill = skill
          self.test_case_file = test_case_file
          self.emitter = emitter
          self.dict = dict
          self.output_file = None
          self.returned_intent = False
          self.test_status = test_status
          self.failure_msg = None
  
      def run(self, loader):
          """ Execute the test
  
          Run a test for a skill. The skill, test_case_file and emitter is
          already set up in the __init__ method.
  
          This method does all the preparation and cleanup and calls
          self.execute_test() to perform the actual test.
  
          Args:
              bool: Test results -- only True if all passed
          """
          s = [s for s in loader.skills if s and s.root_dir == self.skill]
          if s:
              s = s[0]
          else:
              # The skill wasn't loaded, print the load log for the skill
              if self.skill in loader.load_log:
                  print('\n {} Captured Logs from loading {}'.format('=' * 15,
                                                                     '=' * 15))
                  print(loader.load_log.pop(self.skill))
  
              raise SkillTestError('Skill couldn\'t be loaded')
  
          orig_get_response = s.get_response
          original_settings = s.settings
          try:
              return self.execute_test(s)
          finally:
              s.get_response = orig_get_response
              s.settings = original_settings
  
      def send_play_query(self, s, test_case):
          """Emit an event triggering the a check for playback possibilities."""
          play_query = test_case['play_query']
          print('PLAY QUERY', color.USER_UTT + play_query + color.RESET)
          self.emitter.emit('play:query', Message('play:query:',
                                                  {'phrase': play_query}))
  
      def send_play_start(self, s, test_case):
          """Emit an event starting playback from the skill."""
          print('PLAY START')
          callback_data = test_case['play_start']
          callback_data['skill_id'] = s.skill_id
          self.emitter.emit('play:start',
                            Message('play:start', callback_data))
  
      def send_question(self, test_case):
          """Emit a Question to the loaded skills."""
          print("QUESTION: {}".format(test_case['question']))
          callback_data = {'phrase': test_case['question']}
          self.emitter.emit('question:query',
                            Message('question:query', data=callback_data))
  
      def send_utterance(self, test_case):
          """Emit an utterance to the loaded skills."""
          utt = test_case['utterance']
          print("UTTERANCE:", color.USER_UTT + utt + color.RESET)
          self.emitter.emit('recognizer_loop:utterance',
                            Message('recognizer_loop:utterance',
                                    {'utterances': [utt]}))
  
      def apply_test_settings(self, s, test_case):
          """Replace the skills settings with settings from the test_case."""
          s.settings = TestSettings('/tmp/', self.test_case_file)
          for key in test_case['settings']:
              s.settings[key] = test_case['settings'][key]
          print(color.YELLOW, 'will run test with custom settings:',
                              '\n{}'.format(s.settings), color.RESET)
  
      def setup_get_response(self, s, test_case):
          """Setup interception of get_response calls."""
          def get_response(dialog='', data=None, announcement='',
                           validator=None, on_fail=None, num_retries=-1):
              data = data or {}
              utt = announcement or s.dialog_renderer.render(dialog, data)
              print(color.MYCROFT + ">> " + utt + color.RESET)
              s.speak(utt)
  
              response = test_case['responses'].pop(0)
              print("SENDING RESPONSE:",
                    color.USER_UTT + response + color.RESET)
              return response
  
          s.get_response = get_response
  
      def remove_context(self, s, cxt):
          """remove an adapt context."""
          if isinstance(cxt, list):
              for x in cxt:
                  MycroftSkill.remove_context(s, x)
          else:
              MycroftSkill.remove_context(s, cxt)
  
      def set_context(self, s, cxt):
          """Set an adapt context."""
          for key, value in cxt.items():
              MycroftSkill.set_context(s, key, value)
  
      def send_test_input(self, s, test_case):
          """Emit an utterance, just like the STT engine does. This sends the
          provided text to the skill engine for intent matching and it then
          invokes the skill.
  
          It also handles some special cases for common play skills and common
          query skills.
          """
          if 'utterance' in test_case:
              self.send_utterance(test_case)
          elif 'play_query' in test_case:
              self.send_play_query(s, test_case)
          elif 'play_start' in test_case:
              self.send_play_start(s, test_case)
          elif 'question' in test_case:
              self.send_question(test_case)
          else:
              raise SkillTestError('No input provided in test case')
  
      def execute_test(self, s):
          """ Execute test case.
  
          Arguments:
              s (MycroftSkill): mycroft skill to test
  
          Returns:
              (bool) True if the test succeeded completely.
          """
          test_case = load_test_case_file(self.test_case_file)
  
          if 'settings' in test_case:
              self.apply_test_settings(s, test_case)
  
          if 'responses' in test_case:
              self.setup_get_response(s, test_case)
  
          # If we keep track of test status for the entire skill, then
          # get all intents from the skill, and mark current intent
          # tested
          if self.test_status:
              self.test_status.append_intent(s)
              if 'intent_type' in test_case:
                  self.test_status.set_tested(test_case['intent_type'])
  
          evaluation_rule = EvaluationRule(test_case, s)
  
          # Set up queue for emitted events. Because
          # the evaluation method expects events to be received in convoy,
          # and be handled one by one. We cant make assumptions about threading
          # in the core or the skill
          q = Queue()
          s.bus.q = q
  
          # Set up context before calling intent
          # This option makes it possible to better isolate (reduce dependance)
          # between test_cases
          cxt = test_case.get('remove_context', None)
          if cxt:
              self.remove_context(s, cxt)
  
          cxt = test_case.get('set_context', None)
          if cxt:
              self.set_context(s, cxt)
  
          self.send_test_input(s, test_case)
          # Wait up to X seconds for the test_case to complete
          timeout = self.get_timeout(test_case)
  
          while not evaluation_rule.all_succeeded():
              # Process the queue until a skill handler sends a complete message
              if self.check_queue(q, evaluation_rule) or time.time() > timeout:
                  break
  
          self.shutdown_emitter(s)
  
          # Report test result if failed
          return self.results(evaluation_rule)
  
      def get_timeout(self, test_case):
          """Find any timeout specified in test case.
  
          If no timeout is specified return the default.
          """
          if (test_case.get('evaluation_timeout', None) and
                  isinstance(test_case['evaluation_timeout'], int)):
              return time.time() + int(test_case.get('evaluation_timeout'))
          else:
              return time.time() + DEFAULT_EVALUAITON_TIMEOUT
  
      def check_queue(self, q, evaluation_rule):
          """Check the queue for events.
  
          If event indicating skill completion is found returns True, else False.
          """
          try:
              event = q.get(timeout=1)
              if ':' in event.type:
                  event.data['__type__'] = event.type.split(':')[1]
              else:
                  event.data['__type__'] = event.type
  
              evaluation_rule.evaluate(event.data)
              if event.type == 'mycroft.skill.handler.complete':
                  return True
          except Empty:
              pass
          return False
  
      def shutdown_emitter(self, s):
          """Shutdown the skill connection to the bus."""
          # Stop emiter from sending on queue
          s.bus.q = None
  
          # remove the skill which is not responding
          self.emitter.remove_all_listeners('speak')
          self.emitter.remove_all_listeners('mycroft.skill.handler.complete')
  
      def results(self, evaluation_rule):
          """Display and report the results."""
          if not evaluation_rule.all_succeeded():
              self.failure_msg = str(evaluation_rule.get_failure())
              print(color.FAIL + "Evaluation
  
  [... the rest is omitted ...]
files: test/integrationtests/skills/skill_tester.py
files: # 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.
  #
  """The module execute a test of one skill intent.
  
  Using a mocked message bus this module is responsible for sending utterences
  and testing that the intent is called.
  
  The module runner can test:
      That the expected intent in the skill is activated
      That the expected parameters are extracted from the utterance
      That Mycroft contexts are set or removed
      That the skill speak the intended answer
      The content of any message exchanged between the skill and the mycroft core
  
  To set up a test the test runner can
      Send an utterance, as the user would normally speak
      Set up and remove context
      Set up a custom timeout for the test runner, to allow for skills that runs
      for a very long time
  
  """
  from queue import Queue, Empty
  import json
  import time
  import os
  import re
  import ast
  from os.path import join, isdir, basename
  from pyee import EventEmitter
  from numbers import Number
  from mycroft.messagebus.message import Message
  from mycroft.skills.core import MycroftSkill, FallbackSkill
  from mycroft.skills.settings import SkillSettings
  from mycroft.skills.skill_loader import SkillLoader
  from mycroft.configuration import Configuration
  from mycroft.util.format import expand_options
  
  from logging import StreamHandler
  from io import StringIO
  from contextlib import contextmanager
  
  MainModule = '__init__'
  
  DEFAULT_EVALUAITON_TIMEOUT = 30
  
  # Set a configuration value to allow skills to check if they're in a test
  Configuration.get()['test_env'] = True
  
  
  class SkillTestError(Exception):
      pass
  
  
  # Easy way to show colors on terminals
  class clr:
      PINK = '\033[95m'
      BLUE = '\033[94m'
      CYAN = '\033[96m'
      GREEN = '\033[92m'
      YELLOW = '\033[93m'
      RED = '\033[91m'
      DKGRAY = '\033[90m'
      # Classes
      USER_UTT = '\033[96m'  # cyan
      MYCROFT = '\033[33m'   # bright yellow
      HEADER = '\033[94m'    # blue
      WARNING = '\033[93m'   # yellow
      FAIL = '\033[91m'      # red
      RESET = '\033[0m'
  
  
  class no_clr:
      PINK = ''
      BLUE = ''
      CYAN = ''
      GREEN = ''
      YELLOW = ''
      RED = ''
      DKGRAY = ''
      USER_UTT = ''
      MYCROFT = ''
      HEADER = ''
      WARNING = ''
      FAIL = ''
      RESET = ''
  
  
  # MST as in Mycroft Skill Tester
  if 'MST_NO_COLOR' not in os.environ:
      color = clr
  else:
      color = no_clr
  
  
  @contextmanager
  def temporary_handler(log, handler):
      """Context manager to replace the default logger with a temporary logger.
  
      Args:
          log (LOG): mycroft LOG object
          handler (logging.Handler): Handler object to use
      """
      old_handler = log.handler
      log.handler = handler
      yield
      log.handler = old_handler
  
  
  def create_skill_descriptor(skill_path):
      return {"path": skill_path}
  
  
  def get_skills(skills_folder):
      """Find skills in the skill folder or sub folders.
  
          Recursive traversal into subfolders stop when a __init__.py file
          is discovered
  
          Args:
              skills_folder:  Folder to start a search for skills __init__.py
                              files
  
          Returns:
              list: the skills
      """
  
      skills = []
  
      def _get_skill_descriptor(skills_folder):
          if not isdir(skills_folder):
              return
          if MainModule + ".py" in os.listdir(skills_folder):
              skills.append(create_skill_descriptor(skills_folder))
              return
  
          possible_skills = os.listdir(skills_folder)
          for i in possible_skills:
              _get_skill_descriptor(join(skills_folder, i))
  
      _get_skill_descriptor(skills_folder)
  
      skills = sorted(skills, key=lambda p: basename(p['path']))
      return skills
  
  
  def load_skills(emitter, skills_root):
      """Load all skills and set up emitter
  
          Args:
              emitter: The emmitter to use
              skills_root: Directory of the skills __init__.py
  
          Returns:
              tuple: (list of loaded skills, dict with logs for each skill)
  
      """
      skill_list = []
      log = {}
      for skill in get_skills(skills_root):
          path = skill["path"]
          skill_id = 'test-' + basename(path)
  
          # Catch the logs during skill loading
          from mycroft.skills.core import LOG as skills_log
          buf = StringIO()
          with temporary_handler(skills_log, StreamHandler(buf)):
              skill_loader = SkillLoader(emitter, path)
              skill_loader.skill_id = skill_id
              skill_loader.load()
              skill_list.append(skill_loader.instance)
          log[path] = buf.getvalue()
  
      return skill_list, log
  
  
  def unload_skills(skills):
      for s in skills:
          s.default_shutdown()
  
  
  class InterceptEmitter(object):
      """
      This class intercepts and allows emitting events between the
      skill_tester and the skill being tested.
      When a test is running emitted communication is intercepted for analysis
      """
  
      def __init__(self):
          self.emitter = EventEmitter()
          self.q = None
  
      def on(self, event, f):
          # run all events
          print("Event: ", event)
          self.emitter.on(event, f)
  
      def emit(self, event, *args, **kwargs):
          event_name = event.type
          if self.q:
              self.q.put(event)
          self.emitter.emit(event_name, event, *args, **kwargs)
  
      def wait_for_response(self, event, *args, **kwargs):
          return None
  
      def once(self, event, f):
          self.emitter.once(event, f)
  
      def remove(self, event_name, func):
          pass
  
      def remove_all_listeners(self, event_name):
          pass
  
  
  class TestSettings(SkillSettings):
      """ SkillSettings instance without saving/loading capability.
      """
      def save_skill_settings(self, skill_settings):
          pass
  
      def _poll_skill_settings(self):
          pass
  
      def _load_settings_meta(self):
          return None
  
      def update(self, settings_meta):
          pass
  
      def load_skill_settings_from_file(self):
          pass
  
      def store(self, force=False):
          pass
  
  
  class MockSkillsLoader(object):
      """Load a skill and set up emitter
      """
  
      def __init__(self, skills_root):
          self.load_log = None
  
          self.skills_root = skills_root
          self.emitter = InterceptEmitter()
          from mycroft.skills.intent_service import IntentService
          from mycroft.skills.padatious_service import PadatiousService
          self.ih = IntentService(self.emitter)
          self.ps = PadatiousService(self.emitter, self.ih)
          self.skills = None
          self.emitter.on(
              'intent_failure',
              FallbackSkill.make_intent_failure_handler(self.emitter))
  
          def make_response(message):
              skill_id = message.data.get('skill_id', '')
              data = dict(result=False, skill_id=skill_id)
              self.emitter.emit(Message('skill.converse.response', data))
          self.emitter.on('skill.converse.request', make_response)
  
      def load_skills(self):
          skills, self.load_log = load_skills(self.emitter, self.skills_root)
          self.skills = [s for s in skills if s]
          self.ps.train(Message('', data=dict(single_thread=True)))
          return self.emitter.emitter  # kick out the underlying emitter
  
      def unload_skills(self):
          unload_skills(self.skills)
  
  
  class SkillTest(object):
      """
          This class is instantiated for each skill being tested. It holds the
          data needed for the test, and contains the methods doing the test
  
      """
  
      def __init__(self, skill, test_case_file, emitter, test_status=None):
          self.skill = skill
          self.test_case_file = test_case_file
          self.emitter = emitter
          self.dict = dict
          self.output_file = None
          self.returned_intent = False
          self.test_status = test_status
          self.failure_msg = None
  
      def run(self, loader):
          """ Execute the test
  
          Run a test for a skill. The skill, test_case_file and emitter is
          already set up in the __init__ method.
  
          This method does all the preparation and cleanup and calls
          self.execute_test() to perform the actual test.
  
          Args:
              bool: Test results -- only True if all passed
          """
          s = [s for s in loader.skills if s and s.root_dir == self.skill]
          if s:
              s = s[0]
          else:
              # The skill wasn't loaded, print the load log for the skill
              if self.skill in loader.load_log:
                  print('\n {} Captured Logs from loading {}'.format('=' * 15,
                                                                     '=' * 15))
                  print(loader.load_log.pop(self.skill))
  
              raise SkillTestError('Skill couldn\'t be loaded')
  
          orig_get_response = s.get_response
          original_settings = s.settings
          try:
              return self.execute_test(s)
          finally:
              s.get_response = orig_get_response
              s.settings = original_settings
  
      def execute_test(self, s):
          """ Execute test case.
  
          Arguments:
              s (MycroftSkill): mycroft skill to test
  
          Returns:
              (bool) True if the test succeeded completely.
          """
          print("")
          print(color.HEADER + "="*20 + " RUNNING TEST " + "="*20 + color.RESET)
          print('Test file: ', self.test_case_file)
          with open(self.test_case_file, 'r') as f:
              test_case = json.load(f)
          print('Test:', json.dumps(test_case, indent=4, sort_keys=False))
  
          if 'settings' in test_case:
              s.settings = TestSettings('/tmp/', self.test_case_file)
              for key in test_case['settings']:
                  s.settings[key] = test_case['settings'][key]
              print(color.YELLOW, 'will run test with custom settings:',
                    '\n{}'.format(s.settings), color.RESET)
  
          if 'responses' in test_case:
              def get_response(dialog='', data=None, announcement='',
                               validator=None, on_fail=None, num_retries=-1):
                  data = data or {}
                  utt = announcement or s.dialog_renderer.render(dialog, data)
                  print(color.MYCROFT + ">> " + utt + color.RESET)
                  s.speak(utt)
  
                  response = test_case['responses'].pop(0)
                  print("SENDING RESPONSE:",
                        color.USER_UTT + response + color.RESET)
                  return response
  
              s.get_response = get_response
  
          # If we keep track of test status for the entire skill, then
          # get all intents from the skill, and mark current intent
          # tested
          if self.test_status:
              self.test_status.append_intent(s)
              if 'intent_type' in test_case:
                  self.test_status.set_tested(test_case['intent_type'])
  
          evaluation_rule = EvaluationRule(test_case, s)
  
          # Set up queue for emitted events. Because
          # the evaluation method expects events to be received in convoy,
          # and be handled one by one. We cant make assumptions about threading
          # in the core or the skill
          q = Queue()
          s.bus.q = q
  
          # Set up context before calling intent
          # This option makes it possible to better isolate (reduce dependance)
          # between test_cases
          cxt = test_case.get('remove_context', None)
          if cxt:
              if isinstance(cxt, list):
                  for x in cxt:
                      MycroftSkill.remove_context(s, x)
              else:
                  MycroftSkill.remove_context(s, cxt)
  
          cxt = test_case.get('set_context', None)
          if cxt:
              for key, value in cxt.items():
                  MycroftSkill.set_context(s, key, value)
  
          # Emit an utterance, just like the STT engine does.  This sends the
          # provided text to the skill engine for intent matching and it then
          # invokes the skill.
          if 'utterance' in test_case:
              utt = test_case['utterance']
              print("UTTERANCE:", color.USER_UTT + utt + color.RESET)
              self.emitter.emit('recognizer_loop:utterance',
                                Message('recognizer_loop:utterance',
                                        {'utterances': [utt]}))
          elif 'play_query' in test_case:
              play_query = test_case['play_query']
              print('PLAY QUERY', color.USER_UTT + play_query + color.RESET)
              self.emitter.emit('play:query', Message('play:query:',
                                                      {'phrase': play_query}))
          elif 'play_start' in test_case:
              print('PLAY START')
              callback_data = test_case['play_start']
              callback_data['skill_id'] = s.skill_id
              self.emitter.emit('play:start',
                                Message('play:start', callback_data))
          elif 'question' in test_case:
              print("QUESTION: {}".format(test_case['question']))
              callback_data = {'phrase': test_case['question']}
              self.emitter.emit('question:query',
                                Message('question:query', data=callback_data))
          else:
              raise SkillTestError('No input utterance provided')
  
          # Wait up to X seconds for the test_case to complete
          timeout = time.time() + int(test_case.get('evaluation_timeout')) \
              if test_case.get('evaluation_timeout', None) and \
              isinstance(test_case['evaluation_timeout'], int) \
              else time.time() + DEFAULT_EVALUAITON_TIMEOUT
          while not evaluation_rule.all_succeeded():
              try:
                  event = q.get(timeout=1)
                  if ':' in event.type:
                      event.data['__type__'] = event.type.split(':')[1]
                  else:
                      event.data['__type__'] = event.type
  
                  evaluation_rule.evaluate(event.data)
                  if event.type == 'mycroft.skill.handler.complete':
                      break
              except Empty:
                  pass
              if time.time() > timeout:
                  break
  
          # Stop emmiter from sending on queue
          s.bus.q = None
  
          # remove the skill which is not responding
          self.emitter.remove_all_listeners('speak')
          self.emitter.remove_all_listeners('mycroft.skill.handler.complete')
          # Report test result if failed
          if not evaluation_rule.all_succeeded():
              self.failure_msg = str(evaluation_rule.get_failure())
              print(color.FAIL + "Evaluation failed" + color.RESET)
              print(color.FAIL + "Failure:", self.failure_msg + color.RESET)
              return False
  
          return True
  
  
  # Messages that should not print debug info
  HIDDEN_MESSAGES = ['skill.converse.request', 'skill.converse.response',
                     'gui.page.show', 'gui.value.set']
  
  
  def load_dialog_list(skill, dialog):
      """ Load dialog from files into a single list.
  
      Args:
          skill (MycroftSkill): skill to load dialog from
          dialog (list): Dialog names (str) to load
  
      Returns:
          list: Expanded dialog strings
      """
      dialogs = []
      try:
          for d in dialog:
              for e in skill.dialog_renderer.templates[d]:
                  dialogs += expand_options(e)
      except Exception as template_load_exception:
          print(color.FAIL +
                "Failed to load dialog template " +
                "'dialog/en-us/" + d + ".dialog'" +
                color.RESET)
          raise Exception("Can't load 'excepted_dialog': "
                          "file '" + d + ".dialog'") \
              from template_load_exception
      return dialogs
  
  
  class EvaluationRule(object):
      """
          This class initially convert the test_case json file to internal rule
          format, which is stored throughout the testcase run. All Messages on
          the event bus can be evaluated against the rules (test_case)
  
          This approach makes it easier to add new tests, since Message and rule
          traversal is already set up for the internal rule format.
          The test writer can use the internal rule format directly in the
          test_case using the assert keyword, which allows for more
          powerfull/individual test cases than the standard dictionaly
      """
  
      def __init__(self, test_case, skill=None):
          """ Convert test_case read from file to internal rule format
  
          Args:
              test_case:  The loaded test case
              skill:      optional skill to test, used to fetch dialogs
          """
          self.rule = []
  
          _x = ['and']
          if 'utterance' in test_case and 'intent_type' in test_case:
              intent_type = str(test_case['intent_type'])
              _x.append(['or'] +
                        [['endsWith', 'intent
  
  [... the rest is omitted ...]
files: test/integrationtests/skills/skill_tester.py