test_sr_nict.py
Go to the documentation of this file.
00001 #!/usr/bin/env python
00002 # -*- coding: utf-8 -*-
00003 
00004 # import python libraries
00005 import os
00006 import logging
00007 import unittest
00008 import ConfigParser
00009 import time
00010 
00011 from nose.tools import raises
00012 from rospeex_core.sr.base.session import PacketType
00013 from rospeex_core.sr.nict.sync_client import SyncClient
00014 from rospeex_core.sr.nict.mime_message import MIMEMessage
00015 from rospeex_core.sr.nict.streaming_client import StreamingClient
00016 from rospeex_core import exceptions as ext
00017 
00018 
00019 # setup logging
00020 FORMAT = '[%(levelname)s] %(message)s @%(filename)s:%(funcName)s:%(lineno)d'
00021 logging.basicConfig(format=FORMAT, level=logging.DEBUG)
00022 logger = logging.getLogger(__name__)
00023 
00024 
00025 class TestSyncClient(unittest.TestCase):
00026     def setUp(self):
00027         base_dir = os.path.dirname(__file__)
00028         filename = os.path.join(base_dir, 'config.cfg')
00029         settings = ConfigParser.ConfigParser()
00030         settings.read(filename)
00031         self.flac_file = os.path.join(
00032             base_dir,
00033             settings.get('SpeechRecognition', 'flac_file')
00034         )
00035         self.broken_wav_file = os.path.join(
00036             base_dir,
00037             settings.get('SpeechRecognition', 'broken_wav_file')
00038         )
00039         self.wav_file = os.path.join(
00040             base_dir,
00041             settings.get('SpeechRecognition', 'wav_file')
00042         )
00043 
00044         multi_lang_wav = settings.get('SpeechRecognition', 'wav_hello')
00045         self.wav_hello = self.get_opt_subsection(multi_lang_wav.splitlines())
00046 
00047         for key in self.wav_hello.keys():
00048             f = self.wav_hello[key]
00049             self.wav_hello[key] = os.path.join(
00050                     base_dir,
00051                     f)
00052 
00053         multi_lang_text = settings.get('SpeechRecognition', 'text_hello')
00054         self.text_hello = self.get_opt_subsection(multi_lang_text.splitlines())
00055         for key in self.text_hello.keys():
00056             t = self.text_hello[key]
00057             self.text_hello[key] = unicode(t, 'utf-8')
00058 
00059         self._streaming_result = None
00060 
00061     def get_opt_subsection(self, lines):
00062         conf = {}
00063         for item in lines:
00064             pair = item.split(':')
00065             if pair[0]:
00066                 conf[pair[0]] = pair[1].strip()
00067         return conf
00068 
00069     @raises(ext.ParameterException)
00070     def test_request_invalid_audio_format_flac(self):
00071         request_data = open(self.flac_file, 'rb').read()
00072         client = SyncClient()
00073         client.request(request_data)
00074 
00075     @raises(ext.ParameterException)
00076     def test_request_invalid_audio_format_broken_wav(self):
00077         language = 'ja'
00078         request_data = open(self.broken_wav_file, 'rb').read()
00079         client = SyncClient()
00080         client.request(data=request_data, language=language)
00081 
00082     @raises(ext.ParameterException)
00083     def test_request_invalid_language(self):
00084         language = 'hoge'
00085         request_data = open(self.wav_file, 'rb').read()
00086         client = SyncClient()
00087         client.request(data=request_data, language=language)
00088 
00089     @raises(ext.RequestTimeoutException)
00090     def test_request_invalid_request_timeout(self):
00091         language = 'ja'
00092         request_data = open(self.wav_file, 'rb').read()
00093         client = SyncClient()
00094         client.request(data=request_data, language=language, timeout=1)
00095 
00096     def test_request_valid_language(self):
00097         for language in SyncClient.LANGUAGES:
00098             request_data = open(self.wav_hello[language], 'rb').read()
00099             client = SyncClient()
00100             msg = client.request(data=request_data, language=language)
00101             logger.info(msg)
00102             assert msg == self.text_hello[language]
00103 
00104     def test_support_streaming(self):
00105         client = SyncClient()
00106         assert not client.support_streaming()
00107 
00108 
00109 class TestMIMEMessage(unittest.TestCase):
00110     def setUp(self):
00111         pass
00112 
00113     def tearDown(self):
00114         pass
00115 
00116     def test_valid_mime_message(self):
00117         MIMEMessage.create_http_header()
00118         MIMEMessage.create_http_header('HOGE')
00119         MIMEMessage.create_header_request('ja', 1)
00120         MIMEMessage.create_header_request('en', 100)
00121         MIMEMessage.create_finish_request()
00122         MIMEMessage.create_data_request('hoge')
00123         MIMEMessage.create_bundle_request('hoge', 'hoge')
00124 
00125     @raises(Exception)
00126     def test_invalid_create_data_request(self):
00127         MIMEMessage.create_data_request(None)
00128 
00129     @raises(Exception)
00130     def test_invalid_create_bundle_request(self):
00131         MIMEMessage.create_bundle_request('hoge', None)
00132 
00133 
00134 class TestStreamingClient(unittest.TestCase):
00135     def setUp(self):
00136         base_dir = os.path.dirname(__file__)
00137         filename = os.path.join(base_dir, 'config.cfg')
00138         settings = ConfigParser.ConfigParser()
00139         settings.read(filename)
00140         self.flac_file = os.path.join(
00141             base_dir,
00142             settings.get('SpeechRecognition', 'flac_file')
00143         )
00144         self.broken_wav_file = os.path.join(
00145             base_dir,
00146             settings.get('SpeechRecognition', 'broken_wav_file')
00147         )
00148         self.wav_file = os.path.join(
00149             base_dir,
00150             settings.get('SpeechRecognition', 'wav_file')
00151         )
00152 
00153         multi_lang_wav = settings.get('SpeechRecognition', 'wav_hello')
00154         self.wav_hello = self.get_opt_subsection(multi_lang_wav.splitlines())
00155 
00156         for key in self.wav_hello.keys():
00157             f = self.wav_hello[key]
00158             self.wav_hello[key] = os.path.join(
00159                     base_dir,
00160                     f)
00161 
00162         multi_lang_text = settings.get('SpeechRecognition', 'text_hello')
00163         self.text_hello = self.get_opt_subsection(multi_lang_text.splitlines())
00164         for key in self.text_hello.keys():
00165             t = self.text_hello[key]
00166             self.text_hello[key] = unicode(t, 'utf-8')
00167 
00168     def get_opt_subsection(self, lines):
00169         conf = {}
00170         for item in lines:
00171             pair = item.split(':')
00172             if pair[0]:
00173                 conf[pair[0]] = pair[1].strip()
00174         return conf
00175 
00176     @raises(ext.ParameterException)
00177     def test_request_invalid_audio_format_flac(self):
00178         request_data = open(self.flac_file, 'rb').read()
00179         client = StreamingClient()
00180         client.request(request_data)
00181 
00182     @raises(ext.ParameterException)
00183     def test_request_invalid_audio_format_broken_wav(self):
00184         language = 'ja'
00185         request_data = open(self.broken_wav_file, 'rb').read()
00186         client = StreamingClient()
00187         client.request(data=request_data, language=language)
00188 
00189     @raises(ext.ParameterException)
00190     def test_request_invalid_language(self):
00191         language = 'hoge'
00192         request_data = open(self.wav_file, 'rb').read()
00193         client = StreamingClient()
00194         client.request(data=request_data, language=language)
00195 
00196     @raises(ext.RequestTimeoutException)
00197     def test_request_invalid_request_timeout(self):
00198         language = 'ja'
00199         request_data = open(self.wav_file, 'rb').read()
00200         client = StreamingClient()
00201         client.request(data=request_data, language=language, timeout=1)
00202 
00203     def test_request_valid_language(self):
00204         for language in StreamingClient.LANGUAGES:
00205             request_data = open(self.wav_file, 'rb').read()
00206             client = StreamingClient()
00207             msg = client.request(data=request_data, language=language)
00208             logger.info(msg)
00209 
00210     def test_support_streaming(self):
00211         client = StreamingClient()
00212         assert client.support_streaming()
00213 
00214     def _streaming_cb(self, msg):
00215         self._streaming_result = msg
00216 
00217     def test_add_streaming_packet(self):
00218         for language in StreamingClient.LANGUAGES:
00219             request_data = open(self.wav_hello[language], 'rb').read()
00220             client = StreamingClient(language)
00221             client.set_streaming_config(language)
00222             client.register_streaming_cb(self._streaming_cb)
00223 
00224             client.add_streaming_packet(PacketType.START, None)
00225             client.add_streaming_packet(PacketType.DATA, request_data)
00226             client.add_streaming_packet(PacketType.END, None)
00227             time.sleep(10)
00228 
00229             logger.info(self._streaming_result)
00230             assert self._streaming_result == self.text_hello[language]
00231 
00232 if __name__ == '__main__':
00233     import rosunit
00234     rosunit.unitrun(
00235         'rospeex_core',
00236         'sr_nict_sync_client',
00237         TestSyncClient,
00238         None,
00239         coverage_packages=['rospeex_core.sr']
00240     )
00241 
00242     rosunit.unitrun(
00243         'rospeex_core',
00244         'sr_nict_streaming_client',
00245         TestStreamingClient,
00246         None,
00247         coverage_packages=['rospeex_core.sr']
00248     )
00249 
00250     rosunit.unitrun(
00251         'rospeex_core',
00252         'sr_nict_mime_message',
00253         TestMIMEMessage,
00254         None,
00255         coverage_packages=['rospeex_core.sr']
00256     )


rospeex_core
Author(s): Komei Sugiura
autogenerated on Thu Apr 20 2017 03:08:53