<s> from __future__ import with_statement <EOL> import os <EOL> import sys <EOL> import unittest <EOL> import datetime <EOL> from libcloud . utils . iso8601 import UTC <EOL> try : <EOL> import simplejson as json <EOL> except ImportError : <EOL> import json <EOL> from mock import Mock , patch <EOL> from libcloud . utils . py3 import httplib <EOL> from libcloud . utils . py3 import method_type <EOL> from libcloud . utils . py3 import u <EOL> from libcloud . common . types import InvalidCredsError , MalformedResponseError , LibcloudError <EOL> from libcloud . compute . types import Provider , KeyPairDoesNotExistError , StorageVolumeState , VolumeSnapshotState <EOL> from libcloud . compute . providers import get_driver <EOL> from libcloud . compute . drivers . openstack import ( <EOL> OpenStack_1_0_NodeDriver , OpenStack_1_0_Response , <EOL> OpenStack_1_1_NodeDriver , OpenStackSecurityGroup , <EOL> OpenStackSecurityGroupRule , OpenStack_1_1_FloatingIpPool , <EOL> OpenStack_1_1_FloatingIpAddress , OpenStackKeyPair <EOL> ) <EOL> from libcloud . compute . base import Node , NodeImage , NodeSize <EOL> from libcloud . pricing import set_pricing , clear_pricing_data <EOL> from libcloud . test import MockResponse , MockHttpTestCase , XML_HEADERS <EOL> from libcloud . test . file_fixtures import ComputeFileFixtures , OpenStackFixtures <EOL> from libcloud . test . compute import TestCaseMixin <EOL> from libcloud . test . secrets import OPENSTACK_PARAMS <EOL> BASE_DIR = os . path . abspath ( os . path . split ( __file__ ) [ <NUM_LIT:0> ] ) <EOL> class OpenStack_1_0_ResponseTestCase ( unittest . TestCase ) : <EOL> XML = """<STR_LIT>""" <EOL> def test_simple_xml_content_type_handling ( self ) : <EOL> http_response = MockResponse ( <EOL> <NUM_LIT:200> , OpenStack_1_0_ResponseTestCase . XML , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> body = OpenStack_1_0_Response ( http_response , None ) . parse_body ( ) <EOL> self . assertTrue ( hasattr ( body , '<STR_LIT>' ) , "<STR_LIT>" ) <EOL> def test_extended_xml_content_type_handling ( self ) : <EOL> http_response = MockResponse ( <NUM_LIT:200> , <EOL> OpenStack_1_0_ResponseTestCase . XML , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> body = OpenStack_1_0_Response ( http_response , None ) . parse_body ( ) <EOL> self . assertTrue ( hasattr ( body , '<STR_LIT>' ) , "<STR_LIT>" ) <EOL> def test_non_xml_content_type_handling ( self ) : <EOL> RESPONSE_BODY = "<STR_LIT>" <EOL> http_response = MockResponse ( <EOL> <NUM_LIT> , RESPONSE_BODY , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> body = OpenStack_1_0_Response ( http_response , None ) . parse_body ( ) <EOL> self . assertEqual ( <EOL> body , RESPONSE_BODY , "<STR_LIT>" ) <EOL> class OpenStack_1_0_Tests ( unittest . TestCase , TestCaseMixin ) : <EOL> should_list_locations = False <EOL> should_list_volumes = False <EOL> driver_klass = OpenStack_1_0_NodeDriver <EOL> driver_args = OPENSTACK_PARAMS <EOL> driver_kwargs = { } <EOL> @ classmethod <EOL> def create_driver ( self ) : <EOL> if self is not OpenStack_1_0_FactoryMethodTests : <EOL> self . driver_type = self . driver_klass <EOL> return self . driver_type ( * self . driver_args , ** self . driver_kwargs ) <EOL> def setUp ( self ) : <EOL> def get_endpoint ( * args , ** kwargs ) : <EOL> return "<STR_LIT>" <EOL> self . driver_klass . connectionCls . get_endpoint = get_endpoint <EOL> self . driver_klass . connectionCls . conn_classes = ( OpenStackMockHttp , <EOL> OpenStackMockHttp ) <EOL> self . driver_klass . connectionCls . auth_url = "<STR_LIT>" <EOL> OpenStackMockHttp . type = None <EOL> self . driver = self . create_driver ( ) <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> clear_pricing_data ( ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_populate_hosts_and_requests_path ( self , _ ) : <EOL> tomorrow = datetime . datetime . today ( ) + datetime . timedelta ( <NUM_LIT:1> ) <EOL> cls = self . driver_klass . connectionCls <EOL> count = <NUM_LIT:5> <EOL> con = cls ( '<STR_LIT:username>' , '<STR_LIT:key>' ) <EOL> osa = con . get_auth_class ( ) <EOL> mocked_auth_method = Mock ( ) <EOL> osa . authenticate = mocked_auth_method <EOL> for i in range ( <NUM_LIT:0> , count ) : <EOL> con . _populate_hosts_and_request_paths ( ) <EOL> if i == <NUM_LIT:0> : <EOL> osa . auth_token = '<STR_LIT>' <EOL> osa . auth_token_expires = tomorrow <EOL> self . assertEqual ( mocked_auth_method . call_count , <NUM_LIT:1> ) <EOL> osa . auth_token = None <EOL> osa . auth_token_expires = None <EOL> con = cls ( '<STR_LIT:username>' , '<STR_LIT:key>' , ex_force_base_url = '<STR_LIT>' , <EOL> ex_force_auth_token = '<STR_LIT>' ) <EOL> osa = con . get_auth_class ( ) <EOL> mocked_auth_method = Mock ( ) <EOL> osa . authenticate = mocked_auth_method <EOL> for i in range ( <NUM_LIT:0> , count ) : <EOL> con . _populate_hosts_and_request_paths ( ) <EOL> self . assertEqual ( mocked_auth_method . call_count , <NUM_LIT:0> ) <EOL> def test_auth_token_is_set ( self ) : <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> self . assertEqual ( <EOL> self . driver . connection . auth_token , "<STR_LIT>" ) <EOL> def test_auth_token_expires_is_set ( self ) : <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> expires = self . driver . connection . auth_token_expires <EOL> self . assertEqual ( expires . isoformat ( ) , "<STR_LIT>" ) <EOL> def test_auth ( self ) : <EOL> if self . driver . connection . _auth_version == '<STR_LIT>' : <EOL> return <EOL> OpenStackMockHttp . type = '<STR_LIT>' <EOL> try : <EOL> self . driver = self . create_driver ( ) <EOL> self . driver . list_nodes ( ) <EOL> except InvalidCredsError : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . assertEqual ( True , isinstance ( e , InvalidCredsError ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_auth_missing_key ( self ) : <EOL> if self . driver . connection . _auth_version == '<STR_LIT>' : <EOL> return <EOL> OpenStackMockHttp . type = '<STR_LIT>' <EOL> try : <EOL> self . driver = self . create_driver ( ) <EOL> self . driver . list_nodes ( ) <EOL> except MalformedResponseError : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . assertEqual ( True , isinstance ( e , MalformedResponseError ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_auth_server_error ( self ) : <EOL> if self . driver . connection . _auth_version == '<STR_LIT>' : <EOL> return <EOL> OpenStackMockHttp . type = '<STR_LIT>' <EOL> try : <EOL> self . driver = self . create_driver ( ) <EOL> self . driver . list_nodes ( ) <EOL> except MalformedResponseError : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . assertEqual ( True , isinstance ( e , MalformedResponseError ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_error_parsing_when_body_is_missing_message ( self ) : <EOL> OpenStackMockHttp . type = '<STR_LIT>' <EOL> try : <EOL> self . driver . list_images ( ) <EOL> except Exception : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . assertEqual ( True , isinstance ( e , Exception ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_list_locations ( self ) : <EOL> locations = self . driver . list_locations ( ) <EOL> self . assertEqual ( len ( locations ) , <NUM_LIT:1> ) <EOL> def test_list_nodes ( self ) : <EOL> OpenStackMockHttp . type = '<STR_LIT>' <EOL> ret = self . driver . list_nodes ( ) <EOL> self . assertEqual ( len ( ret ) , <NUM_LIT:0> ) <EOL> OpenStackMockHttp . type = None <EOL> ret = self . driver . list_nodes ( ) <EOL> self . assertEqual ( len ( ret ) , <NUM_LIT:1> ) <EOL> node = ret [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , node . public_ips [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( '<STR_LIT>' in node . private_ips ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , '<STR_LIT:1>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( type ( node . extra . get ( '<STR_LIT>' ) ) , type ( dict ( ) ) ) <EOL> OpenStackMockHttp . type = '<STR_LIT>' <EOL> ret = self . driver . list_nodes ( ) <EOL> self . assertEqual ( len ( ret ) , <NUM_LIT:1> ) <EOL> node = ret [ <NUM_LIT:0> ] <EOL> self . assertEqual ( type ( node . extra . get ( '<STR_LIT>' ) ) , type ( dict ( ) ) ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' ) <EOL> OpenStackMockHttp . type = None <EOL> def test_list_images ( self ) : <EOL> ret = self . driver . list_images ( ) <EOL> expected = { <NUM_LIT:10> : { '<STR_LIT>' : None , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None } , <EOL> <NUM_LIT:11> : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:100>' , <EOL> '<STR_LIT>' : '<STR_LIT:5>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } } <EOL> for ret_idx , extra in list ( expected . items ( ) ) : <EOL> for key , value in list ( extra . items ( ) ) : <EOL> self . assertEqual ( ret [ ret_idx ] . extra [ key ] , value ) <EOL> def test_create_node ( self ) : <EOL> image = NodeImage ( id = <NUM_LIT:11> , name = '<STR_LIT>' , <EOL> driver = self . driver ) <EOL> size = NodeSize ( <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , <EOL> driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT:password>' ) , '<STR_LIT>' ) <EOL> def test_create_node_without_adminPass ( self ) : <EOL> OpenStackMockHttp . type = '<STR_LIT>' <EOL> image = NodeImage ( id = <NUM_LIT:11> , name = '<STR_LIT>' , <EOL> driver = self . driver ) <EOL> size = NodeSize ( <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , <EOL> driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT:password>' ) , None ) <EOL> def test_create_node_ex_shared_ip_group ( self ) : <EOL> OpenStackMockHttp . type = '<STR_LIT>' <EOL> image = NodeImage ( id = <NUM_LIT:11> , name = '<STR_LIT>' , <EOL> driver = self . driver ) <EOL> size = NodeSize ( <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , <EOL> driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size , <EOL> ex_shared_ip_group_id = '<STR_LIT>' ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT:password>' ) , '<STR_LIT>' ) <EOL> def test_create_node_with_metadata ( self ) : <EOL> OpenStackMockHttp . type = '<STR_LIT>' <EOL> image = NodeImage ( id = <NUM_LIT:11> , name = '<STR_LIT>' , <EOL> driver = self . driver ) <EOL> size = NodeSize ( <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , <EOL> driver = self . driver ) <EOL> metadata = { '<STR_LIT:a>' : '<STR_LIT:b>' , '<STR_LIT:c>' : '<STR_LIT:d>' } <EOL> files = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size , <EOL> metadata = metadata , files = files ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT:password>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , metadata ) <EOL> def test_reboot_node ( self ) : <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> ret = node . reboot ( ) <EOL> self . assertTrue ( ret is True ) <EOL> def test_destroy_node ( self ) : <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> ret = node . destroy ( ) <EOL> self . assertTrue ( ret is True ) <EOL> def test_ex_limits ( self ) : <EOL> limits = self . driver . ex_limits ( ) <EOL> self . assertTrue ( "<STR_LIT>" in limits ) <EOL> self . assertTrue ( "<STR_LIT>" in limits ) <EOL> def test_create_image ( self ) : <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> image = self . driver . create_image ( node , "<STR_LIT>" ) <EOL> self . assertEqual ( image . name , "<STR_LIT>" ) <EOL> self . assertEqual ( image . id , "<STR_LIT>" ) <EOL> def test_delete_image ( self ) : <EOL> image = NodeImage ( id = <NUM_LIT> , name = '<STR_LIT>' , <EOL> driver = self . driver ) <EOL> ret = self . driver . delete_image ( image ) <EOL> self . assertTrue ( ret ) <EOL> def test_ex_list_ip_addresses ( self ) : <EOL> ret = self . driver . ex_list_ip_addresses ( node_id = <NUM_LIT> ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ret . public_addresses ) ) <EOL> self . assertTrue ( '<STR_LIT>' in ret . public_addresses ) <EOL> self . assertTrue ( '<STR_LIT>' in ret . public_addresses ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( ret . private_addresses ) ) <EOL> self . assertTrue ( '<STR_LIT>' in ret . private_addresses ) <EOL> def test_ex_list_ip_groups ( self ) : <EOL> ret = self . driver . ex_list_ip_groups ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ret ) ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:0> ] . id ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:0> ] . name ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:1> ] . id ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:1> ] . name ) <EOL> self . assertTrue ( ret [ <NUM_LIT:0> ] . servers is None ) <EOL> def test_ex_list_ip_groups_detail ( self ) : <EOL> ret = self . driver . ex_list_ip_groups ( details = True ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ret ) ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:0> ] . id ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:0> ] . name ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ret [ <NUM_LIT:0> ] . servers ) ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:0> ] . servers [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:0> ] . servers [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:1> ] . id ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:1> ] . name ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( ret [ <NUM_LIT:1> ] . servers ) ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:1> ] . servers [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:1> ] . servers [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , ret [ <NUM_LIT:1> ] . servers [ <NUM_LIT:2> ] ) <EOL> def test_ex_create_ip_group ( self ) : <EOL> ret = self . driver . ex_create_ip_group ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , ret . id ) <EOL> self . assertEqual ( '<STR_LIT>' , ret . name ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( ret . servers ) ) <EOL> self . assertEqual ( '<STR_LIT>' , ret . servers [ <NUM_LIT:0> ] ) <EOL> def test_ex_delete_ip_group ( self ) : <EOL> ret = self . driver . ex_delete_ip_group ( '<STR_LIT>' ) <EOL> self . assertEqual ( True , ret ) <EOL> def test_ex_share_ip ( self ) : <EOL> ret = self . driver . ex_share_ip ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( True , ret ) <EOL> def test_ex_unshare_ip ( self ) : <EOL> ret = self . driver . ex_unshare_ip ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( True , ret ) <EOL> def test_ex_resize ( self ) : <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> size = NodeSize ( <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , <EOL> driver = self . driver ) <EOL> self . assertTrue ( self . driver . ex_resize ( node = node , size = size ) ) <EOL> def test_ex_confirm_resize ( self ) : <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> self . assertTrue ( self . driver . ex_confirm_resize ( node = node ) ) <EOL> def test_ex_revert_resize ( self ) : <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> self . assertTrue ( self . driver . ex_revert_resize ( node = node ) ) <EOL> def test_list_sizes ( self ) : <EOL> sizes = self . driver . list_sizes ( ) <EOL> self . assertEqual ( len ( sizes ) , <NUM_LIT:7> , '<STR_LIT>' ) <EOL> for size in sizes : <EOL> self . assertTrue ( isinstance ( size . price , float ) , <EOL> '<STR_LIT>' ) <EOL> if self . driver . api_name == '<STR_LIT>' : <EOL> self . assertEqual ( size . price , <NUM_LIT:0> , <EOL> '<STR_LIT>' ) <EOL> def test_list_sizes_with_specified_pricing ( self ) : <EOL> if self . driver . api_name != '<STR_LIT>' : <EOL> return <EOL> pricing = dict ( ( str ( i ) , i ) for i in range ( <NUM_LIT:1> , <NUM_LIT:8> ) ) <EOL> set_pricing ( driver_type = '<STR_LIT>' , driver_name = '<STR_LIT>' , <EOL> pricing = pricing ) <EOL> sizes = self . driver . list_sizes ( ) <EOL> self . assertEqual ( len ( sizes ) , <NUM_LIT:7> , '<STR_LIT>' ) <EOL> for size in sizes : <EOL> self . assertTrue ( isinstance ( size . price , float ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( float ( size . price ) , float ( pricing [ size . id ] ) ) <EOL> class OpenStack_1_0_FactoryMethodTests ( OpenStack_1_0_Tests ) : <EOL> should_list_locations = False <EOL> should_list_volumes = False <EOL> driver_klass = OpenStack_1_0_NodeDriver <EOL> driver_type = get_driver ( Provider . OPENSTACK ) <EOL> driver_args = OPENSTACK_PARAMS + ( '<STR_LIT:1.0>' , ) <EOL> def test_factory_method_invalid_version ( self ) : <EOL> try : <EOL> self . driver_type ( * ( OPENSTACK_PARAMS + ( '<STR_LIT>' , ) ) ) <EOL> except NotImplementedError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> class OpenStackMockHttp ( MockHttpTestCase ) : <EOL> fixtures = ComputeFileFixtures ( '<STR_LIT>' ) <EOL> auth_fixtures = OpenStackFixtures ( ) <EOL> json_content_headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> def _v1_0 ( self , method , url , body , headers ) : <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , headers , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> def _v1_0_UNAUTHORIZED ( self , method , url , body , headers ) : <EOL> return ( httplib . UNAUTHORIZED , "<STR_LIT>" , { } , httplib . responses [ httplib . UNAUTHORIZED ] ) <EOL> def _v1_0_INTERNAL_SERVER_ERROR ( self , method , url , body , headers ) : <EOL> return ( httplib . INTERNAL_SERVER_ERROR , "<STR_LIT>" , { } , <EOL> httplib . responses [ httplib . INTERNAL_SERVER_ERROR ] ) <EOL> def _v1_0_slug_images_detail_NO_MESSAGE_IN_ERROR_BODY ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . MULTIPLE_CHOICES , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0_UNAUTHORIZED_MISSING_KEY ( self , method , url , body , headers ) : <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , headers , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> def _v2_0_tokens ( self , method , url , body , headers ) : <EOL> body = self . auth_fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0_slug_servers_detail_EMPTY ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , XML_HEADERS , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0_slug_servers_detail ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , XML_HEADERS , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0_slug_servers_detail_METADATA ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , XML_HEADERS , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0_slug_images_333111 ( self , method , url , body , headers ) : <EOL> if method != "<STR_LIT>" : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , "<STR_LIT>" , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> def _v1_0_slug_images ( self , method , url , body , headers ) : <EOL> if method != "<STR_LIT:POST>" : <EOL> raise NotImplementedError ( ) <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , XML_HEADERS , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_0_slug_images_detail ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , XML_HEADERS , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0_slug_servers ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , XML_HEADERS , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_0_slug_servers_NO_ADMIN_PASS ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , XML_HEADERS , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_0_slug_servers_EX_SHARED_IP_GROUP ( self , method , url , body , headers ) : <EOL> body = u ( body ) <EOL> self . assertTrue ( body . find ( '<STR_LIT>' ) != - <NUM_LIT:1> ) <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , XML_HEADERS , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_0_slug_servers_METADATA ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , XML_HEADERS , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_0_slug_servers_72258_action ( self , method , url , body , headers ) : <EOL> if method != "<STR_LIT:POST>" or body [ : <NUM_LIT:8> ] != "<STR_LIT>" : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , { } , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_0_slug_limits ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , XML_HEADERS , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_0_slug_servers_72258 ( self , method , url , body , headers ) : <EOL> if method != "<STR_LIT>" : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , { } , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_0_slug_servers_72258_ips ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , XML_HEADERS , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0_slug_shared_ip_groups_5467 ( self , method , url , body , headers ) : <EOL> if method != '<STR_LIT>' : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , { } , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> def _v1_0_slug_shared_ip_groups ( self , method , url , body , headers ) : <EOL> fixture = '<STR_LIT>' if method == '<STR_LIT:POST>' else '<STR_LIT>' <EOL> body = self . fixtures . load ( fixture ) <EOL> return ( httplib . OK , body , XML_HEADERS , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0_slug_shared_ip_groups_detail ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , XML_HEADERS , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0_slug_servers_3445_ips_public_67_23_21_133 ( self , method , url , body , headers ) : <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , { } , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_0_slug_servers_444222_action ( self , method , url , body , headers ) : <EOL> body = u ( body ) <EOL> if body . find ( '<STR_LIT>' ) != - <NUM_LIT:1> : <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , headers , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> elif body . find ( '<STR_LIT>' ) != - <NUM_LIT:1> : <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , headers , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> elif body . find ( '<STR_LIT>' ) != - <NUM_LIT:1> : <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , headers , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> def _v1_0_slug_flavors_detail ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> headers = { <EOL> '<STR_LIT:date>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> headers . update ( XML_HEADERS ) <EOL> return ( httplib . OK , body , headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_auth ( self , method , url , body , headers ) : <EOL> body = self . auth_fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_auth_UNAUTHORIZED ( self , method , url , body , headers ) : <EOL> body = self . auth_fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . UNAUTHORIZED , body , self . json_content_headers , httplib . responses [ httplib . UNAUTHORIZED ] ) <EOL> def _v1_1_auth_UNAUTHORIZED_MISSING_KEY ( self , method , url , body , headers ) : <EOL> body = self . auth_fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_auth_INTERNAL_SERVER_ERROR ( self , method , url , body , headers ) : <EOL> return ( httplib . INTERNAL_SERVER_ERROR , "<STR_LIT>" , { '<STR_LIT>' : '<STR_LIT>' } , <EOL> httplib . responses [ httplib . INTERNAL_SERVER_ERROR ] ) <EOL> class OpenStack_1_1_Tests ( unittest . TestCase , TestCaseMixin ) : <EOL> should_list_locations = False <EOL> should_list_volumes = True <EOL> driver_klass = OpenStack_1_1_NodeDriver <EOL> driver_type = OpenStack_1_1_NodeDriver <EOL> driver_args = OPENSTACK_PARAMS <EOL> driver_kwargs = { '<STR_LIT>' : '<STR_LIT>' } <EOL> @ classmethod <EOL> def create_driver ( self ) : <EOL> if self is not OpenStack_1_1_FactoryMethodTests : <EOL> self . driver_type = self . driver_klass <EOL> return self . driver_type ( * self . driver_args , ** self . driver_kwargs ) <EOL> def setUp ( self ) : <EOL> self . driver_klass . connectionCls . conn_classes = ( <EOL> OpenStack_2_0_MockHttp , OpenStack_2_0_MockHttp ) <EOL> self . driver_klass . connectionCls . auth_url = "<STR_LIT>" <EOL> OpenStackMockHttp . type = None <EOL> OpenStack_1_1_MockHttp . type = None <EOL> OpenStack_2_0_MockHttp . type = None <EOL> self . driver = self . create_driver ( ) <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> clear_pricing_data ( ) <EOL> self . node = self . driver . list_nodes ( ) [ <NUM_LIT:1> ] <EOL> def _force_reauthentication ( self ) : <EOL> """<STR_LIT>""" <EOL> self . driver . connection . _ex_force_base_url = '<STR_LIT>' <EOL> self . driver . connection . auth_token = None <EOL> self . driver . connection . auth_token_expires = None <EOL> self . driver . connection . _osa . auth_token = None <EOL> self . driver . connection . _osa . auth_token_expires = None <EOL> def test_auth_token_is_set ( self ) : <EOL> self . _force_reauthentication ( ) <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> self . assertEqual ( <EOL> self . driver . connection . auth_token , "<STR_LIT>" ) <EOL> def test_auth_token_expires_is_set ( self ) : <EOL> self . _force_reauthentication ( ) <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> expires = self . driver . connection . auth_token_expires <EOL> self . assertEqual ( expires . isoformat ( ) , "<STR_LIT>" ) <EOL> def test_ex_force_base_url ( self ) : <EOL> self . driver . connection . _ex_force_base_url = '<STR_LIT>' <EOL> self . driver . connection . auth_token = None <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> self . assertEqual ( self . driver . connection . host , '<STR_LIT>' ) <EOL> self . assertEqual ( self . driver . connection . port , <NUM_LIT> ) <EOL> self . assertEqual ( self . driver . connection . request_path , '<STR_LIT>' ) <EOL> def test_get_endpoint_populates_host_port_and_request_path ( self ) : <EOL> self . driver . connection . get_endpoint = lambda : '<STR_LIT>' <EOL> self . driver . connection . auth_token = None <EOL> self . driver . connection . _ex_force_base_url = None <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> self . assertEqual ( self . driver . connection . host , '<STR_LIT>' ) <EOL> self . assertEqual ( self . driver . connection . port , <NUM_LIT> ) <EOL> self . assertEqual ( self . driver . connection . request_path , '<STR_LIT>' ) <EOL> def test_set_auth_token_populates_host_port_and_request_path ( self ) : <EOL> self . driver . connection . _ex_force_base_url = '<STR_LIT>' <EOL> self . driver . connection . auth_token = "<STR_LIT>" <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> self . assertEqual ( <EOL> self . driver . connection . host , '<STR_LIT>' ) <EOL> self . assertEqual ( self . driver . connection . port , <NUM_LIT> ) <EOL> self . assertEqual ( self . driver . connection . request_path , '<STR_LIT>' ) <EOL> def test_auth_token_without_base_url_raises_exception ( self ) : <EOL> kwargs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> try : <EOL> self . driver_type ( * self . driver_args , ** kwargs ) <EOL> self . fail ( '<STR_LIT>' ) <EOL> except LibcloudError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_ex_force_auth_token_passed_to_connection ( self ) : <EOL> base_url = '<STR_LIT>' <EOL> kwargs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : base_url <EOL> } <EOL> driver = self . driver_type ( * self . driver_args , ** kwargs ) <EOL> driver . list_nodes ( ) <EOL> self . assertEqual ( kwargs [ '<STR_LIT>' ] , <EOL> driver . connection . auth_token ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> driver . connection . host ) <EOL> self . assertEqual ( '<STR_LIT>' , driver . connection . request_path ) <EOL> self . assertEqual ( <NUM_LIT> , driver . connection . port ) <EOL> def test_list_nodes ( self ) : <EOL> nodes = self . driver . list_nodes ( ) <EOL> self . assertEqual ( len ( nodes ) , <NUM_LIT:2> ) <EOL> node = nodes [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , node . id ) <EOL> self . assertTrue ( '<STR_LIT>' in node . public_ips ) <EOL> self . assertTrue ( '<STR_LIT>' in node . public_ips ) <EOL> self . assertTrue ( '<STR_LIT>' in node . public_ips ) <EOL> self . assertTrue ( '<STR_LIT>' in node . public_ips ) <EOL> self . assertTrue ( <EOL> '<STR_LIT>' in node . public_ips ) <EOL> self . assertTrue ( '<STR_LIT>' in node . private_ips ) <EOL> self . assertTrue ( '<STR_LIT>' in node . private_ips ) <EOL> self . assertTrue ( '<STR_LIT>' in node . private_ips ) <EOL> self . assertTrue ( '<STR_LIT>' in node . private_ips ) <EOL> self . assertTrue ( <EOL> '<STR_LIT>' in node . private_ips ) <EOL> self . assertEqual ( node . created_at , datetime . datetime ( <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT:11> , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> , tzinfo = UTC ) ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , '<STR_LIT:2>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , { } ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) , <NUM_LIT> ) <EOL> self . assertEqual ( node . extra . get ( '<STR_LIT>' ) [ '<STR_LIT:id>' ] , <NUM_LIT> ) <EOL> self . assertTrue ( node . extra . get ( '<STR_LIT>' ) is not None ) <EOL> self . assertTrue ( node . extra . get ( '<STR_LIT>' ) is not None ) <EOL> def test_list_nodes_no_image_id_attribute ( self ) : <EOL> self . driver_klass . connectionCls . conn_classes [ <NUM_LIT:0> ] . type = '<STR_LIT>' <EOL> self . driver_klass . connectionCls . conn_classes [ <NUM_LIT:1> ] . type = '<STR_LIT>' <EOL> nodes = self . driver . list_nodes ( ) <EOL> self . assertEqual ( nodes [ <NUM_LIT:0> ] . extra [ '<STR_LIT>' ] , None ) <EOL> def test_list_volumes ( self ) : <EOL> volumes = self . driver . list_volumes ( ) <EOL> self . assertEqual ( len ( volumes ) , <NUM_LIT:2> ) <EOL> volume = volumes [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , volume . id ) <EOL> self . assertEqual ( '<STR_LIT>' , volume . name ) <EOL> self . assertEqual ( StorageVolumeState . AVAILABLE , volume . state ) <EOL> self . assertEqual ( <NUM_LIT:2> , volume . size ) <EOL> self . assertEqual ( volume . extra , { <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } ] , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT:state>' : '<STR_LIT>' , <EOL> '<STR_LIT:location>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:None>' , <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> volume = volumes [ <NUM_LIT:1> ] <EOL> self . assertEqual ( '<STR_LIT>' , volume . id ) <EOL> self . assertEqual ( '<STR_LIT>' , volume . name ) <EOL> self . assertEqual ( <NUM_LIT:50> , volume . size ) <EOL> self . assertEqual ( StorageVolumeState . UNKNOWN , volume . state ) <EOL> self . assertEqual ( volume . extra , { <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:state>' : '<STR_LIT>' , <EOL> '<STR_LIT:location>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:None>' , <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> def test_list_sizes ( self ) : <EOL> sizes = self . driver . list_sizes ( ) <EOL> self . assertEqual ( len ( sizes ) , <NUM_LIT:8> , '<STR_LIT>' ) <EOL> for size in sizes : <EOL> self . assertTrue ( isinstance ( size . price , float ) , <EOL> '<STR_LIT>' ) <EOL> self . assertTrue ( isinstance ( size . ram , int ) ) <EOL> self . assertTrue ( isinstance ( size . vcpus , int ) ) <EOL> self . assertTrue ( isinstance ( size . disk , int ) ) <EOL> self . assertTrue ( isinstance ( size . swap , int ) ) <EOL> self . assertTrue ( isinstance ( size . ephemeral_disk , int ) or <EOL> size . ephemeral_disk is None ) <EOL> self . assertTrue ( isinstance ( size . extra , dict ) ) <EOL> if size . id == '<STR_LIT:1>' : <EOL> self . assertEqual ( size . ephemeral_disk , <NUM_LIT> ) <EOL> self . assertEqual ( size . extra , { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:class>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:2>" , <EOL> "<STR_LIT>" : "<STR_LIT:0>" <EOL> } ) <EOL> self . assertEqual ( sizes [ <NUM_LIT:0> ] . vcpus , <NUM_LIT:8> ) <EOL> def test_list_sizes_with_specified_pricing ( self ) : <EOL> pricing = dict ( ( str ( i ) , i * <NUM_LIT> ) for i in range ( <NUM_LIT:1> , <NUM_LIT:9> ) ) <EOL> set_pricing ( driver_type = '<STR_LIT>' , <EOL> driver_name = self . driver . api_name , pricing = pricing ) <EOL> sizes = self . driver . list_sizes ( ) <EOL> self . assertEqual ( len ( sizes ) , <NUM_LIT:8> , '<STR_LIT>' ) <EOL> for size in sizes : <EOL> self . assertTrue ( isinstance ( size . price , float ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( size . price , pricing [ size . id ] , <EOL> '<STR_LIT>' ) <EOL> def test_list_images ( self ) : <EOL> images = self . driver . list_images ( ) <EOL> self . assertEqual ( len ( images ) , <NUM_LIT> , '<STR_LIT>' ) <EOL> image = images [ <NUM_LIT:0> ] <EOL> self . assertEqual ( image . id , '<STR_LIT>' ) <EOL> self . assertEqual ( image . name , '<STR_LIT>' ) <EOL> self . assertEqual ( image . extra [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( image . extra [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( image . extra [ '<STR_LIT:status>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( image . extra [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> image . extra [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( image . extra [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( image . extra [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> def test_create_node ( self ) : <EOL> image = NodeImage ( <EOL> id = <NUM_LIT:11> , name = '<STR_LIT>' , driver = self . driver ) <EOL> size = NodeSize ( <EOL> <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size ) <EOL> self . assertEqual ( node . id , '<STR_LIT>' ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT:password>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_create_node_with_ex_keyname_and_ex_userdata ( self ) : <EOL> image = NodeImage ( <EOL> id = <NUM_LIT:11> , name = '<STR_LIT>' , driver = self . driver ) <EOL> size = NodeSize ( <EOL> <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size , <EOL> ex_keyname = '<STR_LIT>' , <EOL> ex_userdata = '<STR_LIT>' ) <EOL> self . assertEqual ( node . id , '<STR_LIT>' ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT:password>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_create_node_with_availability_zone ( self ) : <EOL> image = NodeImage ( <EOL> id = <NUM_LIT:11> , name = '<STR_LIT>' , driver = self . driver ) <EOL> size = NodeSize ( <EOL> <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size , <EOL> availability_zone = '<STR_LIT>' ) <EOL> self . assertEqual ( node . id , '<STR_LIT>' ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT:password>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_create_node_with_ex_disk_config ( self ) : <EOL> OpenStack_1_1_MockHttp . type = '<STR_LIT>' <EOL> image = NodeImage ( <EOL> id = <NUM_LIT:11> , name = '<STR_LIT>' , driver = self . driver ) <EOL> size = NodeSize ( <EOL> <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size , <EOL> ex_disk_config = '<STR_LIT>' ) <EOL> self . assertEqual ( node . id , '<STR_LIT>' ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> self . assertEqual ( node . extra [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_create_node_with_ex_config_drive ( self ) : <EOL> OpenStack_1_1_MockHttp . type = '<STR_LIT>' <EOL> image = NodeImage ( <EOL> id = <NUM_LIT:11> , name = '<STR_LIT>' , driver = self . driver ) <EOL> size = NodeSize ( <EOL> <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size , <EOL> ex_config_drive = True ) <EOL> self . assertEqual ( node . id , '<STR_LIT>' ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> self . assertTrue ( node . extra [ '<STR_LIT>' ] ) <EOL> def test_destroy_node ( self ) : <EOL> self . assertTrue ( self . node . destroy ( ) ) <EOL> def test_reboot_node ( self ) : <EOL> self . assertTrue ( self . node . reboot ( ) ) <EOL> def test_create_volume ( self ) : <EOL> volume = self . driver . create_volume ( <NUM_LIT:1> , '<STR_LIT:test>' ) <EOL> self . assertEqual ( volume . name , '<STR_LIT:test>' ) <EOL> self . assertEqual ( volume . size , <NUM_LIT:1> ) <EOL> def test_destroy_volume ( self ) : <EOL> volume = self . driver . ex_get_volume ( <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( self . driver . destroy_volume ( volume ) , True ) <EOL> def test_attach_volume ( self ) : <EOL> node = self . driver . list_nodes ( ) [ <NUM_LIT:0> ] <EOL> volume = self . driver . ex_get_volume ( <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . driver . attach_volume ( node , volume , '<STR_LIT>' ) , True ) <EOL> def test_detach_volume ( self ) : <EOL> node = self . driver . list_nodes ( ) [ <NUM_LIT:0> ] <EOL> volume = self . driver . ex_get_volume ( <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . driver . attach_volume ( node , volume , '<STR_LIT>' ) , True ) <EOL> self . assertEqual ( self . driver . detach_volume ( volume ) , True ) <EOL> def test_ex_set_password ( self ) : <EOL> self . assertTrue ( self . driver . ex_set_password ( self . node , '<STR_LIT>' ) ) <EOL> def test_ex_rebuild ( self ) : <EOL> image = NodeImage ( id = <NUM_LIT:11> , name = '<STR_LIT>' , <EOL> driver = self . driver ) <EOL> success = self . driver . ex_rebuild ( self . node , image = image ) <EOL> self . assertTrue ( success ) <EOL> def test_ex_rebuild_with_ex_disk_config ( self ) : <EOL> image = NodeImage ( id = <NUM_LIT> , name = '<STR_LIT>' , <EOL> driver = self . driver ) <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> success = self . driver . ex_rebuild ( node , image = image , <EOL> ex_disk_config = '<STR_LIT>' ) <EOL> self . assertTrue ( success ) <EOL> def test_ex_rebuild_with_ex_config_drive ( self ) : <EOL> image = NodeImage ( id = <NUM_LIT> , name = '<STR_LIT>' , <EOL> driver = self . driver ) <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> success = self . driver . ex_rebuild ( node , image = image , <EOL> ex_disk_config = '<STR_LIT>' , <EOL> ex_config_drive = True ) <EOL> self . assertTrue ( success ) <EOL> def test_ex_resize ( self ) : <EOL> size = NodeSize ( <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , <EOL> driver = self . driver ) <EOL> try : <EOL> self . driver . ex_resize ( self . node , size ) <EOL> except Exception : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . fail ( '<STR_LIT>' + repr ( e ) ) <EOL> def test_ex_confirm_resize ( self ) : <EOL> try : <EOL> self . driver . ex_confirm_resize ( self . node ) <EOL> except Exception : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . fail ( '<STR_LIT>' + repr ( e ) ) <EOL> def test_ex_revert_resize ( self ) : <EOL> try : <EOL> self . driver . ex_revert_resize ( self . node ) <EOL> except Exception : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . fail ( '<STR_LIT>' + repr ( e ) ) <EOL> def test_create_image ( self ) : <EOL> image = self . driver . create_image ( self . node , '<STR_LIT>' ) <EOL> self . assertEqual ( image . name , '<STR_LIT>' ) <EOL> self . assertEqual ( image . id , '<STR_LIT>' ) <EOL> def test_ex_set_server_name ( self ) : <EOL> old_node = Node ( <EOL> id = '<STR_LIT>' , name = None , state = None , <EOL> public_ips = None , private_ips = None , driver = self . driver , <EOL> ) <EOL> new_node = self . driver . ex_set_server_name ( old_node , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , new_node . name ) <EOL> def test_ex_set_metadata ( self ) : <EOL> old_node = Node ( <EOL> id = '<STR_LIT>' , name = None , state = None , <EOL> public_ips = None , private_ips = None , driver = self . driver , <EOL> ) <EOL> metadata = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> returned_metadata = self . driver . ex_set_metadata ( old_node , metadata ) <EOL> self . assertEqual ( metadata , returned_metadata ) <EOL> def test_ex_get_metadata ( self ) : <EOL> node = Node ( <EOL> id = '<STR_LIT>' , name = None , state = None , <EOL> public_ips = None , private_ips = None , driver = self . driver , <EOL> ) <EOL> metadata = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> returned_metadata = self . driver . ex_get_metadata ( node ) <EOL> self . assertEqual ( metadata , returned_metadata ) <EOL> def test_ex_update_node ( self ) : <EOL> old_node = Node ( <EOL> id = '<STR_LIT>' , <EOL> name = None , state = None , public_ips = None , private_ips = None , driver = self . driver , <EOL> ) <EOL> new_node = self . driver . ex_update_node ( old_node , name = '<STR_LIT>' ) <EOL> self . assertTrue ( new_node ) <EOL> self . assertEqual ( '<STR_LIT>' , new_node . name ) <EOL> self . assertEqual ( '<STR_LIT>' , new_node . public_ips [ <NUM_LIT:0> ] ) <EOL> def test_ex_get_node_details ( self ) : <EOL> node_id = '<STR_LIT>' <EOL> node = self . driver . ex_get_node_details ( node_id ) <EOL> self . assertEqual ( node . id , '<STR_LIT>' ) <EOL> self . assertEqual ( node . name , '<STR_LIT>' ) <EOL> def test_ex_get_size ( self ) : <EOL> size_id = '<STR_LIT>' <EOL> size = self . driver . ex_get_size ( size_id ) <EOL> self . assertEqual ( size . id , size_id ) <EOL> self . assertEqual ( size . name , '<STR_LIT>' ) <EOL> def test_get_image ( self ) : <EOL> image_id = '<STR_LIT>' <EOL> image = self . driver . get_image ( image_id ) <EOL> self . assertEqual ( image . id , image_id ) <EOL> self . assertEqual ( image . name , '<STR_LIT>' ) <EOL> self . assertEqual ( image . extra [ '<STR_LIT>' ] , None ) <EOL> self . assertEqual ( image . extra [ '<STR_LIT>' ] , "<STR_LIT:5>" ) <EOL> self . assertEqual ( image . extra [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> def test_delete_image ( self ) : <EOL> image = NodeImage ( <EOL> id = '<STR_LIT>' , name = '<STR_LIT>' , driver = self . driver ) <EOL> result = self . driver . delete_image ( image ) <EOL> self . assertTrue ( result ) <EOL> def test_extract_image_id_from_url ( self ) : <EOL> url = '<STR_LIT>' <EOL> url_two = '<STR_LIT>' <EOL> image_id = self . driver . _extract_image_id_from_url ( url ) <EOL> image_id_two = self . driver . _extract_image_id_from_url ( url_two ) <EOL> self . assertEqual ( image_id , '<STR_LIT>' ) <EOL> self . assertEqual ( image_id_two , '<STR_LIT>' ) <EOL> def test_ex_rescue_with_password ( self ) : <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> n = self . driver . ex_rescue ( node , '<STR_LIT:foo>' ) <EOL> self . assertEqual ( n . extra [ '<STR_LIT:password>' ] , '<STR_LIT:foo>' ) <EOL> def test_ex_rescue_no_password ( self ) : <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> n = self . driver . ex_rescue ( node ) <EOL> self . assertEqual ( n . extra [ '<STR_LIT:password>' ] , '<STR_LIT:foo>' ) <EOL> def test_ex_unrescue ( self ) : <EOL> node = Node ( id = <NUM_LIT> , name = None , state = None , public_ips = None , <EOL> private_ips = None , driver = self . driver ) <EOL> result = self . driver . ex_unrescue ( node ) <EOL> self . assertTrue ( result ) <EOL> def test_ex_get_node_security_groups ( self ) : <EOL> node = Node ( id = '<STR_LIT>' , name = None , <EOL> state = None , public_ips = None , private_ips = None , driver = self . driver ) <EOL> security_groups = self . driver . ex_get_node_security_groups ( node ) <EOL> self . assertEqual ( <EOL> len ( security_groups ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> security_group = security_groups [ <NUM_LIT:1> ] <EOL> self . assertEqual ( security_group . id , <NUM_LIT:4> ) <EOL> self . assertEqual ( security_group . tenant_id , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group . name , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> security_group . description , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . id , <NUM_LIT:1> ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . parent_group_id , <NUM_LIT:4> ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . ip_protocol , "<STR_LIT>" ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . from_port , <NUM_LIT:20> ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . to_port , <NUM_LIT> ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . ip_range , '<STR_LIT>' ) <EOL> def test_ex_list_security_groups ( self ) : <EOL> security_groups = self . driver . ex_list_security_groups ( ) <EOL> self . assertEqual ( <EOL> len ( security_groups ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> security_group = security_groups [ <NUM_LIT:1> ] <EOL> self . assertEqual ( security_group . id , <NUM_LIT:4> ) <EOL> self . assertEqual ( security_group . tenant_id , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group . name , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> security_group . description , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . id , <NUM_LIT:1> ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . parent_group_id , <NUM_LIT:4> ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . ip_protocol , "<STR_LIT>" ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . from_port , <NUM_LIT:20> ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . to_port , <NUM_LIT> ) <EOL> self . assertEqual ( security_group . rules [ <NUM_LIT:0> ] . ip_range , '<STR_LIT>' ) <EOL> def test_ex_create_security_group ( self ) : <EOL> name = '<STR_LIT:test>' <EOL> description = '<STR_LIT>' <EOL> security_group = self . driver . ex_create_security_group ( <EOL> name , description ) <EOL> self . assertEqual ( security_group . id , <NUM_LIT:6> ) <EOL> self . assertEqual ( security_group . tenant_id , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group . name , name ) <EOL> self . assertEqual ( security_group . description , description ) <EOL> self . assertEqual ( len ( security_group . rules ) , <NUM_LIT:0> ) <EOL> def test_ex_delete_security_group ( self ) : <EOL> security_group = OpenStackSecurityGroup ( <EOL> id = <NUM_LIT:6> , tenant_id = None , name = None , description = None , driver = self . driver ) <EOL> result = self . driver . ex_delete_security_group ( security_group ) <EOL> self . assertTrue ( result ) <EOL> def test_ex_create_security_group_rule ( self ) : <EOL> security_group = OpenStackSecurityGroup ( <EOL> id = <NUM_LIT:6> , tenant_id = None , name = None , description = None , driver = self . driver ) <EOL> security_group_rule = self . driver . ex_create_security_group_rule ( <EOL> security_group , '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:16> , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group_rule . id , <NUM_LIT:2> ) <EOL> self . assertEqual ( security_group_rule . parent_group_id , <NUM_LIT:6> ) <EOL> self . assertEqual ( security_group_rule . ip_protocol , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group_rule . from_port , <NUM_LIT> ) <EOL> self . assertEqual ( security_group_rule . to_port , <NUM_LIT:16> ) <EOL> self . assertEqual ( security_group_rule . ip_range , '<STR_LIT>' ) <EOL> self . assertEqual ( security_group_rule . tenant_id , None ) <EOL> def test_ex_delete_security_group_rule ( self ) : <EOL> security_group_rule = OpenStackSecurityGroupRule ( <EOL> id = <NUM_LIT:2> , parent_group_id = None , ip_protocol = None , from_port = None , to_port = None , driver = self . driver ) <EOL> result = self . driver . ex_delete_security_group_rule ( security_group_rule ) <EOL> self . assertTrue ( result ) <EOL> def test_list_key_pairs ( self ) : <EOL> keypairs = self . driver . list_key_pairs ( ) <EOL> self . assertEqual ( len ( keypairs ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> keypair = keypairs [ <NUM_LIT:1> ] <EOL> self . assertEqual ( keypair . name , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> keypair . fingerprint , '<STR_LIT>' ) <EOL> self . assertTrue ( len ( keypair . public_key ) > <NUM_LIT:10> ) <EOL> self . assertEqual ( keypair . private_key , None ) <EOL> def test_get_key_pair ( self ) : <EOL> key_pair = self . driver . get_key_pair ( name = '<STR_LIT>' ) <EOL> self . assertEqual ( key_pair . name , '<STR_LIT>' ) <EOL> def test_get_key_pair_doesnt_exist ( self ) : <EOL> self . assertRaises ( KeyPairDoesNotExistError , <EOL> self . driver . get_key_pair , <EOL> name = '<STR_LIT>' ) <EOL> def test_create_key_pair ( self ) : <EOL> name = '<STR_LIT>' <EOL> keypair = self . driver . create_key_pair ( name = name ) <EOL> self . assertEqual ( keypair . name , name ) <EOL> self . assertEqual ( keypair . fingerprint , <EOL> '<STR_LIT>' ) <EOL> self . assertTrue ( len ( keypair . public_key ) > <NUM_LIT:10> ) <EOL> self . assertTrue ( len ( keypair . private_key ) > <NUM_LIT:10> ) <EOL> def test_import_key_pair_from_file ( self ) : <EOL> name = '<STR_LIT>' <EOL> path = os . path . join ( <EOL> os . path . dirname ( __file__ ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> pub_key = open ( path , '<STR_LIT:r>' ) . read ( ) <EOL> keypair = self . driver . import_key_pair_from_file ( name = name , <EOL> key_file_path = path ) <EOL> self . assertEqual ( keypair . name , name ) <EOL> self . assertEqual ( <EOL> keypair . fingerprint , '<STR_LIT>' ) <EOL> self . assertEqual ( keypair . public_key , pub_key ) <EOL> self . assertEqual ( keypair . private_key , None ) <EOL> def test_import_key_pair_from_string ( self ) : <EOL> name = '<STR_LIT>' <EOL> path = os . path . join ( <EOL> os . path . dirname ( __file__ ) , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> pub_key = open ( path , '<STR_LIT:r>' ) . read ( ) <EOL> keypair = self . driver . import_key_pair_from_string ( name = name , <EOL> key_material = pub_key ) <EOL> self . assertEqual ( keypair . name , name ) <EOL> self . assertEqual ( <EOL> keypair . fingerprint , '<STR_LIT>' ) <EOL> self . assertEqual ( keypair . public_key , pub_key ) <EOL> self . assertEqual ( keypair . private_key , None ) <EOL> def test_delete_key_pair ( self ) : <EOL> keypair = OpenStackKeyPair ( <EOL> name = '<STR_LIT>' , fingerprint = None , public_key = None , driver = self . driver ) <EOL> result = self . driver . delete_key_pair ( key_pair = keypair ) <EOL> self . assertTrue ( result ) <EOL> def test_ex_list_floating_ip_pools ( self ) : <EOL> ret = self . driver . ex_list_floating_ip_pools ( ) <EOL> self . assertEqual ( ret [ <NUM_LIT:0> ] . name , '<STR_LIT>' ) <EOL> self . assertEqual ( ret [ <NUM_LIT:1> ] . name , '<STR_LIT>' ) <EOL> def test_ex_attach_floating_ip_to_node ( self ) : <EOL> image = NodeImage ( <EOL> id = <NUM_LIT:11> , name = '<STR_LIT>' , driver = self . driver ) <EOL> size = NodeSize ( <EOL> <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size ) <EOL> node . id = <NUM_LIT> <EOL> ip = '<STR_LIT>' <EOL> self . assertTrue ( self . driver . ex_attach_floating_ip_to_node ( node , ip ) ) <EOL> def test_detach_floating_ip_from_node ( self ) : <EOL> image = NodeImage ( <EOL> id = <NUM_LIT:11> , name = '<STR_LIT>' , driver = self . driver ) <EOL> size = NodeSize ( <EOL> <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT>' , image = image , size = size ) <EOL> node . id = <NUM_LIT> <EOL> ip = '<STR_LIT>' <EOL> self . assertTrue ( self . driver . ex_detach_floating_ip_from_node ( node , ip ) ) <EOL> def test_OpenStack_1_1_FloatingIpPool_list_floating_ips ( self ) : <EOL> pool = OpenStack_1_1_FloatingIpPool ( '<STR_LIT:foo>' , self . driver . connection ) <EOL> ret = pool . list_floating_ips ( ) <EOL> self . assertEqual ( ret [ <NUM_LIT:0> ] . id , '<STR_LIT>' ) <EOL> self . assertEqual ( ret [ <NUM_LIT:0> ] . pool , pool ) <EOL> self . assertEqual ( ret [ <NUM_LIT:0> ] . ip_address , '<STR_LIT>' ) <EOL> self . assertEqual ( ret [ <NUM_LIT:0> ] . node_id , None ) <EOL> self . assertEqual ( ret [ <NUM_LIT:1> ] . id , '<STR_LIT>' ) <EOL> self . assertEqual ( ret [ <NUM_LIT:1> ] . pool , pool ) <EOL> self . assertEqual ( ret [ <NUM_LIT:1> ] . ip_address , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> ret [ <NUM_LIT:1> ] . node_id , '<STR_LIT>' ) <EOL> def test_OpenStack_1_1_FloatingIpPool_get_floating_ip ( self ) : <EOL> pool = OpenStack_1_1_FloatingIpPool ( '<STR_LIT:foo>' , self . driver . connection ) <EOL> ret = pool . get_floating_ip ( '<STR_LIT>' ) <EOL> self . assertEqual ( ret . id , '<STR_LIT>' ) <EOL> self . assertEqual ( ret . pool , pool ) <EOL> self . assertEqual ( ret . ip_address , '<STR_LIT>' ) <EOL> self . assertEqual ( ret . node_id , None ) <EOL> def test_OpenStack_1_1_FloatingIpPool_create_floating_ip ( self ) : <EOL> pool = OpenStack_1_1_FloatingIpPool ( '<STR_LIT:foo>' , self . driver . connection ) <EOL> ret = pool . create_floating_ip ( ) <EOL> self . assertEqual ( ret . id , '<STR_LIT>' ) <EOL> self . assertEqual ( ret . pool , pool ) <EOL> self . assertEqual ( ret . ip_address , '<STR_LIT>' ) <EOL> self . assertEqual ( ret . node_id , None ) <EOL> def test_OpenStack_1_1_FloatingIpPool_delete_floating_ip ( self ) : <EOL> pool = OpenStack_1_1_FloatingIpPool ( '<STR_LIT:foo>' , self . driver . connection ) <EOL> ip = OpenStack_1_1_FloatingIpAddress ( '<STR_LIT>' , '<STR_LIT>' , pool ) <EOL> self . assertTrue ( pool . delete_floating_ip ( ip ) ) <EOL> def test_OpenStack_1_1_FloatingIpAddress_delete ( self ) : <EOL> pool = OpenStack_1_1_FloatingIpPool ( '<STR_LIT:foo>' , self . driver . connection ) <EOL> pool . delete_floating_ip = Mock ( ) <EOL> ip = OpenStack_1_1_FloatingIpAddress ( '<STR_LIT>' , '<STR_LIT>' , pool ) <EOL> ip . pool . delete_floating_ip ( ) <EOL> self . assertEqual ( pool . delete_floating_ip . call_count , <NUM_LIT:1> ) <EOL> def test_ex_list_network ( self ) : <EOL> networks = self . driver . ex_list_networks ( ) <EOL> network = networks [ <NUM_LIT:0> ] <EOL> self . assertEqual ( len ( networks ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( network . name , '<STR_LIT>' ) <EOL> self . assertEqual ( network . cidr , '<STR_LIT>' ) <EOL> def test_ex_create_network ( self ) : <EOL> network = self . driver . ex_create_network ( name = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' ) <EOL> self . assertEqual ( network . name , '<STR_LIT>' ) <EOL> self . assertEqual ( network . cidr , '<STR_LIT>' ) <EOL> def test_ex_delete_network ( self ) : <EOL> network = self . driver . ex_list_networks ( ) [ <NUM_LIT:0> ] <EOL> self . assertTrue ( self . driver . ex_delete_network ( network = network ) ) <EOL> def test_ex_get_metadata_for_node ( self ) : <EOL> image = NodeImage ( id = <NUM_LIT:11> , name = '<STR_LIT>' , driver = self . driver ) <EOL> size = NodeSize ( <NUM_LIT:1> , '<STR_LIT>' , None , None , None , None , driver = self . driver ) <EOL> node = self . driver . create_node ( name = '<STR_LIT:foo>' , <EOL> image = image , <EOL> size = size ) <EOL> metadata = self . driver . ex_get_metadata_for_node ( node ) <EOL> self . assertEqual ( metadata [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( metadata ) , <NUM_LIT:1> ) <EOL> def test_ex_pause_node ( self ) : <EOL> node = Node ( <EOL> id = '<STR_LIT>' , name = None , state = None , <EOL> public_ips = None , private_ips = None , driver = self . driver , <EOL> ) <EOL> ret = self . driver . ex_pause_node ( node ) <EOL> self . assertTrue ( ret is True ) <EOL> def test_ex_unpause_node ( self ) : <EOL> node = Node ( <EOL> id = '<STR_LIT>' , name = None , state = None , <EOL> public_ips = None , private_ips = None , driver = self . driver , <EOL> ) <EOL> ret = self . driver . ex_unpause_node ( node ) <EOL> self . assertTrue ( ret is True ) <EOL> def test_ex_suspend_node ( self ) : <EOL> node = Node ( <EOL> id = '<STR_LIT>' , name = None , state = None , <EOL> public_ips = None , private_ips = None , driver = self . driver , <EOL> ) <EOL> ret = self . driver . ex_suspend_node ( node ) <EOL> self . assertTrue ( ret is True ) <EOL> def test_ex_resume_node ( self ) : <EOL> node = Node ( <EOL> id = '<STR_LIT>' , name = None , state = None , <EOL> public_ips = None , private_ips = None , driver = self . driver , <EOL> ) <EOL> ret = self . driver . ex_resume_node ( node ) <EOL> self . assertTrue ( ret is True ) <EOL> def test_ex_get_console_output ( self ) : <EOL> node = Node ( <EOL> id = '<STR_LIT>' , name = None , state = None , <EOL> public_ips = None , private_ips = None , driver = self . driver , <EOL> ) <EOL> resp = self . driver . ex_get_console_output ( node ) <EOL> expected_output = '<STR_LIT>' <EOL> self . assertEqual ( resp [ '<STR_LIT>' ] , expected_output ) <EOL> def test_ex_list_snapshots ( self ) : <EOL> if self . driver_type . type == '<STR_LIT>' : <EOL> self . conn_classes [ <NUM_LIT:0> ] . type = '<STR_LIT>' <EOL> self . conn_classes [ <NUM_LIT:1> ] . type = '<STR_LIT>' <EOL> snapshots = self . driver . ex_list_snapshots ( ) <EOL> self . assertEqual ( len ( snapshots ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( snapshots [ <NUM_LIT:0> ] . created , datetime . datetime ( <NUM_LIT> , <NUM_LIT:2> , <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT:50> , <NUM_LIT:7> , tzinfo = UTC ) ) <EOL> self . assertEqual ( snapshots [ <NUM_LIT:0> ] . extra [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> self . assertEqual ( snapshots [ <NUM_LIT:0> ] . extra [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( snapshots [ <NUM_LIT:0> ] . state , VolumeSnapshotState . AVAILABLE ) <EOL> assert snapshots [ <NUM_LIT:2> ] . created is None <EOL> def test_list_volume_snapshots ( self ) : <EOL> volume = self . driver . list_volumes ( ) [ <NUM_LIT:0> ] <EOL> if self . driver_type . type == '<STR_LIT>' : <EOL> self . conn_classes [ <NUM_LIT:0> ] . type = '<STR_LIT>' <EOL> self . conn_classes [ <NUM_LIT:1> ] . type = '<STR_LIT>' <EOL> snapshots = self . driver . list_volume_snapshots ( volume ) <EOL> self . assertEqual ( len ( snapshots ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( snapshots [ <NUM_LIT:0> ] . id , '<STR_LIT>' ) <EOL> def test_create_volume_snapshot ( self ) : <EOL> volume = self . driver . list_volumes ( ) [ <NUM_LIT:0> ] <EOL> if self . driver_type . type == '<STR_LIT>' : <EOL> self . conn_classes [ <NUM_LIT:0> ] . type = '<STR_LIT>' <EOL> self . conn_classes [ <NUM_LIT:1> ] . type = '<STR_LIT>' <EOL> ret = self . driver . create_volume_snapshot ( volume , <EOL> '<STR_LIT>' , <EOL> ex_description = '<STR_LIT>' , <EOL> ex_force = True ) <EOL> self . assertEqual ( ret . id , '<STR_LIT>' ) <EOL> def test_ex_create_snapshot ( self ) : <EOL> volume = self . driver . list_volumes ( ) [ <NUM_LIT:0> ] <EOL> if self . driver_type . type == '<STR_LIT>' : <EOL> self . conn_classes [ <NUM_LIT:0> ] . type = '<STR_LIT>' <EOL> self . conn_classes [ <NUM_LIT:1> ] . type = '<STR_LIT>' <EOL> ret = self . driver . ex_create_snapshot ( volume , <EOL> '<STR_LIT>' , <EOL> description = '<STR_LIT>' , <EOL> force = True ) <EOL> self . assertEqual ( ret . id , '<STR_LIT>' ) <EOL> def test_destroy_volume_snapshot ( self ) : <EOL> if self . driver_type . type == '<STR_LIT>' : <EOL> self . conn_classes [ <NUM_LIT:0> ] . type = '<STR_LIT>' <EOL> self . conn_classes [ <NUM_LIT:1> ] . type = '<STR_LIT>' <EOL> snapshot = self . driver . ex_list_snapshots ( ) [ <NUM_LIT:0> ] <EOL> ret = self . driver . destroy_volume_snapshot ( snapshot ) <EOL> self . assertTrue ( ret ) <EOL> def test_ex_delete_snapshot ( self ) : <EOL> if self . driver_type . type == '<STR_LIT>' : <EOL> self . conn_classes [ <NUM_LIT:0> ] . type = '<STR_LIT>' <EOL> self . conn_classes [ <NUM_LIT:1> ] . type = '<STR_LIT>' <EOL> snapshot = self . driver . ex_list_snapshots ( ) [ <NUM_LIT:0> ] <EOL> ret = self . driver . ex_delete_snapshot ( snapshot ) <EOL> self . assertTrue ( ret ) <EOL> class OpenStack_1_1_FactoryMethodTests ( OpenStack_1_1_Tests ) : <EOL> should_list_locations = False <EOL> should_list_volumes = True <EOL> driver_klass = OpenStack_1_1_NodeDriver <EOL> driver_type = get_driver ( Provider . OPENSTACK ) <EOL> driver_args = OPENSTACK_PARAMS + ( '<STR_LIT>' , ) <EOL> driver_kwargs = { '<STR_LIT>' : '<STR_LIT>' } <EOL> class OpenStack_1_1_MockHttp ( MockHttpTestCase ) : <EOL> fixtures = ComputeFileFixtures ( '<STR_LIT>' ) <EOL> auth_fixtures = OpenStackFixtures ( ) <EOL> json_content_headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> def _v2_0_tokens ( self , method , url , body , headers ) : <EOL> body = self . auth_fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_0 ( self , method , url , body , headers ) : <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , headers , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> def _v1_1_slug_servers_detail ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers_detail_ERROR_STATE_NO_IMAGE_ID ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_flavors_detail ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_images_detail ( self , method , url , body , headers ) : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:POST>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> elif method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers_26f7fbee_8ce1_4c28_887a_bfe8e4bb10fe ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( <EOL> '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers_12065_action ( self , method , url , body , headers ) : <EOL> if method != "<STR_LIT:POST>" : <EOL> self . fail ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , { } , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_1_slug_servers_12064_action ( self , method , url , body , headers ) : <EOL> if method != "<STR_LIT:POST>" : <EOL> self . fail ( '<STR_LIT>' ) <EOL> if "<STR_LIT>" in json . loads ( body ) : <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , <EOL> { "<STR_LIT:location>" : "<STR_LIT>" } , <EOL> httplib . responses [ httplib . ACCEPTED ] ) <EOL> elif "<STR_LIT>" in json . loads ( body ) : <EOL> return ( httplib . OK , '<STR_LIT>' , { } , <EOL> httplib . responses [ httplib . OK ] ) <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , { } , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_1_slug_servers_12066_action ( self , method , url , body , headers ) : <EOL> if method != "<STR_LIT:POST>" : <EOL> self . fail ( '<STR_LIT>' ) <EOL> if "<STR_LIT>" not in json . loads ( body ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . assertTrue ( '<STR_LIT>' in body , <EOL> msg = "<STR_LIT>" + body ) <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , { } , httplib . responses [ httplib . ACCEPTED ] ) <EOL> def _v1_1_slug_servers_12065 ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT>" : <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , { } , httplib . responses [ httplib . ACCEPTED ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_servers_12064 ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> elif method == "<STR_LIT>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> elif method == "<STR_LIT>" : <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , { } , httplib . responses [ httplib . ACCEPTED ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_servers_12062 ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers_12063_metadata ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> elif method == "<STR_LIT>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers_EX_DISK_CONFIG ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:POST>" : <EOL> body = u ( body ) <EOL> self . assertTrue ( body . find ( '<STR_LIT>' ) ) <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_flavors_7 ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_images_13 ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_images_26365521_8c62_11f9_2c33_283d153ecc3a ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT>" : <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , { } , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_images_4949f9ee_2421_4c81_8b49_13119446008b ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( <EOL> '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_servers_1c01300f_ef97_4937_8f03_ac676d6234be_os_security_groups ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( <EOL> '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_security_groups ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> elif method == "<STR_LIT:POST>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_security_groups_6 ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT>" : <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , { } , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_os_security_group_rules ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:POST>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_security_group_rules_2 ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT>" : <EOL> return ( httplib . NO_CONTENT , "<STR_LIT>" , { } , httplib . responses [ httplib . NO_CONTENT ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_os_keypairs ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> elif method == "<STR_LIT:POST>" : <EOL> if '<STR_LIT>' in body : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_keypairs_test_key_pair ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT:GET>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_keypairs_doesnt_exist ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT:GET>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . NOT_FOUND , body , self . json_content_headers , <EOL> httplib . responses [ httplib . NOT_FOUND ] ) <EOL> def _v1_1_slug_os_keypairs_key1 ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT>" : <EOL> return ( httplib . ACCEPTED , "<STR_LIT>" , { } , httplib . responses [ httplib . ACCEPTED ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_os_volumes ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> elif method == "<STR_LIT:POST>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_volumes_cd76a3a1_c4ce_40f6_9b9f_07a61508938d ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( <EOL> '<STR_LIT>' ) <EOL> elif method == "<STR_LIT>" : <EOL> body = '<STR_LIT>' <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers_12065_os_volume_attachments ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:POST>" : <EOL> body = self . fixtures . load ( <EOL> '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers_12065_os_volume_attachments_cd76a3a1_c4ce_40f6_9b9f_07a61508938d ( self , method , url , body , <EOL> headers ) : <EOL> if method == "<STR_LIT>" : <EOL> body = '<STR_LIT>' <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_floating_ip_pools ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_os_floating_ips_foo_bar_id ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT>" : <EOL> body = '<STR_LIT>' <EOL> return ( httplib . ACCEPTED , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_os_floating_ips ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:GET>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> elif method == "<STR_LIT:POST>" : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_servers_4242_action ( self , method , url , body , headers ) : <EOL> if method == "<STR_LIT:POST>" : <EOL> body = '<STR_LIT>' <EOL> return ( httplib . ACCEPTED , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_networks ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT:GET>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> elif method == '<STR_LIT:POST>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_os_networks_f13e5051_feea_416b_827a_1a0acc2dad14 ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT>' : <EOL> body = '<STR_LIT>' <EOL> return ( httplib . ACCEPTED , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_servers_72258_action ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT:POST>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers_12063_action ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT:POST>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_servers_12086_action ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT:POST>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> return ( httplib . ACCEPTED , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> def _v1_1_slug_os_snapshots ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT:GET>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> elif method == '<STR_LIT:POST>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_snapshots_RACKSPACE ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT:GET>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> elif method == '<STR_LIT:POST>' : <EOL> body = self . fixtures . load ( '<STR_LIT>' ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( httplib . OK , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_snapshots_3fbbcccf_d058_4502_8844_6feeffdf4cb5 ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT>' : <EOL> body = '<STR_LIT>' <EOL> status_code = httplib . NO_CONTENT <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( status_code , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> def _v1_1_slug_os_snapshots_3fbbcccf_d058_4502_8844_6feeffdf4cb5_RACKSPACE ( self , method , url , body , headers ) : <EOL> if method == '<STR_LIT>' : <EOL> body = '<STR_LIT>' <EOL> status_code = httplib . NO_CONTENT <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> return ( status_code , body , self . json_content_headers , httplib . responses [ httplib . OK ] ) <EOL> class OpenStack_2_0_MockHttp ( OpenStack_1_1_MockHttp ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( OpenStack_2_0_MockHttp , self ) . __init__ ( * args , ** kwargs ) <EOL> methods1 = OpenStack_1_1_MockHttp . __dict__ <EOL> names1 = [ m for m in methods1 if m . find ( '<STR_LIT>' ) == <NUM_LIT:0> ] <EOL> for name in names1 : <EOL> method = methods1 [ name ] <EOL> new_name = name . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> setattr ( self , new_name , method_type ( method , self , <EOL> OpenStack_2_0_MockHttp ) ) <EOL> class OpenStack_1_1_Auth_2_0_Tests ( OpenStack_1_1_Tests ) : <EOL> driver_args = OPENSTACK_PARAMS + ( '<STR_LIT>' , ) <EOL> driver_kwargs = { '<STR_LIT>' : '<STR_LIT>' } <EOL> def setUp ( self ) : <EOL> self . driver_klass . connectionCls . conn_classes = ( OpenStack_2_0_MockHttp , OpenStack_2_0_MockHttp ) <EOL> self . driver_klass . connectionCls . auth_url = "<STR_LIT>" <EOL> OpenStackMockHttp . type = None <EOL> OpenStack_1_1_MockHttp . type = None <EOL> OpenStack_2_0_MockHttp . type = None <EOL> self . driver = self . create_driver ( ) <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> clear_pricing_data ( ) <EOL> self . node = self . driver . list_nodes ( ) [ <NUM_LIT:1> ] <EOL> def test_auth_user_info_is_set ( self ) : <EOL> self . driver . connection . _populate_hosts_and_request_paths ( ) <EOL> self . assertEqual ( self . driver . connection . auth_user_info , { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' } ] } ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> sys . exit ( unittest . main ( ) ) </s>
