summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/__init__.py0
-rw-r--r--tests/test_brain.py49
-rw-r--r--tests/test_diagnose.py12
-rw-r--r--tests/test_g2p.py71
-rw-r--r--tests/test_modules.py97
-rw-r--r--tests/test_stt.py51
-rw-r--r--tests/test_tts.py11
-rw-r--r--tests/test_vocabcompiler.py139
8 files changed, 430 insertions, 0 deletions
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/tests/__init__.py
diff --git a/tests/test_brain.py b/tests/test_brain.py
new file mode 100644
index 0000000..7b4c482
--- /dev/null
+++ b/tests/test_brain.py
@@ -0,0 +1,49 @@
+#!/usr/bin/env python2
+# -*- coding: utf-8-*-
+import unittest
+import mock
+from client import brain, test_mic
+
+
+DEFAULT_PROFILE = {
+ 'prefers_email': False,
+ 'location': 'Cape Town',
+ 'timezone': 'US/Eastern',
+ 'phone_number': '012344321'
+}
+
+
+class TestBrain(unittest.TestCase):
+
+ @staticmethod
+ def _emptyBrain():
+ mic = test_mic.Mic([])
+ profile = DEFAULT_PROFILE
+ return brain.Brain(mic, profile)
+
+ def testLog(self):
+ """Does Brain correctly log errors when raised by modules?"""
+ my_brain = TestBrain._emptyBrain()
+ unclear = my_brain.modules[-1]
+ with mock.patch.object(unclear, 'handle') as mocked_handle:
+ with mock.patch.object(my_brain._logger, 'error') as mocked_log:
+ mocked_handle.side_effect = KeyError('foo')
+ my_brain.query("zzz gibberish zzz")
+ self.assertTrue(mocked_log.called)
+
+ def testSortByPriority(self):
+ """Does Brain sort modules by priority?"""
+ my_brain = TestBrain._emptyBrain()
+ priorities = filter(lambda m: hasattr(m, 'PRIORITY'), my_brain.modules)
+ target = sorted(priorities, key=lambda m: m.PRIORITY, reverse=True)
+ self.assertEqual(target, priorities)
+
+ def testPriority(self):
+ """Does Brain correctly send query to higher-priority module?"""
+ my_brain = TestBrain._emptyBrain()
+ hn_module = 'HN'
+ hn = filter(lambda m: m.__name__ == hn_module, my_brain.modules)[0]
+
+ with mock.patch.object(hn, 'handle') as mocked_handle:
+ my_brain.query(["hacker news"])
+ self.assertTrue(mocked_handle.called)
diff --git a/tests/test_diagnose.py b/tests/test_diagnose.py
new file mode 100644
index 0000000..ba49618
--- /dev/null
+++ b/tests/test_diagnose.py
@@ -0,0 +1,12 @@
+#!/usr/bin/env python2
+# -*- coding: utf-8-*-
+import unittest
+from client import diagnose
+
+
+class TestDiagnose(unittest.TestCase):
+ def testPythonImportCheck(self):
+ # This a python stdlib module that definitely exists
+ self.assertTrue(diagnose.check_python_import("os"))
+ # I sincerly hope nobody will ever create a package with that name
+ self.assertFalse(diagnose.check_python_import("nonexistant_package"))
diff --git a/tests/test_g2p.py b/tests/test_g2p.py
new file mode 100644
index 0000000..7904cbf
--- /dev/null
+++ b/tests/test_g2p.py
@@ -0,0 +1,71 @@
+#!/usr/bin/env python2
+# -*- coding: utf-8-*-
+import unittest
+import tempfile
+import mock
+from client import g2p
+
+
+def phonetisaurus_installed():
+ try:
+ g2p.PhonetisaurusG2P(**g2p.PhonetisaurusG2P.get_config())
+ except OSError:
+ return False
+ else:
+ return True
+
+
+@unittest.skipUnless(phonetisaurus_installed(),
+ "Phonetisaurus or fst_model not present")
+class TestG2P(unittest.TestCase):
+
+ def setUp(self):
+ self.g2pconverter = g2p.PhonetisaurusG2P(
+ **g2p.PhonetisaurusG2P.get_config())
+ self.words = ['GOOD', 'BAD', 'UGLY']
+
+ def testTranslateWord(self):
+ for word in self.words:
+ self.assertIn(word, self.g2pconverter.translate(word).keys())
+
+ def testTranslateWords(self):
+ results = self.g2pconverter.translate(self.words).keys()
+ for word in self.words:
+ self.assertIn(word, results)
+
+
+class TestPatchedG2P(TestG2P):
+ class DummyProc(object):
+ def __init__(self, *args, **kwargs):
+ self.returncode = 0
+
+ def communicate(self):
+ return ("GOOD\t9.20477\t<s> G UH D </s>\n" +
+ "GOOD\t14.4036\t<s> G UW D </s>\n" +
+ "GOOD\t16.0258\t<s> G UH D IY </s>\n" +
+ "BAD\t0.7416\t<s> B AE D </s>\n" +
+ "BAD\t12.5495\t<s> B AA D </s>\n" +
+ "BAD\t13.6745\t<s> B AH D </s>\n" +
+ "UGLY\t12.572\t<s> AH G L IY </s>\n" +
+ "UGLY\t17.9278\t<s> Y UW G L IY </s>\n" +
+ "UGLY\t18.9617\t<s> AH G L AY </s>\n", "")
+
+ def setUp(self):
+ with mock.patch('client.g2p.diagnose.check_executable',
+ return_value=True):
+ with tempfile.NamedTemporaryFile() as f:
+ conf = g2p.PhonetisaurusG2P.get_config().items()
+ with mock.patch.object(g2p.PhonetisaurusG2P, 'get_config',
+ classmethod(lambda cls: dict(conf +
+ [('fst_model', f.name)]))):
+ super(self.__class__, self).setUp()
+
+ def testTranslateWord(self):
+ with mock.patch('subprocess.Popen',
+ return_value=TestPatchedG2P.DummyProc()):
+ super(self.__class__, self).testTranslateWord()
+
+ def testTranslateWords(self):
+ with mock.patch('subprocess.Popen',
+ return_value=TestPatchedG2P.DummyProc()):
+ super(self.__class__, self).testTranslateWords()
diff --git a/tests/test_modules.py b/tests/test_modules.py
new file mode 100644
index 0000000..8f606de
--- /dev/null
+++ b/tests/test_modules.py
@@ -0,0 +1,97 @@
+#!/usr/bin/env python2
+# -*- coding: utf-8-*-
+import unittest
+from client import test_mic, diagnose, jasperpath
+from client.modules import Life, Joke, Time, Gmail, HN, News, Weather
+
+DEFAULT_PROFILE = {
+ 'prefers_email': False,
+ 'location': 'Cape Town',
+ 'timezone': 'US/Eastern',
+ 'phone_number': '012344321'
+}
+
+
+class TestModules(unittest.TestCase):
+
+ def setUp(self):
+ self.profile = DEFAULT_PROFILE
+ self.send = False
+
+ def runConversation(self, query, inputs, module):
+ """Generic method for spoofing conversation.
+
+ Arguments:
+ query -- The initial input to the server.
+ inputs -- Additional input, if conversation is extended.
+
+ Returns:
+ The server's responses, in a list.
+ """
+ self.assertTrue(module.isValid(query))
+ mic = test_mic.Mic(inputs)
+ module.handle(query, mic, self.profile)
+ return mic.outputs
+
+ def testLife(self):
+ query = "What is the meaning of life?"
+ inputs = []
+ outputs = self.runConversation(query, inputs, Life)
+ self.assertEqual(len(outputs), 1)
+ self.assertTrue("42" in outputs[0])
+
+ def testJoke(self):
+ query = "Tell me a joke."
+ inputs = ["Who's there?", "Random response"]
+ outputs = self.runConversation(query, inputs, Joke)
+ self.assertEqual(len(outputs), 3)
+ allJokes = open(jasperpath.data('text', 'JOKES.txt'), 'r').read()
+ self.assertTrue(outputs[2] in allJokes)
+
+ def testTime(self):
+ query = "What time is it?"
+ inputs = []
+ self.runConversation(query, inputs, Time)
+
+ @unittest.skipIf(not diagnose.check_network_connection(),
+ "No internet connection")
+ def testGmail(self):
+ key = 'gmail_password'
+ if key not in self.profile or not self.profile[key]:
+ return
+
+ query = "Check my email"
+ inputs = []
+ self.runConversation(query, inputs, Gmail)
+
+ @unittest.skipIf(not diagnose.check_network_connection(),
+ "No internet connection")
+ def testHN(self):
+ query = "find me some of the top hacker news stories"
+ if self.send:
+ inputs = ["the first and third"]
+ else:
+ inputs = ["no"]
+ outputs = self.runConversation(query, inputs, HN)
+ self.assertTrue("front-page articles" in outputs[1])
+
+ @unittest.skipIf(not diagnose.check_network_connection(),
+ "No internet connection")
+ def testNews(self):
+ query = "find me some of the top news stories"
+ if self.send:
+ inputs = ["the first"]
+ else:
+ inputs = ["no"]
+ outputs = self.runConversation(query, inputs, News)
+ self.assertTrue("top headlines" in outputs[1])
+
+ @unittest.skipIf(not diagnose.check_network_connection(),
+ "No internet connection")
+ def testWeather(self):
+ query = "what's the weather like tomorrow"
+ inputs = []
+ outputs = self.runConversation(query, inputs, Weather)
+ self.assertTrue(
+ "can't see that far ahead" in outputs[0]
+ or "Tomorrow" in outputs[0])
diff --git a/tests/test_stt.py b/tests/test_stt.py
new file mode 100644
index 0000000..cb33ff6
--- /dev/null
+++ b/tests/test_stt.py
@@ -0,0 +1,51 @@
+#!/usr/bin/env python2
+# -*- coding: utf-8-*-
+import unittest
+import imp
+from client import stt, jasperpath
+
+
+def cmuclmtk_installed():
+ try:
+ imp.find_module('cmuclmtk')
+ except ImportError:
+ return False
+ else:
+ return True
+
+
+def pocketsphinx_installed():
+ try:
+ imp.find_module('pocketsphinx')
+ except ImportError:
+ return False
+ else:
+ return True
+
+
+@unittest.skipUnless(cmuclmtk_installed(), "CMUCLMTK not present")
+@unittest.skipUnless(pocketsphinx_installed(), "Pocketsphinx not present")
+class TestSTT(unittest.TestCase):
+
+ def setUp(self):
+ self.jasper_clip = jasperpath.data('audio', 'jasper.wav')
+ self.time_clip = jasperpath.data('audio', 'time.wav')
+
+ self.passive_stt_engine = stt.PocketSphinxSTT.get_passive_instance()
+ self.active_stt_engine = stt.PocketSphinxSTT.get_active_instance()
+
+ def testTranscribeJasper(self):
+ """
+ Does Jasper recognize his name (i.e., passive listen)?
+ """
+ with open(self.jasper_clip, mode="rb") as f:
+ transcription = self.passive_stt_engine.transcribe(f)
+ self.assertIn("JASPER", transcription)
+
+ def testTranscribe(self):
+ """
+ Does Jasper recognize 'time' (i.e., active listen)?
+ """
+ with open(self.time_clip, mode="rb") as f:
+ transcription = self.active_stt_engine.transcribe(f)
+ self.assertIn("TIME", transcription)
diff --git a/tests/test_tts.py b/tests/test_tts.py
new file mode 100644
index 0000000..9894357
--- /dev/null
+++ b/tests/test_tts.py
@@ -0,0 +1,11 @@
+#!/usr/bin/env python2
+# -*- coding: utf-8-*-
+import unittest
+from client import tts
+
+
+class TestTTS(unittest.TestCase):
+ def testTTS(self):
+ tts_engine = tts.get_engine_by_slug('dummy-tts')
+ tts_instance = tts_engine()
+ tts_instance.say('This is a test.')
diff --git a/tests/test_vocabcompiler.py b/tests/test_vocabcompiler.py
new file mode 100644
index 0000000..f9e427f
--- /dev/null
+++ b/tests/test_vocabcompiler.py
@@ -0,0 +1,139 @@
+#!/usr/bin/env python2
+# -*- coding: utf-8-*-
+import unittest
+import tempfile
+import contextlib
+import logging
+import shutil
+import mock
+from client import vocabcompiler
+
+
+class TestVocabCompiler(unittest.TestCase):
+
+ def testPhraseExtraction(self):
+ expected_phrases = ['MOCK']
+
+ mock_module = mock.Mock()
+ mock_module.WORDS = ['MOCK']
+
+ with mock.patch('client.brain.Brain.get_modules',
+ classmethod(lambda cls: [mock_module])):
+ extracted_phrases = vocabcompiler.get_all_phrases()
+ self.assertEqual(expected_phrases, extracted_phrases)
+
+ def testKeywordPhraseExtraction(self):
+ expected_phrases = ['MOCK']
+
+ with tempfile.TemporaryFile() as f:
+ # We can't use mock_open here, because it doesn't seem to work
+ # with the 'for line in f' syntax
+ f.write("MOCK\n")
+ f.seek(0)
+ with mock.patch('%s.open' % vocabcompiler.__name__,
+ return_value=f, create=True):
+ extracted_phrases = vocabcompiler.get_keyword_phrases()
+ self.assertEqual(expected_phrases, extracted_phrases)
+
+
+class TestVocabulary(unittest.TestCase):
+ VOCABULARY = vocabcompiler.DummyVocabulary
+
+ @contextlib.contextmanager
+ def do_in_tempdir(self):
+ tempdir = tempfile.mkdtemp()
+ yield tempdir
+ shutil.rmtree(tempdir)
+
+ def testVocabulary(self):
+ phrases = ['GOOD BAD UGLY']
+ with self.do_in_tempdir() as tempdir:
+ self.vocab = self.VOCABULARY(path=tempdir)
+ self.assertIsNone(self.vocab.compiled_revision)
+ self.assertFalse(self.vocab.is_compiled)
+ self.assertFalse(self.vocab.matches_phrases(phrases))
+
+ # We're now testing error handling. To avoid flooding the
+ # output with error messages that are catched anyway,
+ # we'll temporarly disable logging. Otherwise, error log
+ # messages and traceback would be printed so that someone
+ # might think that tests failed even though they succeeded.
+ logging.disable(logging.ERROR)
+ with self.assertRaises(OSError):
+ with mock.patch('os.makedirs', side_effect=OSError('test')):
+ self.vocab.compile(phrases)
+ with self.assertRaises(OSError):
+ with mock.patch('%s.open' % vocabcompiler.__name__,
+ create=True,
+ side_effect=OSError('test')):
+ self.vocab.compile(phrases)
+
+ class StrangeCompilationError(Exception):
+ pass
+ with mock.patch.object(self.vocab, '_compile_vocabulary',
+ side_effect=StrangeCompilationError('test')
+ ):
+ with self.assertRaises(StrangeCompilationError):
+ self.vocab.compile(phrases)
+ with self.assertRaises(StrangeCompilationError):
+ with mock.patch('os.remove',
+ side_effect=OSError('test')):
+ self.vocab.compile(phrases)
+ # Re-enable logging again
+ logging.disable(logging.NOTSET)
+
+ self.vocab.compile(phrases)
+ self.assertIsInstance(self.vocab.compiled_revision, str)
+ self.assertTrue(self.vocab.is_compiled)
+ self.assertTrue(self.vocab.matches_phrases(phrases))
+ self.vocab.compile(phrases)
+ self.vocab.compile(phrases, force=True)
+
+
+@unittest.skipUnless(hasattr(vocabcompiler, 'cmuclmtk'),
+ "CMUCLMTK not present")
+class TestPocketsphinxVocabulary(TestVocabulary):
+
+ VOCABULARY = vocabcompiler.PocketsphinxVocabulary
+
+ def testVocabulary(self):
+ super(TestPocketsphinxVocabulary, self).testVocabulary()
+ self.assertIsInstance(self.vocab.decoder_kwargs, dict)
+ self.assertIn('lm', self.vocab.decoder_kwargs)
+ self.assertIn('dict', self.vocab.decoder_kwargs)
+
+
+class TestPatchedPocketsphinxVocabulary(TestPocketsphinxVocabulary):
+
+ def testVocabulary(self):
+
+ def write_test_vocab(text, output_file):
+ with open(output_file, "w") as f:
+ for word in text.split(' '):
+ f.write("%s\n" % word)
+
+ def write_test_lm(text, output_file, **kwargs):
+ with open(output_file, "w") as f:
+ f.write("TEST")
+
+ class DummyG2P(object):
+ def __init__(self, *args, **kwargs):
+ pass
+
+ @classmethod
+ def get_config(self, *args, **kwargs):
+ return {}
+
+ def translate(self, *args, **kwargs):
+ return {'GOOD': ['G UH D',
+ 'G UW D'],
+ 'BAD': ['B AE D'],
+ 'UGLY': ['AH G L IY']}
+
+ with mock.patch('client.vocabcompiler.cmuclmtk',
+ create=True) as mocked_cmuclmtk:
+ mocked_cmuclmtk.text2vocab = write_test_vocab
+ mocked_cmuclmtk.text2lm = write_test_lm
+ with mock.patch('client.vocabcompiler.PhonetisaurusG2P', DummyG2P):
+ super(TestPatchedPocketsphinxVocabulary,
+ self).testVocabulary()
generated by cgit v1.3.1 (git 2.54.0) at 2026-08-02 02:47:58 +0000