def test_set_port_mac(self):
     attr = self.generate_device_details()
     self.manage_device(attr)
     plumber = trunk_plumber.Plumber(namespace=attr.namespace)
     # force it to return name of above
     plumber._get_tap_device_name = lambda x: attr.name
     new_mac = utils.get_random_mac('fa:16:3e:00:00:00'.split(':'))
     self.assertTrue(plumber.set_port_mac('port_id', new_mac))
     self.assertFalse(plumber.set_port_mac('port_id', new_mac))
     new_mac = utils.get_random_mac('fa:16:3e:00:00:00'.split(':'))
     self.assertTrue(plumber.set_port_mac('port_id', new_mac))
     self.assertFalse(plumber.set_port_mac('port_id', new_mac))

 def _create_dvr_mac_address(self, context, host):
     """Create DVR mac address for a given host."""
     base_mac = cfg.CONF.dvr_base_mac.split(':')
     max_retries = cfg.CONF.mac_generation_retries
     for attempt in reversed(range(max_retries)):
         try:
             with context.session.begin(subtransactions=True):
                 mac_address = utils.get_random_mac(base_mac)
                 dvr_mac_binding = DistributedVirtualRouterMacAddress(
                     host=host, mac_address=mac_address)
                 context.session.add(dvr_mac_binding)
                 LOG.debug("Generated DVR mac for host %(host)s "
                           "is %(mac_address)s",
                           {'host': host, 'mac_address': mac_address})
             dvr_macs = self.get_dvr_mac_address_list(context)
             # TODO(vivek): improve scalability of this fanout by
             # sending a single mac address rather than the entire set
             self.notifier.dvr_mac_address_update(context, dvr_macs)
             return self._make_dvr_mac_address_dict(dvr_mac_binding)
         except db_exc.DBDuplicateEntry:
             LOG.debug("Generated DVR mac %(mac)s exists."
                       " Remaining attempts %(attempts_left)s.",
                       {'mac': mac_address, 'attempts_left': attempt})
     LOG.error(_LE("MAC generation error after %s attempts"), max_retries)
     raise ext_dvr.MacAddressGenerationFailure(host=host)

 def generate_device_details(self, name=None, ip_cidrs=None,
                             mac_address=None, namespace=None):
     return Device(name or base.get_rand_name(),
                   ip_cidrs or ["%s/24" % TEST_IP],
                   mac_address or
                   utils.get_random_mac('fa:16:3e:00:00:00'.split(':')),
                   namespace or base.get_rand_name())

 def generate_device_details(self, name=None, ip_cidr=None,
                             mac_address=None, namespace=None):
     return Device(name or base.get_rand_name(),
                   ip_cidr or '240.0.0.1/24',
                   mac_address or
                   utils.get_random_mac('fa:16:3e:00:00:00'.split(':')),
                   namespace or base.get_rand_name())

    def test_plug_with_namespace_sets_mtu_higher_than_bridge(self):
        device_mtu = 1450

        # Create a new OVS bridge
        ovs_bridge = self.useFixture(net_helpers.OVSBridgeFixture()).bridge
        self.assertFalse(ovs_bridge.get_port_name_list())

        # Add a new linuxbridge port with reduced MTU to OVS bridge
        lb_bridge = self.useFixture(
            net_helpers.LinuxBridgeFixture()).bridge
        lb_bridge_port = self.useFixture(
            net_helpers.LinuxBridgePortFixture(lb_bridge))
        lb_bridge_port.port.link.set_mtu(device_mtu - 1)
        ovs_bridge.add_port(lb_bridge_port.port.name)

        # Now plug a device with intended MTU that is higher than for the port
        # above and validate that its MTU is not reduced to the least MTU on
        # the bridge
        device_name = utils.get_rand_name()
        mac_address = utils.get_random_mac('fa:16:3e:00:00:00'.split(':'))
        namespace = self.useFixture(net_helpers.NamespaceFixture()).name
        self.interface.plug(network_id=uuidutils.generate_uuid(),
                            port_id=uuidutils.generate_uuid(),
                            device_name=device_name,
                            mac_address=mac_address,
                            bridge=ovs_bridge.br_name,
                            namespace=namespace,
                            mtu=device_mtu)

        self.assertIn(device_name, ovs_bridge.get_port_name_list())
        self.assertTrue(ip_lib.device_exists(device_name, namespace))
        self.assertEqual(
            device_mtu,
            ip_lib.IPDevice(device_name, namespace=namespace).link.mtu
        )

    def setUp(self):
        super(TestTaasPlugin, self).setUp()
        mock.patch.object(n_rpc, 'create_connection', auto_spec=True).start()
        mock.patch.object(taas_plugin, 'TaasCallbacks', auto_spec=True).start()
        mock.patch.object(taas_plugin, 'TaasAgentApi', auto_spec=True).start()
        self._plugin = taas_plugin.TaasPlugin()
        self._context = context.get_admin_context()

        self._tenant_id = 'tenant-X'
        self._network_id = uuidutils.generate_uuid()
        self._host_id = 'host-A'
        self._port_id = uuidutils.generate_uuid()
        self._port_details = {
            'tenant_id': self._tenant_id,
            'binding:host_id': self._host_id,
            'mac_address': n_utils.get_random_mac(
                'fa:16:3e:00:00:00'.split(':')),
        }
        self._tap_service = {
            'tenant_id': self._tenant_id,
            'name': 'MyTap',
            'description': 'This is my tap service',
            'port_id': self._port_id,
            'network_id': self._network_id,
        }
        self._tap_flow = {
            'description': 'This is my tap flow',
            'direction': 'BOTH',
            'name': 'MyTapFlow',
            'source_port': self._port_id,
            'tenant_id': self._tenant_id,
        }

 def generate_router_info(self):
     self.info = copy.deepcopy(FAKE_ROUTER)
     self.info["id"] = _uuid()
     self.info["_interfaces"] = [self._generate_private_interface_for_router(subnet) for subnet in self.private_nets]
     self.info["gw_port"]["id"] = _uuid()
     self.info["gw_port"]["fixed_ips"][0]["ip_address"] = str(self.public_net)
     self.info["gw_port"]["mac_address"] = common_utils.get_random_mac(MAC_BASE)
     self.info["ha"] = False

 def setUp(self):
     super(TrunkParentPortTestCase, self).setUp()
     # Mock out connecting to ovsdb
     mock.patch(NATIVE_OVSDB_CONNECTION).start()
     trunk_id = uuidutils.generate_uuid()
     port_id = uuidutils.generate_uuid()
     trunk_mac = common_utils.get_random_mac('fa:16:3e:00:00:00'.split(':'))
     self.trunk = trunk_manager.TrunkParentPort(
         trunk_id, port_id, trunk_mac)

 def _create_test_port_dict(self):
     return {'id': uuidutils.generate_uuid(),
             'mac_address': utils.get_random_mac(
                 'fa:16:3e:00:00:00'.split(':')),
             'fixed_ips': [{
                 'ip_address': '10.%d.%d.%d' % (
                      random.randint(3, 254),
                      random.randint(3, 254),
                      random.randint(3, 254))}],
             'vif_name': base.get_rand_name(
                 self.driver.DEV_NAME_LEN, self.driver.DEV_NAME_PREFIX)}

 def generate_router_info(self):
     self.info = copy.deepcopy(FAKE_ROUTER)
     self.info['id'] = _uuid()
     self.info['_interfaces'] = [
         self._generate_private_interface_for_router(subnet)
         for subnet in self.private_nets]
     self.info['gw_port']['id'] = _uuid()
     self.info['gw_port']['fixed_ips'][0]['ip_address'] = str(
         self.public_net)
     self.info['gw_port']['mac_address'] = (
         common_utils.get_random_mac(MAC_BASE))
     self.info['ha'] = False

 def _generate_info(self, public_ip, private_cidr, enable_ha=False):
     """Generate router info"""
     info = copy.deepcopy(FAKE_ROUTER)
     info['id'] = _uuid()
     info['_interfaces'][0]['id'] = _uuid()
     (info['_interfaces'][0]
      ['mac_address']) = common_utils.get_random_mac(MAC_BASE)
     (info['_interfaces'][0]['fixed_ips'][0]
      ['ip_address']) = str(private_cidr[4])
     info['_interfaces'][0]['subnets'][0].update({
         'cidr': str(private_cidr),
         'gateway_ip': str(private_cidr[1])})
     info['gw_port']['id'] = _uuid()
     info['gw_port']['fixed_ips'][0]['ip_address'] = str(public_ip)
     info['gw_port']['mac_address'] = common_utils.get_random_mac(MAC_BASE)
     if enable_ha:
         info['ha'] = True
         info['ha_vr_id'] = 1
         info[l3_constants.HA_INTERFACE_KEY] = (
             l3_test_common.get_ha_interface())
     else:
         info['ha'] = False
     return info

 def _create_dvr_mac_address_retry(self, context, host, base_mac):
     with context.session.begin(subtransactions=True):
         mac_address = utils.get_random_mac(base_mac)
         dvr_mac_binding = dvr_models.DistributedVirtualRouterMacAddress(
             host=host, mac_address=mac_address)
         context.session.add(dvr_mac_binding)
         LOG.debug("Generated DVR mac for host %(host)s "
                   "is %(mac_address)s",
                   {'host': host, 'mac_address': mac_address})
     dvr_macs = self.get_dvr_mac_address_list(context)
     # TODO(vivek): improve scalability of this fanout by
     # sending a single mac address rather than the entire set
     self.notifier.dvr_mac_address_update(context, dvr_macs)
     return self._make_dvr_mac_address_dict(dvr_mac_binding)

 def test_arp_correct_protection_allowed_address_pairs(self):
     smac = self.source.port.link.address
     port = {'mac_address': '00:11:22:33:44:55',
             'allowed_address_pairs': [{'mac_address': smac,
                                        'ip_address': self.source.ip}]}
     # make sure a large number of allowed address pairs works
     for i in range(100000):
         port['allowed_address_pairs'].append(
             {'mac_address': utils.get_random_mac(
                  'fa:16:3e:00:00:00'.split(':')),
              'ip_address': '10.10.10.10'})
     self._add_arp_protection(self.source, ['1.2.2.2'], port)
     self._add_arp_protection(self.destination, [self.destination.ip])
     arping(self.source.namespace, self.destination.ip)
     arping(self.destination.namespace, self.source.ip)

 def test_arp_correct_protection_allowed_address_pairs(self):
     smac = self.source.port.link.address
     port = {
         "mac_address": "00:11:22:33:44:55",
         "allowed_address_pairs": [{"mac_address": smac, "ip_address": self.source.ip}],
     }
     # make sure a large number of allowed address pairs works
     for i in range(100000):
         port["allowed_address_pairs"].append(
             {"mac_address": utils.get_random_mac("fa:16:3e:00:00:00".split(":")), "ip_address": "10.10.10.10"}
         )
     self._add_arp_protection(self.source, ["1.2.2.2"], port)
     self._add_arp_protection(self.destination, [self.destination.ip])
     arping(self.source.namespace, self.destination.ip)
     arping(self.destination.namespace, self.source.ip)

 def _generate_private_interface_for_router(self, subnet):
     subnet_id = _uuid()
     return {
         "id": _uuid(),
         "admin_state_up": True,
         "network_id": _uuid(),
         "mac_address": common_utils.get_random_mac(MAC_BASE),
         "subnets": [
             {
                 "ipv6_ra_mode": None,
                 "cidr": str(subnet),
                 "gateway_ip": str(subnet[1]),
                 "id": subnet_id,
                 "ipv6_address_mode": None,
             }
         ],
         "fixed_ips": [{"subnet_id": subnet_id, "prefixlen": 24, "ip_address": str(subnet[4])}],
     }

    def _get_ovn_dhcpv4_opts(self, subnet, network, server_mac=None):
        if not subnet['gateway_ip']:
            return {}

        default_lease_time = str(config.get_ovn_dhcp_default_lease_time())
        mtu = network['mtu']
        options = {
            'server_id': subnet['gateway_ip'],
            'lease_time': default_lease_time,
            'mtu': str(mtu),
            'router': subnet['gateway_ip']
        }

        if server_mac:
            options['server_mac'] = server_mac
        else:
            options['server_mac'] = n_utils.get_random_mac(
                cfg.CONF.base_mac.split(':'))

        if subnet['dns_nameservers']:
            dns_servers = '{'
            for dns in subnet["dns_nameservers"]:
                dns_servers += dns + ', '
            dns_servers = dns_servers.strip(', ')
            dns_servers += '}'
            options['dns_server'] = dns_servers

        # If subnet hostroutes are defined, add them in the
        # 'classless_static_route' dhcp option
        classless_static_routes = "{"
        for route in subnet['host_routes']:
            classless_static_routes += ("%s,%s, ") % (
                route['destination'], route['nexthop'])

        if classless_static_routes != "{":
            # if there are static routes, then we need to add the
            # default route in this option. As per RFC 3442 dhcp clients
            # should ignore 'router' dhcp option (option 3)
            # if option 121 is present.
            classless_static_routes += "0.0.0.0/0,%s}" % (subnet['gateway_ip'])
            options['classless_static_route'] = classless_static_routes

        return options

 def _create_dvr_mac_for_extern_ip(self, context, host):
     """Create dvr mac address for a extern net ip."""
     base_mac = cfg.CONF.dvr_base_mac.split(':')
     max_retries = cfg.CONF.mac_generation_retries
     for attempt in reversed(range(max_retries)):
         try:
             with context.session.begin(subtransactions=True):
                 mac_address = utils.get_random_mac(base_mac)
                 dvr_mac_binding = DistributedVirtualRouterMacAddress(
                     host=host, mac_address=mac_address)
                 context.session.add(dvr_mac_binding)
                 LOG.debug("Generated dvr mac for host %(host)s "
                           "is %(mac_address)s",
                           {'host': host, 'mac_address': mac_address})
             return self._make_dvr_mac_address_dict(dvr_mac_binding)
         except db_exc.DBDuplicateEntry:
             LOG.debug("Generated dvr mac %(mac)s exists."
                       " Remaining attempts %(attempts_left)s.",
                       {'mac': mac_address, 'attempts_left': attempt})
     LOG.error(_("MAC generation error after %s attempts"), max_retries)
     raise ext_dvr.MacAddressGenerationFailure(host=host)

    def _get_ovn_dhcpv4_opts(self, subnet, network):
        if not subnet["gateway_ip"]:
            return {}

        default_lease_time = str(config.get_ovn_dhcp_default_lease_time())
        mtu = network["mtu"]
        options = {
            "server_id": subnet["gateway_ip"],
            "server_mac": n_utils.get_random_mac(cfg.CONF.base_mac.split(":")),
            "lease_time": default_lease_time,
            "mtu": str(mtu),
            "router": subnet["gateway_ip"],
        }

        if subnet["dns_nameservers"]:
            dns_servers = "{"
            for dns in subnet["dns_nameservers"]:
                dns_servers += dns + ", "
            dns_servers = dns_servers.strip(", ")
            dns_servers += "}"
            options["dns_server"] = dns_servers

        # If subnet hostroutes are defined, add them in the
        # 'classless_static_route' dhcp option
        classless_static_routes = "{"
        for route in subnet["host_routes"]:
            classless_static_routes += ("%s,%s, ") % (route["destination"], route["nexthop"])

        if classless_static_routes != "{":
            # if there are static routes, then we need to add the
            # default route in this option. As per RFC 3442 dhcp clients
            # should ignore 'router' dhcp option (option 3)
            # if option 121 is present.
            classless_static_routes += "0.0.0.0/0,%s}" % (subnet["gateway_ip"])
            options["classless_static_route"] = classless_static_routes

        return options

 def _generate_private_interface_for_router(self, subnet):
     subnet_id = _uuid()
     return {
         'id': _uuid(),
         'admin_state_up': True,
         'network_id': _uuid(),
         'mac_address': common_utils.get_random_mac(MAC_BASE),
         'subnets': [
             {
                 'ipv6_ra_mode': None,
                 'cidr': str(subnet),
                 'gateway_ip': str(subnet[1]),
                 'id': subnet_id,
                 'ipv6_address_mode': None
             }
         ],
         'fixed_ips': [
             {
                 'subnet_id': subnet_id,
                 'prefixlen': 24,
                 'ip_address': str(subnet[4])
             }
         ]
     }

    def _setup_port_binding(self, **kwargs):
        with self.ctx.session.begin(subtransactions=True):
            mac = utils.get_random_mac('fa:16:3e:00:00:00'.split(':'))
            port_id = uuidutils.generate_uuid()
            network_id = kwargs.get('network_id', TEST_NETWORK_ID)
            device_owner = kwargs.get('device_owner', '')
            device_id = kwargs.get('device_id', '')
            host = kwargs.get('host', helpers.HOST)

            self.ctx.session.add(models_v2.Port(
                id=port_id, network_id=network_id, mac_address=mac,
                admin_state_up=True, status=constants.PORT_STATUS_ACTIVE,
                device_id=device_id, device_owner=device_owner))

            port_binding_cls = models.PortBinding
            binding_kwarg = {'port_id': port_id,
                             'host': host,
                             'vif_type': portbindings.VIF_TYPE_UNBOUND,
                             'vnic_type': portbindings.VNIC_NORMAL}

            if device_owner == constants.DEVICE_OWNER_DVR_INTERFACE:
                port_binding_cls = models.DistributedPortBinding
                binding_kwarg['router_id'] = TEST_ROUTER_ID
                binding_kwarg['status'] = constants.PORT_STATUS_DOWN

            self.ctx.session.add(port_binding_cls(**binding_kwarg))

            if network_id == TEST_HA_NETWORK_ID:
                agent = self.get_l3_agent_by_host(host)
                haport_bindings_cls = l3ha_model.L3HARouterAgentPortBinding
                habinding_kwarg = {'port_id': port_id,
                                   'router_id': device_id,
                                   'l3_agent_id': agent['id'],
                                   'state': kwargs.get('host_state',
                                       n_const.HA_ROUTER_STATE_ACTIVE)}
                self.ctx.session.add(haport_bindings_cls(**habinding_kwarg))