files.new_contents: # 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
  
  [... the rest is omitted ...]
files.new_path: test/integrationtests/skills/skill_tester.py
files.old_contents: # 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
  
  [... the rest is omitted ...]
files.old_path: test/integrationtests/skills/skill_tester.py