<s> import mock <EOL> import fixtures <EOL> from oslo_config import cfg <EOL> from oslo_log import log as logging <EOL> from nova import test <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> CONF = cfg . CONF <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 NUMAServersTest ( ServersTestBase ) : <EOL> def setUp ( self ) : <EOL> super ( NUMAServersTest , 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> def _setup_compute_service ( self ) : <EOL> pass <EOL> def _setup_scheduler_service ( self ) : <EOL> self . flags ( compute_driver = '<STR_LIT>' ) <EOL> self . flags ( scheduler_driver = '<STR_LIT>' ) <EOL> self . flags ( scheduler_default_filters = CONF . scheduler_default_filters <EOL> + [ '<STR_LIT>' ] ) <EOL> return self . start_service ( '<STR_LIT>' ) <EOL> def _run_build_test ( self , flavor_id , filter_mock , end_status = '<STR_LIT>' ) : <EOL> self . compute = self . start_service ( '<STR_LIT>' , host = '<STR_LIT>' ) <EOL> fake_network . set_stub_network_methods ( self ) <EOL> good_server = self . _build_server ( flavor_id ) <EOL> post = { '<STR_LIT>' : good_server } <EOL> created_server = self . api . post_server ( post ) <EOL> LOG . debug ( "<STR_LIT>" % created_server ) <EOL> self . assertTrue ( created_server [ '<STR_LIT:id>' ] ) <EOL> created_server_id = created_server [ '<STR_LIT:id>' ] <EOL> found_server = self . api . get_server ( created_server_id ) <EOL> self . assertEqual ( created_server_id , found_server [ '<STR_LIT:id>' ] ) <EOL> servers = self . api . get_servers ( ) <EOL> server_ids = [ s [ '<STR_LIT:id>' ] for s in servers ] <EOL> self . assertIn ( created_server_id , server_ids ) <EOL> self . assertTrue ( filter_mock . called ) <EOL> found_server = self . _wait_for_state_change ( found_server , '<STR_LIT>' ) <EOL> self . assertEqual ( end_status , found_server [ '<STR_LIT:status>' ] ) <EOL> self . _delete_server ( created_server_id ) <EOL> def _get_topology_filter_spy ( self ) : <EOL> host_manager = self . scheduler . manager . driver . host_manager <EOL> numa_filter_class = host_manager . filter_cls_map [ '<STR_LIT>' ] <EOL> host_pass_mock = mock . Mock ( wraps = numa_filter_class ( ) . host_passes ) <EOL> return host_pass_mock <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_create_server_with_numa_topology ( 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> extra_spec = { '<STR_LIT>' : '<STR_LIT:2>' } <EOL> flavor_id = self . _create_flavor ( extra_spec = extra_spec ) <EOL> host_pass_mock = self . _get_topology_filter_spy ( ) <EOL> with test . nested ( <EOL> mock . patch ( '<STR_LIT>' , <EOL> return_value = fake_connection ) , <EOL> mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> side_effect = host_pass_mock ) ) as ( conn_mock , <EOL> filter_mock ) : <EOL> self . _run_build_test ( flavor_id , filter_mock ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_create_server_with_numa_fails ( self , img_mock ) : <EOL> host_info = NumaHostInfo ( cpu_nodes = <NUM_LIT:1> , cpu_sockets = <NUM_LIT:1> , cpu_cores = <NUM_LIT:2> , <EOL> kB_mem = <NUM_LIT> ) <EOL> fake_connection = fakelibvirt . Connection ( '<STR_LIT>' , <EOL> version = <NUM_LIT> , <EOL> host_info = host_info ) <EOL> extra_spec = { '<STR_LIT>' : '<STR_LIT:2>' } <EOL> flavor_id = self . _create_flavor ( extra_spec = extra_spec ) <EOL> host_pass_mock = self . _get_topology_filter_spy ( ) <EOL> with test . nested ( <EOL> mock . patch ( '<STR_LIT>' , <EOL> return_value = fake_connection ) , <EOL> mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> side_effect = host_pass_mock ) ) as ( conn_mock , <EOL> filter_mock ) : <EOL> self . _run_build_test ( flavor_id , filter_mock , end_status = '<STR_LIT>' ) </s>
