text:
import contextlib
import functools
import mock
import six
import testtools
import uuid
import webob

import fixtures
from oslo_db import exception as db_exc

from neutron.callbacks import registry
from neutron.common import constants
from neutron.common import exceptions as exc
from neutron.common import utils
from neutron import context
from neutron.db import api as db_api
from neutron.db import db_base_plugin_v2 as base_plugin
from neutron.db import l3_db
from neutron.db import models_v2
from neutron.extensions import external_net as external_net
from neutron.extensions import multiprovidernet as mpnet
from neutron.extensions import portbindings
from neutron.extensions import providernet as pnet
from neutron import manager
from neutron.plugins.common import constants as service_constants
from neutron.plugins.ml2.common import exceptions as ml2_exc
from neutron.plugins.ml2 import config
from neutron.plugins.ml2 import db as ml2_db
from neutron.plugins.ml2 import driver_api
from neutron.plugins.ml2 import driver_context
from neutron.plugins.ml2.drivers import type_vlan
from neutron.plugins.ml2 import models
from neutron.plugins.ml2 import plugin as ml2_plugin
from neutron.tests import base
from neutron.tests.unit import _test_extension_portbindings as test_bindings
from neutron.tests.unit.agent import test_securitygroups_rpc as test_sg_rpc
from neutron.tests.unit.db import test_allowedaddresspairs_db as test_pair
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
from neutron.tests.unit.extensions import test_extra_dhcp_opt as test_dhcpopts
from neutron.tests.unit.plugins.ml2.drivers import mechanism_logger as \
     mech_logger
from neutron.tests.unit.plugins.ml2.drivers import mechanism_test as mech_test

config.cfg.CONF.import_opt('network_vlan_ranges',
                           'neutron.plugins.ml2.drivers.type_vlan',
                           group='ml2_type_vlan')

PLUGIN_NAME = 'neutron.plugins.ml2.plugin.Ml2Plugin'

DEVICE_OWNER_COMPUTE = 'compute:None'
HOST = 'fake_host'

# TODO(marun) - Move to somewhere common for reuse
class PluginConfFixture(fixtures.Fixture):
    """Plugin configuration shared across the unit and functional tests."""

    def __init__(self, plugin_name, parent_setup=None):
        self.plugin_name = plugin_name
        self.parent_setup = parent_setup

    def setUp(self):
        super(PluginConfFixture, self).setUp()
        if self.parent_setup:
            self.parent_setup()

class Ml2ConfFixture(PluginConfFixture):

    def __init__(self, parent_setup=None):
        super(Ml2ConfFixture, self).__init__(PLUGIN_NAME, parent_setup)

class Ml2PluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):

    _mechanism_drivers = ['logger', 'test']

    def setup_parent(self):
        """Perform parent setup with the common plugin configuration class."""
        l3_plugin = ('neutron.tests.unit.extensions.test_l3.'
                     'TestL3NatServicePlugin')
        service_plugins = {'l3_plugin_name': l3_plugin}
        # Ensure that the parent setup can be called without arguments
        # by the common configuration setUp.
        parent_setup = functools.partial(
            super(Ml2PluginV2TestCase, self).setUp,
            plugin=PLUGIN_NAME,
            service_plugins=service_plugins,
        )
        self.useFixture(Ml2ConfFixture(parent_setup))
        self.port_create_status = 'DOWN'

    def setUp(self):
        # Enable the test mechanism driver to ensure that
        # we can successfully call through to all mechanism
        # driver apis.
        config.cfg.CONF.set_override('mechanism_drivers',
                                     self._mechanism_drivers,
                                     group='ml2')
        self.physnet = 'physnet1'
        self.vlan_range = '1:100'
        self.vlan_range2 = '200:300'
        self.physnet2 = 'physnet2'
        self.phys_vrange = ':'.join([self.physnet, self.vlan_range])
        self.phys2_vrange = ':'.join([self.physnet2, self.vlan_range2])
        config.cfg.CONF.set_override('network_vlan_ranges',
                                     [self.phys_vrange, self.phys2_vrange],
                                     group='ml2_type_vlan')
        self.setup_parent()
        self.driver = ml2_plugin.Ml2Plugin()
        self.context = context.get_admin_context()

class TestMl2BulkToggleWithoutBulkless(Ml2PluginV2TestCase):

    _mechanism_drivers = ['logger', 'test']

    def test_bulk_enabled_with_bulk_drivers(self):
        self.assertFalse(self._skip_native_bulk)

class TestMl2BasicGet(test_plugin.TestBasicGet,
                      Ml2PluginV2TestCase):
    pass

class TestMl2V2HTTPResponse(test_plugin.TestV2HTTPResponse,
                            Ml2PluginV2TestCase):
    pass

class TestMl2NetworksV2(test_plugin.TestNetworksV2,
                        Ml2PluginV2TestCase):
    def setUp(self, plugin=None):
        super(TestMl2NetworksV2, self).setUp()
        # provider networks
        self.pnets = [{'name': 'net1',
                       pnet.NETWORK_TYPE: 'vlan',
                       pnet.PHYSICAL_NETWORK: 'physnet1',
                       pnet.SEGMENTATION_ID: 1,
                       'tenant_id': 'tenant_one'},
                      {'name': 'net2',
                       pnet.NETWORK_TYPE: 'vlan',
                       pnet.PHYSICAL_NETWORK: 'physnet2',
                       pnet.SEGMENTATION_ID: 210,
                       'tenant_id': 'tenant_one'},
                      {'name': 'net3',
                       pnet.NETWORK_TYPE: 'vlan',
                       pnet.PHYSICAL_NETWORK: 'physnet2',
                       pnet.SEGMENTATION_ID: 220,
                       'tenant_id': 'tenant_one'}
                      ]
        # multiprovider networks
        self.mp_nets = [{'name': 'net4',
                         mpnet.SEGMENTS:
                             [{pnet.NETWORK_TYPE: 'vlan',
                               pnet.PHYSICAL_NETWORK: 'physnet2',
                               pnet.SEGMENTATION_ID: 1},
                              {pnet.NETWORK_TYPE: 'vlan',
                               pnet.PHYSICAL_NETWORK: 'physnet2',
                               pnet.SEGMENTATION_ID: 202}],
                         'tenant_id': 'tenant_one'}
                        ]
        self.nets = self.mp_nets + self.pnets

    def test_port_delete_helper_tolerates_failure(self):
        plugin = manager.NeutronManager.get_plugin()
        with mock.patch.object(plugin, "delete_port",
                               side_effect=exc.PortNotFound(port_id="123")):
            plugin._delete_ports(None, [mock.MagicMock()])

    def test_subnet_delete_helper_tolerates_failure(self):
        plugin = manager.NeutronManager.get_plugin()
        with mock.patch.object(plugin, "delete_subnet",
                               side_effect=exc.SubnetNotFound(subnet_id="1")):
            plugin._delete_subnets(None, [mock.MagicMock()])

    def _create_and_verify_networks(self, networks):
        for net_idx, net in enumerate(networks):
            # create
            req = self.new_create_request('networks',
                                          {'network': net})
            # verify
            network = self.deserialize(self.fmt,
                                       req.get_response(self.api))['network']
            if mpnet.SEGMENTS not in net:
                for k, v in six.iteritems(net):
                    self.assertEqual(net[k], network[k])
                    self.assertNotIn(mpnet.SEGMENTS, network)
            else:
                segments = network[mpnet.SEGMENTS]
                expected_segments = net[mpnet.SEGMENTS]
                self.assertEqual(len(expected_segments), len(segments))
                for expected, actual in zip(expected_segments, segments):
                    self.assertEqual(expected, actual)

    def _lookup_network_by_segmentation_id(self, seg_id, num_expected_nets):
        params_str = "%s=%s" % (pnet.SEGMENTATION_ID, seg_id)
        net_req = self.new_list_request('networks', None,
                                        params=params_str)
        networks = self.deserialize(self.fmt, net_req.get_response(self.api))
        if num_expected_nets:
            self.assertIsNotNone(networks)
            self.assertEqual(num_expected_nets, len(networks['networks']))
        else:
            self.assertIsNone(networks)
        return networks

    def test_list_networks_with_segmentation_id(self):
        self._create_and_verify_networks(self.pnets)
        # verify we can find the network that we expect
        lookup_vlan_id = 1
        expected_net = [n for n in self.pnets
                        if n[pnet.SEGMENTATION_ID] == lookup_vlan_id].pop()
        networks = self._lookup_network_by_segmentation_id(lookup_vlan_id, 1)
        # verify all provider attributes
        network = networks['networks'][0]
        for attr in pnet.ATTRIBUTES:
            self.assertEqual(expected_net[attr], network[attr])

    def test_list_mpnetworks_with_segmentation_id(self):
        self._create_and_verify_networks(self.nets)

        # get all networks with seg_id=1 (including multisegment networks)
        lookup_vlan_id = 1
        networks = self._lookup_network_by_segmentation_id(lookup_vlan_id, 2)

        # get the mpnet
        networks = [n for n in networks['networks'] if mpnet.SEGMENTS in n]
        network = networks.pop()
        # verify attributes of the looked up item
        segments = network[mpnet.SEGMENTS]
        expected_segments = self.mp_nets[0][mpnet.SEGMENTS]
        self.assertEqual(len(expected_segments), len(segments))
        for expected, actual in zip(expected_segments, segments):
            self.assertEqual(expected, actual)

    def test_create_network_segment_allocation_fails(self):
        plugin = manager.NeutronManager.get_plugin()
        with mock.patch.object(plugin.type_manager, 'create_network_segments',
            side_effect=db_exc.RetryRequest(ValueError())) as f:
            self.assertRaises(ValueError,
                              plugin.create_network,
                              context.get_admin_context(),
                              {'network': {'tenant_id': 'sometenant',
                                           'name': 'dummy',
                                           'admin_state_up': True,
                                           'shared': False}})
            self.assertEqual(db_api.MAX_RETRIES + 1, f.call_count)

class TestMl2SubnetsV2(test_plugin.TestSubnetsV2,
                       Ml2PluginV2TestCase):
    def test_delete_subnet_race_with_dhcp_port_creation(self):
        with self.network() as network:
            with self.subnet(network=network) as subnet:
                subnet_id = subnet['subnet']['id']
                attempt = [0]

                def check_and_create_ports(context, subnet_id):
                    """A method to emulate race condition.

                    Adds dhcp port in the middle of subnet delete
                    """
                    if attempt[0] > 0:
                        return False
                    attempt[0] += 1
                    data = {'port': {'network_id': network['network']['id'],
                                     'tenant_id':
                                     network['network']['tenant_id'],
                                     'name': 'port1',
                                     'admin_state_up': 1,
                                     'device_owner':
                                     constants.DEVICE_OWNER_DHCP,
                                     'fixed_ips': [{'subnet_id': subnet_id}]}}
                    port_req = self.new_create_request('ports', data)
                    port_res = port_req.get_response(self.api)
                    self.assertEqual(201, port_res.status_int)
                    return (context.session.query(models_v2.IPAllocation).
                            filter_by(subnet_id=subnet_id).
                            join(models_v2.Port).first())

                plugin = manager.NeutronManager.get_plugin()
                # we mock _subnet_check_ip_allocations with method
                # that creates DHCP port 'in the middle' of subnet_delete
                # causing retry this way subnet is deleted on the
                # second attempt
                with mock.patch.object(plugin, '_subnet_check_ip_allocations',
                                       side_effect=check_and_create_ports):
                    req = self.new_delete_request('subnets', subnet_id)
                    res = req.get_response(self.api)
                    self.assertEqual(204, res.status_int)

