<s> import unittest2 <EOL> class TestClient ( unittest2 . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from gcloud . bigtable . client import Client <EOL> return Client <EOL> def _makeOne ( self , * args , ** kwargs ) : <EOL> return self . _getTargetClass ( ) ( * args , ** kwargs ) <EOL> def _constructor_test_helper ( self , expected_scopes , creds , <EOL> read_only = False , admin = False , <EOL> user_agent = None , timeout_seconds = None , <EOL> expected_creds = None ) : <EOL> from gcloud . bigtable import client as MUT <EOL> user_agent = user_agent or MUT . DEFAULT_USER_AGENT <EOL> timeout_seconds = timeout_seconds or MUT . DEFAULT_TIMEOUT_SECONDS <EOL> PROJECT = '<STR_LIT>' <EOL> client = self . _makeOne ( project = PROJECT , credentials = creds , <EOL> read_only = read_only , admin = admin , <EOL> user_agent = user_agent , <EOL> timeout_seconds = timeout_seconds ) <EOL> expected_creds = expected_creds or creds <EOL> self . assertTrue ( client . _credentials is expected_creds ) <EOL> if expected_scopes is not None : <EOL> self . assertEqual ( client . _credentials . scopes , expected_scopes ) <EOL> self . assertEqual ( client . project , PROJECT ) <EOL> self . assertEqual ( client . timeout_seconds , timeout_seconds ) <EOL> self . assertEqual ( client . user_agent , user_agent ) <EOL> self . assertEqual ( client . _data_stub_internal , None ) <EOL> self . assertEqual ( client . _cluster_stub_internal , None ) <EOL> self . assertEqual ( client . _operations_stub_internal , None ) <EOL> self . assertEqual ( client . _table_stub_internal , None ) <EOL> def test_constructor_default_scopes ( self ) : <EOL> from gcloud . bigtable import client as MUT <EOL> expected_scopes = [ MUT . DATA_SCOPE ] <EOL> creds = _Credentials ( ) <EOL> self . _constructor_test_helper ( expected_scopes , creds ) <EOL> def test_constructor_custom_user_agent_and_timeout ( self ) : <EOL> from gcloud . bigtable import client as MUT <EOL> timeout_seconds = <NUM_LIT> <EOL> user_agent = '<STR_LIT>' <EOL> expected_scopes = [ MUT . DATA_SCOPE ] <EOL> creds = _Credentials ( ) <EOL> self . _constructor_test_helper ( expected_scopes , creds , <EOL> user_agent = user_agent , <EOL> timeout_seconds = timeout_seconds ) <EOL> def test_constructor_with_admin ( self ) : <EOL> from gcloud . bigtable import client as MUT <EOL> expected_scopes = [ MUT . DATA_SCOPE , MUT . ADMIN_SCOPE ] <EOL> creds = _Credentials ( ) <EOL> self . _constructor_test_helper ( expected_scopes , creds , admin = True ) <EOL> def test_constructor_with_read_only ( self ) : <EOL> from gcloud . bigtable import client as MUT <EOL> expected_scopes = [ MUT . READ_ONLY_SCOPE ] <EOL> creds = _Credentials ( ) <EOL> self . _constructor_test_helper ( expected_scopes , creds , read_only = True ) <EOL> def test_constructor_both_admin_and_read_only ( self ) : <EOL> creds = _Credentials ( ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . _constructor_test_helper ( [ ] , creds , admin = True , <EOL> read_only = True ) <EOL> def test_constructor_implicit_credentials ( self ) : <EOL> from gcloud . _testing import _Monkey <EOL> from gcloud . bigtable import client as MUT <EOL> creds = _Credentials ( ) <EOL> expected_scopes = [ MUT . DATA_SCOPE ] <EOL> def mock_get_credentials ( ) : <EOL> return creds <EOL> with _Monkey ( MUT , get_credentials = mock_get_credentials ) : <EOL> self . _constructor_test_helper ( expected_scopes , None , <EOL> expected_creds = creds ) <EOL> def test_constructor_credentials_wo_create_scoped ( self ) : <EOL> creds = object ( ) <EOL> expected_scopes = None <EOL> self . _constructor_test_helper ( expected_scopes , creds ) <EOL> def _copy_test_helper ( self , read_only = False , admin = False ) : <EOL> credentials = _Credentials ( '<STR_LIT:value>' ) <EOL> project = '<STR_LIT>' <EOL> timeout_seconds = <NUM_LIT> <EOL> user_agent = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> read_only = read_only , admin = admin , <EOL> timeout_seconds = timeout_seconds , <EOL> user_agent = user_agent ) <EOL> client . _data_stub_internal = object ( ) <EOL> client . _cluster_stub_internal = object ( ) <EOL> client . _operations_stub_internal = object ( ) <EOL> client . _table_stub_internal = object ( ) <EOL> new_client = client . copy ( ) <EOL> self . assertEqual ( new_client . _admin , client . _admin ) <EOL> self . assertEqual ( new_client . _credentials , client . _credentials ) <EOL> self . assertEqual ( new_client . project , client . project ) <EOL> self . assertEqual ( new_client . user_agent , client . user_agent ) <EOL> self . assertEqual ( new_client . timeout_seconds , client . timeout_seconds ) <EOL> self . assertEqual ( new_client . _data_stub_internal , None ) <EOL> self . assertEqual ( new_client . _cluster_stub_internal , None ) <EOL> self . assertEqual ( new_client . _operations_stub_internal , None ) <EOL> self . assertEqual ( new_client . _table_stub_internal , None ) <EOL> def test_copy ( self ) : <EOL> self . _copy_test_helper ( ) <EOL> def test_copy_admin ( self ) : <EOL> self . _copy_test_helper ( admin = True ) <EOL> def test_copy_read_only ( self ) : <EOL> self . _copy_test_helper ( read_only = True ) <EOL> def test_credentials_getter ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> self . assertTrue ( client . credentials is credentials ) <EOL> def test_project_name_property ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> project_name = '<STR_LIT>' + project <EOL> self . assertEqual ( client . project_name , project_name ) <EOL> def test_data_stub_getter ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> client . _data_stub_internal = object ( ) <EOL> self . assertTrue ( client . _data_stub is client . _data_stub_internal ) <EOL> def test_data_stub_failure ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> with self . assertRaises ( ValueError ) : <EOL> getattr ( client , '<STR_LIT>' ) <EOL> def test_cluster_stub_getter ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = True ) <EOL> client . _cluster_stub_internal = object ( ) <EOL> self . assertTrue ( client . _cluster_stub is client . _cluster_stub_internal ) <EOL> def test_cluster_stub_non_admin_failure ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = False ) <EOL> with self . assertRaises ( ValueError ) : <EOL> getattr ( client , '<STR_LIT>' ) <EOL> def test_cluster_stub_unset_failure ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = True ) <EOL> with self . assertRaises ( ValueError ) : <EOL> getattr ( client , '<STR_LIT>' ) <EOL> def test_operations_stub_getter ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = True ) <EOL> client . _operations_stub_internal = object ( ) <EOL> self . assertTrue ( client . _operations_stub is <EOL> client . _operations_stub_internal ) <EOL> def test_operations_stub_non_admin_failure ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = False ) <EOL> with self . assertRaises ( ValueError ) : <EOL> getattr ( client , '<STR_LIT>' ) <EOL> def test_operations_stub_unset_failure ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = True ) <EOL> with self . assertRaises ( ValueError ) : <EOL> getattr ( client , '<STR_LIT>' ) <EOL> def test_table_stub_getter ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = True ) <EOL> client . _table_stub_internal = object ( ) <EOL> self . assertTrue ( client . _table_stub is client . _table_stub_internal ) <EOL> def test_table_stub_non_admin_failure ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = False ) <EOL> with self . assertRaises ( ValueError ) : <EOL> getattr ( client , '<STR_LIT>' ) <EOL> def test_table_stub_unset_failure ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = True ) <EOL> with self . assertRaises ( ValueError ) : <EOL> getattr ( client , '<STR_LIT>' ) <EOL> def test__make_data_stub ( self ) : <EOL> from gcloud . _testing import _Monkey <EOL> from gcloud . bigtable import client as MUT <EOL> from gcloud . bigtable . client import DATA_API_HOST <EOL> from gcloud . bigtable . client import DATA_API_PORT <EOL> from gcloud . bigtable . client import DATA_STUB_FACTORY <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> fake_stub = object ( ) <EOL> make_stub_args = [ ] <EOL> def mock_make_stub ( * args ) : <EOL> make_stub_args . append ( args ) <EOL> return fake_stub <EOL> with _Monkey ( MUT , _make_stub = mock_make_stub ) : <EOL> result = client . _make_data_stub ( ) <EOL> self . assertTrue ( result is fake_stub ) <EOL> self . assertEqual ( make_stub_args , [ <EOL> ( <EOL> client , <EOL> DATA_STUB_FACTORY , <EOL> DATA_API_HOST , <EOL> DATA_API_PORT , <EOL> ) , <EOL> ] ) <EOL> def test__make_cluster_stub ( self ) : <EOL> from gcloud . _testing import _Monkey <EOL> from gcloud . bigtable import client as MUT <EOL> from gcloud . bigtable . client import CLUSTER_ADMIN_HOST <EOL> from gcloud . bigtable . client import CLUSTER_ADMIN_PORT <EOL> from gcloud . bigtable . client import CLUSTER_STUB_FACTORY <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> fake_stub = object ( ) <EOL> make_stub_args = [ ] <EOL> def mock_make_stub ( * args ) : <EOL> make_stub_args . append ( args ) <EOL> return fake_stub <EOL> with _Monkey ( MUT , _make_stub = mock_make_stub ) : <EOL> result = client . _make_cluster_stub ( ) <EOL> self . assertTrue ( result is fake_stub ) <EOL> self . assertEqual ( make_stub_args , [ <EOL> ( <EOL> client , <EOL> CLUSTER_STUB_FACTORY , <EOL> CLUSTER_ADMIN_HOST , <EOL> CLUSTER_ADMIN_PORT , <EOL> ) , <EOL> ] ) <EOL> def test__make_operations_stub ( self ) : <EOL> from gcloud . _testing import _Monkey <EOL> from gcloud . bigtable import client as MUT <EOL> from gcloud . bigtable . client import CLUSTER_ADMIN_HOST <EOL> from gcloud . bigtable . client import CLUSTER_ADMIN_PORT <EOL> from gcloud . bigtable . client import OPERATIONS_STUB_FACTORY <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> fake_stub = object ( ) <EOL> make_stub_args = [ ] <EOL> def mock_make_stub ( * args ) : <EOL> make_stub_args . append ( args ) <EOL> return fake_stub <EOL> with _Monkey ( MUT , _make_stub = mock_make_stub ) : <EOL> result = client . _make_operations_stub ( ) <EOL> self . assertTrue ( result is fake_stub ) <EOL> self . assertEqual ( make_stub_args , [ <EOL> ( <EOL> client , <EOL> OPERATIONS_STUB_FACTORY , <EOL> CLUSTER_ADMIN_HOST , <EOL> CLUSTER_ADMIN_PORT , <EOL> ) , <EOL> ] ) <EOL> def test__make_table_stub ( self ) : <EOL> from gcloud . _testing import _Monkey <EOL> from gcloud . bigtable import client as MUT <EOL> from gcloud . bigtable . client import TABLE_ADMIN_HOST <EOL> from gcloud . bigtable . client import TABLE_ADMIN_PORT <EOL> from gcloud . bigtable . client import TABLE_STUB_FACTORY <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> fake_stub = object ( ) <EOL> make_stub_args = [ ] <EOL> def mock_make_stub ( * args ) : <EOL> make_stub_args . append ( args ) <EOL> return fake_stub <EOL> with _Monkey ( MUT , _make_stub = mock_make_stub ) : <EOL> result = client . _make_table_stub ( ) <EOL> self . assertTrue ( result is fake_stub ) <EOL> self . assertEqual ( make_stub_args , [ <EOL> ( <EOL> client , <EOL> TABLE_STUB_FACTORY , <EOL> TABLE_ADMIN_HOST , <EOL> TABLE_ADMIN_PORT , <EOL> ) , <EOL> ] ) <EOL> def test_is_started ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> self . assertFalse ( client . is_started ( ) ) <EOL> client . _data_stub_internal = object ( ) <EOL> self . assertTrue ( client . is_started ( ) ) <EOL> client . _data_stub_internal = None <EOL> self . assertFalse ( client . is_started ( ) ) <EOL> def _start_method_helper ( self , admin ) : <EOL> from gcloud . _testing import _Monkey <EOL> from gcloud . bigtable . _testing import _FakeStub <EOL> from gcloud . bigtable import client as MUT <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = admin ) <EOL> stub = _FakeStub ( ) <EOL> make_stub_args = [ ] <EOL> def mock_make_stub ( * args ) : <EOL> make_stub_args . append ( args ) <EOL> return stub <EOL> with _Monkey ( MUT , _make_stub = mock_make_stub ) : <EOL> client . start ( ) <EOL> self . assertTrue ( client . _data_stub_internal is stub ) <EOL> if admin : <EOL> self . assertTrue ( client . _cluster_stub_internal is stub ) <EOL> self . assertTrue ( client . _operations_stub_internal is stub ) <EOL> self . assertTrue ( client . _table_stub_internal is stub ) <EOL> self . assertEqual ( stub . _entered , <NUM_LIT:4> ) <EOL> self . assertEqual ( len ( make_stub_args ) , <NUM_LIT:4> ) <EOL> else : <EOL> self . assertTrue ( client . _cluster_stub_internal is None ) <EOL> self . assertTrue ( client . _operations_stub_internal is None ) <EOL> self . assertTrue ( client . _table_stub_internal is None ) <EOL> self . assertEqual ( stub . _entered , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( make_stub_args ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( stub . _exited , [ ] ) <EOL> def test_start_non_admin ( self ) : <EOL> self . _start_method_helper ( admin = False ) <EOL> def test_start_with_admin ( self ) : <EOL> self . _start_method_helper ( admin = True ) <EOL> def test_start_while_started ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> client . _data_stub_internal = data_stub = object ( ) <EOL> self . assertTrue ( client . is_started ( ) ) <EOL> client . start ( ) <EOL> self . assertEqual ( client . _data_stub_internal , data_stub ) <EOL> def _stop_method_helper ( self , admin ) : <EOL> from gcloud . bigtable . _testing import _FakeStub <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = admin ) <EOL> stub1 = _FakeStub ( ) <EOL> stub2 = _FakeStub ( ) <EOL> client . _data_stub_internal = stub1 <EOL> client . _cluster_stub_internal = stub2 <EOL> client . _operations_stub_internal = stub2 <EOL> client . _table_stub_internal = stub2 <EOL> client . stop ( ) <EOL> self . assertTrue ( client . _data_stub_internal is None ) <EOL> self . assertTrue ( client . _cluster_stub_internal is None ) <EOL> self . assertTrue ( client . _operations_stub_internal is None ) <EOL> self . assertTrue ( client . _table_stub_internal is None ) <EOL> self . assertEqual ( stub1 . _entered , <NUM_LIT:0> ) <EOL> self . assertEqual ( stub2 . _entered , <NUM_LIT:0> ) <EOL> exc_none_triple = ( None , None , None ) <EOL> self . assertEqual ( stub1 . _exited , [ exc_none_triple ] ) <EOL> if admin : <EOL> self . assertEqual ( stub2 . _exited , [ exc_none_triple ] * <NUM_LIT:3> ) <EOL> else : <EOL> self . assertEqual ( stub2 . _exited , [ ] ) <EOL> def test_stop_non_admin ( self ) : <EOL> self . _stop_method_helper ( admin = False ) <EOL> def test_stop_with_admin ( self ) : <EOL> self . _stop_method_helper ( admin = True ) <EOL> def test_stop_while_stopped ( self ) : <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> self . assertFalse ( client . is_started ( ) ) <EOL> client . _cluster_stub_internal = cluster_stub = object ( ) <EOL> client . stop ( ) <EOL> self . assertEqual ( client . _cluster_stub_internal , cluster_stub ) <EOL> def test_cluster_factory ( self ) : <EOL> from gcloud . bigtable . cluster import Cluster <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> client = self . _makeOne ( project = project , credentials = credentials ) <EOL> zone = '<STR_LIT>' <EOL> cluster_id = '<STR_LIT>' <EOL> display_name = '<STR_LIT>' <EOL> serve_nodes = <NUM_LIT> <EOL> cluster = client . cluster ( zone , cluster_id , display_name = display_name , <EOL> serve_nodes = serve_nodes ) <EOL> self . assertTrue ( isinstance ( cluster , Cluster ) ) <EOL> self . assertEqual ( cluster . zone , zone ) <EOL> self . assertEqual ( cluster . cluster_id , cluster_id ) <EOL> self . assertEqual ( cluster . display_name , display_name ) <EOL> self . assertEqual ( cluster . serve_nodes , serve_nodes ) <EOL> self . assertTrue ( cluster . _client is client ) <EOL> def _list_zones_helper ( self , zone_status ) : <EOL> from gcloud . bigtable . _generated import ( <EOL> bigtable_cluster_data_pb2 as data_pb2 ) <EOL> from gcloud . bigtable . _generated import ( <EOL> bigtable_cluster_service_messages_pb2 as messages_pb2 ) <EOL> from gcloud . bigtable . _testing import _FakeStub <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> timeout_seconds = <NUM_LIT> <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = True , timeout_seconds = timeout_seconds ) <EOL> request_pb = messages_pb2 . ListZonesRequest ( <EOL> name = '<STR_LIT>' + project , <EOL> ) <EOL> zone1 = '<STR_LIT:foo>' <EOL> zone2 = '<STR_LIT:bar>' <EOL> response_pb = messages_pb2 . ListZonesResponse ( <EOL> zones = [ <EOL> data_pb2 . Zone ( display_name = zone1 , status = zone_status ) , <EOL> data_pb2 . Zone ( display_name = zone2 , status = zone_status ) , <EOL> ] , <EOL> ) <EOL> client . _cluster_stub_internal = stub = _FakeStub ( response_pb ) <EOL> expected_result = [ zone1 , zone2 ] <EOL> result = client . list_zones ( ) <EOL> self . assertEqual ( result , expected_result ) <EOL> self . assertEqual ( stub . method_calls , [ ( <EOL> '<STR_LIT>' , <EOL> ( request_pb , timeout_seconds ) , <EOL> { } , <EOL> ) ] ) <EOL> def test_list_zones ( self ) : <EOL> from gcloud . bigtable . _generated import ( <EOL> bigtable_cluster_data_pb2 as data_pb2 ) <EOL> self . _list_zones_helper ( data_pb2 . Zone . OK ) <EOL> def test_list_zones_failure ( self ) : <EOL> from gcloud . bigtable . _generated import ( <EOL> bigtable_cluster_data_pb2 as data_pb2 ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . _list_zones_helper ( data_pb2 . Zone . EMERGENCY_MAINENANCE ) <EOL> def test_list_clusters ( self ) : <EOL> from gcloud . bigtable . _generated import ( <EOL> bigtable_cluster_data_pb2 as data_pb2 ) <EOL> from gcloud . bigtable . _generated import ( <EOL> bigtable_cluster_service_messages_pb2 as messages_pb2 ) <EOL> from gcloud . bigtable . _testing import _FakeStub <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> timeout_seconds = <NUM_LIT> <EOL> client = self . _makeOne ( project = project , credentials = credentials , <EOL> admin = True , timeout_seconds = timeout_seconds ) <EOL> request_pb = messages_pb2 . ListClustersRequest ( <EOL> name = '<STR_LIT>' + project , <EOL> ) <EOL> zone = '<STR_LIT:foo>' <EOL> failed_zone = '<STR_LIT:bar>' <EOL> cluster_id1 = '<STR_LIT>' <EOL> cluster_id2 = '<STR_LIT>' <EOL> cluster_name1 = ( '<STR_LIT>' + project + '<STR_LIT>' + zone + <EOL> '<STR_LIT>' + cluster_id1 ) <EOL> cluster_name2 = ( '<STR_LIT>' + project + '<STR_LIT>' + zone + <EOL> '<STR_LIT>' + cluster_id2 ) <EOL> response_pb = messages_pb2 . ListClustersResponse ( <EOL> failed_zones = [ <EOL> data_pb2 . Zone ( display_name = failed_zone ) , <EOL> ] , <EOL> clusters = [ <EOL> data_pb2 . Cluster ( <EOL> name = cluster_name1 , <EOL> display_name = cluster_name1 , <EOL> serve_nodes = <NUM_LIT:3> , <EOL> ) , <EOL> data_pb2 . Cluster ( <EOL> name = cluster_name2 , <EOL> display_name = cluster_name2 , <EOL> serve_nodes = <NUM_LIT:3> , <EOL> ) , <EOL> ] , <EOL> ) <EOL> client . _cluster_stub_internal = stub = _FakeStub ( response_pb ) <EOL> failed_zones = [ failed_zone ] <EOL> clusters = [ <EOL> client . cluster ( zone , cluster_id1 ) , <EOL> client . cluster ( zone , cluster_id2 ) , <EOL> ] <EOL> expected_result = ( clusters , failed_zones ) <EOL> result = client . list_clusters ( ) <EOL> self . assertEqual ( result , expected_result ) <EOL> self . assertEqual ( stub . method_calls , [ ( <EOL> '<STR_LIT>' , <EOL> ( request_pb , timeout_seconds ) , <EOL> { } , <EOL> ) ] ) <EOL> class Test_MetadataPlugin ( unittest2 . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from gcloud . bigtable . client import _MetadataPlugin <EOL> return _MetadataPlugin <EOL> def _makeOne ( self , * args , ** kwargs ) : <EOL> return self . _getTargetClass ( ) ( * args , ** kwargs ) <EOL> def test_constructor ( self ) : <EOL> from gcloud . bigtable . client import Client <EOL> from gcloud . bigtable . client import DATA_SCOPE <EOL> credentials = _Credentials ( ) <EOL> project = '<STR_LIT>' <EOL> user_agent = '<STR_LIT>' <EOL> client = Client ( project = project , credentials = credentials , <EOL> user_agent = user_agent ) <EOL> transformer = self . _makeOne ( client ) <EOL> self . assertTrue ( transformer . _credentials is credentials ) <EOL> self . assertEqual ( transformer . _user_agent , user_agent ) <EOL> self . assertEqual ( credentials . scopes , [ DATA_SCOPE ] ) <EOL> def test___call__ ( self ) : <EOL> from gcloud . bigtable . client import Client <EOL> from gcloud . bigtable . client import DATA_SCOPE <EOL> from gcloud . bigtable . client import DEFAULT_USER_AGENT <EOL> access_token_expected = '<STR_LIT>' <EOL> credentials = _Credentials ( access_token = access_token_expected ) <EOL> project = '<STR_LIT>' <EOL> client = Client ( project = project , credentials = credentials ) <EOL> callback_args = [ ] <EOL> def callback ( * args ) : <EOL> callback_args . append ( args ) <EOL> transformer = self . _makeOne ( client ) <EOL> result = transformer ( None , callback ) <EOL> cb_headers = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' + access_token_expected ) , <EOL> ( '<STR_LIT>' , DEFAULT_USER_AGENT ) , <EOL> ] <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( callback_args , [ ( cb_headers , None ) ] ) <EOL> self . assertEqual ( credentials . scopes , [ DATA_SCOPE ] ) <EOL> self . assertEqual ( len ( credentials . _tokens ) , <NUM_LIT:1> ) <EOL> class Test__make_stub ( unittest2 . TestCase ) : <EOL> def _callFUT ( self , * args , ** kwargs ) : <EOL> from gcloud . bigtable . client import _make_stub <EOL> return _make_stub ( * args , ** kwargs ) <EOL> def test_it ( self ) : <EOL> from gcloud . _testing import _Monkey <EOL> from gcloud . bigtable import client as MUT <EOL> mock_result = object ( ) <EOL> stub_inputs = [ ] <EOL> SSL_CREDS = object ( ) <EOL> METADATA_CREDS = object ( ) <EOL> COMPOSITE_CREDS = object ( ) <EOL> CHANNEL = object ( ) <EOL> class _ImplementationsModule ( object ) : <EOL> def __init__ ( self ) : <EOL> self . ssl_channel_credentials_args = None <EOL> self . metadata_call_credentials_args = None <EOL> self . composite_channel_credentials_args = None <EOL> self . secure_channel_args = None <EOL> def ssl_channel_credentials ( self , * args ) : <EOL> self . ssl_channel_credentials_args = args <EOL> return SSL_CREDS <EOL> def metadata_call_credentials ( self , * args , ** kwargs ) : <EOL> self . metadata_call_credentials_args = ( args , kwargs ) <EOL> return METADATA_CREDS <EOL> def composite_channel_credentials ( self , * args ) : <EOL> self . composite_channel_credentials_args = args <EOL> return COMPOSITE_CREDS <EOL> def secure_channel ( self , * args ) : <EOL> self . secure_channel_args = args <EOL> return CHANNEL <EOL> implementations_mod = _ImplementationsModule ( ) <EOL> def mock_stub_factory ( channel ) : <EOL> stub_inputs . append ( channel ) <EOL> return mock_result <EOL> metadata_plugin = object ( ) <EOL> clients = [ ] <EOL> def mock_plugin ( client ) : <EOL> clients . append ( client ) <EOL> return metadata_plugin <EOL> host = '<STR_LIT>' <EOL> port = <NUM_LIT> <EOL> client = object ( ) <EOL> with _Monkey ( MUT , implementations = implementations_mod , <EOL> _MetadataPlugin = mock_plugin ) : <EOL> result = self . _callFUT ( client , mock_stub_factory , host , port ) <EOL> self . assertTrue ( result is mock_result ) <EOL> self . assertEqual ( stub_inputs , [ CHANNEL ] ) <EOL> self . assertEqual ( clients , [ client ] ) <EOL> self . assertEqual ( implementations_mod . ssl_channel_credentials_args , <EOL> ( None , None , None ) ) <EOL> self . assertEqual ( implementations_mod . metadata_call_credentials_args , <EOL> ( ( metadata_plugin , ) , { '<STR_LIT:name>' : '<STR_LIT>' } ) ) <EOL> self . assertEqual ( <EOL> implementations_mod . composite_channel_credentials_args , <EOL> ( SSL_CREDS , METADATA_CREDS ) ) <EOL> self . assertEqual ( implementations_mod . secure_channel_args , <EOL> ( host , port , COMPOSITE_CREDS ) ) <EOL> class _Credentials ( object ) : <EOL> scopes = None <EOL> def __init__ ( self , access_token = None ) : <EOL> self . _access_token = access_token <EOL> self . _tokens = [ ] <EOL> def get_access_token ( self ) : <EOL> from oauth2client . client import AccessTokenInfo <EOL> token = AccessTokenInfo ( access_token = self . _access_token , <EOL> expires_in = None ) <EOL> self . _tokens . append ( token ) <EOL> return token <EOL> def create_scoped ( self , scope ) : <EOL> self . scopes = scope <EOL> return self <EOL> def __eq__ ( self , other ) : <EOL> return self . _access_token == other . _access_token </s>
