Source code for apache_beam.ml.gcp.naturallanguageml_test_it

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.
#
# pytype: skip-file

from __future__ import absolute_import

import unittest

from nose.plugins.attrib import attr

import apache_beam as beam
from apache_beam.testing.test_pipeline import TestPipeline
from apache_beam.testing.util import assert_that
from apache_beam.testing.util import equal_to

# Protect against environments where Google Cloud Natural Language client is
# not available.
try:
  from apache_beam.ml.gcp.naturallanguageml import AnnotateText
  from apache_beam.ml.gcp.naturallanguageml import Document
  from apache_beam.ml.gcp.naturallanguageml import enums
  from apache_beam.ml.gcp.naturallanguageml import types
except ImportError:
  AnnotateText = None


[docs]def extract(response): yield beam.pvalue.TaggedOutput('language', response.language) yield beam.pvalue.TaggedOutput( 'parts_of_speech', [ enums.PartOfSpeech.Tag(x.part_of_speech.tag).name for x in response.tokens ])
[docs]@attr('IT') @unittest.skipIf(AnnotateText is None, 'GCP dependencies are not installed') class NaturalLanguageMlTestIT(unittest.TestCase):
[docs] def test_analyzing_syntax(self): with TestPipeline(is_integration_test=True) as p: output = ( p | beam.Create([Document('Unified programming model.')]) | AnnotateText( types.AnnotateTextRequest.Features(extract_syntax=True)) | beam.ParDo(extract).with_outputs('language', 'parts_of_speech')) assert_that(output.language, equal_to(['en']), label='verify_language') assert_that( output.parts_of_speech, equal_to([['ADJ', 'NOUN', 'NOUN', 'PUNCT']]), label='verify_parts_of_speech')
if __name__ == '__main__': unittest.main()