<s> import copy <EOL> import uuid <EOL> import mock <EOL> from oslo_serialization import jsonutils <EOL> from nova import exception <EOL> from nova import objects <EOL> from nova . objects import fields <EOL> from nova . tests . unit . objects import test_objects <EOL> fake_instance_uuid = str ( uuid . uuid4 ( ) ) <EOL> fake_obj_numa_topology = objects . InstanceNUMATopology ( <EOL> instance_uuid = fake_instance_uuid , <EOL> cells = [ <EOL> objects . InstanceNUMACell ( <EOL> id = <NUM_LIT:0> , cpuset = set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) , memory = <NUM_LIT> , pagesize = <NUM_LIT> ) , <EOL> objects . InstanceNUMACell ( <EOL> id = <NUM_LIT:1> , cpuset = set ( [ <NUM_LIT:3> , <NUM_LIT:4> ] ) , memory = <NUM_LIT> , pagesize = <NUM_LIT> ) <EOL> ] ) <EOL> fake_numa_topology = fake_obj_numa_topology . _to_dict ( ) <EOL> fake_db_topology = { <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT:id>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : fake_instance_uuid , <EOL> '<STR_LIT>' : fake_obj_numa_topology . _to_json ( ) <EOL> } <EOL> fake_old_db_topology = dict ( fake_db_topology ) <EOL> fake_old_db_topology [ '<STR_LIT>' ] = jsonutils . dumps ( fake_numa_topology ) <EOL> def get_fake_obj_numa_topology ( context ) : <EOL> fake_obj_numa_topology_cpy = fake_obj_numa_topology . obj_clone ( ) <EOL> fake_obj_numa_topology_cpy . _context = context <EOL> return fake_obj_numa_topology_cpy <EOL> class _TestInstanceNUMATopology ( object ) : <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_create ( self , mock_update ) : <EOL> topo_obj = get_fake_obj_numa_topology ( self . context ) <EOL> topo_obj . instance_uuid = fake_db_topology [ '<STR_LIT>' ] <EOL> topo_obj . create ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( mock_update . call_args_list ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_save ( self , mock_update ) : <EOL> topo_obj = get_fake_obj_numa_topology ( self . context ) <EOL> topo_obj . instance_uuid = fake_db_topology [ '<STR_LIT>' ] <EOL> topo_obj . _save ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( mock_update . call_args_list ) ) <EOL> def _test_get_by_instance_uuid ( self ) : <EOL> numa_topology = objects . InstanceNUMATopology . get_by_instance_uuid ( <EOL> self . context , fake_db_topology [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( fake_db_topology [ '<STR_LIT>' ] , <EOL> numa_topology . instance_uuid ) <EOL> for obj_cell , topo_cell in zip ( <EOL> numa_topology . cells , fake_obj_numa_topology [ '<STR_LIT>' ] ) : <EOL> self . assertIsInstance ( obj_cell , objects . InstanceNUMACell ) <EOL> self . assertEqual ( topo_cell . id , obj_cell . id ) <EOL> self . assertEqual ( topo_cell . cpuset , obj_cell . cpuset ) <EOL> self . assertEqual ( topo_cell . memory , obj_cell . memory ) <EOL> self . assertEqual ( topo_cell . pagesize , obj_cell . pagesize ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_by_instance_uuid ( self , mock_get ) : <EOL> mock_get . return_value = fake_db_topology <EOL> self . _test_get_by_instance_uuid ( ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_by_instance_uuid_old ( self , mock_get ) : <EOL> mock_get . return_value = fake_old_db_topology <EOL> self . _test_get_by_instance_uuid ( ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_by_instance_uuid_missing ( self , mock_get ) : <EOL> mock_get . return_value = None <EOL> self . assertRaises ( <EOL> exception . NumaTopologyNotFound , <EOL> objects . InstanceNUMATopology . get_by_instance_uuid , <EOL> self . context , '<STR_LIT>' ) <EOL> def test_siblings ( self ) : <EOL> inst_cell = objects . InstanceNUMACell ( <EOL> cpuset = set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> self . assertEqual ( [ ] , inst_cell . siblings ) <EOL> topo = objects . VirtCPUTopology ( sockets = <NUM_LIT:1> , cores = <NUM_LIT:3> , threads = <NUM_LIT:0> ) <EOL> inst_cell = objects . InstanceNUMACell ( <EOL> cpuset = set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) , cpu_topology = topo ) <EOL> self . assertEqual ( [ ] , inst_cell . siblings ) <EOL> topo = objects . VirtCPUTopology ( sockets = <NUM_LIT:1> , cores = <NUM_LIT:3> , threads = <NUM_LIT:1> ) <EOL> inst_cell = objects . InstanceNUMACell ( <EOL> cpuset = set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) , cpu_topology = topo ) <EOL> self . assertEqual ( [ ] , inst_cell . siblings ) <EOL> topo = objects . VirtCPUTopology ( sockets = <NUM_LIT:1> , cores = <NUM_LIT:2> , threads = <NUM_LIT:2> ) <EOL> inst_cell = objects . InstanceNUMACell ( <EOL> cpuset = set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , cpu_topology = topo ) <EOL> self . assertEqual ( [ set ( [ <NUM_LIT:0> , <NUM_LIT:1> ] ) , set ( [ <NUM_LIT:2> , <NUM_LIT:3> ] ) ] , inst_cell . siblings ) <EOL> topo = objects . VirtCPUTopology ( sockets = <NUM_LIT:1> , cores = <NUM_LIT:1> , threads = <NUM_LIT:4> ) <EOL> inst_cell = objects . InstanceNUMACell ( <EOL> cpuset = set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , cpu_topology = topo ) <EOL> self . assertEqual ( [ set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) ] , inst_cell . siblings ) <EOL> def test_pin ( self ) : <EOL> inst_cell = objects . InstanceNUMACell ( cpuset = set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , <EOL> cpu_pinning = None ) <EOL> inst_cell . pin ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . assertEqual ( { <NUM_LIT:0> : <NUM_LIT> } , inst_cell . cpu_pinning ) <EOL> inst_cell . pin ( <NUM_LIT:12> , <NUM_LIT> ) <EOL> self . assertEqual ( { <NUM_LIT:0> : <NUM_LIT> } , inst_cell . cpu_pinning ) <EOL> inst_cell . pin ( <NUM_LIT:1> , <NUM_LIT:16> ) <EOL> self . assertEqual ( { <NUM_LIT:0> : <NUM_LIT> , <NUM_LIT:1> : <NUM_LIT:16> } , inst_cell . cpu_pinning ) <EOL> def test_pin_vcpus ( self ) : <EOL> inst_cell = objects . InstanceNUMACell ( cpuset = set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , <EOL> cpu_pinning = None ) <EOL> inst_cell . pin_vcpus ( ( <NUM_LIT:0> , <NUM_LIT> ) , ( <NUM_LIT:1> , <NUM_LIT:15> ) , ( <NUM_LIT:2> , <NUM_LIT:16> ) , ( <NUM_LIT:3> , <NUM_LIT> ) ) <EOL> self . assertEqual ( { <NUM_LIT:0> : <NUM_LIT> , <NUM_LIT:1> : <NUM_LIT:15> , <NUM_LIT:2> : <NUM_LIT:16> , <NUM_LIT:3> : <NUM_LIT> } , inst_cell . cpu_pinning ) <EOL> def test_default_behavior ( self ) : <EOL> inst_cell = objects . InstanceNUMACell ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( inst_cell . obj_get_changes ( ) ) ) <EOL> def test_cpu_pinning_requested_cell ( self ) : <EOL> inst_cell = objects . InstanceNUMACell ( cpuset = set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , <EOL> cpu_pinning = None ) <EOL> self . assertFalse ( inst_cell . cpu_pinning_requested ) <EOL> inst_cell . cpu_policy = fields . CPUAllocationPolicy . DEDICATED <EOL> self . assertTrue ( inst_cell . cpu_pinning_requested ) <EOL> def test_cpu_pinning_requested ( self ) : <EOL> fake_topo_obj = copy . deepcopy ( fake_obj_numa_topology ) <EOL> self . assertFalse ( fake_topo_obj . cpu_pinning_requested ) <EOL> for cell in fake_topo_obj . cells : <EOL> cell . cpu_policy = fields . CPUAllocationPolicy . DEDICATED <EOL> self . assertTrue ( fake_topo_obj . cpu_pinning_requested ) <EOL> def test_clear_host_pinning ( self ) : <EOL> topo_obj = get_fake_obj_numa_topology ( self . context ) <EOL> topo_obj . cells [ <NUM_LIT:0> ] . pin_vcpus ( ( <NUM_LIT:1> , <NUM_LIT:10> ) , ( <NUM_LIT:2> , <NUM_LIT:11> ) ) <EOL> topo_obj . cells [ <NUM_LIT:0> ] . id = <NUM_LIT:3> <EOL> topo_obj . cells [ <NUM_LIT:1> ] . pin_vcpus ( ( <NUM_LIT:3> , <NUM_LIT:0> ) , ( <NUM_LIT:4> , <NUM_LIT:1> ) ) <EOL> topo_obj . cells [ <NUM_LIT:1> ] . id = <NUM_LIT:0> <EOL> topo_obj . clear_host_pinning ( ) <EOL> self . assertEqual ( { } , topo_obj . cells [ <NUM_LIT:0> ] . cpu_pinning ) <EOL> self . assertEqual ( - <NUM_LIT:1> , topo_obj . cells [ <NUM_LIT:0> ] . id ) <EOL> self . assertEqual ( { } , topo_obj . cells [ <NUM_LIT:1> ] . cpu_pinning ) <EOL> self . assertEqual ( - <NUM_LIT:1> , topo_obj . cells [ <NUM_LIT:1> ] . id ) <EOL> class TestInstanceNUMATopology ( test_objects . _LocalTest , <EOL> _TestInstanceNUMATopology ) : <EOL> pass <EOL> class TestInstanceNUMATopologyRemote ( test_objects . _RemoteTest , <EOL> _TestInstanceNUMATopology ) : <EOL> pass </s>
