<s> import time <EOL> import unittest <EOL> from azure . common import ( <EOL> AzureHttpError , <EOL> AzureMissingResourceHttpError , <EOL> ) <EOL> from azure . servicemanagement import ServiceBusManagementService <EOL> from testutils . common_recordingtestcase import ( <EOL> TestMode , <EOL> record , <EOL> ) <EOL> from tests . legacy_mgmt_testcase import LegacyMgmtTestCase <EOL> class LegacyMgmtServiceBusTest ( LegacyMgmtTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( LegacyMgmtServiceBusTest , self ) . setUp ( ) <EOL> self . sms = self . create_service_management ( ServiceBusManagementService ) <EOL> self . sb_namespace = self . get_resource_name ( '<STR_LIT>' ) <EOL> def tearDown ( self ) : <EOL> if not self . is_playback ( ) : <EOL> try : <EOL> self . sms . delete_namespace ( self . sb_namespace ) <EOL> except : <EOL> pass <EOL> return super ( LegacyMgmtServiceBusTest , self ) . tearDown ( ) <EOL> def _namespace_exists ( self , name ) : <EOL> try : <EOL> ns = self . sms . get_namespace ( name ) <EOL> return ns . status != '<STR_LIT>' <EOL> except : <EOL> return False <EOL> def _wait_for_namespace_active ( self , name ) : <EOL> count = <NUM_LIT:0> <EOL> ns = self . sms . get_namespace ( name ) <EOL> while ns . status != '<STR_LIT>' : <EOL> count = count + <NUM_LIT:1> <EOL> if count > <NUM_LIT> : <EOL> self . assertTrue ( <EOL> False , <EOL> '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> ns = self . sms . get_namespace ( name ) <EOL> @ record <EOL> def test_get_regions ( self ) : <EOL> result = self . sms . get_regions ( ) <EOL> self . assertIsNotNone ( result ) <EOL> self . assertTrue ( len ( result ) > <NUM_LIT:0> ) <EOL> for region in result : <EOL> self . assertTrue ( len ( region . code ) > <NUM_LIT:0> ) <EOL> self . assertTrue ( len ( region . fullname ) > <NUM_LIT:0> ) <EOL> @ record <EOL> def test_list_namespaces ( self ) : <EOL> result = self . sms . list_namespaces ( ) <EOL> self . assertIsNotNone ( result ) <EOL> self . assertTrue ( len ( result ) > <NUM_LIT:0> ) <EOL> for ns in result : <EOL> self . assertTrue ( len ( ns . name ) > <NUM_LIT:0> ) <EOL> self . assertTrue ( len ( ns . region ) > <NUM_LIT:0> ) <EOL> @ record <EOL> def test_get_namespace ( self ) : <EOL> name = self . settings . SERVICEBUS_NAME <EOL> result = self . sms . get_namespace ( name ) <EOL> self . assertIsNotNone ( result ) <EOL> self . assertEqual ( result . name , name ) <EOL> self . assertIsNotNone ( result . region ) <EOL> self . assertIsNotNone ( result . default_key ) <EOL> self . assertIsNotNone ( result . status ) <EOL> self . assertIsNotNone ( result . created_at ) <EOL> self . assertIsNotNone ( result . acs_management_endpoint ) <EOL> self . assertIsNotNone ( result . servicebus_endpoint ) <EOL> self . assertIsNotNone ( result . connection_string ) <EOL> self . assertTrue ( result . enabled ) <EOL> @ record <EOL> def test_get_namespace_with_non_existing_namespace ( self ) : <EOL> name = self . sb_namespace <EOL> with self . assertRaises ( AzureMissingResourceHttpError ) : <EOL> self . sms . get_namespace ( name ) <EOL> @ record <EOL> def test_check_namespace_availability_not_available ( self ) : <EOL> name = self . settings . SERVICEBUS_NAME <EOL> availability = self . sms . check_namespace_availability ( name ) <EOL> self . assertFalse ( availability . result ) <EOL> @ record <EOL> def test_check_namespace_availability_available ( self ) : <EOL> name = '<STR_LIT>' <EOL> availability = self . sms . check_namespace_availability ( name ) <EOL> self . assertTrue ( availability . result ) <EOL> @ record <EOL> def test_create_namespace ( self ) : <EOL> name = self . sb_namespace <EOL> region = '<STR_LIT>' <EOL> result = self . sms . create_namespace ( name , region ) <EOL> self . _wait_for_namespace_active ( name ) <EOL> self . assertIsNone ( result ) <EOL> self . assertTrue ( self . _namespace_exists ( name ) ) <EOL> @ record <EOL> def test_list_topics ( self ) : <EOL> name = self . settings . SERVICEBUS_NAME <EOL> result = self . sms . list_topics ( name ) <EOL> self . assertIsNotNone ( result ) <EOL> self . assertIsInstance ( result , list ) <EOL> @ record <EOL> def test_list_queues ( self ) : <EOL> name = self . settings . SERVICEBUS_NAME <EOL> result = self . sms . list_queues ( name ) <EOL> self . assertIsNotNone ( result ) <EOL> self . assertIsInstance ( result , list ) <EOL> @ record <EOL> def test_list_notification_hubs ( self ) : <EOL> name = self . settings . SERVICEBUS_NAME <EOL> result = self . sms . list_notification_hubs ( name ) <EOL> self . assertIsNotNone ( result ) <EOL> self . assertIsInstance ( result , list ) <EOL> @ record <EOL> def test_list_relays ( self ) : <EOL> name = self . settings . SERVICEBUS_NAME <EOL> result = self . sms . list_relays ( name ) <EOL> self . assertIsNotNone ( result ) <EOL> self . assertIsInstance ( result , list ) <EOL> @ record <EOL> def test_create_namespace_with_existing_namespace ( self ) : <EOL> name = self . sb_namespace <EOL> region = '<STR_LIT>' <EOL> self . sms . create_namespace ( name , region ) <EOL> self . _wait_for_namespace_active ( name ) <EOL> with self . assertRaises ( AzureHttpError ) : <EOL> self . sms . create_namespace ( name , region ) <EOL> @ record <EOL> def test_delete_namespace ( self ) : <EOL> name = self . sb_namespace <EOL> region = '<STR_LIT>' <EOL> self . sms . create_namespace ( name , region ) <EOL> self . _wait_for_namespace_active ( name ) <EOL> result = self . sms . delete_namespace ( name ) <EOL> self . assertIsNone ( result ) <EOL> self . assertFalse ( self . _namespace_exists ( name ) ) <EOL> @ record <EOL> def test_delete_namespace_with_non_existing_namespace ( self ) : <EOL> name = self . sb_namespace <EOL> with self . assertRaises ( AzureMissingResourceHttpError ) : <EOL> self . sms . delete_namespace ( name ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
