<s> import mock <EOL> import six <EOL> from oslo_serialization import jsonutils <EOL> from nailgun import consts <EOL> from nailgun import objects <EOL> from nailgun . orchestrator import deployment_serializers <EOL> from nailgun . plugins import adapters <EOL> from nailgun . utils import reverse <EOL> from nailgun . orchestrator . neutron_serializers import NeutronNetworkDeploymentSerializer90 <EOL> from nailgun . orchestrator . neutron_serializers import NeutronNetworkTemplateSerializer90 <EOL> from nailgun . test . integration import test_orchestrator_serializer_80 <EOL> class TestSerializer90Mixin ( object ) : <EOL> env_version = "<STR_LIT>" <EOL> task_deploy = True <EOL> @ classmethod <EOL> def create_serializer ( cls , cluster ) : <EOL> serializer_type = deployment_serializers . get_serializer_for_cluster ( <EOL> cluster <EOL> ) <EOL> return serializer_type ( None ) <EOL> class TestDeploymentAttributesSerialization90 ( <EOL> TestSerializer90Mixin , <EOL> test_orchestrator_serializer_80 . TestDeploymentAttributesSerialization80 <EOL> ) : <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_serialization_with_dpdk ( self , drivers_mock ) : <EOL> drivers_mock . return_value = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> } <EOL> node = self . env . create_nodes_w_interfaces_count ( <EOL> <NUM_LIT:1> , <NUM_LIT:3> , <EOL> cluster_id = self . cluster_db . id , <EOL> roles = [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] <EOL> nic_1 = node . interfaces [ <NUM_LIT:0> ] <EOL> nic_2 = node . interfaces [ <NUM_LIT:2> ] <EOL> nets_1 = nic_1 . assigned_networks_list <EOL> nets_2 = nic_2 . assigned_networks_list <EOL> dpdk_interface_name = nic_2 . name <EOL> for i , net in enumerate ( nets_1 ) : <EOL> if net [ '<STR_LIT:name>' ] == '<STR_LIT>' : <EOL> nets_2 . append ( nets_1 . pop ( i ) ) <EOL> break <EOL> objects . NIC . assign_networks ( nic_1 , nets_1 ) <EOL> objects . NIC . assign_networks ( nic_2 , nets_2 ) <EOL> objects . NIC . update ( nic_2 , <EOL> { '<STR_LIT>' : <EOL> { <EOL> '<STR_LIT>' : { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True } , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } } ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialised_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> self . assertEqual ( len ( serialised_for_astute ) , <NUM_LIT:1> ) <EOL> node = serialised_for_astute [ <NUM_LIT:0> ] <EOL> dpdk = node . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( dpdk ) <EOL> self . assertTrue ( dpdk . get ( '<STR_LIT>' ) ) <EOL> transformations = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> private_br = filter ( lambda t : t . get ( '<STR_LIT:name>' ) == <EOL> consts . DEFAULT_BRIDGES_NAMES . br_prv , <EOL> transformations ) [ <NUM_LIT:0> ] <EOL> dpdk_ports = filter ( lambda t : t . get ( '<STR_LIT:name>' ) == <EOL> dpdk_interface_name , <EOL> transformations ) <EOL> all_ports = filter ( lambda t : t . get ( '<STR_LIT:action>' ) == <EOL> '<STR_LIT>' , <EOL> transformations ) <EOL> self . assertEqual ( private_br . get ( '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( len ( all_ports ) - len ( dpdk_ports ) , len ( nets_1 ) ) <EOL> self . assertEqual ( len ( dpdk_ports ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( dpdk_ports [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> consts . DEFAULT_BRIDGES_NAMES . br_prv ) <EOL> self . assertEqual ( dpdk_ports [ <NUM_LIT:0> ] . get ( '<STR_LIT>' ) , <EOL> consts . NEUTRON_L23_PROVIDERS . dpdkovs ) <EOL> interfaces = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> dpdk_interface = interfaces [ dpdk_interface_name ] <EOL> vendor_specific = dpdk_interface . get ( '<STR_LIT>' , { } ) <EOL> self . assertEqual ( vendor_specific . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> def _check_dpdk_bond_serializing ( self , bond_properties ) : <EOL> node = self . env . create_nodes_w_interfaces_count ( <EOL> <NUM_LIT:1> , <NUM_LIT:4> , <EOL> cluster_id = self . cluster_db . id , <EOL> roles = [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] <EOL> node . attributes [ '<STR_LIT>' ] = { <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT:value>' : <NUM_LIT> } , <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT:value>' : { '<STR_LIT>' : <NUM_LIT:1> } } <EOL> } <EOL> cluster_attrs = objects . Cluster . get_editable_attributes ( node . cluster ) <EOL> cluster_attrs [ '<STR_LIT>' ] [ '<STR_LIT>' ] . update ( <EOL> { '<STR_LIT:value>' : consts . HYPERVISORS . kvm } ) <EOL> objects . Cluster . update_attributes ( <EOL> node . cluster , { '<STR_LIT>' : cluster_attrs } ) <EOL> for iface in node . interfaces : <EOL> iface [ '<STR_LIT>' ] . update ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> iface [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] = True <EOL> interfaces = self . env . node_nics_get ( node . id ) . json_body <EOL> first_nic = interfaces [ <NUM_LIT:0> ] <EOL> nics_for_bond = [ interfaces . pop ( ) , interfaces . pop ( ) ] <EOL> networks_for_bond = [ ] <EOL> bond_interface_name = '<STR_LIT>' <EOL> first_nic_networks = first_nic [ '<STR_LIT>' ] <EOL> for i , net in enumerate ( first_nic_networks ) : <EOL> if net [ '<STR_LIT:name>' ] == '<STR_LIT>' : <EOL> networks_for_bond . append ( first_nic_networks . pop ( i ) ) <EOL> break <EOL> bond_interface = { <EOL> '<STR_LIT:name>' : bond_interface_name , <EOL> '<STR_LIT:type>' : consts . NETWORK_INTERFACE_TYPES . bond , <EOL> '<STR_LIT>' : nics_for_bond , <EOL> '<STR_LIT>' : networks_for_bond , <EOL> '<STR_LIT>' : bond_properties , <EOL> '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : True } } } <EOL> interfaces . append ( bond_interface ) <EOL> self . env . node_nics_put ( node . id , interfaces ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialised_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> self . assertEqual ( len ( serialised_for_astute ) , <NUM_LIT:1> ) <EOL> node = serialised_for_astute [ <NUM_LIT:0> ] <EOL> dpdk = node . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( dpdk ) <EOL> self . assertTrue ( dpdk . get ( '<STR_LIT>' ) ) <EOL> transformations = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> private_br = filter ( lambda t : t . get ( '<STR_LIT:name>' ) == <EOL> consts . DEFAULT_BRIDGES_NAMES . br_prv , <EOL> transformations ) [ <NUM_LIT:0> ] <EOL> dpdk_bonds = filter ( lambda t : t . get ( '<STR_LIT:name>' ) == <EOL> bond_interface_name , <EOL> transformations ) <EOL> self . assertEqual ( len ( dpdk_bonds ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( dpdk_bonds [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> consts . DEFAULT_BRIDGES_NAMES . br_prv ) <EOL> self . assertEqual ( private_br . get ( '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( dpdk_bonds [ <NUM_LIT:0> ] . get ( '<STR_LIT>' ) , <EOL> consts . NEUTRON_L23_PROVIDERS . dpdkovs ) <EOL> self . assertEqual ( dpdk_bonds [ <NUM_LIT:0> ] . get ( '<STR_LIT>' ) . get ( '<STR_LIT>' ) , <EOL> bond_interface [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) ) <EOL> interfaces = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for iface in nics_for_bond : <EOL> dpdk_interface = interfaces [ iface [ '<STR_LIT:name>' ] ] <EOL> vendor_specific = dpdk_interface . get ( '<STR_LIT>' , { } ) <EOL> self . assertEqual ( vendor_specific . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_serialization_with_dpdk_on_bond ( self , drivers_mock ) : <EOL> drivers_mock . return_value = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> } <EOL> bond_properties = { <EOL> '<STR_LIT>' : consts . BOND_MODES . balance_slb , <EOL> '<STR_LIT>' : consts . BOND_TYPES . dpdkovs , <EOL> } <EOL> self . _check_dpdk_bond_serializing ( bond_properties ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_serialization_with_dpdk_on_lacp_bond ( self , drivers_mock ) : <EOL> drivers_mock . return_value = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> } <EOL> bond_properties = { <EOL> '<STR_LIT>' : consts . BOND_MODES . balance_tcp , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : consts . BOND_TYPES . dpdkovs } <EOL> self . _check_dpdk_bond_serializing ( bond_properties ) <EOL> def test_attributes_cpu_pinning ( self ) : <EOL> numa_nodes = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> , '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> , '<STR_LIT>' : [ <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , <NUM_LIT:8> ] } <EOL> ] <EOL> node = self . env . create_node ( cluster_id = self . cluster_db . id , <EOL> roles = [ '<STR_LIT>' ] ) <EOL> node . meta [ '<STR_LIT>' ] [ '<STR_LIT>' ] = numa_nodes <EOL> node . attributes . update ( { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { '<STR_LIT:value>' : <NUM_LIT:2> } , <EOL> '<STR_LIT>' : { '<STR_LIT:value>' : <NUM_LIT:2> } , <EOL> } <EOL> } ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> serialized_node = serialized_for_astute [ <NUM_LIT:0> ] <EOL> self . assertEqual ( serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] , [ <NUM_LIT:3> , <NUM_LIT:4> ] ) <EOL> node_name = objects . Node . get_slave_name ( node ) <EOL> node_common_attrs = serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ node_name ] <EOL> self . assertTrue ( node_common_attrs [ '<STR_LIT>' ] ) <EOL> def test_pinning_one_cpu_for_dpdk ( self ) : <EOL> numa_nodes = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> , '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> , '<STR_LIT>' : [ <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , <NUM_LIT:8> ] } <EOL> ] <EOL> node = self . env . create_node ( cluster_id = self . cluster_db . id , <EOL> roles = [ '<STR_LIT>' ] ) <EOL> node . meta [ '<STR_LIT>' ] [ '<STR_LIT>' ] = numa_nodes <EOL> node . attributes . update ( { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { '<STR_LIT:value>' : <NUM_LIT:1> } , <EOL> } <EOL> } ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> serialized_node = serialized_for_astute [ <NUM_LIT:0> ] <EOL> self . assertEqual ( serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertNotIn ( '<STR_LIT>' , serialized_node [ '<STR_LIT>' ] ) <EOL> self . assertNotIn ( '<STR_LIT>' , serialized_node [ '<STR_LIT>' ] ) <EOL> node_name = objects . Node . get_slave_name ( node ) <EOL> node_common_attrs = serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ node_name ] <EOL> self . assertFalse ( node_common_attrs [ '<STR_LIT>' ] ) <EOL> def test_dpdk_hugepages ( self ) : <EOL> numa_nodes = [ ] <EOL> for i in six . moves . range ( <NUM_LIT:3> ) : <EOL> numa_nodes . append ( { <EOL> '<STR_LIT:id>' : i , <EOL> '<STR_LIT>' : [ i ] , <EOL> '<STR_LIT>' : <NUM_LIT> ** <NUM_LIT:3> <EOL> } ) <EOL> meta = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ <NUM_LIT> ] , <EOL> '<STR_LIT>' : numa_nodes <EOL> } <EOL> } <EOL> node = self . env . create_node ( <EOL> cluster_id = self . cluster_db . id , <EOL> roles = [ '<STR_LIT>' ] , <EOL> meta = meta ) <EOL> node . attributes . update ( { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:value>' : <NUM_LIT> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:value>' : { '<STR_LIT>' : <NUM_LIT:1> } } } } <EOL> ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> serialized_node = serialized_for_astute [ <NUM_LIT:0> ] <EOL> self . assertEquals ( <EOL> "<STR_LIT>" , <EOL> serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> def test_attributes_no_hugepages_distribution_with_gig_hugepage ( self ) : <EOL> meta = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ <NUM_LIT> , <NUM_LIT> ] , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> , '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> , '<STR_LIT>' : [ <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , <NUM_LIT:8> ] } ] , <EOL> } <EOL> } <EOL> node = self . env . create_node ( <EOL> cluster_id = self . cluster_db . id , <EOL> roles = [ '<STR_LIT>' ] , <EOL> meta = meta ) <EOL> node . attributes . update ( { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT:value>' : { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> <EOL> } <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT:value>' : <NUM_LIT> , <EOL> } <EOL> } <EOL> } ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> serialized_node = serialized_for_astute [ <NUM_LIT:0> ] <EOL> self . assertNotIn ( '<STR_LIT>' , serialized_node ) <EOL> def test_attributes_hugepages_distribution ( self ) : <EOL> meta = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ <NUM_LIT> , <NUM_LIT> ] , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> , '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> , '<STR_LIT>' : [ <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , <NUM_LIT:8> ] } ] , <EOL> } <EOL> } <EOL> node = self . env . create_node ( <EOL> cluster_id = self . cluster_db . id , <EOL> roles = [ '<STR_LIT>' ] , <EOL> meta = meta ) <EOL> node . attributes . update ( { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT:value>' : { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT:value>' : <NUM_LIT> , <EOL> } <EOL> } <EOL> } ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> serialized_node = serialized_for_astute [ <NUM_LIT:0> ] <EOL> expected = [ <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT:size>' : <NUM_LIT> , '<STR_LIT:count>' : <NUM_LIT> } , <EOL> { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:size>' : <NUM_LIT> , '<STR_LIT:count>' : <NUM_LIT> } , <EOL> ] <EOL> self . assertEqual ( serialized_node [ '<STR_LIT>' ] , expected ) <EOL> def test_cpu_pinning_disabled ( self ) : <EOL> nodes_roles = [ [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ] <EOL> for roles in nodes_roles : <EOL> self . env . create_node ( <EOL> cluster_id = self . cluster_db . id , <EOL> roles = roles ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> for serialized_node in serialized_for_astute : <EOL> nova = serialized_node . get ( '<STR_LIT>' , { } ) <EOL> self . assertNotIn ( '<STR_LIT>' , nova ) <EOL> dpdk = serialized_node . get ( '<STR_LIT>' , { } ) <EOL> self . assertNotIn ( '<STR_LIT>' , dpdk ) <EOL> self . assertNotIn ( '<STR_LIT>' , dpdk ) <EOL> nodes_attrs = serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for node_attrs in six . itervalues ( nodes_attrs ) : <EOL> self . assertFalse ( node_attrs [ '<STR_LIT>' ] ) <EOL> def test_hugepages_disabled ( self ) : <EOL> nodes_roles = [ [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ] <EOL> for roles in nodes_roles : <EOL> self . env . create_node ( <EOL> cluster_id = self . cluster_db . id , <EOL> roles = roles ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> for serialized_node in serialized_for_astute : <EOL> nova = serialized_node . get ( '<STR_LIT>' , { } ) <EOL> self . assertFalse ( nova . get ( '<STR_LIT>' , False ) ) <EOL> dpdk = serialized_node . get ( '<STR_LIT>' , { } ) <EOL> self . assertNotIn ( '<STR_LIT>' , dpdk ) <EOL> self . assertNotIn ( '<STR_LIT>' , serialized_node ) <EOL> nodes_attrs = serialized_node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for node_attrs in six . itervalues ( nodes_attrs ) : <EOL> self . assertFalse ( node_attrs [ '<STR_LIT>' ] ) <EOL> def test_immutable_metadata_key ( self ) : <EOL> node = self . env . create_node ( <EOL> api = True , <EOL> cluster_id = self . cluster_db . id , <EOL> pending_roles = [ '<STR_LIT>' ] , <EOL> pending_addition = True ) <EOL> self . db . flush ( ) <EOL> resp = self . app . put ( <EOL> reverse ( '<STR_LIT>' , kwargs = { '<STR_LIT>' : node [ '<STR_LIT:id>' ] } ) , <EOL> jsonutils . dumps ( { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> headers = self . default_headers ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> for node_data in serialized_for_astute : <EOL> for k , v in six . iteritems ( node_data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) : <EOL> node = objects . Node . get_by_uid ( v [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( objects . Node . permanent_id ( node ) , k ) <EOL> class TestDeploymentLCMSerialization90 ( <EOL> TestSerializer90Mixin , <EOL> test_orchestrator_serializer_80 . BaseDeploymentSerializer <EOL> ) : <EOL> def setUp ( self ) : <EOL> super ( TestDeploymentLCMSerialization90 , self ) . setUp ( ) <EOL> self . cluster = self . env . create ( <EOL> release_kwargs = { <EOL> '<STR_LIT:version>' : self . env_version , <EOL> '<STR_LIT>' : consts . RELEASE_OS . ubuntu <EOL> } , <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : consts . CLUSTER_MODES . ha_compact , <EOL> '<STR_LIT>' : consts . CLUSTER_NET_PROVIDERS . neutron , <EOL> '<STR_LIT>' : consts . NEUTRON_SEGMENT_TYPES . vlan } ) <EOL> self . cluster_db = self . env . clusters [ - <NUM_LIT:1> ] <EOL> self . node = self . env . create_node ( <EOL> cluster_id = self . cluster_db . id , roles = [ '<STR_LIT>' ] <EOL> ) <EOL> self . serializer = self . create_serializer ( self . cluster_db ) <EOL> @ classmethod <EOL> def create_serializer ( cls , cluster ) : <EOL> return deployment_serializers . DeploymentLCMSerializer ( ) <EOL> def test_openstack_configuration_in_serialized ( self ) : <EOL> self . env . create_openstack_config ( <EOL> cluster_id = self . cluster_db . id , <EOL> configuration = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> ) <EOL> self . env . create_openstack_config ( <EOL> cluster_id = self . cluster_db . id , node_role = '<STR_LIT>' , <EOL> configuration = { '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> self . env . create_openstack_config ( <EOL> cluster_id = self . cluster_db . id , node_id = self . node . id , <EOL> configuration = { '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized = self . serializer . serialize ( self . cluster_db , [ self . node ] ) <EOL> self . assertEqual ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> serialized [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> ) <EOL> def test_cluster_attributes_in_serialized ( self ) : <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized = self . serializer . serialize ( self . cluster_db , [ self . node ] ) <EOL> for node_info in serialized : <EOL> self . assertEqual ( <EOL> objects . Cluster . to_dict ( self . cluster_db ) , <EOL> node_info [ '<STR_LIT>' ] <EOL> ) <EOL> self . assertEqual ( <EOL> objects . Release . to_dict ( self . cluster_db . release ) , <EOL> node_info [ '<STR_LIT>' ] <EOL> ) <EOL> self . assertEqual ( <EOL> [ '<STR_LIT>' ] , <EOL> serialized [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> ) <EOL> self . assertEqual ( <EOL> [ consts . TASK_ROLES . master ] , <EOL> serialized [ <NUM_LIT:1> ] [ '<STR_LIT>' ] <EOL> ) <EOL> @ mock . patch . object ( <EOL> adapters . PluginAdapterBase , '<STR_LIT>' , <EOL> mock . MagicMock ( return_value = True ) <EOL> ) <EOL> def test_plugins_in_serialized ( self ) : <EOL> releases = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:version>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ] <EOL> plugin1 = self . env . create_plugin ( <EOL> cluster = self . cluster_db , <EOL> name = '<STR_LIT>' , <EOL> attributes_metadata = { '<STR_LIT>' : { '<STR_LIT:name>' : '<STR_LIT>' } } , <EOL> package_version = '<STR_LIT>' , <EOL> releases = releases <EOL> ) <EOL> plugin2 = self . env . create_plugin ( <EOL> cluster = self . cluster_db , <EOL> name = '<STR_LIT>' , <EOL> attributes_metadata = { '<STR_LIT>' : { '<STR_LIT:name>' : '<STR_LIT>' } } , <EOL> package_version = '<STR_LIT>' , <EOL> releases = releases <EOL> ) <EOL> self . env . create_plugin ( <EOL> cluster = self . cluster_db , <EOL> enabled = False , <EOL> name = '<STR_LIT>' , <EOL> attributes_metadata = { '<STR_LIT>' : { '<STR_LIT:name>' : '<STR_LIT>' } } , <EOL> package_version = '<STR_LIT>' , <EOL> releases = releases <EOL> ) <EOL> self . env . create_node ( <EOL> cluster_id = self . cluster_db . id , <EOL> roles = [ '<STR_LIT>' ] <EOL> ) <EOL> plugins_data = [ <EOL> { <EOL> '<STR_LIT:name>' : p . name , <EOL> '<STR_LIT>' : [ { <EOL> '<STR_LIT>' : p . master_scripts_path ( self . cluster_db ) , <EOL> '<STR_LIT>' : p . slaves_scripts_path <EOL> } ] , <EOL> '<STR_LIT>' : [ { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : p . full_name , <EOL> '<STR_LIT>' : p . repo_url ( self . cluster_db ) , <EOL> '<STR_LIT>' : '<STR_LIT:/>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> <EOL> } ] <EOL> } <EOL> for p in six . moves . map ( adapters . wrap_plugin , [ plugin1 , plugin2 ] ) <EOL> ] <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> for node in serialized : <EOL> self . assertIn ( '<STR_LIT>' , node ) <EOL> self . datadiff ( plugins_data , node [ '<STR_LIT>' ] , compare_sorted = True ) <EOL> def test_serialize_with_customized ( self ) : <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serialized = self . serializer . serialize ( self . cluster_db , [ self . node ] ) <EOL> objects . Cluster . replace_deployment_info ( self . cluster_db , serialized ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> cust_serialized = self . serializer . serialize ( <EOL> self . cluster_db , [ self . node ] ) <EOL> for item in serialized : <EOL> if item [ '<STR_LIT>' ] != consts . MASTER_NODE_UID : <EOL> self . assertIn ( item , cust_serialized ) <EOL> else : <EOL> item [ '<STR_LIT>' ] [ '<STR_LIT>' ] = True <EOL> self . assertIn ( item , cust_serialized ) <EOL> class TestDeploymentHASerializer90 ( <EOL> TestSerializer90Mixin , <EOL> test_orchestrator_serializer_80 . TestDeploymentHASerializer80 <EOL> ) : <EOL> def test_glance_properties ( self ) : <EOL> self . check_no_murano_data ( ) <EOL> def test_ceph_keys ( self ) : <EOL> storage_attrs = self . serializer . get_common_attrs ( <EOL> self . cluster <EOL> ) [ '<STR_LIT>' ] <EOL> expected_keys = ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ) <EOL> for ceph_key in expected_keys : <EOL> self . assertIn ( ceph_key , storage_attrs ) <EOL> def test_serialize_with_customized ( self ) : <EOL> cluster = self . env . clusters [ <NUM_LIT:0> ] <EOL> serializer = self . create_serializer ( cluster ) <EOL> objects . Cluster . prepare_for_deployment ( cluster ) <EOL> serialized = serializer . serialize ( cluster , cluster . nodes ) <EOL> objects . Cluster . replace_deployment_info ( cluster , serialized ) <EOL> objects . Cluster . prepare_for_deployment ( cluster ) <EOL> cust_serialized = serializer . serialize ( <EOL> cluster , cluster . nodes ) <EOL> for item in serialized : <EOL> self . assertIn ( item , cust_serialized ) <EOL> class TestDeploymentTasksSerialization90 ( <EOL> TestSerializer90Mixin , <EOL> test_orchestrator_serializer_80 . TestDeploymentTasksSerialization80 <EOL> ) : <EOL> pass <EOL> class TestMultiNodeGroupsSerialization90 ( <EOL> TestSerializer90Mixin , <EOL> test_orchestrator_serializer_80 . TestMultiNodeGroupsSerialization80 <EOL> ) : <EOL> pass <EOL> class TestNetworkTemplateSerializer90CompatibleWith80 ( <EOL> TestSerializer90Mixin , <EOL> test_orchestrator_serializer_80 . <EOL> TestNetworkTemplateSerializer80CompatibleWith70 <EOL> ) : <EOL> general_serializer = NeutronNetworkDeploymentSerializer90 <EOL> template_serializer = NeutronNetworkTemplateSerializer90 <EOL> def check_vendor_specific_is_not_set ( self , use_net_template = False ) : <EOL> node = self . env . create_node ( <EOL> cluster_id = self . cluster . id , <EOL> roles = [ '<STR_LIT>' ] , primary_roles = [ '<STR_LIT>' ] <EOL> ) <EOL> objects . Cluster . set_network_template ( <EOL> self . cluster , <EOL> self . net_template if use_net_template else None ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> serializer = deployment_serializers . get_serializer_for_cluster ( <EOL> self . cluster ) <EOL> net_serializer = serializer . get_net_provider_serializer ( self . cluster ) <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> networks = nm . get_node_networks ( node ) <EOL> endpoints = net_serializer . generate_network_scheme ( <EOL> node , networks ) [ '<STR_LIT>' ] <EOL> for name in endpoints : <EOL> if endpoints [ name ] . get ( '<STR_LIT>' ) : <EOL> self . assertItemsEqual ( [ '<STR_LIT>' ] , <EOL> endpoints [ name ] [ '<STR_LIT>' ] ) <EOL> def test_multiple_node_roles_network_metadata_attrs ( self ) : <EOL> for node_data in self . serialized_for_astute : <EOL> self . assertItemsEqual ( <EOL> node_data [ '<STR_LIT>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> nodes = node_data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for node_name , node_attrs in nodes . items ( ) : <EOL> self . assertTrue ( <EOL> { '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' } . issubset ( node_attrs ) <EOL> ) <EOL> node = objects . Node . get_by_uid ( node_attrs [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( objects . Node . get_slave_name ( node ) , node_name ) <EOL> self . assertEqual ( node_attrs [ '<STR_LIT>' ] , node . uid ) <EOL> self . assertEqual ( node_attrs [ '<STR_LIT>' ] , <EOL> objects . Node . get_node_fqdn ( node ) ) <EOL> self . assertEqual ( node_attrs [ '<STR_LIT:name>' ] , node_name ) <EOL> self . assertEqual ( node_attrs [ '<STR_LIT>' ] , node . name ) <EOL> self . assertEqual ( node_attrs [ '<STR_LIT>' ] , node . uid ) <EOL> self . assertEqual ( node_attrs [ '<STR_LIT>' ] , False ) <EOL> class TestNetworkTemplateSerializer90 ( <EOL> TestSerializer90Mixin , <EOL> test_orchestrator_serializer_80 . TestNetworkTemplateSerializer80 <EOL> ) : <EOL> legacy_serializer = NeutronNetworkDeploymentSerializer90 <EOL> template_serializer = NeutronNetworkTemplateSerializer90 <EOL> def check_selective_gateway ( self , use_net_template = False ) : <EOL> node = self . env . create_node ( <EOL> cluster_id = self . cluster . id , <EOL> roles = [ '<STR_LIT>' ] , primary_roles = [ '<STR_LIT>' ] <EOL> ) <EOL> objects . Cluster . set_network_template ( <EOL> self . cluster , <EOL> self . net_template if use_net_template else None ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> serializer = deployment_serializers . get_serializer_for_cluster ( <EOL> self . cluster ) <EOL> net_serializer = serializer . get_net_provider_serializer ( self . cluster ) <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> networks_list = nm . get_node_networks ( node ) <EOL> networks = { net [ '<STR_LIT:name>' ] : net for net in networks_list } <EOL> endpoints = net_serializer . generate_network_scheme ( <EOL> node , networks_list ) [ '<STR_LIT>' ] <EOL> na = self . net_template [ <EOL> '<STR_LIT>' ] [ '<STR_LIT:default>' ] [ '<STR_LIT>' ] <EOL> ep_net_map = { na [ net_name ] [ '<STR_LIT>' ] : net_name for net_name in na } <EOL> for name in endpoints : <EOL> if name not in ep_net_map : <EOL> self . assertNotIn ( '<STR_LIT>' , endpoints [ name ] ) <EOL> continue <EOL> if networks [ ep_net_map [ name ] ] . get ( '<STR_LIT>' ) is None : <EOL> self . assertNotIn ( '<STR_LIT>' , endpoints [ name ] ) <EOL> else : <EOL> self . assertIn ( '<STR_LIT>' , endpoints [ name ] ) <EOL> self . assertEqual ( <EOL> endpoints [ name ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> networks [ ep_net_map [ name ] ] [ '<STR_LIT>' ] ) <EOL> def test_selective_gateway_in_deployment_serializer ( self ) : <EOL> self . check_selective_gateway ( ) <EOL> def test_selective_gateway_in_template_serializer ( self ) : <EOL> self . check_selective_gateway ( use_net_template = True ) <EOL> class TestSerializeInterfaceDriversData90 ( <EOL> TestSerializer90Mixin , <EOL> test_orchestrator_serializer_80 . TestSerializeInterfaceDriversData80 <EOL> ) : <EOL> pass <EOL> class TestSriovSerialization90 ( <EOL> TestSerializer90Mixin , <EOL> test_orchestrator_serializer_80 . BaseDeploymentSerializer <EOL> ) : <EOL> def setUp ( self , * args ) : <EOL> super ( TestSriovSerialization90 , self ) . setUp ( ) <EOL> self . cluster_db = self . env . create ( <EOL> release_kwargs = { '<STR_LIT:version>' : self . env_version } , <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : consts . CLUSTER_MODES . ha_compact , <EOL> '<STR_LIT>' : consts . CLUSTER_NET_PROVIDERS . neutron , <EOL> '<STR_LIT>' : consts . NEUTRON_SEGMENT_TYPES . vlan , <EOL> '<STR_LIT:status>' : consts . CLUSTER_STATUSES . new } , <EOL> ) <EOL> self . env . create_nodes_w_interfaces_count ( <EOL> nodes_count = <NUM_LIT:1> , if_count = <NUM_LIT:3> , cluster_id = self . cluster_db . id , <EOL> pending_roles = [ '<STR_LIT>' ] , pending_addition = True ) <EOL> def serialize ( self ) : <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serializer = self . create_serializer ( self . cluster_db ) <EOL> return serializer . serialize ( self . cluster_db , self . env . nodes ) <EOL> def test_nic_sriov_info_is_serialized ( self ) : <EOL> for nic in self . env . nodes [ <NUM_LIT:0> ] . nic_interfaces : <EOL> if not nic . assigned_networks_list : <EOL> nic_sriov = nic <EOL> nic . interface_properties [ '<STR_LIT>' ] = { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : <NUM_LIT:8> , <EOL> '<STR_LIT>' : <NUM_LIT:8> , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> objects . NIC . update ( <EOL> nic , { '<STR_LIT>' : nic . interface_properties } ) <EOL> break <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> node0 = self . serialize ( ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( <EOL> node0 [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' ] <EOL> ) <EOL> for trans in node0 [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> if trans . get ( '<STR_LIT:name>' ) == nic_sriov . name : <EOL> self . assertEqual ( <EOL> trans [ '<STR_LIT>' ] , <EOL> { <EOL> '<STR_LIT>' : <NUM_LIT:8> , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> ) <EOL> self . assertEqual ( trans [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> break <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) </s>
