<s> import mock <EOL> import fixtures <EOL> from oslo_log import log as logging <EOL> from nova . tests . functional . api import client <EOL> from nova . tests . functional . test_servers import ServersTestBase <EOL> from nova . tests . unit import fake_network <EOL> from nova . tests . unit . virt . libvirt import fake_libvirt_utils <EOL> from nova . tests . unit . virt . libvirt import fakelibvirt <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class NumaHostInfo ( fakelibvirt . HostInfo ) : <EOL> def __init__ ( self , ** kwargs ) : <EOL> super ( NumaHostInfo , self ) . __init__ ( ** kwargs ) <EOL> self . numa_mempages_list = [ ] <EOL> def get_numa_topology ( self ) : <EOL> if self . numa_topology : <EOL> return self . numa_topology <EOL> topology = self . _gen_numa_topology ( self . cpu_nodes , self . cpu_sockets , <EOL> self . cpu_cores , self . cpu_threads , <EOL> self . kB_mem ) <EOL> self . numa_topology = topology <EOL> cpu_count = len ( topology . cells ) * len ( topology . cells [ <NUM_LIT:0> ] . cpus ) <EOL> self . cpus = cpu_count - len ( self . disabled_cpus_list ) <EOL> return topology <EOL> def set_custom_numa_toplogy ( self , topology ) : <EOL> self . numa_topology = topology <EOL> class RealTimeServersTest ( ServersTestBase ) : <EOL> def setUp ( self ) : <EOL> super ( RealTimeServersTest , self ) . setUp ( ) <EOL> self . useFixture ( fixtures . MonkeyPatch ( <EOL> '<STR_LIT>' , <EOL> fake_libvirt_utils ) ) <EOL> self . useFixture ( fixtures . MonkeyPatch ( <EOL> '<STR_LIT>' , <EOL> fakelibvirt ) ) <EOL> self . useFixture ( fixtures . MonkeyPatch ( <EOL> '<STR_LIT>' , <EOL> fakelibvirt ) ) <EOL> self . useFixture ( fixtures . MonkeyPatch ( <EOL> '<STR_LIT>' , <EOL> fakelibvirt ) ) <EOL> self . useFixture ( fakelibvirt . FakeLibvirtFixture ( ) ) <EOL> self . flags ( sysinfo_serial = '<STR_LIT:none>' , group = '<STR_LIT>' ) <EOL> def _setup_compute_service ( self ) : <EOL> self . flags ( compute_driver = '<STR_LIT>' ) <EOL> def test_no_dedicated_cpu ( self ) : <EOL> flavor = self . _create_flavor ( extra_spec = { '<STR_LIT>' : '<STR_LIT:yes>' } ) <EOL> server = self . _build_server ( flavor ) <EOL> self . assertRaises ( <EOL> client . OpenStackApiException , <EOL> self . api . post_server , { '<STR_LIT>' : server } ) <EOL> def test_no_realtime_mask ( self ) : <EOL> flavor = self . _create_flavor ( extra_spec = { <EOL> '<STR_LIT>' : '<STR_LIT:yes>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> server = self . _build_server ( flavor ) <EOL> self . assertRaises ( <EOL> client . OpenStackApiException , <EOL> self . api . post_server , { '<STR_LIT>' : server } ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_invalid_libvirt_version ( self , img_mock ) : <EOL> host_info = NumaHostInfo ( cpu_nodes = <NUM_LIT:2> , cpu_sockets = <NUM_LIT:1> , cpu_cores = <NUM_LIT:2> , <EOL> cpu_threads = <NUM_LIT:2> , kB_mem = <NUM_LIT> ) <EOL> fake_connection = fakelibvirt . Connection ( '<STR_LIT>' , <EOL> version = <NUM_LIT> , <EOL> hv_version = <NUM_LIT> , <EOL> host_info = host_info ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = fake_connection ) : <EOL> self . compute = self . start_service ( '<STR_LIT>' , host = '<STR_LIT>' ) <EOL> fake_network . set_stub_network_methods ( self ) <EOL> flavor = self . _create_flavor ( extra_spec = { <EOL> '<STR_LIT>' : '<STR_LIT:yes>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> server = self . _build_server ( flavor ) <EOL> created = self . api . post_server ( { '<STR_LIT>' : server } ) <EOL> instance = self . api . get_server ( created [ '<STR_LIT:id>' ] ) <EOL> instance = self . _wait_for_state_change ( instance , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , instance [ '<STR_LIT:status>' ] ) <EOL> self . _delete_server ( instance [ '<STR_LIT:id>' ] ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_success ( self , img_mock ) : <EOL> host_info = NumaHostInfo ( cpu_nodes = <NUM_LIT:2> , cpu_sockets = <NUM_LIT:1> , cpu_cores = <NUM_LIT:2> , <EOL> cpu_threads = <NUM_LIT:2> , kB_mem = <NUM_LIT> ) <EOL> fake_connection = fakelibvirt . Connection ( '<STR_LIT>' , <EOL> version = <NUM_LIT> , <EOL> hv_version = <NUM_LIT> , <EOL> host_info = host_info ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = fake_connection ) : <EOL> self . compute = self . start_service ( '<STR_LIT>' , host = '<STR_LIT>' ) <EOL> fake_network . set_stub_network_methods ( self ) <EOL> flavor = self . _create_flavor ( extra_spec = { <EOL> '<STR_LIT>' : '<STR_LIT:yes>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> server = self . _build_server ( flavor ) <EOL> created = self . api . post_server ( { '<STR_LIT>' : server } ) <EOL> instance = self . api . get_server ( created [ '<STR_LIT:id>' ] ) <EOL> instance = self . _wait_for_state_change ( instance , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , instance [ '<STR_LIT:status>' ] ) <EOL> self . _delete_server ( instance [ '<STR_LIT:id>' ] ) </s>
