# Copyright 2016 Google 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.

import unittest

import mock

def _make_credentials():

import google.auth.credentials

return mock.Mock(spec=google.auth.credentials.Credentials)

class TestGAPICSpeechAPI(unittest.TestCase):

@staticmethod

def _get_target_class():

from google.cloud.speech._gax import GAPICSpeechAPI

return GAPICSpeechAPI

def _make_one(self, *args, **kw):

return self._get_target_class()(*args, **kw)

@mock.patch('google.cloud._helpers.make_secure_channel',

return_value=mock.sentinel.channel)

@mock.patch(

'google.cloud.gapic.speech.v1.speech_client.SpeechClient.__init__',

return_value=None)

@mock.patch('google.cloud._helpers.make_secure_stub',

return_value=mock.sentinel.stub)

def test_constructor(self, mocked_stub, mocked_init, mocked_channel):

from google.longrunning import operations_grpc

from google.cloud._http import DEFAULT_USER_AGENT

from google.cloud.gapic.speech.v1.speech_client import SpeechClient

from google.cloud.speech import __version__

from google.cloud.speech._gax import OPERATIONS_API_HOST

credentials = _make_credentials()

mock_cnxn = mock.Mock(credentials=credentials, spec=['credentials'])

mock_client = mock.Mock(

_connection=mock_cnxn, _credentials=credentials,

spec=['_connection', '_credentials'])

speech_api = self._make_one(mock_client)

self.assertIs(speech_api._client, mock_client)

self.assertIsInstance(speech_api._gapic_api, SpeechClient)

mocked_stub.assert_called_once_with(

mock_cnxn.credentials, DEFAULT_USER_AGENT,

operations_grpc.OperationsStub, OPERATIONS_API_HOST)

mocked_init.assert_called_once_with(

channel=mock.sentinel.channel, lib_name='gccl',

lib_version=__version__)

mocked_channel.assert_called_once_with(

mock_cnxn.credentials, DEFAULT_USER_AGENT,

'speech.googleapis.com')

class TestSpeechGAXMakeRequests(unittest.TestCase):

SAMPLE_RATE = 16000

HINTS = ['hi']

AUDIO_CONTENT = b'/9j/4QNURXhpZgAASUkq'

def _call_fut(self, sample, language_code, max_alternatives,

profanity_filter, speech_contexts, single_utterance,

interim_results):

from google.cloud.speech._gax import _make_streaming_request

return _make_streaming_request(

sample=sample, language_code=language_code,

max_alternatives=max_alternatives,

profanity_filter=profanity_filter, speech_contexts=speech_contexts,

single_utterance=single_utterance, interim_results=interim_results)

def test_ctor(self):

from google.cloud import speech

from google.cloud.speech.sample import Sample

from google.cloud.proto.speech.v1.cloud_speech_pb2 import (

RecognitionConfig, SpeechContext, StreamingRecognitionConfig,

StreamingRecognizeRequest)

sample = Sample(

content=self.AUDIO_CONTENT, encoding=speech.Encoding.FLAC,

sample_rate_hertz=self.SAMPLE_RATE)

language_code = 'US-en'

max_alternatives = 2

profanity_filter = True

speech_contexts = [SpeechContext(phrases=self.HINTS)]

single_utterance = True

interim_results = False

streaming_request = self._call_fut(

sample, language_code, max_alternatives, profanity_filter,

speech_contexts, single_utterance, interim_results)

self.assertIsInstance(streaming_request, StreamingRecognizeRequest)

# This isn't set by _make_streaming_request().

# The first request can only have `streaming_config` set.

# The following requests can only have `audio_content` set.

self.assertEqual(streaming_request.audio_content, b'')

self.assertIsInstance(

streaming_request.streaming_config, StreamingRecognitionConfig)

streaming_config = streaming_request.streaming_config

self.assertTrue(streaming_config.single_utterance)

self.assertFalse(streaming_config.interim_results)

config = streaming_config.config

self.assertIsInstance(config, RecognitionConfig)

self.assertEqual(config.encoding, 2) # speech.Encoding.FLAC maps to 2.

self.assertEqual(config.sample_rate_hertz, self.SAMPLE_RATE)

self.assertEqual(config.language_code, language_code)

self.assertEqual(config.max_alternatives, max_alternatives)

self.assertTrue(config.profanity_filter)

self.assertEqual(config.speech_contexts[0].phrases, self.HINTS)

class TestSpeechGAXMakeRequestsStream(unittest.TestCase):

SAMPLE_RATE = 16000

HINTS = ['hi']

AUDIO_CONTENT = b'/9j/4QNURXhpZgAASUkq'

def _call_fut(self, sample, language_code, max_alternatives,

profanity_filter, speech_contexts, single_utterance,

interim_results):

from google.cloud.speech._gax import _stream_requests

return _stream_requests(

sample=sample, language_code=language_code,

max_alternatives=max_alternatives,

profanity_filter=profanity_filter, speech_contexts=speech_contexts,

single_utterance=single_utterance, interim_results=interim_results)

def test_stream_requests(self):

from io import BytesIO

from google.cloud import speech

from google.cloud.speech.sample import Sample

from google.cloud.proto.speech.v1.cloud_speech_pb2 import (

StreamingRecognitionConfig, StreamingRecognizeRequest)

sample = Sample(

stream=BytesIO(self.AUDIO_CONTENT), encoding=speech.Encoding.FLAC,

sample_rate_hertz=self.SAMPLE_RATE)

language_code = 'US-en'

max_alternatives = 2

profanity_filter = True

speech_contexts = self.HINTS

single_utterance = True

interim_results = False

streaming_requests = self._call_fut(

sample, language_code, max_alternatives, profanity_filter,

speech_contexts, single_utterance, interim_results)

all_requests = []

for streaming_request in streaming_requests:

self.assertIsInstance(streaming_request, StreamingRecognizeRequest)

all_requests.append(streaming_request)

self.assertEqual(len(all_requests), 2)

config_request = all_requests[0]

streaming_request = all_requests[1]

# This isn't set by _make_streaming_request().

# The first request can only have `streaming_config` set.

# The following requests can only have `audio_content` set.

self.assertEqual(config_request.audio_content, b'')

self.assertEqual(streaming_request.audio_content, self.AUDIO_CONTENT)

self.assertIsInstance(

config_request.streaming_config, StreamingRecognitionConfig)