<s> import collections <EOL> import copy <EOL> import mock <EOL> import netaddr <EOL> import six <EOL> import yaml <EOL> from oslo_serialization import jsonutils <EOL> from nailgun import consts <EOL> from nailgun . db import db <EOL> from nailgun . db . sqlalchemy import models <EOL> from nailgun . extensions . network_manager . manager import NetworkManager <EOL> from nailgun import objects <EOL> from nailgun . orchestrator import stages <EOL> from nailgun . test import base <EOL> from nailgun . test . base import DeploymentTasksTestMixin <EOL> from nailgun . utils import reverse <EOL> from nailgun . orchestrator . deployment_serializers import get_serializer_for_cluster <EOL> from nailgun . orchestrator . neutron_serializers import NeutronNetworkDeploymentSerializer70 <EOL> from nailgun . orchestrator . neutron_serializers import NeutronNetworkTemplateSerializer70 <EOL> from nailgun . orchestrator . orchestrator_graph import AstuteGraph <EOL> from nailgun . test . integration . test_orchestrator_serializer import BaseDeploymentSerializer <EOL> from nailgun . test . integration . test_orchestrator_serializer import TestDeploymentHASerializer61 <EOL> from nailgun . test . integration . test_orchestrator_serializer import TestNovaOrchestratorSerializer <EOL> from nailgun . test . integration . test_orchestrator_serializer import TestSerializeInterfaceDriversData <EOL> class PrepareDataMixin ( object ) : <EOL> def patch_net_roles_for_release ( self ) : <EOL> rel_id = self . env . create_release ( version = self . env_version ) . id <EOL> rel_db = self . db . query ( models . Release ) . filter_by ( id = rel_id ) . one ( ) <EOL> to_patch = yaml . safe_load ( """<STR_LIT>""" ) <EOL> rel_db . network_roles_metadata = to_patch <EOL> self . db . flush ( ) <EOL> return rel_db <EOL> class BaseTestDeploymentAttributesSerialization70 ( BaseDeploymentSerializer , <EOL> PrepareDataMixin ) : <EOL> management = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> fuelweb_admin = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> neutron = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> storage = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> public = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> private = [ '<STR_LIT>' ] <EOL> networks = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> segmentation_type = None <EOL> env_version = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> super ( BaseTestDeploymentAttributesSerialization70 , self ) . setUp ( ) <EOL> self . cluster = self . create_env ( consts . CLUSTER_MODES . ha_compact ) <EOL> objects . Cluster . prepare_for_deployment ( self . env . clusters [ - <NUM_LIT:1> ] ) <EOL> self . cluster_db = self . db . query ( models . Cluster ) . get ( self . cluster [ '<STR_LIT:id>' ] ) <EOL> serializer_type = get_serializer_for_cluster ( self . cluster_db ) <EOL> self . serializer = serializer_type ( AstuteGraph ( self . cluster_db ) ) <EOL> self . serialized_for_astute = self . serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes ) <EOL> self . vm_data = self . env . read_fixtures ( [ '<STR_LIT>' ] ) <EOL> def create_env ( self , mode ) : <EOL> release = self . patch_net_roles_for_release ( ) <EOL> return self . env . create ( <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : release . id , <EOL> '<STR_LIT>' : mode , <EOL> '<STR_LIT>' : consts . CLUSTER_NET_PROVIDERS . neutron , <EOL> '<STR_LIT>' : self . segmentation_type } , <EOL> nodes_kwargs = [ <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : True } , <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : True } ] ) <EOL> def check_vips_serialized ( self , node_data ) : <EOL> vips_names = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertTrue ( all ( vip_name not in node_data <EOL> for vip_name in vips_names ) ) <EOL> vips_data = node_data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assertItemsEqual ( vips_data , <EOL> vips_names ) <EOL> for vip in vips_names : <EOL> self . assertItemsEqual ( <EOL> vips_data [ vip ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> class TestDeploymentAttributesSerialization70 ( <EOL> BaseTestDeploymentAttributesSerialization70 <EOL> ) : <EOL> segmentation_type = consts . NEUTRON_SEGMENT_TYPES . vlan <EOL> custom_network = { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:50> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> plugin_network_roles = yaml . safe_load ( """<STR_LIT>""" . format ( ** custom_network ) ) <EOL> def test_non_default_bridge_mapping ( self ) : <EOL> expected_mapping = { <EOL> u'<STR_LIT:test>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' <EOL> } <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : self . env . releases [ <NUM_LIT:0> ] . id , <EOL> '<STR_LIT>' : consts . CLUSTER_MODES . ha_compact , <EOL> '<STR_LIT>' : consts . CLUSTER_NET_PROVIDERS . neutron , <EOL> '<STR_LIT>' : self . segmentation_type } ) <EOL> self . cluster_db = objects . Cluster . get_by_uid ( cluster [ '<STR_LIT:id>' ] ) <EOL> for name in expected_mapping : <EOL> self . env . _create_network_group ( cluster = self . cluster_db , <EOL> name = name ) <EOL> self . env . create_node ( <EOL> api = True , <EOL> cluster_id = cluster [ '<STR_LIT:id>' ] , <EOL> pending_roles = [ '<STR_LIT>' ] , <EOL> pending_addition = True ) <EOL> net_serializer = self . serializer . get_net_provider_serializer ( <EOL> self . cluster_db ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> mapping = net_serializer . get_node_non_default_bridge_mapping ( <EOL> self . cluster_db . nodes [ <NUM_LIT:0> ] ) <EOL> br_collision = collections . defaultdict ( list ) <EOL> self . assertEqual ( len ( mapping ) , len ( expected_mapping ) ) <EOL> self . assertEqual ( len ( expected_mapping ) , len ( set ( mapping . values ( ) ) ) ) <EOL> for netname in expected_mapping : <EOL> self . assertTrue ( <EOL> mapping [ netname ] . startswith ( expected_mapping [ netname ] ) ) <EOL> br_collision [ expected_mapping [ netname ] ] . append ( netname ) <EOL> for bridge , netnames in six . iteritems ( br_collision ) : <EOL> bridges = set ( ( mapping [ netname ] for netname in netnames ) ) <EOL> self . assertEqual ( len ( bridges ) , len ( netnames ) ) <EOL> def test_network_scheme_custom_networks ( self ) : <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : self . env . releases [ <NUM_LIT:0> ] . id , <EOL> '<STR_LIT>' : consts . CLUSTER_MODES . ha_compact , <EOL> '<STR_LIT>' : consts . CLUSTER_NET_PROVIDERS . neutron , <EOL> '<STR_LIT>' : self . segmentation_type } ) <EOL> self . cluster_db = objects . Cluster . get_by_uid ( cluster [ '<STR_LIT:id>' ] ) <EOL> self . env . _create_network_group ( cluster = self . cluster_db , <EOL> name = self . custom_network [ '<STR_LIT:name>' ] , <EOL> cidr = self . custom_network [ '<STR_LIT>' ] , <EOL> vlan_start = self . custom_network [ <EOL> '<STR_LIT>' <EOL> ] ) <EOL> self . env . create_plugin ( <EOL> cluster = self . cluster_db , <EOL> network_roles_metadata = self . plugin_network_roles , <EOL> package_version = '<STR_LIT>' ) <EOL> self . env . create_node ( <EOL> api = True , <EOL> cluster_id = cluster [ '<STR_LIT:id>' ] , <EOL> pending_roles = [ '<STR_LIT>' ] , <EOL> pending_addition = True ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> serializer = self . create_serializer ( self . cluster_db ) <EOL> serialized_for_astute = serializer . serialize ( <EOL> self . cluster_db , self . cluster_db . nodes <EOL> ) <EOL> for node in serialized_for_astute : <EOL> vips = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> roles = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> transformations = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> node_network_roles = ( node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> [ '<STR_LIT>' + node [ '<STR_LIT>' ] ] [ '<STR_LIT>' ] ) <EOL> custom_ip = node_network_roles . get ( self . custom_network [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> custom_brs = filter ( lambda t : t . get ( '<STR_LIT:name>' ) == <EOL> self . custom_network [ '<STR_LIT>' ] , <EOL> transformations ) <EOL> custom_ports = filter ( lambda t : t . get ( '<STR_LIT:name>' ) == <EOL> ( "<STR_LIT>" % <EOL> self . custom_network [ '<STR_LIT>' ] ) , <EOL> transformations ) <EOL> self . assertEqual ( roles . get ( self . custom_network [ '<STR_LIT>' ] ) , <EOL> self . custom_network [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( vips . get ( self . custom_network [ '<STR_LIT:name>' ] , <EOL> { } ) . get ( '<STR_LIT>' ) , <EOL> self . custom_network [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( netaddr . IPAddress ( custom_ip ) in <EOL> netaddr . IPNetwork ( self . custom_network [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( len ( custom_brs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( custom_ports ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( custom_ports [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> self . custom_network [ '<STR_LIT>' ] ) <EOL> def test_network_scheme ( self ) : <EOL> for node in self . serialized_for_astute : <EOL> roles = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> node = objects . Node . get_by_uid ( node [ '<STR_LIT>' ] ) <EOL> expected_roles = zip ( <EOL> self . management , [ '<STR_LIT>' ] * len ( self . management ) ) <EOL> expected_roles += zip ( <EOL> self . fuelweb_admin , [ '<STR_LIT>' ] * len ( self . fuelweb_admin ) ) <EOL> expected_roles += zip ( <EOL> self . storage , [ '<STR_LIT>' ] * len ( self . storage ) ) <EOL> if objects . Node . should_have_public ( node ) : <EOL> expected_roles += zip ( <EOL> self . public , [ '<STR_LIT>' ] * len ( self . public ) ) <EOL> expected_roles += [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> if node . cluster . network_config . segmentation_type == consts . NEUTRON_SEGMENT_TYPES . vlan : <EOL> expected_roles += [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> if node . cluster . network_config . segmentation_type in ( consts . NEUTRON_SEGMENT_TYPES . gre , <EOL> consts . NEUTRON_SEGMENT_TYPES . tun ) : <EOL> expected_roles += [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> self . assertEqual ( roles , dict ( expected_roles ) ) <EOL> def test_offloading_modes_serialize ( self ) : <EOL> meta = self . env . default_metadata ( ) <EOL> changed_offloading_modes = { } <EOL> for interface in meta [ '<STR_LIT>' ] : <EOL> changed_offloading_modes [ interface [ '<STR_LIT:name>' ] ] = NetworkManager . _get_modified_offloading_modes ( <EOL> interface . get ( '<STR_LIT>' ) ) <EOL> for node in self . serialized_for_astute : <EOL> interfaces = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for iface_name in interfaces : <EOL> ethtool_blk = interfaces [ iface_name ] . get ( '<STR_LIT>' , None ) <EOL> self . assertIsNotNone ( <EOL> ethtool_blk , <EOL> "<STR_LIT>" ) <EOL> offload_blk = ethtool_blk . get ( '<STR_LIT>' , None ) <EOL> self . assertIsNotNone ( <EOL> offload_blk , <EOL> "<STR_LIT>" ) <EOL> self . assertDictEqual ( offload_blk , <EOL> changed_offloading_modes [ iface_name ] ) <EOL> def test_network_metadata ( self ) : <EOL> neutron_serializer = self . serializer . get_net_provider_serializer ( <EOL> self . cluster_db ) <EOL> for node_data in self . serialized_for_astute : <EOL> self . assertItemsEqual ( <EOL> node_data [ '<STR_LIT>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> for k , v in six . iteritems ( node_data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) : <EOL> self . assertItemsEqual ( <EOL> v , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:name>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> node = objects . Node . get_by_uid ( v [ '<STR_LIT>' ] ) <EOL> ip_by_net = neutron_serializer . get_network_to_ip_mapping ( node ) <EOL> self . assertEqual ( objects . Node . get_slave_name ( node ) , k ) <EOL> self . assertEqual ( v [ '<STR_LIT>' ] , node . uid ) <EOL> self . assertEqual ( v [ '<STR_LIT>' ] , objects . Node . get_node_fqdn ( node ) ) <EOL> self . assertEqual ( v [ '<STR_LIT:name>' ] , k ) <EOL> self . assertEqual ( v [ '<STR_LIT>' ] , node . name ) <EOL> self . assertEqual ( v [ '<STR_LIT>' ] , node . uid ) <EOL> network_roles = zip ( self . management , <EOL> [ ip_by_net [ '<STR_LIT>' ] ] * len ( <EOL> self . management ) ) <EOL> network_roles += zip ( self . fuelweb_admin , <EOL> [ ip_by_net [ '<STR_LIT>' ] ] * len ( <EOL> self . fuelweb_admin ) ) <EOL> network_roles += zip ( <EOL> self . storage , [ ip_by_net [ '<STR_LIT>' ] ] * len ( self . storage ) ) <EOL> network_roles += zip ( self . neutron , [ None ] * len ( self . neutron ) ) <EOL> if objects . Node . should_have_public ( node ) : <EOL> network_roles += zip ( <EOL> self . public , [ ip_by_net [ '<STR_LIT>' ] ] * len ( self . public ) ) <EOL> if node . cluster . network_config . segmentation_type in ( consts . NEUTRON_SEGMENT_TYPES . gre , <EOL> consts . NEUTRON_SEGMENT_TYPES . tun ) : <EOL> network_roles += zip ( <EOL> self . private , <EOL> [ ip_by_net [ '<STR_LIT>' ] ] * len ( self . private ) ) <EOL> self . assertEqual ( v [ '<STR_LIT>' ] , dict ( network_roles ) ) <EOL> self . check_vips_serialized ( node_data ) <EOL> def test_generate_vmware_attributes_data ( self ) : <EOL> self . check_generate_vmware_attributes_data ( ) <EOL> result = self . serializer . serialize_node ( <EOL> self . env . nodes [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> result [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> result [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:2> ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> class TestDeploymentAttributesSerializationSegmentationGre70 ( <EOL> TestDeploymentAttributesSerialization70 <EOL> ) : <EOL> segmentation_type = consts . NEUTRON_SEGMENT_TYPES . gre <EOL> class TestDeploymentAttributesSerializationSegmentationTun70 ( <EOL> TestDeploymentAttributesSerialization70 <EOL> ) : <EOL> segmentation_type = consts . NEUTRON_SEGMENT_TYPES . tun <EOL> class TestDeploymentSerializationForNovaNetwork70 ( <EOL> BaseTestDeploymentAttributesSerialization70 <EOL> ) : <EOL> def create_env ( self , mode ) : <EOL> release = self . patch_net_roles_for_release ( ) <EOL> return self . env . create ( <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : release . id , <EOL> '<STR_LIT>' : mode , <EOL> '<STR_LIT>' : consts . CLUSTER_NET_PROVIDERS . nova_network } , <EOL> nodes_kwargs = [ <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT:name>' : self . node_name , <EOL> } <EOL> ] ) <EOL> def test_network_scheme ( self ) : <EOL> for node in self . serialized_for_astute : <EOL> roles = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> expected_roles = { <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> '<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> '<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> '<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> '<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> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . assertEqual ( roles , expected_roles ) <EOL> def test_network_metadata ( self ) : <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> ip_by_net = { <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None <EOL> } <EOL> node = self . env . nodes [ <NUM_LIT:0> ] <EOL> networks = nm . get_node_networks ( node ) <EOL> for net in ip_by_net : <EOL> netgroup = nm . get_network_by_netname ( net , networks ) <EOL> if netgroup . get ( '<STR_LIT>' ) : <EOL> ip_by_net [ net ] = netgroup [ '<STR_LIT>' ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] <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 . assertItemsEqual ( <EOL> node_attrs , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:name>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <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> network_roles = { <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> } <EOL> self . assertEqual ( <EOL> node_attrs [ '<STR_LIT>' ] , <EOL> network_roles <EOL> ) <EOL> self . check_vips_serialized ( node_data ) <EOL> def test_generate_vmware_attributes_data ( self ) : <EOL> self . check_generate_vmware_attributes_data ( ) <EOL> result = self . serializer . serialize_node ( <EOL> self . env . nodes [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> result [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> result [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:2> ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> class TestPluginDeploymentTasksInjection70 ( base . BaseIntegrationTest ) : <EOL> env_version = '<STR_LIT>' <EOL> release_deployment_tasks = [ <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT>' } } , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> ] <EOL> def setUp ( self ) : <EOL> super ( TestPluginDeploymentTasksInjection70 , self ) . setUp ( ) <EOL> self . cluster = self . env . create ( <EOL> release_kwargs = { <EOL> '<STR_LIT>' : self . release_deployment_tasks , <EOL> '<STR_LIT:version>' : self . env_version <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> } , <EOL> nodes_kwargs = [ <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] , '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : True } <EOL> ] <EOL> ) <EOL> self . plugin_data = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:version>' : self . cluster . release . version , <EOL> '<STR_LIT>' : self . cluster . release . operating_system . lower ( ) , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> ] , <EOL> } <EOL> def prepare_plugins_for_cluster ( self , cluster , plugins_kw_list ) : <EOL> for kw in plugins_kw_list : <EOL> kw . update ( self . plugin_data ) <EOL> self . env . create_plugin ( cluster = cluster , ** kw ) <EOL> def _check_pre_deployment_tasks ( self , serialized , task_type ) : <EOL> self . assertTrue ( serialized ) <EOL> needed_task = next ( <EOL> t for t in serialized <EOL> if t [ '<STR_LIT:type>' ] == task_type ) <EOL> self . assertIsNotNone ( needed_task ) <EOL> self . assertIsNotNone ( needed_task . get ( '<STR_LIT>' ) ) <EOL> self . assertItemsEqual ( <EOL> ( n . uid for n in self . cluster . nodes ) , <EOL> needed_task [ '<STR_LIT>' ] <EOL> ) <EOL> def test_plugin_depl_tasks_proper_injections ( self ) : <EOL> self . prepare_plugins_for_cluster ( <EOL> self . cluster , <EOL> [ <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> } , <EOL> ] , <EOL> } , <EOL> ] <EOL> ) <EOL> graph = AstuteGraph ( self . cluster ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> serializer = get_serializer_for_cluster ( self . cluster ) ( graph ) <EOL> serialized = serializer . serialize ( self . cluster , self . cluster . nodes ) <EOL> serialized_tasks = serialized [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> expected_priority = { <EOL> <NUM_LIT:100> : '<STR_LIT>' , <EOL> <NUM_LIT:200> : '<STR_LIT>' , <EOL> <NUM_LIT> : '<STR_LIT>' , <EOL> } <EOL> for task in serialized_tasks : <EOL> task_identificator = task [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assertEqual ( <EOL> task_identificator , expected_priority [ task [ '<STR_LIT>' ] ] <EOL> ) <EOL> def test_plugin_depl_task_overwrite_from_rel ( self ) : <EOL> self . prepare_plugins_for_cluster ( <EOL> self . cluster , <EOL> [ <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> } , <EOL> ] , <EOL> } , <EOL> ] <EOL> ) <EOL> graph = AstuteGraph ( self . cluster ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> serializer = get_serializer_for_cluster ( self . cluster ) ( graph ) <EOL> serialized = serializer . serialize ( self . cluster , self . cluster . nodes ) <EOL> serialized_tasks = serialized [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> needed_task_priority = next ( <EOL> t [ '<STR_LIT>' ] for t in serialized_tasks <EOL> if t [ '<STR_LIT>' ] [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> ) <EOL> self . assertEqual ( needed_task_priority , <NUM_LIT:100> ) <EOL> def test_plugin_depl_task_in_pre_depl ( self ) : <EOL> self . prepare_plugins_for_cluster ( <EOL> self . cluster , <EOL> [ <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> } , <EOL> ] , <EOL> } , <EOL> ] <EOL> ) <EOL> graph = AstuteGraph ( self . cluster ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> mock . Mock ( return_value = '<STR_LIT>' ) ) : <EOL> pre_deployment = stages . pre_deployment_serialize ( <EOL> graph , self . cluster , self . cluster . nodes ) <EOL> for task_type in ( consts . ORCHESTRATOR_TASK_TYPES . sync , <EOL> consts . ORCHESTRATOR_TASK_TYPES . upload_file ) : <EOL> self . _check_pre_deployment_tasks ( pre_deployment , task_type ) <EOL> def test_plugin_depl_task_for_master_not_in_pre_depl ( self ) : <EOL> self . prepare_plugins_for_cluster ( <EOL> self . cluster , <EOL> [ <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : consts . MASTER_NODE_UID , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> } , <EOL> ] , <EOL> } , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ consts . MASTER_NODE_UID , <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> } , <EOL> ] , <EOL> } , <EOL> ] <EOL> ) <EOL> graph = AstuteGraph ( self . cluster ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> mock . Mock ( return_value = '<STR_LIT>' ) ) : <EOL> pre_deployment = stages . pre_deployment_serialize ( <EOL> graph , self . cluster , self . cluster . nodes ) <EOL> for st in pre_deployment : <EOL> self . assertNotIn ( consts . MASTER_NODE_UID , st [ '<STR_LIT>' ] ) <EOL> def test_plugin_depl_task_in_post_depl ( self ) : <EOL> self . prepare_plugins_for_cluster ( <EOL> self . cluster , <EOL> [ <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> } , <EOL> ] , <EOL> } , <EOL> ] <EOL> ) <EOL> graph = AstuteGraph ( self . cluster ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> post_deployment = stages . post_deployment_serialize ( <EOL> graph , self . cluster , self . cluster . nodes ) <EOL> self . assertEqual ( <EOL> post_deployment [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' <EOL> ) <EOL> def test_process_skipped_task ( self ) : <EOL> self . prepare_plugins_for_cluster ( <EOL> self . cluster , <EOL> [ <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> } , <EOL> ] , <EOL> } , <EOL> ] <EOL> ) <EOL> graph = AstuteGraph ( self . cluster ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> serializer = get_serializer_for_cluster ( self . cluster ) ( graph ) <EOL> serialized = serializer . serialize ( self . cluster , self . cluster . nodes ) <EOL> tasks = serialized [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> release_depl_tasks_ids = ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> serialized_tasks_ids = ( t [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for t in tasks ) <EOL> self . assertItemsEqual ( release_depl_tasks_ids , serialized_tasks_ids ) <EOL> class TestRolesSerializationWithPlugins ( BaseDeploymentSerializer , <EOL> PrepareDataMixin , <EOL> DeploymentTasksTestMixin ) : <EOL> env_version = '<STR_LIT>' <EOL> ROLES = yaml . safe_load ( """<STR_LIT>""" ) <EOL> DEPLOYMENT_TASKS = yaml . safe_load ( """<STR_LIT>""" ) <EOL> def setUp ( self ) : <EOL> super ( TestRolesSerializationWithPlugins , self ) . setUp ( ) <EOL> release = self . patch_net_roles_for_release ( ) <EOL> self . cluster = self . env . create ( <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : release . id , <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> } ) <EOL> self . plugin_data = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:version>' : self . cluster . release . version , <EOL> '<STR_LIT>' : self . cluster . release . operating_system . lower ( ) , <EOL> '<STR_LIT>' : [ self . cluster . mode ] , <EOL> } <EOL> ] <EOL> } <EOL> def _get_serializer ( self , cluster ) : <EOL> return get_serializer_for_cluster ( cluster ) ( AstuteGraph ( cluster ) ) <EOL> def test_tasks_were_serialized ( self ) : <EOL> plugin_data = { <EOL> '<STR_LIT>' : self . ROLES , <EOL> '<STR_LIT>' : self . DEPLOYMENT_TASKS <EOL> } <EOL> plugin_data . update ( self . plugin_data ) <EOL> self . env . create_plugin ( cluster = self . cluster , ** plugin_data ) <EOL> self . env . create_node ( <EOL> api = True , <EOL> cluster_id = self . cluster . id , <EOL> pending_roles = [ '<STR_LIT>' ] , <EOL> pending_addition = True ) <EOL> self . db . flush ( ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> serializer = self . _get_serializer ( self . cluster ) <EOL> serialized_data = serializer . serialize ( <EOL> self . cluster , self . cluster . nodes ) <EOL> self . assertItemsEqual ( serialized_data [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , [ { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : None , <EOL> } , <EOL> '<STR_LIT>' : <NUM_LIT:100> , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ self . cluster . nodes [ <NUM_LIT:0> ] . uid ] , <EOL> '<STR_LIT:id>' : '<STR_LIT>' <EOL> } ] ) <EOL> def test_tasks_were_not_serialized ( self ) : <EOL> plugin_data = { <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : self . DEPLOYMENT_TASKS <EOL> } <EOL> plugin_data . update ( self . plugin_data ) <EOL> self . env . create_plugin ( cluster = self . cluster , ** plugin_data ) <EOL> self . env . create_node ( <EOL> api = True , <EOL> cluster_id = self . cluster . id , <EOL> pending_roles = [ '<STR_LIT>' ] , <EOL> pending_addition = True ) <EOL> self . db . flush ( ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> serializer = self . _get_serializer ( self . cluster ) <EOL> serialized_data = serializer . serialize ( <EOL> self . cluster , self . cluster . nodes ) <EOL> self . maxDiff = None <EOL> self . _compare_tasks ( [ <EOL> { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT:/>' } , <EOL> '<STR_LIT>' : <NUM_LIT:100> , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ self . cluster . nodes [ <NUM_LIT:0> ] . uid ] , <EOL> } , { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT:/>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : None } , <EOL> '<STR_LIT>' : <NUM_LIT:200> , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ self . cluster . nodes [ <NUM_LIT:0> ] . uid ] , <EOL> } , { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT:/>' } , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ self . cluster . nodes [ <NUM_LIT:0> ] . uid ] , <EOL> } ] , serialized_data [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> class TestNetworkTemplateSerializer70 ( BaseDeploymentSerializer , <EOL> PrepareDataMixin ) : <EOL> env_version = '<STR_LIT>' <EOL> general_serializer = NeutronNetworkDeploymentSerializer70 <EOL> template_serializer = NeutronNetworkTemplateSerializer70 <EOL> def setUp ( self , * args ) : <EOL> super ( TestNetworkTemplateSerializer70 , self ) . setUp ( ) <EOL> self . cluster = self . create_env ( consts . NEUTRON_SEGMENT_TYPES . vlan ) <EOL> self . net_template = self . env . read_fixtures ( [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] <EOL> objects . Cluster . set_network_template ( <EOL> self . cluster , <EOL> self . net_template <EOL> ) <EOL> cluster_db = self . db . query ( models . Cluster ) . get ( self . cluster [ '<STR_LIT:id>' ] ) <EOL> objects . Cluster . prepare_for_deployment ( cluster_db ) <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> self . serialized_for_astute = serializer ( <EOL> AstuteGraph ( cluster_db ) ) . serialize ( self . cluster , cluster_db . nodes ) <EOL> def create_env ( self , segment_type ) : <EOL> release = self . patch_net_roles_for_release ( ) <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : release . id , <EOL> '<STR_LIT>' : consts . CLUSTER_MODES . ha_compact , <EOL> '<STR_LIT>' : consts . CLUSTER_NET_PROVIDERS . neutron , <EOL> '<STR_LIT>' : segment_type } , <EOL> ) <EOL> nodes_kwargs = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT:name>' : self . node_name , <EOL> '<STR_LIT>' : cluster [ '<STR_LIT:id>' ] <EOL> } <EOL> self . env . create_nodes_w_interfaces_count ( <NUM_LIT:1> , <NUM_LIT:6> , ** nodes_kwargs ) <EOL> nodes_kwargs [ '<STR_LIT>' ] = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . env . create_nodes_w_interfaces_count ( <NUM_LIT:1> , <NUM_LIT:6> , ** nodes_kwargs ) <EOL> return cluster <EOL> def create_more_nodes ( self , iface_count = <NUM_LIT:3> ) : <EOL> self . env . create_nodes_w_interfaces_count ( <EOL> <NUM_LIT:1> , iface_count , roles = [ '<STR_LIT>' ] , cluster_id = self . cluster . id ) <EOL> self . env . create_nodes_w_interfaces_count ( <EOL> <NUM_LIT:1> , iface_count , <EOL> roles = [ '<STR_LIT>' , '<STR_LIT>' ] , cluster_id = self . cluster . id ) <EOL> self . env . create_nodes_w_interfaces_count ( <EOL> <NUM_LIT:1> , iface_count , roles = [ '<STR_LIT>' ] , cluster_id = self . cluster . id ) <EOL> def check_node_ips_on_certain_networks ( self , node , net_names ) : <EOL> ips = db ( ) . query ( models . IPAddr ) . filter_by ( node = node . id ) <EOL> self . assertEqual ( ips . count ( ) , len ( net_names ) ) <EOL> for ip in ips : <EOL> self . assertIn ( ip . network_data . name , net_names ) <EOL> def test_get_net_provider_serializer ( self ) : <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> self . cluster . network_config . configuration_template = None <EOL> net_serializer = serializer . get_net_provider_serializer ( self . cluster ) <EOL> self . assertIs ( net_serializer , self . general_serializer ) <EOL> self . cluster . network_config . configuration_template = self . net_template <EOL> net_serializer = serializer . get_net_provider_serializer ( self . cluster ) <EOL> self . assertIs ( net_serializer , self . template_serializer ) <EOL> def test_ip_assignment_according_to_template ( self ) : <EOL> self . create_more_nodes ( iface_count = <NUM_LIT:4> ) <EOL> node_roles_vs_net_names = [ <EOL> ( [ '<STR_LIT>' ] , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> ( [ '<STR_LIT>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> ( [ '<STR_LIT>' ] , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) ] <EOL> template_meta = self . net_template [ "<STR_LIT>" ] [ "<STR_LIT:default>" ] <EOL> for node_role , template_list in six . iteritems ( <EOL> template_meta [ "<STR_LIT>" ] ) : <EOL> if node_role == '<STR_LIT>' : <EOL> template_list . remove ( '<STR_LIT>' ) <EOL> objects . Cluster . set_network_template ( <EOL> self . cluster , <EOL> self . net_template <EOL> ) <EOL> cluster_db = objects . Cluster . get_by_uid ( self . cluster [ '<STR_LIT:id>' ] ) <EOL> objects . Cluster . prepare_for_deployment ( cluster_db ) <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> serialized_for_astute = serializer ( <EOL> AstuteGraph ( cluster_db ) ) . serialize ( self . cluster , cluster_db . nodes ) <EOL> self . assertEqual ( len ( serialized_for_astute ) , <NUM_LIT:7> ) <EOL> for node_data in serialized_for_astute : <EOL> node = objects . Node . get_by_uid ( node_data [ '<STR_LIT>' ] ) <EOL> for node_roles , net_names in node_roles_vs_net_names : <EOL> if node . all_roles == set ( node_roles ) : <EOL> self . check_node_ips_on_certain_networks ( node , net_names ) <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" . format ( <EOL> node . all_roles ) ) <EOL> def test_gateway_not_set_for_none_ip ( self ) : <EOL> attrs = copy . deepcopy ( self . cluster . attributes . editable ) <EOL> attrs [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:value>' ] = True <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : self . cluster . id } ) , <EOL> params = jsonutils . dumps ( { '<STR_LIT>' : attrs } ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertTrue ( objects . Cluster . neutron_dvr_enabled ( self . cluster ) ) <EOL> objects . Cluster . set_network_template ( self . cluster , None ) <EOL> computes = filter ( lambda n : '<STR_LIT>' in n . roles , self . cluster . nodes ) <EOL> self . assertTrue ( len ( computes ) > <NUM_LIT:0> ) <EOL> compute = computes [ <NUM_LIT:0> ] <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> net_serializer = serializer . get_net_provider_serializer ( self . cluster ) <EOL> self . assertIs ( net_serializer , self . general_serializer ) <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> networks = nm . get_node_networks ( compute ) <EOL> self . assertFalse ( objects . Node . should_have_public_with_ip ( compute ) ) <EOL> network_scheme = net_serializer . generate_network_scheme ( <EOL> compute , networks ) <EOL> self . assertNotIn ( '<STR_LIT>' , network_scheme [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT:none>' , network_scheme [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> def test_public_iface_added_to_br_ex_in_dvr ( self ) : <EOL> attrs = copy . deepcopy ( self . cluster . attributes . editable ) <EOL> attrs [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:value>' ] = True <EOL> resp = self . app . patch ( <EOL> reverse ( <EOL> '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : self . cluster . id } ) , <EOL> params = jsonutils . dumps ( { '<STR_LIT>' : attrs } ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertTrue ( objects . Cluster . neutron_dvr_enabled ( self . cluster ) ) <EOL> objects . Cluster . set_network_template ( self . cluster , None ) <EOL> computes = filter ( lambda n : '<STR_LIT>' in n . roles , self . cluster . nodes ) <EOL> self . assertTrue ( len ( computes ) > <NUM_LIT:0> ) <EOL> compute = computes [ <NUM_LIT:0> ] <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> net_serializer = serializer . get_net_provider_serializer ( self . cluster ) <EOL> self . assertIs ( net_serializer , self . general_serializer ) <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> networks = nm . get_node_networks ( compute ) <EOL> public_net = next ( ( net for net in networks if net [ '<STR_LIT:name>' ] == '<STR_LIT>' ) , <EOL> None ) <EOL> self . assertIsNotNone ( public_net ) <EOL> self . assertFalse ( objects . Node . should_have_public_with_ip ( compute ) ) <EOL> network_scheme = net_serializer . generate_network_scheme ( <EOL> compute , networks ) <EOL> expected = { '<STR_LIT:action>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : public_net [ '<STR_LIT>' ] } <EOL> self . assertIn ( expected , network_scheme [ '<STR_LIT>' ] ) <EOL> def test_replacements_in_network_assignments ( self ) : <EOL> node_roles_vs_net_names = [ <EOL> ( [ '<STR_LIT>' ] , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) ] <EOL> template_meta = self . net_template [ "<STR_LIT>" ] [ "<STR_LIT:default>" ] <EOL> iface_var = template_meta [ '<STR_LIT>' ] [ '<STR_LIT:default>' ] . keys ( ) [ <NUM_LIT:0> ] <EOL> ep_with_var = "<STR_LIT>" . format ( iface_var ) <EOL> template_meta [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] = ep_with_var <EOL> template_meta [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] = [ ep_with_var ] <EOL> objects . Cluster . set_network_template ( <EOL> self . cluster , <EOL> self . net_template <EOL> ) <EOL> cluster_db = objects . Cluster . get_by_uid ( self . cluster [ '<STR_LIT:id>' ] ) <EOL> objects . Cluster . prepare_for_deployment ( cluster_db ) <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> serialized_for_astute = serializer ( <EOL> AstuteGraph ( cluster_db ) ) . serialize ( self . cluster , cluster_db . nodes ) <EOL> for node_data in serialized_for_astute : <EOL> node = objects . Node . get_by_uid ( node_data [ '<STR_LIT>' ] ) <EOL> for node_roles , net_names in node_roles_vs_net_names : <EOL> if node . all_roles == set ( node_roles ) : <EOL> self . check_node_ips_on_certain_networks ( node , net_names ) <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" . format ( <EOL> node . all_roles ) ) <EOL> def test_multiple_node_roles_network_roles ( self ) : <EOL> expected_roles = { <EOL> objects . Node . get_node_fqdn ( self . cluster . nodes [ <NUM_LIT:0> ] ) : { <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> '<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> '<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> '<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> '<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> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> objects . Node . get_node_fqdn ( self . cluster . nodes [ <NUM_LIT:1> ] ) : { <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> '<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> '<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> '<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> '<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> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } <EOL> for node in self . serialized_for_astute : <EOL> roles = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assertEqual ( roles , expected_roles [ node [ '<STR_LIT>' ] ] ) <EOL> def test_routes_while_using_multiple_node_groups ( self ) : <EOL> self . env . create_node_group ( ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster ) <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> facts = serializer ( AstuteGraph ( self . cluster ) ) . serialize ( <EOL> self . cluster , self . cluster . nodes ) <EOL> for node in facts : <EOL> node_db = objects . Node . get_by_uid ( node [ '<STR_LIT>' ] ) <EOL> is_public = objects . Node . should_have_public ( node_db ) <EOL> endpoints = node [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in node_db . roles : <EOL> self . assertEqual ( endpoints [ '<STR_LIT>' ] , { '<STR_LIT>' : '<STR_LIT:none>' } ) <EOL> endpoints . pop ( '<STR_LIT>' ) <EOL> else : <EOL> self . assertNotIn ( '<STR_LIT>' , endpoints ) <EOL> if is_public : <EOL> self . assertEqual ( endpoints [ '<STR_LIT>' ] , { '<STR_LIT>' : '<STR_LIT:none>' } ) <EOL> endpoints . pop ( '<STR_LIT>' ) <EOL> for name , descr in six . iteritems ( endpoints ) : <EOL> self . assertTrue ( { '<STR_LIT>' , '<STR_LIT>' } . issubset ( descr ) ) <EOL> self . assertIn ( len ( descr [ '<STR_LIT>' ] ) , [ <NUM_LIT:0> , <NUM_LIT:1> ] ) <EOL> if descr [ '<STR_LIT>' ] : <EOL> self . assertEqual ( name , '<STR_LIT>' ) <EOL> for route in descr [ '<STR_LIT>' ] : <EOL> self . assertItemsEqual ( [ '<STR_LIT>' , '<STR_LIT>' ] , route ) <EOL> def test_multiple_node_roles_transformations ( self ) : <EOL> node = self . cluster . nodes [ <NUM_LIT:1> ] <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> net_serializer = serializer . get_net_provider_serializer ( self . cluster ) <EOL> transformations = net_serializer . generate_transformations ( node ) <EOL> admin_brs = filter ( lambda t : t . get ( '<STR_LIT:name>' ) == '<STR_LIT>' , <EOL> transformations ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( admin_brs ) ) <EOL> self . assertEqual ( '<STR_LIT>' , transformations [ <NUM_LIT:4> ] [ '<STR_LIT:name>' ] ) <EOL> port_seen = False <EOL> for tx in transformations : <EOL> if tx . get ( '<STR_LIT:name>' ) == '<STR_LIT>' and tx [ '<STR_LIT:action>' ] == '<STR_LIT>' and port_seen : <EOL> self . fail ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if tx . get ( '<STR_LIT>' ) == '<STR_LIT>' and tx [ '<STR_LIT:action>' ] == '<STR_LIT>' : <EOL> port_seen = True <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 . assertItemsEqual ( <EOL> node_attrs , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:name>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <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> def test_multiple_node_roles_network_metadata_roles ( self ) : <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> ip_by_net = { } <EOL> for node_data in self . serialized_for_astute : <EOL> nodes = node_data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for node_name , node_attrs in nodes . items ( ) : <EOL> node = objects . Node . get_by_uid ( node_attrs [ '<STR_LIT>' ] ) <EOL> networks = nm . get_node_networks ( node ) <EOL> node_nets = [ n [ '<STR_LIT:name>' ] for n in networks ] <EOL> for net in node_nets : <EOL> netgroup = nm . get_network_by_netname ( net , networks ) <EOL> if netgroup . get ( '<STR_LIT>' ) : <EOL> ip_by_net [ net ] = netgroup [ '<STR_LIT>' ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] <EOL> network_roles = { <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] <EOL> } <EOL> if node . all_roles == set ( [ '<STR_LIT>' ] ) : <EOL> network_roles . update ( { <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] <EOL> } ) <EOL> else : <EOL> network_roles . update ( { <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : ip_by_net [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : None <EOL> } ) <EOL> self . assertEqual ( <EOL> node_attrs [ '<STR_LIT>' ] , <EOL> network_roles <EOL> ) <EOL> def test_delete_default_network_group ( self ) : <EOL> net_name = "<STR_LIT>" <EOL> node_group = objects . Cluster . get_default_group ( self . cluster ) <EOL> storage_net = objects . NetworkGroup . get_from_node_group_by_name ( <EOL> node_group . id , net_name ) <EOL> objects . NetworkGroup . delete ( storage_net ) <EOL> net_template = self . env . read_fixtures ( [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] <EOL> template_meta = net_template [ "<STR_LIT>" ] [ "<STR_LIT:default>" ] <EOL> del ( template_meta [ "<STR_LIT>" ] [ net_name ] ) <EOL> for k , v in six . iteritems ( template_meta [ "<STR_LIT>" ] ) : <EOL> if net_name in v : <EOL> v . remove ( net_name ) <EOL> del ( template_meta [ "<STR_LIT>" ] [ net_name ] ) <EOL> objects . Cluster . set_network_template ( <EOL> self . cluster , <EOL> net_template <EOL> ) <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> net_serializer = serializer . get_net_provider_serializer ( self . cluster ) <EOL> net_serializer . generate_network_metadata ( self . cluster ) <EOL> def test_network_not_mapped_to_nics_w_template ( self ) : <EOL> resp = self . app . get ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = self . cluster ) , <EOL> headers = self . default_headers , <EOL> expect_errors = False <EOL> ) <EOL> management = None <EOL> for ng in jsonutils . loads ( resp . body ) : <EOL> if ng [ '<STR_LIT:name>' ] == '<STR_LIT>' : <EOL> management = ng <EOL> break <EOL> self . app . delete ( <EOL> reverse ( <EOL> '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : management . pop ( '<STR_LIT:id>' ) } <EOL> ) , <EOL> headers = self . default_headers <EOL> ) <EOL> self . app . post ( <EOL> reverse ( '<STR_LIT>' ) , <EOL> jsonutils . dumps ( management ) , <EOL> headers = self . default_headers , <EOL> expect_errors = False , <EOL> ) <EOL> resp = self . app . get ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = self . cluster ) , <EOL> headers = self . default_headers , <EOL> expect_errors = False <EOL> ) <EOL> cluster_db = self . db . query ( models . Cluster ) . get ( self . cluster [ '<STR_LIT:id>' ] ) <EOL> objects . Cluster . prepare_for_deployment ( cluster_db ) <EOL> serializer = get_serializer_for_cluster ( cluster_db ) <EOL> self . serialized_for_astute = serializer ( <EOL> AstuteGraph ( cluster_db ) ) . serialize ( cluster_db , cluster_db . nodes ) <EOL> network_roles = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> for node_data in self . serialized_for_astute : <EOL> for n in node_data [ '<STR_LIT>' ] : <EOL> n_db = objects . Node . get_by_uid ( n [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in n_db . roles : <EOL> self . assertIn ( '<STR_LIT>' , n ) <EOL> self . assertIn ( '<STR_LIT>' , n ) <EOL> self . assertIn ( '<STR_LIT>' , n ) <EOL> self . assertIn ( '<STR_LIT>' , n ) <EOL> self . assertNotIn ( '<STR_LIT>' , n ) <EOL> self . assertNotIn ( '<STR_LIT>' , n ) <EOL> else : <EOL> self . assertIn ( '<STR_LIT>' , n ) <EOL> self . assertIn ( '<STR_LIT>' , n ) <EOL> self . assertNotIn ( '<STR_LIT>' , n ) <EOL> self . assertNotIn ( '<STR_LIT>' , n ) <EOL> self . assertIn ( '<STR_LIT>' , n ) <EOL> self . assertIn ( '<STR_LIT>' , n ) <EOL> nodes = node_data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for node_name , node_attrs in nodes . items ( ) : <EOL> for role in network_roles : <EOL> self . assertIsNotNone ( node_attrs [ '<STR_LIT>' ] [ role ] ) <EOL> def test_floating_role_belongs_to_public_bridge ( self ) : <EOL> net_template = self . env . read_fixtures ( [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] <EOL> schemes = net_template [ "<STR_LIT>" ] [ "<STR_LIT:default>" ] [ "<STR_LIT>" ] <EOL> schemes [ "<STR_LIT>" ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> objects . Cluster . set_network_template ( <EOL> self . cluster , <EOL> net_template <EOL> ) <EOL> cluster_db = self . db . query ( models . Cluster ) . get ( self . cluster [ '<STR_LIT:id>' ] ) <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> serializer = get_serializer_for_cluster ( self . cluster ) <EOL> self . serialized_for_astute = serializer ( <EOL> AstuteGraph ( cluster_db ) ) . serialize ( self . cluster , cluster_db . nodes ) <EOL> for node_data in self . serialized_for_astute : <EOL> node = objects . Node . get_by_uid ( node_data [ '<STR_LIT>' ] ) <EOL> if objects . Node . should_have_public_with_ip ( node ) : <EOL> nets = nm . get_node_networks ( node ) <EOL> ng = nm . get_network_by_netname ( '<STR_LIT>' , nets ) <EOL> endpoints = node_data [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> self . assertEqual ( endpoints [ "<STR_LIT>" ] [ "<STR_LIT>" ] , [ ng . get ( '<STR_LIT>' ) ] ) <EOL> def test_get_node_network_mapping ( self ) : <EOL> self . create_more_nodes ( ) <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> node_roles_vs_networks = [ <EOL> ( [ '<STR_LIT>' ] , [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) , <EOL> ( [ '<STR_LIT>' ] , [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) , <EOL> ( [ '<STR_LIT>' ] , [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' ] , [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' ] , [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) ] <EOL> for node in self . env . nodes : <EOL> net_names_and_eps = nm . get_node_network_mapping ( node ) <EOL> for node_roles , networks in node_roles_vs_networks : <EOL> if node . all_roles == set ( node_roles ) : <EOL> self . assertItemsEqual ( net_names_and_eps , networks ) <EOL> def test_get_network_name_to_endpoint_mappings ( self ) : <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> group_id = objects . Cluster . get_default_group ( self . cluster ) . id <EOL> self . assertEqual ( <EOL> nm . get_network_name_to_endpoint_mappings ( self . cluster ) , <EOL> { <EOL> group_id : { <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> } <EOL> } <EOL> ) <EOL> def test_assign_ips_in_node_group ( self ) : <EOL> mgmt = self . db . query ( models . NetworkGroup ) . filter_by ( name = '<STR_LIT>' ) . first ( ) <EOL> ips_2_db = self . db . query ( models . IPAddr . ip_addr ) . filter ( models . IPAddr . network == mgmt . id , <EOL> models . IPAddr . node . isnot ( None ) ) <EOL> self . assertEqual ( ips_2_db . count ( ) , <NUM_LIT:2> ) <EOL> ips_2_str = set ( ips_2_db ) <EOL> self . create_more_nodes ( ) <EOL> node_ids = set ( n . id for n in self . env . nodes ) <EOL> ip_ranges = [ netaddr . IPRange ( r . first , r . last ) <EOL> for r in mgmt . ip_ranges ] <EOL> nm = objects . Cluster . get_network_manager ( self . cluster ) <EOL> nm . assign_ips_in_node_group ( <EOL> mgmt . id , mgmt . name , node_ids , ip_ranges ) <EOL> ips_5_db = self . db . query ( models . IPAddr . ip_addr ) . filter ( models . IPAddr . network == mgmt . id , <EOL> models . IPAddr . node . isnot ( None ) ) <EOL> self . assertEqual ( ips_5_db . count ( ) , <NUM_LIT:5> ) <EOL> ips_5_str = set ( ips_5_db ) <EOL> self . assertEqual ( len ( ips_5_str . difference ( ips_2_str ) ) , <NUM_LIT:3> ) <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 = get_serializer_for_cluster ( 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> self . assertNotIn ( '<STR_LIT>' , endpoints [ name ] ) <EOL> def test_vendor_specific_in_deployment_serializer ( self ) : <EOL> self . check_vendor_specific_is_not_set ( ) <EOL> def test_vendor_specific_in_template_serializer ( self ) : <EOL> self . check_vendor_specific_is_not_set ( use_net_template = True ) <EOL> class TestCustomNetGroupIpAllocation ( BaseDeploymentSerializer ) : <EOL> env_version = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> super ( TestCustomNetGroupIpAllocation , self ) . setUp ( ) <EOL> self . cluster = self . create_env ( ) <EOL> self . cluster_db = self . env . clusters [ - <NUM_LIT:1> ] <EOL> def create_env ( self ) : <EOL> return self . env . create ( <EOL> release_kwargs = { '<STR_LIT:version>' : self . env_version } , <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : consts . CLUSTER_NET_PROVIDERS . neutron , <EOL> '<STR_LIT>' : consts . NEUTRON_SEGMENT_TYPES . gre } , <EOL> nodes_kwargs = [ <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> ] ) <EOL> def test_ip_allocation ( self ) : <EOL> self . env . _create_network_group ( <EOL> cluster = self . cluster , name = '<STR_LIT:test>' , cidr = '<STR_LIT>' , <EOL> meta = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] } ) <EOL> objects . Cluster . prepare_for_deployment ( self . cluster_db ) <EOL> ip_column_name = models . IPAddr . ip_addr . label ( '<STR_LIT>' ) <EOL> ip_addrs_count = db ( ) . query ( models . IPAddr ) . filter ( <EOL> "<STR_LIT>" . format ( ip_column_name ) <EOL> ) . count ( ) <EOL> self . assertEqual ( ip_addrs_count , <NUM_LIT:2> ) <EOL> class TestSerializer70Mixin ( object ) : <EOL> env_version = "<STR_LIT>" <EOL> class TestNovaOrchestratorSerializer70 ( TestSerializer70Mixin , <EOL> TestNovaOrchestratorSerializer , <EOL> PrepareDataMixin ) : <EOL> def create_env ( self , mode , network_manager = '<STR_LIT>' ) : <EOL> node_args = [ <EOL> { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , '<STR_LIT>' : True } , <EOL> { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , '<STR_LIT>' : True } , <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] , '<STR_LIT>' : True } , <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] , '<STR_LIT>' : True } , <EOL> { '<STR_LIT>' : [ ] , '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : True } ] <EOL> release = self . patch_net_roles_for_release ( ) <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { <EOL> '<STR_LIT>' : release . id , <EOL> '<STR_LIT>' : mode , <EOL> '<STR_LIT>' : network_manager , <EOL> '<STR_LIT>' : consts . CLUSTER_NET_PROVIDERS . nova_network } , <EOL> nodes_kwargs = node_args ) <EOL> cluster_db = self . db . query ( models . Cluster ) . get ( cluster [ '<STR_LIT:id>' ] ) <EOL> objects . Cluster . prepare_for_deployment ( cluster_db ) <EOL> self . db . flush ( ) <EOL> return cluster_db <EOL> class TestSerializeInterfaceDriversData70 ( TestSerializer70Mixin , <EOL> TestSerializeInterfaceDriversData ) : <EOL> pass <EOL> class TestDeploymentHASerializer70 ( TestSerializer70Mixin , <EOL> TestDeploymentHASerializer61 ) : <EOL> pass </s>
