<s> """<STR_LIT>""" <EOL> import contextlib <EOL> import os <EOL> import StringIO <EOL> import sys <EOL> import unittest <EOL> import uuid <EOL> from pubsub_sample import main <EOL> TEST_PROJECT_ID_ENV = '<STR_LIT>' <EOL> DEFAULT_TEST_PROJECT_ID = "<STR_LIT>" <EOL> @ contextlib . contextmanager <EOL> def captured_output ( ) : <EOL> """<STR_LIT>""" <EOL> new_out , new_err = StringIO . StringIO ( ) , StringIO . StringIO ( ) <EOL> old_out , old_err = sys . stdout , sys . stderr <EOL> try : <EOL> sys . stdout , sys . stderr = new_out , new_err <EOL> yield sys . stdout , sys . stderr <EOL> finally : <EOL> sys . stdout , sys . stderr = old_out , old_err <EOL> def get_project_id ( ) : <EOL> """<STR_LIT>""" <EOL> return os . getenv ( TEST_PROJECT_ID_ENV , DEFAULT_TEST_PROJECT_ID ) <EOL> class PubsubSampleTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> """<STR_LIT>""" <EOL> random_id = uuid . uuid4 ( ) <EOL> cls . topic = '<STR_LIT>' % random_id <EOL> cls . sub = '<STR_LIT>' % random_id <EOL> main ( [ '<STR_LIT>' , get_project_id ( ) , '<STR_LIT>' , <EOL> cls . topic ] ) <EOL> main ( [ '<STR_LIT>' , get_project_id ( ) , '<STR_LIT>' , <EOL> cls . sub , cls . topic ] ) <EOL> cls . messages = [ '<STR_LIT>' % uuid . uuid4 ( ) , <EOL> '<STR_LIT>' % uuid . uuid4 ( ) , <EOL> '<STR_LIT>' ] <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> """<STR_LIT>""" <EOL> main ( [ '<STR_LIT>' , get_project_id ( ) , '<STR_LIT>' , cls . topic ] ) <EOL> main ( [ '<STR_LIT>' , get_project_id ( ) , '<STR_LIT>' , <EOL> cls . sub ] ) <EOL> def test_list_topics ( self ) : <EOL> """<STR_LIT>""" <EOL> expected_topic = ( '<STR_LIT>' <EOL> % ( get_project_id ( ) , self . topic ) ) <EOL> with captured_output ( ) as ( out , _ ) : <EOL> main ( [ '<STR_LIT>' , get_project_id ( ) , '<STR_LIT>' ] ) <EOL> output = out . getvalue ( ) . strip ( ) <EOL> self . assertTrue ( expected_topic in output ) <EOL> def test_list_subscriptions ( self ) : <EOL> """<STR_LIT>""" <EOL> expected_sub = ( '<STR_LIT>' <EOL> % ( get_project_id ( ) , self . sub ) ) <EOL> with captured_output ( ) as ( out , _ ) : <EOL> main ( [ '<STR_LIT>' , get_project_id ( ) , '<STR_LIT>' ] ) <EOL> output = out . getvalue ( ) . strip ( ) <EOL> self . assertTrue ( expected_sub in output ) <EOL> def test_publish_message ( self ) : <EOL> """<STR_LIT>""" <EOL> for message in self . messages : <EOL> with captured_output ( ) as ( out , _ ) : <EOL> main ( [ '<STR_LIT>' , get_project_id ( ) , '<STR_LIT>' , <EOL> self . topic , message ] ) <EOL> output = out . getvalue ( ) . strip ( ) <EOL> self . assertTrue ( message in output ) <EOL> def test_pull_message ( self ) : <EOL> """<STR_LIT>""" <EOL> with captured_output ( ) as ( out , _ ) : <EOL> main ( [ '<STR_LIT>' , get_project_id ( ) , '<STR_LIT>' , <EOL> self . sub , '<STR_LIT>' ] ) <EOL> output = out . getvalue ( ) . strip ( ) <EOL> for message in self . messages : <EOL> self . assertTrue ( message in output ) </s>