class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):

    def test_update_port_status_build(self):
        with self.port() as port:
            self.assertEqual('DOWN', port['port']['status'])
            self.assertEqual('DOWN', self.port_create_status)

    def test_update_port_status_short_id(self):
        ctx = context.get_admin_context()
        plugin = manager.NeutronManager.get_plugin()
        with self.port() as port:
            with mock.patch.object(ml2_db, 'get_binding_levels',
                                   return_value=[]) as mock_gbl:
                port_id = port['port']['id']
                short_id = port_id[:11]
                plugin.update_port_status(ctx, short_id, 'UP')
                mock_gbl.assert_called_once_with(mock.ANY, port_id, mock.ANY)

    def test_update_port_mac(self):
        self.check_update_port_mac(
            host_arg={portbindings.HOST_ID: HOST},
            arg_list=(portbindings.HOST_ID,))

    def test_update_non_existent_port(self):
        ctx = context.get_admin_context()
        plugin = manager.NeutronManager.get_plugin()
        data = {'port': {'admin_state_up': False}}
        self.assertRaises(exc.PortNotFound, plugin.update_port, ctx,
                          'invalid-uuid', data)

    def test_delete_non_existent_port(self):
        ctx = context.get_admin_context()
        plugin = manager.NeutronManager.get_plugin()
        with mock.patch.object(ml2_plugin.LOG, 'debug') as log_debug:
            plugin.delete_port(ctx, 'invalid-uuid', l3_port_check=False)
            log_debug.assert_has_calls([
                mock.call(_("Deleting port %s"), 'invalid-uuid'),
                mock.call(_("The port '%s' was deleted"), 'invalid-uuid')
            ])

    def test_l3_cleanup_on_net_delete(self):
        l3plugin = manager.NeutronManager.get_service_plugins().get(
            service_constants.L3_ROUTER_NAT)
        kwargs = {'arg_list': (external_net.EXTERNAL,),
                  external_net.EXTERNAL: True}
        with self.network(**kwargs) as n:
            with self.subnet(network=n, cidr='200.0.0.0/22'):
                l3plugin.create_floatingip(
                    context.get_admin_context(),
                    {'floatingip': {'floating_network_id': n['network']['id'],
                                    'tenant_id': n['network']['tenant_id']}}
                )
        self._delete('networks', n['network']['id'])
        flips = l3plugin.get_floatingips(context.get_admin_context())
        self.assertFalse(flips)

    def test_create_ports_bulk_port_binding_failure(self):
        ctx = context.get_admin_context()
        with self.network() as net:
            plugin = manager.NeutronManager.get_plugin()

            with mock.patch.object(plugin, '_bind_port_if_needed',
                side_effect=ml2_exc.MechanismDriverError(
                    method='create_port_bulk')) as _bind_port_if_needed:

                res = self._create_port_bulk(self.fmt, 2, net['network']['id'],
                                             'test', True, context=ctx)

                self.assertTrue(_bind_port_if_needed.called)
                # We expect a 500 as we injected a fault in the plugin
                self._validate_behavior_on_bulk_failure(
                    res, 'ports', webob.exc.HTTPServerError.code)

    def test_create_ports_bulk_with_sec_grp(self):
        ctx = context.get_admin_context()
        plugin = manager.NeutronManager.get_plugin()
        with contextlib.nested(
            self.network(),
            mock.patch.object(plugin.notifier,
                              'security_groups_member_updated'),
            mock.patch.object(plugin.notifier,
                              'security_groups_provider_updated')
        ) as (net, m_upd, p_upd):

            res = self._create_port_bulk(self.fmt, 3, net['network']['id'],
                                         'test', True, context=ctx)
            ports = self.deserialize(self.fmt, res)
            used_sg = ports['ports'][0]['security_groups']
            m_upd.assert_called_once_with(ctx, used_sg)
            self.assertFalse(p_upd.called)

    def test_create_ports_bulk_with_sec_grp_member_provider_update(self):
        ctx = context.get_admin_context()
        plugin = manager.NeutronManager.get_plugin()
        with contextlib.nested(
            self.network(),
            mock.patch.object(plugin.notifier,
                              'security_groups_member_updated'),
            mock.patch.object(plugin.notifier,
                              'security_groups_provider_updated')
        ) as (net, m_upd, p_upd):

            net_id = net['network']['id']
            data = [{
                    'network_id': net_id,
                    'tenant_id': self._tenant_id
                    },
                    {
                    'network_id': net_id,
                    'tenant_id': self._tenant_id,
                    'device_owner': constants.DEVICE_OWNER_DHCP
                    }
                    ]

            res = self._create_bulk_from_list(self.fmt, 'port',
                                              data, context=ctx)
            ports = self.deserialize(self.fmt, res)
            used_sg = ports['ports'][0]['security_groups']
            m_upd.assert_called_once_with(ctx, used_sg)
            p_upd.assert_called_once_with(ctx)

            m_upd.reset_mock()
            p_upd.reset_mock()
            data[0]['device_owner'] = constants.DEVICE_OWNER_DHCP
            self._create_bulk_from_list(self.fmt, 'port',
                                        data, context=ctx)
            self.assertFalse(m_upd.called)
            p_upd.assert_called_once_with(ctx)

    def test_create_ports_bulk_with_sec_grp_provider_update_ipv6(self):
        ctx = context.get_admin_context()
        plugin = manager.NeutronManager.get_plugin()
        fake_prefix = '2001:db8::/64'
        fake_gateway = 'fe80::1'
        with self.network() as net:
            with contextlib.nested(
                self.subnet(net, gateway_ip=fake_gateway,
                            cidr=fake_prefix, ip_version=6),
                mock.patch.object(
                    plugin.notifier, 'security_groups_member_updated'),
                mock.patch.object(
                    plugin.notifier, 'security_groups_provider_updated')
            ) as (snet_v6, m_upd, p_upd):

                net_id = net['network']['id']
                data = [{
                        'network_id': net_id,
                        'tenant_id': self._tenant_id,
                        'fixed_ips': [{'subnet_id': snet_v6['subnet']['id']}],
                        'device_owner': constants.DEVICE_OWNER_ROUTER_INTF
                        }
                        ]
                self._create_bulk_from_list(self.fmt, 'port',
                                            data, context=ctx)
                self.assertFalse(m_upd.called)
                p_upd.assert_called_once_with(ctx)

    def test_delete_port_no_notify_in_disassociate_floatingips(self):
        ctx = context.get_admin_context()
        plugin = manager.NeutronManager.get_plugin()
        l3plugin = manager.NeutronManager.get_service_plugins().get(
            service_constants.L3_ROUTER_NAT)
        with contextlib.nested(
            self.port(),
            mock.patch.object(l3plugin, 'disassociate_floatingips'),
            mock.patch.object(registry, 'notify')
        ) as (port, disassociate_floatingips, notify):

            port_id = port['port']['id']
            plugin.delete_port(ctx, port_id)

            # check that no notification was requested while under
            # transaction
            disassociate_floatingips.assert_has_calls([
                mock.call(ctx, port_id, do_notify=False)
            ])

            # check that notifier was still triggered
            self.assertTrue(notify.call_counts)

    def test_check_if_compute_port_serviced_by_dvr(self):
        self.assertTrue(utils.is_dvr_serviced('compute:None'))

    def test_check_if_lbaas_vip_port_serviced_by_dvr(self):
        self.assertTrue(utils.is_dvr_serviced(
            constants.DEVICE_OWNER_LOADBALANCER))

    def test_check_if_dhcp_port_serviced_by_dvr(self):
        self.assertTrue(utils.is_dvr_serviced(constants.DEVICE_OWNER_DHCP))

    def test_check_if_port_not_serviced_by_dvr(self):
        self.assertFalse(utils.is_dvr_serviced(
            constants.DEVICE_OWNER_ROUTER_INTF))

    def test_disassociate_floatingips_do_notify_returns_nothing(self):
        ctx = context.get_admin_context()
        l3plugin = manager.NeutronManager.get_service_plugins().get(
            service_constants.L3_ROUTER_NAT)
        with self.port() as port:

            port_id = port['port']['id']
            # check that nothing is returned when notifications are handled
            # by the called method
            self.assertIsNone(l3plugin.disassociate_floatingips(ctx, port_id))

class TestMl2PluginOnly(Ml2PluginV2TestCase):
    """For testing methods that don't call drivers"""

    def _test_check_mac_update_allowed(self, vif_type, expect_change=True):
        plugin = manager.NeutronManager.get_plugin()
        port = {'mac_address': "fake_mac", 'id': "fake_id"}
        if expect_change:
            new_attrs = {"mac_address": "dummy_mac"}
        else:
            new_attrs = {"mac_address": port['mac_address']}
        binding = mock.Mock()
        binding.vif_type = vif_type
        mac_changed = plugin._check_mac_update_allowed(port, new_attrs,
                                                       binding)
        self.assertEqual(expect_change, mac_changed)

    def test_check_mac_update_allowed_if_no_mac_change(self):
        self._test_check_mac_update_allowed(portbindings.VIF_TYPE_UNBOUND,
                                            expect_change=False)

    def test_check_mac_update_allowed_unless_bound(self):
        with testtools.ExpectedException(exc.PortBound):
            self._test_check_mac_update_allowed(portbindings.VIF_TYPE_OVS)

class TestMl2DvrPortsV2(TestMl2PortsV2):
    def setUp(self):
        super(TestMl2DvrPortsV2, self).setUp()
        extensions = ['router',
                      constants.L3_AGENT_SCHEDULER_EXT_ALIAS,
                      constants.L3_DISTRIBUTED_EXT_ALIAS]
        self.plugin = manager.NeutronManager.get_plugin()
        self.l3plugin = mock.Mock()
        type(self.l3plugin).supported_extension_aliases = (
            mock.PropertyMock(return_value=extensions))
        self.service_plugins = {'L3_ROUTER_NAT': self.l3plugin}

    def _test_delete_dvr_serviced_port(self, device_owner, floating_ip=False):
        ns_to_delete = {'host': 'myhost', 'agent_id': 'vm_l3_agent',
                        'router_id': 'my_router'}
        fip_set = set()
        if floating_ip:
            fip_set.add(ns_to_delete['router_id'])

        with contextlib.nested(
            mock.patch.object(manager.NeutronManager,
                              'get_service_plugins',
                              return_value=self.service_plugins),
            self.port(device_owner=device_owner),
            mock.patch.object(registry, 'notify'),
            mock.patch.object(self.l3plugin, 'disassociate_floatingips',
                              return_value=fip_set),
            mock.patch.object(self.l3plugin, 'dvr_deletens_if_no_port',
                              return_value=[ns_to_delete]),
        ) as (get_service_plugin, port, notify, disassociate_floatingips,
              dvr_delns_ifno_port):

            port_id = port['port']['id']
            self.plugin.delete_port(self.context, port_id)

            self.assertTrue(notify.call_count)
            dvr_delns_ifno_port.assert_called_once_with(self.context,
                                                        port['port']['id'])

    def test_delete_last_vm_port(self):
        self._test_delete_dvr_serviced_port(device_owner='compute:None')

    def test_delete_last_vm_port_with_floatingip(self):
        self._test_delete_dvr_serviced_port(device_owner='compute:None',
                                            floating_ip=True)

    def test_delete_lbaas_vip_port(self):
        self._test_delete_dvr_serviced_port(
            device_owner=constants.DEVICE_OWNER_LOADBALANCER)

    def test_concurrent_csnat_port_delete(self):
        plugin = manager.NeutronManager.get_service_plugins()[
            service_constants.L3_ROUTER_NAT]
        r = plugin.create_router(
            self.context,
            {'router': {'name': 'router', 'admin_state_up': True}})
        with self.subnet() as s:
            p = plugin.add_router_interface(self.context, r['id'],
                                            {'subnet_id': s['subnet']['id']})

        # lie to turn the port into an SNAT interface
        with self.context.session.begin():
            rp = self.context.session.query(l3_db.RouterPort).filter_by(
                port_id=p['port_id']).first()
            rp.port_type = constants.DEVICE_OWNER_ROUTER_SNAT

        # take the port away before csnat gets a chance to delete it
        # to simulate a concurrent delete
        orig_get_ports = plugin._core_plugin.get_ports

        def get_ports_with_delete_first(*args, **kwargs):
            plugin._core_plugin.delete_port(self.context,
                                            p['port_id'],
                                            l3_port_check=False)
            return orig_get_ports(*args, **kwargs)
        plugin._core_plugin.get_ports = get_ports_with_delete_first

        # This should be able to handle a concurrent delete without raising
        # an exception
        router = plugin._get_router(self.context, r['id'])
        plugin.delete_csnat_router_interface_ports(self.context, router)

class TestMl2PortBinding(Ml2PluginV2TestCase,
                         test_bindings.PortBindingsTestCase):
    # Test case does not set binding:host_id, so ml2 does not attempt
    # to bind port
    VIF_TYPE = portbindings.VIF_TYPE_UNBOUND
    HAS_PORT_FILTER = False
    ENABLE_SG = True
    FIREWALL_DRIVER = test_sg_rpc.FIREWALL_HYBRID_DRIVER

    def setUp(self, firewall_driver=None):
        test_sg_rpc.set_firewall_driver(self.FIREWALL_DRIVER)
        config.cfg.CONF.set_override(
            'enable_security_group', self.ENABLE_SG,
            group='SECURITYGROUP')
        super(TestMl2PortBinding, self).setUp()

    def _check_port_binding_profile(self, port, profile=None):
        self.assertIn('id', port)
        self.assertIn(portbindings.PROFILE, port)
        value = port[portbindings.PROFILE]
        self.assertEqual(profile or {}, value)

    def test_create_port_binding_profile(self):
        self._test_create_port_binding_profile({'a': 1, 'b': 2})

    def test_update_port_binding_profile(self):
        self._test_update_port_binding_profile({'c': 3})

    def test_create_port_binding_profile_too_big(self):
        s = 'x' * 5000
        profile_arg = {portbindings.PROFILE: {'d': s}}
        try:
            with self.port(expected_res_status=400,
                           arg_list=(portbindings.PROFILE,),
                           **profile_arg):
                pass
        except webob.exc.HTTPClientError:
            pass

    def test_remove_port_binding_profile(self):
        profile = {'e': 5}
        profile_arg = {portbindings.PROFILE: profile}
        with self.port(arg_list=(portbindings.PROFILE,),
                       **profile_arg) as port:
            self._check_port_binding_profile(port['port'], profile)
            port_id = port['port']['id']
            profile_arg = {portbindings.PROFILE: None}
            port = self._update('ports', port_id,
                                {'port': profile_arg})['port']
            self._check_port_binding_profile(port)
            port = self._show('ports', port_id)['port']
            self._check_port_binding_profile(port)

    def test_return_on_concurrent_delete_and_binding(self):
        # create a port and delete it so we have an expired mechanism context
        with self.port() as port:
            plugin = manager.NeutronManager.get_plugin()
            binding = ml2_db.get_locked_port_and_binding(self.context.session,
                                                         port['port']['id'])[1]
            binding['host'] = 'test'
            mech_context = driver_context.PortContext(
                plugin, self.context, port['port'],
                plugin.get_network(self.context, port['port']['network_id']),
                binding, None)
        with contextlib.nested(
            mock.patch('neutron.plugins.ml2.plugin.'
                       'db.get_locked_port_and_binding',
                       return_value=(None, None)),
            mock.patch('neutron.plugins.ml2.plugin.Ml2Plugin._make_port_dict')
        ) as (glpab_mock, mpd_mock):
            plugin._bind_port_if_needed(mech_context)
            # called during deletion to get port
            self.assertTrue(glpab_mock.mock_calls)
            # should have returned before calling _make_port_dict
            self.assertFalse(mpd_mock.mock_calls)

    def test_port_binding_profile_not_changed(self):
        profile = {'e': 5}
        profile_arg = {portbindings.PROFILE: profile}
        with self.port(arg_list=(portbindings.PROFILE,),
                       **profile_arg) as port:
            self._check_port_binding_profile(port['port'], profile)
            port_id = port['port']['id']
            state_arg = {'admin_state_up': True}
            port = self._update('ports', port_id,
                                {'port': state_arg})['port']
            self._check_port_binding_profile(port, profile)
            port = self._show('ports', port_id)['port']
            self._check_port_binding_profile(port, profile)

    def test_process_dvr_port_binding_update_router_id(self):
        host_id = 'host'
        binding = models.DVRPortBinding(
                            port_id='port_id',
                            host=host_id,
                            router_id='old_router_id',
                            vif_type=portbindings.VIF_TYPE_OVS,
                            vnic_type=portbindings.VNIC_NORMAL,
                            status=constants.PORT_STATUS_DOWN)
        plugin = manager.NeutronManager.get_plugin()
        mock_network = {'id': 'net_id'}
        mock_port = {'id': 'port_id'}
        context = mock.Mock()
        new_router_id = 'new_router'
        attrs = {'device_id': new_router_id, portbindings.HOST_ID: host_id}
        with mock.patch.object(plugin, '_update_port_dict_binding'):
            with mock.patch.object(ml2_db, 'get_network_segments',
                                   return_value=[]):
                mech_context = driver_context.PortContext(
                    self, context, mock_port, mock_network, binding, None)
                plugin._process_dvr_port_binding(mech_context, context, attrs)
                self.assertEqual(new_router_id,
                                 mech_context._binding.router_id)
                self.assertEqual(host_id, mech_context._binding.host)

    def test_update_dvr_port_binding_on_non_existent_port(self):
        plugin = manager.NeutronManager.get_plugin()
        port = {
            'id': 'foo_port_id',
            'binding:host_id': 'foo_host',
        }
        with mock.patch.object(ml2_db, 'ensure_dvr_port_binding') as mock_dvr:
            plugin.update_dvr_port_binding(
                self.context, 'foo_port_id', {'port': port})
        self.assertFalse(mock_dvr.called)

class TestMl2PortBindingNoSG(TestMl2PortBinding):
    HAS_PORT_FILTER = False
    ENABLE_SG = False
    FIREWALL_DRIVER = test_sg_rpc.FIREWALL_NOOP_DRIVER

class TestMl2PortBindingHost(Ml2PluginV2TestCase,
                             test_bindings.PortBindingsHostTestCaseMixin):
    pass

class TestMl2PortBindingVnicType(Ml2PluginV2TestCase,
                                 test_bindings.PortBindingsVnicTestCaseMixin):
    pass

class TestMultiSegmentNetworks(Ml2PluginV2TestCase):

    def setUp(self, plugin=None):
        super(TestMultiSegmentNetworks, self).setUp()

    def test_allocate_dynamic_segment(self):
        data = {'network': {'name': 'net1',
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        network = self.deserialize(self.fmt,
                                   network_req.get_response(self.api))
        segment = {driver_api.NETWORK_TYPE: 'vlan',
                   driver_api.PHYSICAL_NETWORK: 'physnet1'}
        network_id = network['network']['id']
        self.driver.type_manager.allocate_dynamic_segment(
            self.context.session, network_id, segment)
        dynamic_segment = ml2_db.get_dynamic_segment(self.context.session,
                                                     network_id,
                                                     'physnet1')
        self.assertEqual('vlan', dynamic_segment[driver_api.NETWORK_TYPE])
        self.assertEqual('physnet1',
                         dynamic_segment[driver_api.PHYSICAL_NETWORK])
        self.assertTrue(dynamic_segment[driver_api.SEGMENTATION_ID] > 0)
        segment2 = {driver_api.NETWORK_TYPE: 'vlan',
                    driver_api.SEGMENTATION_ID: 1234,
                    driver_api.PHYSICAL_NETWORK: 'physnet3'}
        self.driver.type_manager.allocate_dynamic_segment(
            self.context.session, network_id, segment2)
        dynamic_segment = ml2_db.get_dynamic_segment(self.context.session,
                                                     network_id,
                                                     segmentation_id='1234')
        self.assertEqual('vlan', dynamic_segment[driver_api.NETWORK_TYPE])
        self.assertEqual('physnet3',
                         dynamic_segment[driver_api.PHYSICAL_NETWORK])
        self.assertEqual(dynamic_segment[driver_api.SEGMENTATION_ID], 1234)

    def test_allocate_dynamic_segment_multiple_physnets(self):
        data = {'network': {'name': 'net1',
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        network = self.deserialize(self.fmt,
                                   network_req.get_response(self.api))
        segment = {driver_api.NETWORK_TYPE: 'vlan',
                   driver_api.PHYSICAL_NETWORK: 'physnet1'}
        network_id = network['network']['id']
        self.driver.type_manager.allocate_dynamic_segment(
            self.context.session, network_id, segment)
        dynamic_segment = ml2_db.get_dynamic_segment(self.context.session,
                                                     network_id,
                                                     'physnet1')
        self.assertEqual('vlan', dynamic_segment[driver_api.NETWORK_TYPE])
        self.assertEqual('physnet1',
                         dynamic_segment[driver_api.PHYSICAL_NETWORK])
        dynamic_segmentation_id = dynamic_segment[driver_api.SEGMENTATION_ID]
        self.assertTrue(dynamic_segmentation_id > 0)
        dynamic_segment1 = ml2_db.get_dynamic_segment(self.context.session,
                                                      network_id,
                                                      'physnet1')
        dynamic_segment1_id = dynamic_segment1[driver_api.SEGMENTATION_ID]
        self.assertEqual(dynamic_segmentation_id, dynamic_segment1_id)
        segment2 = {driver_api.NETWORK_TYPE: 'vlan',
                    driver_api.PHYSICAL_NETWORK: 'physnet2'}
        self.driver.type_manager.allocate_dynamic_segment(
            self.context.session, network_id, segment2)
        dynamic_segment2 = ml2_db.get_dynamic_segment(self.context.session,
                                                      network_id,
                                                      'physnet2')
        dynamic_segmentation2_id = dynamic_segment2[driver_api.SEGMENTATION_ID]
        self.assertNotEqual(dynamic_segmentation_id, dynamic_segmentation2_id)

    def test_allocate_release_dynamic_segment(self):
        data = {'network': {'name': 'net1',
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        network = self.deserialize(self.fmt,
                                   network_req.get_response(self.api))
        segment = {driver_api.NETWORK_TYPE: 'vlan',
                   driver_api.PHYSICAL_NETWORK: 'physnet1'}
        network_id = network['network']['id']
        self.driver.type_manager.allocate_dynamic_segment(
            self.context.session, network_id, segment)
        dynamic_segment = ml2_db.get_dynamic_segment(self.context.session,
                                                     network_id,
                                                     'physnet1')
        self.assertEqual('vlan', dynamic_segment[driver_api.NETWORK_TYPE])
        self.assertEqual('physnet1',
                         dynamic_segment[driver_api.PHYSICAL_NETWORK])
        dynamic_segmentation_id = dynamic_segment[driver_api.SEGMENTATION_ID]
        self.assertTrue(dynamic_segmentation_id > 0)
        self.driver.type_manager.release_dynamic_segment(
            self.context.session, dynamic_segment[driver_api.ID])
        self.assertIsNone(ml2_db.get_dynamic_segment(
            self.context.session, network_id, 'physnet1'))

    def test_create_network_provider(self):
        data = {'network': {'name': 'net1',
                            pnet.NETWORK_TYPE: 'vlan',
                            pnet.PHYSICAL_NETWORK: 'physnet1',
                            pnet.SEGMENTATION_ID: 1,
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        network = self.deserialize(self.fmt,
                                   network_req.get_response(self.api))
        self.assertEqual('vlan', network['network'][pnet.NETWORK_TYPE])
        self.assertEqual('physnet1', network['network'][pnet.PHYSICAL_NETWORK])
        self.assertEqual(1, network['network'][pnet.SEGMENTATION_ID])
        self.assertNotIn(mpnet.SEGMENTS, network['network'])

    def test_create_network_single_multiprovider(self):
        data = {'network': {'name': 'net1',
                            mpnet.SEGMENTS:
                            [{pnet.NETWORK_TYPE: 'vlan',
                              pnet.PHYSICAL_NETWORK: 'physnet1',
                              pnet.SEGMENTATION_ID: 1}],
                            'tenant_id': 'tenant_one'}}
        net_req = self.new_create_request('networks', data)
        network = self.deserialize(self.fmt, net_req.get_response(self.api))
        self.assertEqual('vlan', network['network'][pnet.NETWORK_TYPE])
        self.assertEqual('physnet1', network['network'][pnet.PHYSICAL_NETWORK])
        self.assertEqual(1, network['network'][pnet.SEGMENTATION_ID])
        self.assertNotIn(mpnet.SEGMENTS, network['network'])

        # Tests get_network()
        net_req = self.new_show_request('networks', network['network']['id'])
        network = self.deserialize(self.fmt, net_req.get_response(self.api))
        self.assertEqual('vlan', network['network'][pnet.NETWORK_TYPE])
        self.assertEqual('physnet1', network['network'][pnet.PHYSICAL_NETWORK])
        self.assertEqual(1, network['network'][pnet.SEGMENTATION_ID])
        self.assertNotIn(mpnet.SEGMENTS, network['network'])

    def test_create_network_multiprovider(self):
        data = {'network': {'name': 'net1',
                            mpnet.SEGMENTS:
                            [{pnet.NETWORK_TYPE: 'vlan',
                              pnet.PHYSICAL_NETWORK: 'physnet1',
                              pnet.SEGMENTATION_ID: 1},
                             {pnet.NETWORK_TYPE: 'vlan',
                              pnet.PHYSICAL_NETWORK: 'physnet1',
                              pnet.SEGMENTATION_ID: 2}],
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        network = self.deserialize(self.fmt,
                                   network_req.get_response(self.api))
        segments = network['network'][mpnet.SEGMENTS]
        for segment_index, segment in enumerate(data['network']
                                                [mpnet.SEGMENTS]):
            for field in [pnet.NETWORK_TYPE, pnet.PHYSICAL_NETWORK,
                          pnet.SEGMENTATION_ID]:
                self.assertEqual(segment.get(field),
                            segments[segment_index][field])

        # Tests get_network()
        net_req = self.new_show_request('networks', network['network']['id'])
        network = self.deserialize(self.fmt, net_req.get_response(self.api))
        segments = network['network'][mpnet.SEGMENTS]
        for segment_index, segment in enumerate(data['network']
                                                [mpnet.SEGMENTS]):
            for field in [pnet.NETWORK_TYPE, pnet.PHYSICAL_NETWORK,
                          pnet.SEGMENTATION_ID]:
                self.assertEqual(segment.get(field),
                            segments[segment_index][field])

    def test_create_network_with_provider_and_multiprovider_fail(self):
        data = {'network': {'name': 'net1',
                            mpnet.SEGMENTS:
                            [{pnet.NETWORK_TYPE: 'vlan',
                              pnet.PHYSICAL_NETWORK: 'physnet1',
                              pnet.SEGMENTATION_ID: 1}],
                            pnet.NETWORK_TYPE: 'vlan',
                            pnet.PHYSICAL_NETWORK: 'physnet1',
                            pnet.SEGMENTATION_ID: 1,
                            'tenant_id': 'tenant_one'}}

        network_req = self.new_create_request('networks', data)
        res = network_req.get_response(self.api)
        self.assertEqual(400, res.status_int)

    def test_create_network_duplicate_full_segments(self):
        data = {'network': {'name': 'net1',
                            mpnet.SEGMENTS:
                            [{pnet.NETWORK_TYPE: 'vlan',
                              pnet.PHYSICAL_NETWORK: 'physnet1',
                              pnet.SEGMENTATION_ID: 1},
                             {pnet.NETWORK_TYPE: 'vlan',
                              pnet.PHYSICAL_NETWORK: 'physnet1',
                              pnet.SEGMENTATION_ID: 1}],
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        res = network_req.get_response(self.api)
        self.assertEqual(400, res.status_int)

    def test_create_network_duplicate_partial_segments(self):
        data = {'network': {'name': 'net1',
                            mpnet.SEGMENTS:
                            [{pnet.NETWORK_TYPE: 'vlan',
                              pnet.PHYSICAL_NETWORK: 'physnet1'},
                             {pnet.NETWORK_TYPE: 'vlan',
                              pnet.PHYSICAL_NETWORK: 'physnet1'}],
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        res = network_req.get_response(self.api)
        self.assertEqual(201, res.status_int)

    def test_release_network_segments(self):
        data = {'network': {'name': 'net1',
                            'admin_state_up': True,
                            'shared': False,
                            pnet.NETWORK_TYPE: 'vlan',
                            pnet.PHYSICAL_NETWORK: 'physnet1',
                            pnet.SEGMENTATION_ID: 1,
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        res = network_req.get_response(self.api)
        network = self.deserialize(self.fmt, res)
        network_id = network['network']['id']
        segment = {driver_api.NETWORK_TYPE: 'vlan',
                   driver_api.PHYSICAL_NETWORK: 'physnet2'}
        self.driver.type_manager.allocate_dynamic_segment(
            self.context.session, network_id, segment)
        dynamic_segment = ml2_db.get_dynamic_segment(self.context.session,
                                                     network_id,
                                                     'physnet2')
        self.assertEqual('vlan', dynamic_segment[driver_api.NETWORK_TYPE])
        self.assertEqual('physnet2',
                         dynamic_segment[driver_api.PHYSICAL_NETWORK])
        self.assertTrue(dynamic_segment[driver_api.SEGMENTATION_ID] > 0)

        with mock.patch.object(type_vlan.VlanTypeDriver,
                               'release_segment') as rs:
            req = self.new_delete_request('networks', network_id)
            res = req.get_response(self.api)
            self.assertEqual(2, rs.call_count)
        self.assertEqual(ml2_db.get_network_segments(
            self.context.session, network_id), [])
        self.assertIsNone(ml2_db.get_dynamic_segment(
            self.context.session, network_id, 'physnet2'))

    def test_release_segment_no_type_driver(self):
        data = {'network': {'name': 'net1',
                            'admin_state_up': True,
                            'shared': False,
                            pnet.NETWORK_TYPE: 'vlan',
                            pnet.PHYSICAL_NETWORK: 'physnet1',
                            pnet.SEGMENTATION_ID: 1,
                            'tenant_id': 'tenant_one'}}
        network_req = self.new_create_request('networks', data)
        res = network_req.get_response(self.api)
        network = self.deserialize(self.fmt, res)
        network_id = network['network']['id']

        segment = {driver_api.NETWORK_TYPE: 'faketype',
                   driver_api.PHYSICAL_NETWORK: 'physnet1',
                   driver_api.ID: 1}
        with mock.patch('neutron.plugins.ml2.managers.LOG') as log:
            with mock.patch('neutron.plugins.ml2.managers.db') as db:
                db.get_network_segments.return_value = (segment,)
                self.driver.type_manager.release_network_segments(
                    self.context.session, network_id)

                log.error.assert_called_once_with(
                    "Failed to release segment '%s' because "
                    "network type is not supported.", segment)

    def test_create_provider_fail(self):
        segment = {pnet.NETWORK_TYPE: None,
                   pnet.PHYSICAL_NETWORK: 'phys_net',
                   pnet.SEGMENTATION_ID: None}
        with testtools.ExpectedException(exc.InvalidInput):
            self.driver.type_manager._process_provider_create(segment)

    def test_create_network_plugin(self):
        data = {'network': {'name': 'net1',
                            'admin_state_up': True,
                            'shared': False,
                            pnet.NETWORK_TYPE: 'vlan',
                            pnet.PHYSICAL_NETWORK: 'physnet1',
                            pnet.SEGMENTATION_ID: 1,
                            'tenant_id': 'tenant_one'}}

        def raise_mechanism_exc(*args, **kwargs):
            raise ml2_exc.MechanismDriverError(
                method='create_network_postcommit')

        with mock.patch('neutron.plugins.ml2.managers.MechanismManager.'
                        'create_network_precommit', new=raise_mechanism_exc):
            with testtools.ExpectedException(ml2_exc.MechanismDriverError):
                self.driver.create_network(self.context, data)

    def test_extend_dictionary_no_segments(self):
        network = dict(name='net_no_segment', id='5', tenant_id='tenant_one')
        self.driver.type_manager.extend_network_dict_provider(self.context,
                                                              network)
        self.assertIsNone(network[pnet.NETWORK_TYPE])
        self.assertIsNone(network[pnet.PHYSICAL_NETWORK])
        self.assertIsNone(network[pnet.SEGMENTATION_ID])

class TestMl2AllowedAddressPairs(Ml2PluginV2TestCase,
                                 test_pair.TestAllowedAddressPairs):
    def setUp(self, plugin=None):
        super(test_pair.TestAllowedAddressPairs, self).setUp(
            plugin=PLUGIN_NAME)

class DHCPOptsTestCase(test_dhcpopts.TestExtraDhcpOpt):

    def setUp(self, plugin=None):
        super(test_dhcpopts.ExtraDhcpOptDBTestCase, self).setUp(
            plugin=PLUGIN_NAME)

class Ml2PluginV2FaultyDriverTestCase(test_plugin.NeutronDbPluginV2TestCase):

    def setUp(self):
        # Enable the test mechanism driver to ensure that
        # we can successfully call through to all mechanism
        # driver apis.
        config.cfg.CONF.set_override('mechanism_drivers',
                                     ['test', 'logger'],
                                     group='ml2')
        super(Ml2PluginV2FaultyDriverTestCase, self).setUp(PLUGIN_NAME)
        self.port_create_status = 'DOWN'

class TestFaultyMechansimDriver(Ml2PluginV2FaultyDriverTestCase):

    def test_create_network_faulty(self):

        with mock.patch.object(mech_test.TestMechanismDriver,
                               'create_network_postcommit',
                               side_effect=ml2_exc.MechanismDriverError):
            tenant_id = str(uuid.uuid4())
            data = {'network': {'name': 'net1',
                                'tenant_id': tenant_id}}
            req = self.new_create_request('networks', data)
            res = req.get_response(self.api)
            self.assertEqual(500, res.status_int)
            error = self.deserialize(self.fmt, res)
            self.assertEqual('MechanismDriverError',
                             error['NeutronError']['type'])
            query_params = "tenant_id=%s" % tenant_id
            nets = self._list('networks', query_params=query_params)
            self.assertFalse(nets['networks'])

    def test_delete_network_faulty(self):

        with mock.patch.object(mech_test.TestMechanismDriver,
                               'delete_network_postcommit',
                               side_effect=ml2_exc.MechanismDriverError):
            with mock.patch.object(mech_logger.LoggerMechanismDriver,
                                   'delete_network_postcommit') as dnp:

                data = {'network': {'name': 'net1',
                                    'tenant_id': 'tenant_one'}}
                network_req = self.new_create_request('networks', data)
                network_res = network_req.get_response(self.api)
                self.assertEqual(201, network_res.status_int)
                network = self.deserialize(self.fmt, network_res)
                net_id = network['network']['id']
                req = self.new_delete_request('networks', net_id)
                res = req.get_response(self.api)
                self.assertEqual(204, res.status_int)
                # Test if other mechanism driver was called
                self.assertTrue(dnp.called)
                self._show('networks', net_id,
                           expected_code=webob.exc.HTTPNotFound.code)

    def test_update_network_faulty(self):

        with mock.patch.object(mech_test.TestMechanismDriver,
                               'update_network_postcommit',
                               side_effect=ml2_exc.MechanismDriverError):
            with mock.patch.object(mech_logger.LoggerMechanismDriver,
                                   'update_network_postcommit') as unp:

                data = {'network': {'name': 'net1',
                                    'tenant_id': 'tenant_one'}}
                network_req = self.new_create_request('networks', data)
                network_res = network_req.get_response(self.api)
                self.assertEqual(201, network_res.status_int)
                network = self.deserialize(self.fmt, network_res)
                net_id = network['network']['id']

                new_name = 'a_brand_new_name'
                data = {'network': {'name': new_name}}
                req = self.new_update_request('networks', data, net_id)
                res = req.get_response(self.api)
                self.assertEqual(500, res.status_int)
                error = self.deserialize(self.fmt, res)
                self.assertEqual('MechanismDriverError',
                                 error['NeutronError']['type'])
                # Test if other mechanism driver was called
                self.assertTrue(unp.called)
                net = self._show('networks', net_id)
                self.assertEqual(new_name, net['network']['name'])

                self._delete('networks', net_id)

    def test_create_subnet_faulty(self):

        with mock.patch.object(mech_test.TestMechanismDriver,
                               'create_subnet_postcommit',
                               side_effect=ml2_exc.MechanismDriverError):

            with self.network() as network:
                net_id = network['network']['id']
                data = {'subnet': {'network_id': net_id,
                                   'cidr': '10.0.20.0/24',
                                   'ip_version': '4',
                                   'name': 'subnet1',
                                   'tenant_id':
                                   network['network']['tenant_id'],
                                   'gateway_ip': '10.0.20.1'}}
                req = self.new_create_request('subnets', data)
                res = req.get_response(self.api)
                self.assertEqual(500, res.status_int)
                error = self.deserialize(self.fmt, res)
                self.assertEqual('MechanismDriverError',
                                 error['NeutronError']['type'])
                query_params = "network_id=%s" % net_id
                subnets = self._list('subnets', query_params=query_params)
                self.assertFalse(subnets['subnets'])

    def test_delete_subnet_faulty(self):

        with mock.patch.object(mech_test.TestMechanismDriver,
                               'delete_subnet_postcommit',
                               side_effect=ml2_exc.MechanismDriverError):
            with mock.patch.object(mech_logger.LoggerMechanismDriver,
                                   'delete_subnet_postcommit') as dsp:

                with self.network() as network:
                    data = {'subnet': {'network_id':
                                       network['network']['id'],
                                       'cidr': '10.0.20.0/24',
                                       'ip_version': '4',
                                       'name': 'subnet1',
                                       'tenant_id':
                                       network['network']['tenant_id'],
                                       'gateway_ip': '10.0.20.1'}}
                    subnet_req = self.new_create_request('subnets', data)
                    subnet_res = subnet_req.get_response(self.api)
                    self.assertEqual(201, subnet_res.status_int)
                    subnet = self.deserialize(self.fmt, subnet_res)
                    subnet_id = subnet['subnet']['id']

                    req = self.new_delete_request('subnets', subnet_id)
                    res = req.get_response(self.api)
                    self.assertEqual(204, res.status_int)
                    # Test if other mechanism driver was called
                    self.assertTrue(dsp.called)
                    self._show('subnets', subnet_id,
                               expected_code=webob.exc.HTTPNotFound.code)

    def test_update_subnet_faulty(self):

        with mock.patch.object(mech_test.TestMechanismDriver,
                               'update_subnet_postcommit',
                               side_effect=ml2_exc.MechanismDriverError):
            with mock.patch.object(mech_logger.LoggerMechanismDriver,
                                   'update_subnet_postcommit') as usp:

                with self.network() as network:
                    data = {'subnet': {'network_id':
                                       network['network']['id'],
                                       'cidr': '10.0.20.0/24',
                                       'ip_version': '4',
                                       'name': 'subnet1',
                                       'tenant_id':
                                       network['network']['tenant_id'],
                                       'gateway_ip': '10.0.20.1'}}
                    subnet_req = self.new_create_request('subnets', data)
                    subnet_res = subnet_req.get_response(self.api)
                    self.assertEqual(201, subnet_res.status_int)
                    subnet = self.deserialize(self.fmt, subnet_res)
                    subnet_id = subnet['subnet']['id']
                    new_name = 'a_brand_new_name'
                    data = {'subnet': {'name': new_name}}
                    req = self.new_update_request('subnets', data, subnet_id)
                    res = req.get_response(self.api)
                    self.assertEqual(500, res.status_int)
                    error = self.deserialize(self.fmt, res)
                    self.assertEqual('MechanismDriverError',
                                     error['NeutronError']['type'])
                    # Test if other mechanism driver was called
                    self.assertTrue(usp.called)
                    subnet = self._show('subnets', subnet_id)
                    self.assertEqual(new_name, subnet['subnet']['name'])

                    self._delete('subnets', subnet['subnet']['id'])

    def test_create_port_faulty(self):

        with mock.patch.object(mech_test.TestMechanismDriver,
                               'create_port_postcommit',
                               side_effect=ml2_exc.MechanismDriverError):

            with self.network() as network:
                net_id = network['network']['id']
                data = {'port': {'network_id': net_id,
                                 'tenant_id':
                                 network['network']['tenant_id'],
                                 'name': 'port1',
                                 'admin_state_up': 1,
                                 'fixed_ips': []}}
                req = self.new_create_request('ports', data)
                res = req.get_response(self.api)
                self.assertEqual(500, res.status_int)
                error = self.deserialize(self.fmt, res)
                self.assertEqual('MechanismDriverError',
                                 error['NeutronError']['type'])
                query_params = "network_id=%s" % net_id
                ports = self._list('ports', query_params=query_params)
                self.assertFalse(ports['ports'])

    def test_update_port_faulty(self):

        with mock.patch.object(mech_test.TestMechanismDriver,
                               'update_port_postcommit',
                               side_effect=ml2_exc.MechanismDriverError):
            with mock.patch.object(mech_logger.LoggerMechanismDriver,
                                   'update_port_postcommit') as upp:

                with self.network() as network:
                    data = {'port': {'network_id': network['network']['id'],
                                     'tenant_id':
                                     network['network']['tenant_id'],
                                     'name': 'port1',
                                     'admin_state_up': 1,
                                     'fixed_ips': []}}
                    port_req = self.new_create_request('ports', data)
                    port_res = port_req.get_response(self.api)
                    self.assertEqual(201, port_res.status_int)
                    port = self.deserialize(self.fmt, port_res)
                    port_id = port['port']['id']

                    new_name = 'a_brand_new_name'
                    data = {'port': {'name': new_name}}
                    req = self.new_update_request('ports', data, port_id)
                    res = req.get_response(self.api)
                    self.assertEqual(500, res.status_int)
                    error = self.deserialize(self.fmt, res)
                    self.assertEqual('MechanismDriverError',
                                     error['NeutronError']['type'])
                    # Test if other mechanism driver was called
                    self.assertTrue(upp.called)
                    port = self._show('ports', port_id)
                    self.assertEqual(new_name, port['port']['name'])

                    self._delete('ports', port['port']['id'])

class TestMl2PluginCreateUpdateDeletePort(base.BaseTestCase):
    def setUp(self):
        super(TestMl2PluginCreateUpdateDeletePort, self).setUp()
        self.context = mock.MagicMock()
        self.notify_p = mock.patch('neutron.callbacks.registry.notify')
        self.notify = self.notify_p.start()

    def _ensure_transaction_is_closed(self):
        transaction = self.context.session.begin(subtransactions=True)
        enter = transaction.__enter__.call_count
        exit = transaction.__exit__.call_count
        self.assertEqual(enter, exit)

    def _create_plugin_for_create_update_port(self, new_host_port):
        plugin = ml2_plugin.Ml2Plugin()
        plugin.extension_manager = mock.Mock()
        plugin.type_manager = mock.Mock()
        plugin.mechanism_manager = mock.Mock()
        plugin.notifier = mock.Mock()
        plugin._get_host_port_if_changed = mock.Mock(
            return_value=new_host_port)
        plugin._check_mac_update_allowed = mock.Mock(return_value=True)

        self.notify.side_effect = (
            lambda r, e, t, **kwargs: self._ensure_transaction_is_closed())

        return plugin

    def test_create_port_rpc_outside_transaction(self):
        with contextlib.nested(
            mock.patch.object(ml2_plugin.Ml2Plugin, '__init__'),
            mock.patch.object(base_plugin.NeutronDbPluginV2, 'create_port'),
        ) as (init, super_create_port):
            init.return_value = None

            new_host_port = mock.Mock()
            plugin = self._create_plugin_for_create_update_port(new_host_port)

            plugin.create_port(self.context, mock.MagicMock())

            kwargs = {'context': self.context, 'port': new_host_port}
            self.notify.assert_called_once_with('port', 'after_create',
                plugin, **kwargs)

    def test_update_port_rpc_outside_transaction(self):
        with contextlib.nested(
            mock.patch.object(ml2_plugin.Ml2Plugin, '__init__'),
            mock.patch.object(base_plugin.NeutronDbPluginV2, 'update_port'),
        ) as (init, super_update_port):
            init.return_value = None
            new_host_port = mock.Mock()
            plugin = self._create_plugin_for_create_update_port(new_host_port)

            plugin.update_port(self.context, 'fake_id', mock.MagicMock())

            kwargs = {
                'context': self.context,
                'port': new_host_port,
                'mac_address_updated': True,
            }
            self.notify.assert_called_once_with('port', 'after_update',
                plugin, **kwargs)

    def test_notify_outside_of_delete_transaction(self):
        self.notify.side_effect = (
            lambda r, e, t, **kwargs: self._ensure_transaction_is_closed())
        l3plugin = mock.Mock()
        l3plugin.supported_extension_aliases = [
            'router', constants.L3_AGENT_SCHEDULER_EXT_ALIAS,
            constants.L3_DISTRIBUTED_EXT_ALIAS
        ]
        with contextlib.nested(
            mock.patch.object(ml2_plugin.Ml2Plugin, '__init__',
                              return_value=None),
            mock.patch.object(manager.NeutronManager,
                              'get_service_plugins',
                              return_value={'L3_ROUTER_NAT': l3plugin}),
        ):
            plugin = self._create_plugin_for_create_update_port(mock.Mock())
            # deleting the port will call registry.notify, which will
            # run the transaction balancing function defined in this test
            plugin.delete_port(self.context, 'fake_id')
            self.assertTrue(self.notify.call_count)


text:
<?php

chdir(realpath(dirname(__FILE__)));

require '../app/config/error_handling.php';
require '../app/config/config.php';

require '../app/lib/app.php';
require '../app/lib/model.php';

function getModel($modelName, $db) {
    require_once '../app/model/' . lcfirst($modelName) . '.php';
    $className = 'Scam\\' . $modelName . 'Model';
    return new $className($db);
}

/* used to initialize the database with admin bip32 key (used for multsig addresses)
 * and admin address (bitcoin auth for admin backend */
function setAdmin($app, $bip32Key, $adminAddress) {
    $db = $app->openDatabaseConnection();

    # validate bip32 key
    $key = getModel('User', $db)->parseBip32ExtendedPK($bip32Key);
    if(!$key || $key['depth'] != "2") {
        die("BIP32 key is not a valid bip32 key or not M/'k/0.\n");
    }

    # validate admin addresss
    if(!getModel('BitcoinTransaction', $db)->isValidBitcoinAddress($adminAddress)) {
        die("Address $adminAddress is not a valid bitcoin address.\n");
    }

    $configModel = getModel('Config', $db);
    if(!$configModel->setAdmin($bip32Key, $adminAddress)) {
        die("Error while saving to database.\n");
    }
}

/* Gets called by bitcoind everytime a new block is added.
 * Saves all transactions in this blocks in the database for later processal (by 'run')
 * */
function blockNotify($app, $block) {
    $db = $app->openDatabaseConnection();
    $bitcoinTransactionModel = getModel('BitcoinTransaction', $db);
    $bitcoinTransactionModel->addNewBlock($block);
}

/* Gets called as a cronjob periodically.
 * Processes all new transactions (saved by blockNotify) and checks if:
 *  - a payment to a multisig address has been made (order is paid by buyer)
 *  - a payment from the multisig address to the vendor has been made (buyer signed & broadcasted transaction = order is finished)
 * and then updates the order states accordingly.
 * */
function run($app) {
    $db = $app->openDatabaseConnection();
    $configModel = getModel('Config', $db);
    # implement locking so no more than one cronjob processes the transactions at a time
    $configModel->tryLock();
    try {
        $bitcoinTransactionModel = getModel('BitcoinTransaction', $db);
        $bitcoinTransactionModel->checkTransactionsForOrderPayments();
        $bitcoinTransactionModel->checkIfOrdersArePaid();
        $configModel->releaseLock();
    }
    catch(\Exception $e) {
        $configModel->releaseLock();
        throw $e;
    }
}

try {
    $app = new Scam\App();

    if(isset($argv[1])) {
        if($argv[1] == 'run') {
            run($app);
        }
        elseif($argv[1] == 'block-notify') {
            if(isset($argv[2])) {
                blockNotify($app, $argv[2]);
            }
            else {
                throw new Exception('block-notify without argument (block) called.');
            }
        }
        elseif($argv[1] == 'set-admin') {
            if(isset($argv[2]) && isset($argv[3])) {
                setAdmin($app, $argv[2], $argv[3]);
            }
            else {
                throw new Exception('set-admin without all arguments (extended public key, admin address) called.');
            }
        }
        else {
            throw new Exception('Unknown action ' . $argv[1]);
        }
    }
    else {
        throw new Exception('No action given !');
    }
}
catch(\Exception $e){
    print "Error: " . $e->getMessage() ."\n";
    print $e->getTraceAsString();
    exit(1);
}

redpajama_set_name: RedPajamaGithub


text:
require('regenerator/runtime');

var PouchDB = require('pouchdb');
var repStream = require('pouchdb-replication-stream');
var load = require('pouchdb-load');
PouchDB.plugin(repStream.plugin);
PouchDB.adapter('writableStream', repStream.adapters.writableStream);
PouchDB.plugin({loadIt: load.load});
var memdown = require('memdown');
var bluebird = require('bluebird');
var fs = bluebird.promisifyAll(require('fs'));
var pick = require('lodash').pick;
var shortRevs = require('short-revs');

var source = new PouchDB('inmem', {db: memdown});
var target = new PouchDB('inmem2', {db: memdown});

async function doIt() {
  await source.loadIt(await fs.readFileAsync('src/assets/monsters.txt', 'utf-8'));

  var docs = await source.allDocs({include_docs: true});
  var outputDocs = docs.rows.map(row => {
    var doc = pick(row.doc, '_id', 'moves');
    doc.moves = doc.moves.map(move => {
      var res = pick(move, 'learn_type', 'level');
      res.id = parseInt(move.resource_uri.match(/(\d+)\/$/)[1]);
      return res;
    });
    return doc;
  });

  await target.bulkDocs(outputDocs);

  var out = fs.createWriteStream('src/assets/monster-moves.txt');
  var stream = shortRevs();
  await target.dump(stream);
}

doIt().catch(console.log.bind(console));


text:
INPUT_FILEPATH = "../../thesis/testsuite/"
OUTPUT_FILEPATH = "../../thesis/testsuite/"

''' second computer git commit test comment '''
# Bitmap class
from m_bitmapClass import BMP

# Filter library
from m_filter import countSignal    # Counts black pixels in image
from m_filter import attenuate      # Removes black signal from image

# Output results helper function
def print_results(result ):
    if( result ):
        print("> PASS\n")
        return 0
    else:
        print("> FAIL\n")
        return 1

total_tests = 0
total_fails = 0
'''
# TEST 001 --------------------------------------
print("> Testcase #001 (import Bitmap Image)")
total_tests += 1
result = True
bmp = BMP("norway.bmp",INPUT_FILEPATH)
if( bmp.width != 500 ):
    result = False
    print("(!) Width Dimension Error")
if( bmp.height != 400 ):
    result = False
    print("(!) Height Dimension Error")
total_fails += print_results(result)

# TEST 002 --------------------------------------
print("> Testcase #002 (test countSignal filter)")
total_tests += 1
result = True
bmp = BMP("norway.bmp",INPUT_FILEPATH)
c = countSignal(bmp)
if( c != 5114 ):
    result = False
    print("(!) Error in countSignal result:",c)
total_fails += print_results(result)
'''
# TEST 003 --------------------------------------
print("> Testcase #003 (test attenuate filter)")
total_tests += 1
result = True
bmp = BMP("norway.bmp",INPUT_FILEPATH)
c1 = countSignal(bmp)
bmp = attenuate(bmp,1000)
c2 = countSignal(bmp)
if( (c1-c2) != 1000 ):
    result = False
    print("(!) Error in attenuate filter:",c1,c2)
bmp = attenuate(bmp,1000)
c3 = countSignal(bmp)
if( (c2-c3) != 1000 ):
    result = False
    print("(!) Error in attenuate filter:",c2,c3)
bmp = attenuate(bmp,1000)
c4 = countSignal(bmp)
if( (c3-c4) != 1000 ):
    result = False
    print("(!) Error in attenuate filter:",c3,c4)
bmp = attenuate(bmp,1000)
c5 = countSignal(bmp)
if( (c4-c5) != 1000 ):
    result = False
    print("(!) Error in attenuate filter:",c4,c5)
bmp = attenuate(bmp,1000)
c6 = countSignal(bmp)
if( (c5-c6) != 1000 ):
    result = False
    print("(!) Error in attenuate filter:",c5,c6)
bmp = attenuate(bmp,1000)
c7 = countSignal(bmp)
if( c7 != 0 ):
    result = False
    print("(!) Error attenuating all signal.")
#print("Final signal = ",c7)
#print(c1,c2,c3,c4,c5,c6)
total_fails += print_results(result)

# Final Results:
print("> Total Tests Run:",total_tests)
print("> # Failed:",total_fails)


text:
package io.github.astrapi69.swing;

import java.awt.*;

import javax.swing.*;

import net.miginfocom.swing.MigLayout;
import io.github.astrapi69.collections.pairs.ValueBox;
import io.github.astrapi69.model.LambdaModel;
import io.github.astrapi69.window.adapter.CloseWindow;

public class JMTextFieldTest
{
	public static void main(String[] args)
	{
		ValueBox<String> stringBox = ValueBox.<String> builder().value("foo").build();
		// Bind with JMTextField that encapsulate a property model
		JMTextField textFieldDecorator = new JMTextField("fff", 20);
		textFieldDecorator
			.setPropertyModel(LambdaModel.of(stringBox::getValue, stringBox::setValue));

		final Frame frame = new Frame("JMCheckBoxTest");
		JButton buttonCheck = new JButton("check it");
		buttonCheck.addActionListener(e -> {
			String selected = textFieldDecorator.getPropertyModel().getObject();
			selected = textFieldDecorator.getText();
			System.out.println(selected);
		});
		frame.addWindowListener(new CloseWindow());

		frame.setLayout(new MigLayout());
		frame.add(buttonCheck);
		frame.add(textFieldDecorator);
		frame.setSize(200, 200);
		frame.setVisible(true);
	}

}


text:
layout: page
title: "Building Wall ⋅ E"
subheadline: ""
teaser: "Wall ⋅ E is the first complex/advanced Lego set I've built in probably a decade. Like many kids, I was nuts for Lego as a child, I'd spend hours and hours endlessly building and rebuilding things out if my giant container of parts (which I still have by the way)."
author: Morgan
breadcrumb: true
---

![Wall⋅E before assembly](http://imgur.com/nz82WaC.jpg)

For Christmas this year I was fortunate enough to receive a Lego Wall ⋅ E from my mother.

Side note: don't bother using Amazon's gift wrapping service, they include a packing slip that spoils the whole thing - that's why I opened it early.

![Beginning Assembly](http://imgur.com/GZRhkYk.jpg)

![Body taking shape](http://imgur.com/VnpArvQ.jpg)

Wall ⋅ E is the first complex/advanced Lego set I've built in probably a decade. Like many kids, I was nuts for Lego as a child, I'd spend hours and hours endlessly building and rebuilding things out if my giant container of parts (which I still have by the way).

![Building Treads](http://imgur.com/2IrYV7A.jpg)

![Body complete](http://imgur.com/ktGeZrJ.jpg)

As an adult, I still have an appreciation for Lego, but up until recently I've never had somewhere to display any sets I might have. Funny how things change as one ages. As a child, what a set was meant to be proved pretty irrelevant - it might exist as something that followed the instructions for a day or too, but it was soon dismantled. Now as an adult, any sets I might find myself wanting will be bought solely to be built and displayed as the instructions intended.

![Building eyes](http://imgur.com/asY8jEs.jpg)

![Body with eyes complete](http://imgur.com/Bq1efPR.jpg)

I was so pleased to have gotten Wall ⋅E  as a gift. I adore the character, and sitting down for a few hours to build a Lego set for the first time in a **long** time was very rewarding. I think this will inspire me to finally purchase and build some of the sets I've been pining after for a while.

![Completed Wall E](http://imgur.com/94KMRmU.jpg)

 ---

Typed on Octopage


text:
HTTP Request library for Arduino and the Ethernet shield.

# Install

Clone (or download and unzip) the repository to `~/Documents/Arduino/libraries`
where `~/Documents/Arduino` is your sketchbook directory.

    > cd ~/Documents/Arduino
    > mkdir libraries
    > cd libraries
    > git clone https://github.com/csquared/arduino-restclient.git RestClient

# Usage

### Include

You need to have the `Ethernet` library already included.

```c++
#include <Ethernet.h>
#include <SPI.h>
#include "RestClient.h"
```

### RestClient(host/ip, [port])

Constructor to create an RestClient object to make requests against.

Use domain name and default to port 80:
```c++
RestClient client = RestClient("arduino-http-lib-test.herokuapp.com");
```

Use a local IP and an explicit port:
```c++
RestClient client = RestClient("192.168.1.50",5000);
```

### dhcp()

Sets up `EthernetClient` with a mac address of `DEADBEEFFEED`

```c++
  client.dhcp()
```

Note: you can have multiple RestClient objects but only need to call
this once.

Note: if you have multiple Arduinos on the same network, you'll need
to give each one a different mac address.

### begin(byte mac[])

It just wraps the `EthernetClient` call to `begin` and DHCPs.
Use this if you need to explicitly set the mac address.

```c++
  byte mac[] = { 0xDE, 0xAD, 0xBE, 0xEF, 0xFE, 0xED };
  if (client.begin(mac) == 0) {
     Serial.println("Failed to configure Ethernet using DHCP");
  }
```

### Manual Ethernet Setup

You can skip the above methods and just configure the EthernetClient yourself:

```c++
  byte mac[] = { 0xDE, 0xAD, 0xBE, 0xEF, 0xFE, 0xED };
  //the IP address for the shield:
  byte ip[] = { 192, 168, 2, 11 };
  Ethernet.begin(mac,ip);
```

```c++
  byte mac[] = { 0xDE, 0xAD, 0xBE, 0xEF, 0xFE, 0xED };
  Ethernet.begin(mac);
```

This is especially useful for debugging network connection issues.

## RESTful methods

All methods return an HTTP status code or 0 if there was an error.

### `get(const char* path)`
### `get(const char* path, String* response)`

Start making requests!

```c++
int statusCode = client.get("/"));
```

Pass in a string *by reference* for the response:
```
String response = "";
int statusCode = client.get("/", &response);
```

### post(const char* path, const char* body)
### post(const char* path, String* response)
### post(const char* path, const char* body, String* response)

```
String response = "";
int statusCode = client.post("/", &response);
statusCode = client.post("/", "foo=bar");
response = "";
statusCode = client.post("/", "foo=bar", &response);
```

### put(const char* path, const char* body)
### put(const char* path, String* response)
### put(const char* path, const char* body, String* response)

```
String response = "";
int statusCode = client.put("/", &response);
statusCode = client.put("/", "foo=bar");
response = "";
statusCode = client.put("/", "foo=bar", &response);
```

### del(const char* path)
### del(const char* path, const char* body)
### del(const char* path, String* response)
### del(const char* path, const char* body, String* response)

```
String response = "";
int statusCode = client.del("/", &response);
```

## Full Example

I test every way of calling the library (against a public heroku app)[https://github.com/csquared/arduino-http-test].

You can find the file in File->Examples->RestClient->full_test_suite

## Debug Mode

If you're having trouble, you can always open `RestClient.cpp` and throw at the top:

```c++
#define HTTP_DEBUG
```

Everything happening in the client will get printed to the Serial port.

# Thanks

[ricardochimal](https://github.com/ricardochimal) For all his c++ help.  Couldn't have done this without you!

[theycallmeswift](https://github.com/theycallmeswift) Helping incept and debug v1.0


text:
#ifndef TENSORFLOW_CORE_PLATFORM_JPEG_H_
#define TENSORFLOW_CORE_PLATFORM_JPEG_H_

#include "tensorflow/core/platform/platform.h"

#if defined(PLATFORM_GOOGLE)
#include "tensorflow/core/platform/google/build_config/jpeg.h"
#elif defined(PLATFORM_POSIX) || defined(PLATFORM_POSIX_ANDROID) || \
    defined(PLATFORM_GOOGLE_ANDROID)
extern "C" {
#include "external/jpeg_archive/jpeg-9a/jerror.h"
#include "external/jpeg_archive/jpeg-9a/jinclude.h"
#include "external/jpeg_archive/jpeg-9a/jpeglib.h"
#include "external/jpeg_archive/jpeg-9a/transupp.h"  // for rotations
}
#else
#error Define the appropriate PLATFORM_<foo> macro for this platform
#endif

#endif  // TENSORFLOW_CORE_PLATFORM_JPEG_H_


text:
using UnityEngine;
using System.Collections;

public class ShortDash : Skill
{
    [Tooltip("两次按键的输入间隔")]
    public float inputInterval = 0.1f;

    [Tooltip("移动距离,单位m")]
    public float MoveDistance = 5.0f;

    [Tooltip("移动时间,单位s")]
    public float MoveTime = 0.3f;

    [Tooltip("动画字符串")]
    public string animationString = "walking";

    protected override void Start()
    {
        base.Start();
        player.GetComponent<MovingGeneric>().ColliEvents += this.OnCollide;
    }

    public override bool InputDetermine()
    {
        //HINT:这个条件可能会有一段时间被多次判定成立,不过并不认为这会造成bug
        return input.TestDash(inputInterval);
    }

    public override void SkillStart(bool isServer)
    {
        base.SkillStart(isServer);
        phase = SkillPhase.Recovery;

        var animator = Model.GetComponent<Animator>();
        animator.SetBool(animationString, true);

        //转向
        player.GetComponent<MovingGeneric>().FaceTo(input.Move);

        //创建移动
        var movSpd = input.Move;
        movSpd.Normalize();
        movSpd *= MoveDistance;
        player.GetComponent<MovingGeneric>().AddDisplace(movSpd, MoveTime, this);

        Invoke("DoEnd", MoveTime);
    }

    public override void SkillBreak(bool isServer)
    {
        base.SkillBreak(isServer);

        var animator = Model.GetComponent<Animator>();
        animator.SetBool(animationString, false);

        //删除移动(如果有的话)
        player.GetComponent<MovingGeneric>().RemoveDisplace(this);

        //取消Invoke
        CancelInvoke();
    }

    void DoEnd()
    {
        manager.SkillCancel();
    }

    void OnCollide(Collision collisionInfo)
    {
        //撞东西后停止
        CancelInvoke();
        manager.SkillCancel();
    }
}


text:
require "spec_helper"

describe BookingSync::API::Client::Taxes do
  let(:client) { BookingSync::API::Client.new(test_access_token) }

  before { |ex| @casette_base_path = casette_path(casette_dir, ex.metadata) }

  describe ".taxes", :vcr do
    it "returns taxes" do
      expect(client.taxes).not_to be_empty
      assert_requested :get, bs_url("taxes")
    end
  end

  describe ".tax", :vcr do
    let(:prefetched_tax_id) {
      find_resource("#{@casette_base_path}_taxes/returns_taxes.yml", "taxes")[:id]
    }

    it "returns a single tax" do
      tax = client.tax(prefetched_tax_id)
      expect(tax.id).to eq prefetched_tax_id
    end
  end
end


text:
(function(root, undefined) {

  var app = root.Spinoza,
    rivets = root.rivets;

  app.module("Views.Associations", function(mod, app, BB, BM, $, _) {

    mod.Editor = BM.Layout.extend({

      template : 'jst:association_editor',

      className : 'fx-facts-editor',

      ui : {
        'targetInput' : 'input.fx-target'
      },

      regions : {
        'targetPreview' : '.fx-target-preview',
        'typeSelector'  : '.fx-type-selector'
      },

      events : {
        'click .fx-cancel' : 'onCancel',
        'click .fx-submit' : 'onSubmit'
      },

      initialize: function(options) {
        // model: editable fact
        this._initializeDomain(options.domain);
        this._initializeAssociationTypes();
      },

      _initializeDomain: function(domain) {
        this.domain = domain; // ASSUME: domain entity type stable
      },

      _initializeAssociationTypes: function() {
        this.factTypes = new app.Collections.LinkTypes();
        this._fetchAssociationTypes();
      },

      _fetchAssociationTypes: function() {
        var factTypes = this.factTypes,
          entityType = this._getDomainEntityType();
        factTypes._fetched = factTypes.fetch({
          domain : entityType,
          range  : 'entity'
        });
        return factTypes._fetched;
      },

      _getDomainEntityType: function() {
        return this.domain.get('entity_type');
      },

      onRender: function() {
        this._configurePostFactTypesRender();
      },

      _configurePostFactTypesRender: function() {
        var view = this;
        view.factTypes._fetched.then(function() {
          view._postFactTypesRender();
        });
      },

      _postFactTypesRender: function() {
        this._renderTypeSelector();
        this._bindRivets();
        this._configureAutocomplete();
        this._previewCurrentDocument();
      },

      _renderTypeSelector: function() {
        var selectorView = new app.Views.Facts.TypeSelector({
          collection : this.factTypes,
          attributes : { 'data-bind-value' : 'fact:link_type' }
        });
        this.typeSelector.show(selectorView);
      },

      _bindRivets: function() {
        if (this.rivets) this.rivets.unbind();
        this.rivets = rivets.bind(this.$el, { fact : this.model });
      },

      // ASSUME: link types loaded
     _configureAutocomplete: function() {
        var view = this;
        view.ui.targetInput.
          autocomplete({
            source : function(req, res) {
              view._autocompleteSource(req, res);
            },
            select : function(evt, ui) {
              return view._onAutocompleteSelect(evt, ui);
            },
            focus: function(evt, ui) {
              return view._onAutocompleteFocus(evt, ui);
            }
          });
      },

      _autocompleteSource: function(req, res) {
        var targetType = this._selectedLinkTargetType();
        if (targetType === 'document') {
          this._autocompleteDocumentTitle(req, res);
        }
        else {
          res([]);
        }
      },

      _selectedLinkTargetType: function() {
        var linkType = this._selectedLinkType(),
          targetType = (linkType ? linkType.get('target-type') : null);
        return targetType;
      },

      _selectedLinkType: function() {
        var linkType = this.model.get('link_type');
        if (linkType) {
          return this.factTypes.findWhere({ slug : linkType });
        }
      },

      _autocompleteDocumentTitle: function(req, res) {
        var params = this._configureAutocompleteParameters(req.term),
          url = app.Config.dataUrl('/public/documents' + '?' + $.param(params));
        $.getJSON(url, function(data) {
          var items = data.items;
          res(_.map(items, function(s) {
            return {
              label  : s.title,
              value  : s.title,
              data   : s
            };
          }));
        });
      },

      _configureAutocompleteParameters: function(term) {
        var params = { prefix : term },
          linkType = this._selectedLinkType(),
          entityType = (linkType ? linkType.get('range') : null);
        if (entityType) { params.entity_type = entityType; }
        return params;
      },

      _onAutocompleteSelect: function(evt, ui) {
        evt.preventDefault();
        this._setLinkForDocument(ui.item.data);
        this._submit();
      },

      _setLinkForDocument: function(doc) {
        if (!this._docIsTarget(doc)) {
          this._setLinkTarget(doc);
        }
      },

      _docIsTarget: function(doc) {
        var target = this.model.get('target');
        return target && (target.target_link == doc.link);
      },

      _setLinkTarget: function(doc) {
        this.model.set({
          title  : doc.title,
          target : this._buildTarget(doc)
        });
      },

      _buildTarget: function(doc) {
        return {
          title       : doc.title,
          target_link : doc.link
        };
      },

      _submit: function() {
        this.submit = true;
        this.$el.trigger('close');
      },

      _onAutocompleteFocus: function(evt, ui) {
        var linkId = ui.item.data.link;
        this._previewDocument(linkId);
      },

      _previewDocument: function(linkId) {
        var view = this,
          document = new app.Models.LinkedDocument({ link : linkId });
        document.fetch().then(function() {
          var documentPreview = new app.Views.Documents.Preview({ model : document });
          view.targetPreview.show(documentPreview);
        });
      },

      // ASSUME: link types loaded
      _previewCurrentDocument: function() {
        var preview = false;
        if (this._selectedLinkTargetType() === 'document') {
          var target = this.model.get('target'),
            linkId = (target ? target.target_link : null);
          if (linkId) {
            preview = true;
            this._previewDocument(linkId);
          }
        }
        if (!preview) {
          this._resetPreview();
        }
      },

      _resetPreview: function() {
        this.targetPreview.close();
      },

      onCancel: function(e) {
        e.preventDefault();
        this.submit = false;
        this.$el.trigger('close');
      },

      onSubmit: function(e) {
        e.preventDefault();
        this._submit();
      }

    });

  });

}).call({}, window);


text:
<?php

namespace SR\Utilities\Transform;

use SR\Utilities\Transform\Argument\Expression\Archetype\RangedArchetype;
use SR\Utilities\Transform\Argument\Expression\Archetype\StringArchetype;
use SR\Utilities\Transform\Argument\Expression\Representative\SearchReplaceRepresentative;

class StringTransform extends AbstractTransform
{
    public function set(mixed $value): self
    {
        if (false === static::isConsumable($value)) {
            throw new \InvalidArgumentException('Value is not a string and could not be coerced to one.');
        }

        return parent::set((string) $value);
    }

    /**
     * Perform string replacement using regular expression and replacement. Optionally turn regex inverse (negative)
     * and enforce case sensitivity.
     */
    public function replace(SearchReplaceRepresentative $config): self
    {
        return $this->apply(function () use ($config) {
            return preg_replace($config->regex(), $config->replacement(), $this->get());
        });
    }

    /**
     * Perform conversion to alpha characters only.
     */
    public function toAlpha(): self
    {
        return $this->replace(new SearchReplaceRepresentative('', new RangedArchetype('a-z', true)));
    }

    /**
     * Perform conversion to numeric characters only.
     */
    public function toNumeric(): self
    {
        return $this->replace(new SearchReplaceRepresentative('', new RangedArchetype('0-9', true)));
    }

    /**
     * Perform conversion to alphanumeric characters only.
     */
    public function toAlphanumeric(): self
    {
        return $this->replace(new SearchReplaceRepresentative('', new RangedArchetype('a-z0-9', true)));
    }

    /**
     * Perform conversion to alphanumeric and dash characters only.
     */
    public function toAlphanumericAndDashes(): self
    {
        return $this->replace(new SearchReplaceRepresentative('', new RangedArchetype('a-z0-9-', true)));
    }

    public function spacesToDashes(): self
    {
        return $this->replace(new SearchReplaceRepresentative('-', new StringArchetype('[ ]+')));
    }

    /**
     * Perform conversion to alphanumeric and spaces to dashes only.
     */
    public function toAlphanumericAndSpacesToDashes(): self
    {
        return $this
            ->replace(new SearchReplaceRepresentative('-', new StringArchetype('[\s\n]+')))
            ->toAlphanumericAndDashes()
            ->replace(new SearchReplaceRepresentative('-', new StringArchetype('[-]+')))
        ;
    }

    public function dashesToSpaces(): self
    {
        return $this->replace(new SearchReplaceRepresentative(' ', new StringArchetype('[-]+')));
    }

    public function slugify(bool $lowercase = true): self
    {
        return $this->apply(function (self $value) {
            $result = $value
                ->setMutable(true)
                ->replace(new SearchReplaceRepresentative('-', new RangedArchetype('a-z0-9-', true)))
                ->replace(new SearchReplaceRepresentative('-', new StringArchetype('[-]+')))
                ->replace((new SearchReplaceRepresentative('', new StringArchetype('[-]')))->enableAnchorLeft())
                ->replace((new SearchReplaceRepresentative('', new StringArchetype('[-]')))->enableAnchorRight())
                ->toLower()
            ;

            return $this->returnInstance($result->get());
        });
    }

    public function toLower(): self
    {
        return $this->apply(function () {
            return mb_strtolower($this->get());
        });
    }

    public function toUpper(): self
    {
        return $this->apply(function () {
            return mb_strtoupper($this->get());
        });
    }

    public function camelToPascalCase(): self
    {
        return $this->apply(function () {
            return ucfirst($this->get());
        });
    }

    public function camelToSnakeCase(): self
    {
        return $this->apply(function () {
            return preg_replace('#(?<=\\w)(?=[A-Z])#', '_$1', $this->get());
        })->toLower();
    }

    public function camelToSpinalCase(): self
    {
        return $this->apply(function () {
            return preg_replace('#(?<=\\w)(?=[A-Z])#', '-$1', $this->get());
        })->toLower();
    }

    public function pascalToCamelCase(): self
    {
        return $this->apply(function () {
            return lcfirst($this->get());
        });
    }

    public function pascalToSnakeCase(): self
    {
        return $this->apply(function () {
            return $this->camelToSnakeCase();
        });
    }

    public function pascalToSpinalCase(): self
    {
        return $this->apply(function () {
            return $this->camelToSpinalCase();
        });
    }

    public function snakeToCamelCase(): self
    {
        return $this->apply(function () {
            return preg_replace_callback('{(_)([a-z])}', function ($match) {
                return mb_strtoupper($match[2]);
            }, $this->get());
        });
    }

    public function snakeToPascalCase(): self
    {
        return $this->apply(function () {
            return $this->snakeToCamelCase()->camelToPascalCase();
        });
    }

    public function snakeToSpinalCase(): self
    {
        return $this->apply(function () {
            return str_replace('_', '-', $this->get());
        });
    }

    public function spinalToCamelCase(): self
    {
        return $this->apply(function () {
            return preg_replace_callback('{(-)([a-z])}', function ($match) {
                return mb_strtoupper($match[2]);
            }, $this->get());
        });
    }

    public function spinalToPascalCase(): self
    {
        return $this->apply(function () {
            return $this->spinalToCamelCase()->camelToPascalCase();
        });
    }

    public function spinalToSnakeCase(): self
    {
        return $this->apply(function () {
            return str_replace('-', '_', $this->get());
        });
    }

    public function toPhoneNumber(string $regex = '~.*(\d{3})[^\d]*(\d{3})[^\d]*(\d{4}).*~'): self
    {
        return $this->apply(function () use ($regex) {
            $number = preg_replace('/[^0-9]/', '', preg_replace($regex, '$1$2$3', $this->get()));

            if (7 > mb_strlen($number)) {
                return $this->get();
            }

            return 10 === mb_strlen($number) ? '1' . $number : $number;
        });
    }

    public function toPhoneFormat(): self
    {
        return $this->apply(function () {
            if (false === in_array(mb_strlen($string = $this->copy()->toPhoneNumber()->get()), [11, 7], true)) {
                return $this->get();
            }

            $number = sprintf('%s-%s', mb_substr($string, -7, 3), mb_substr($string, -4, 4));

            if (11 === mb_strlen($string)) {
                $number = sprintf('+%s (%s) ', mb_substr($string, 0, 1), mb_substr($string, 1, 3)) . $number;
            }

            return $number;
        });
    }

    /**
     * @return string[]
     */
    public function split(): array
    {
        return mb_str_split($this->__toString());
    }
}

/* EOF */


text:
package com.google.common.collect.testing.google;

import static com.google.common.collect.testing.Helpers.assertEqualInOrder;

import com.google.common.annotations.GwtCompatible;
import com.google.common.collect.ListMultimap;
import java.util.Arrays;
import java.util.Collection;
import org.junit.Ignore;

/**
 * Superclass for all {@code ListMultimap} testers.
 *
 * @author Louis Wasserman
 */
@GwtCompatible
@Ignore // Affects only Android test runner, which respects JUnit 4 annotations on JUnit 3 tests.
public class AbstractListMultimapTester<K, V>
    extends AbstractMultimapTester<K, V, ListMultimap<K, V>> {

  @Override
  protected void assertGet(K key, V... values) {
    assertGet(key, Arrays.asList(values));
  }

  @Override
  protected void assertGet(K key, Collection<V> values) {
    assertEqualInOrder(values, multimap().get(key));

    if (!values.isEmpty()) {
      assertEqualInOrder(values, multimap().asMap().get(key));
      assertFalse(multimap().isEmpty());
    } else {
      assertNull(multimap().asMap().get(key));
    }

    assertEquals(values.size(), multimap().get(key).size());
    assertEquals(values.size() > 0, multimap().containsKey(key));
    assertEquals(values.size() > 0, multimap().keySet().contains(key));
    assertEquals(values.size() > 0, multimap().keys().contains(key));
  }
}