def _mk_test_dp(name):
     ofp = importutils.import_module("ryu.ofproto.ofproto_v1_3")
     ofpp = importutils.import_module("ryu.ofproto.ofproto_v1_3_parser")
     dp = mock.Mock()
     dp.ofproto = ofp
     dp.ofproto_parser = ofpp
     dp.__repr__ = lambda _self: name
     return dp

 def _mk_test_dp(self, name):
     ofp = importutils.import_module('ryu.ofproto.ofproto_v1_3')
     ofpp = importutils.import_module('ryu.ofproto.ofproto_v1_3_parser')
     dp = mock.Mock()
     dp.ofproto = ofp
     dp.ofproto_parser = ofpp
     dp.__repr__ = mock.Mock(return_value=name)
     return dp

def _get_impl():
    """Delay import of rpc_backend until configuration is loaded."""
    global _RPCIMPL
    if _RPCIMPL is None:
        try:
            _RPCIMPL = importutils.import_module(CONF.rpc_backend)
        except ImportError:
            # For backwards compatibility with older nova config.
            impl = CONF.rpc_backend.replace("nova.rpc", "nova.openstack.common.rpc")
            _RPCIMPL = importutils.import_module(impl)
    return _RPCIMPL

 def test_setup_default_table(self):
     br = self.br
     with mock.patch.object(br, '_send_msg') as sendmsg:
         br.setup_default_table()
     (dp, ofp, ofpp) = br._get_dp()
     arp = importutils.import_module('ryu.lib.packet.arp')
     ether = importutils.import_module('ryu.ofproto.ether')
     call = mock.call
     expected_calls = [
         call(ofpp.OFPFlowMod(dp, command=ofp.OFPFC_DELETE,
              match=ofpp.OFPMatch(), out_group=ofp.OFPG_ANY,
              out_port=ofp.OFPP_ANY, priority=0, table_id=ofp.OFPTT_ALL)),
         call(ofpp.OFPFlowMod(dp, priority=0, table_id=0)),
         call(ofpp.OFPFlowMod(dp, priority=0, table_id=1)),
         call(ofpp.OFPFlowMod(dp, priority=0, table_id=2)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=7)],
              priority=0, table_id=3)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=5)],
              priority=0, table_id=4)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=6)],
              priority=0, table_id=5)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS,
              [ofpp.OFPActionOutput(ofp.OFPP_CONTROLLER)])],
              match=ofpp.OFPMatch(arp_op=arp.ARP_REQUEST,
              eth_type=ether.ETH_TYPE_ARP), priority=1, table_id=6)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=7)],
              priority=0, table_id=6)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=8)],
              priority=0, table_id=7)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=9)],
              priority=0, table_id=8)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=10)],
              priority=0, table_id=9)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=11)],
              priority=0, table_id=10)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=12)],
              priority=0, table_id=11)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=13)],
              priority=0, table_id=12)),
         call(ofpp.OFPFlowMod(dp, priority=0, table_id=13)),
     ]
     sendmsg.assert_has_calls(expected_calls, any_order=True)

 def setUp(self):
     super(OFAAgentTestBase, self).setUp()
     ryu_cfg = importutils.import_module('ryu.cfg')
     ryu_cfg.CONF = cfg.ConfigOpts()
     ryu_cfg.CONF.register_cli_opts([
         cfg.StrOpt('ofp-listen-host', default='',
                    help='openflow listen host'),
         cfg.IntOpt('ofp-tcp-listen-port', default=6633,
                    help='openflow tcp listen port')
     ])
     self.mod_agent = importutils.import_module(self._AGENT_NAME)
     self.ryuapp = mock.Mock()

    def __call__(self, target, creds, enforcer):
        if self.target_field not in target:
            # policy needs a plugin check
            # target field is in the form resource:field
            # however if they're not separated by a colon, use an underscore
            # as a separator for backward compatibility

            def do_split(separator):
                parent_res, parent_field = self.target_field.split(separator, 1)
                return parent_res, parent_field

            for separator in (":", "_"):
                try:
                    parent_res, parent_field = do_split(separator)
                    break
                except ValueError:
                    LOG.debug(_("Unable to find ':' as separator in %s."), self.target_field)
            else:
                # If we are here split failed with both separators
                err_reason = _("Unable to find resource name in %s") % self.target_field
                LOG.exception(err_reason)
                raise exceptions.PolicyCheckError(policy="%s:%s" % (self.kind, self.match), reason=err_reason)
            parent_foreign_key = attributes.RESOURCE_FOREIGN_KEYS.get("%ss" % parent_res, None)
            if not parent_foreign_key:
                err_reason = _("Unable to verify match:%(match)s as the " "parent resource: %(res)s was not found") % {
                    "match": self.match,
                    "res": parent_res,
                }
                LOG.exception(err_reason)
                raise exceptions.PolicyCheckError(policy="%s:%s" % (self.kind, self.match), reason=err_reason)
            # NOTE(salv-orlando): This check currently assumes the parent
            # resource is handled by the core plugin. It might be worth
            # having a way to map resources to plugins so to make this
            # check more general
            # NOTE(ihrachys): if import is put in global, circular
            # import failure occurs
            manager = importutils.import_module("neutron.manager")
            f = getattr(manager.NeutronManager.get_instance().plugin, "get_%s" % parent_res)
            # f *must* exist, if not found it is better to let neutron
            # explode. Check will be performed with admin context
            context = importutils.import_module("neutron.context")
            try:
                data = f(context.get_admin_context(), target[parent_foreign_key], fields=[parent_field])
                target[self.target_field] = data[parent_field]
            except Exception:
                with excutils.save_and_reraise_exception():
                    LOG.exception(_LE("Policy check error while calling %s!"), f)
        match = self.match % target
        if self.kind in creds:
            return match == unicode(creds[self.kind])
        return False

 def setUp(self):
     self.fake_oflib_of = fake_oflib.patch_fake_oflib_of()
     self.fake_oflib_of.start()
     self.addCleanup(self.fake_oflib_of.stop)
     self.mod_agent = importutils.import_module(self._AGENT_NAME)
     super(OFAAgentTestBase, self).setUp()
     self.ryuapp = mock.Mock()

    def setUp(self,
              plugin=MIDONET_PLUGIN_NAME,
              ext_mgr=None,
              service_plugins=None):

        self.midoclient_mock = mock.MagicMock()
        self.midoclient_mock.midonetclient.neutron.client.return_value = True
        modules = {
            'midonetclient': self.midoclient_mock,
            'midonetclient.neutron': self.midoclient_mock.neutron,
            'midonetclient.neutron.client': self.midoclient_mock.client,
        }

        self.module_patcher = mock.patch.dict('sys.modules', modules)
        self.module_patcher.start()

        # import midonetclient here because it needs proper mock objects to be
        # assigned to this module first.  'midoclient_mock' object is the
        # mock object used for this module.
        from midonetclient.neutron.client import MidonetClient
        client_class = MidonetClient
        self.mock_class = client_class()

        extensions_path = importutils.import_module(
            MIDOKURA_EXT_PATH).__file__
        cfg.CONF.set_override('api_extensions_path',
                              os.path.dirname(extensions_path))
        super(MidonetPluginV2TestCase, self).setUp(plugin=plugin)

 def test_arp_passthrough(self):
     br = self.br
     with mock.patch.object(br, '_send_msg') as sendmsg:
         br.arp_passthrough(network=1234, tpa='192.0.2.1')
     (dp, ofp, ofpp) = br._get_dp()
     arp = importutils.import_module('ryu.lib.packet.arp')
     ether = importutils.import_module('ryu.ofproto.ether')
     call = mock.call
     expected_calls = [
         call(ofpp.OFPFlowMod(dp, idle_timeout=5,
              instructions=[ofpp.OFPInstructionGotoTable(table_id=7)],
              match=ofpp.OFPMatch(arp_op=arp.ARP_REQUEST,
              arp_tpa="192.0.2.1", eth_type=ether.ETH_TYPE_ARP,
              metadata=meta.mk_metadata(1234)), priority=1, table_id=5))
     ]
     sendmsg.assert_has_calls(expected_calls)

 def setUp(self):
     self.socket_mock = mock.patch(SERVERMANAGER + ".socket.create_connection").start()
     self.wrap_mock = mock.patch(SERVERMANAGER + ".ssl.wrap_socket").start()
     super(ServerManagerTests, self).setUp()
     # http patch must not be running or it will mangle the servermanager
     # import where the https connection classes are defined
     self.httpPatch.stop()
     self.sm = importutils.import_module(SERVERMANAGER)

    def test__provision_local_vlan_outbound_flat(self):
        ofp = importutils.import_module("ryu.ofproto.ofproto_v1_3")
        ofpp = importutils.import_module("ryu.ofproto.ofproto_v1_3_parser")
        with mock.patch.object(self.agent, "ryu_send_msg") as sendmsg:
            self.agent._provision_local_vlan_outbound(888, ofp.OFPVID_NONE, "phys-net1")

        expected_msg = ofpp.OFPFlowMod(
            self.agent.phys_brs["phys-net1"].datapath,
            instructions=[
                ofpp.OFPInstructionActions(
                    ofp.OFPIT_APPLY_ACTIONS, [ofpp.OFPActionPopVlan(), ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0)]
                )
            ],
            match=ofpp.OFPMatch(in_port=777, vlan_vid=888 | ofp.OFPVID_PRESENT),
            priority=4,
        )
        sendmsg.assert_has_calls([mock.call(expected_msg)])

 def _load_backend(self):
     with self._lock:
         if not self._backend:
             # Import the untranslated name if we don't have a mapping
             backend_path = self._backend_mapping.get(self._backend_name,
                                                      self._backend_name)
             backend_mod = importutils.import_module(backend_path)
             self._backend = backend_mod.get_backend()

 def nuageclient_init(self):
     server = cfg.CONF.RESTPROXY.server
     serverauth = cfg.CONF.RESTPROXY.serverauth
     serverssl = cfg.CONF.RESTPROXY.serverssl
     base_uri = cfg.CONF.RESTPROXY.base_uri
     auth_resource = cfg.CONF.RESTPROXY.auth_resource
     organization = cfg.CONF.RESTPROXY.organization
     nuageclient = importutils.import_module("nuagenetlib.nuageclient")
     self.nuageclient = nuageclient.NuageClient(server, base_uri, serverssl, serverauth, auth_resource, organization)

    def setUp(self):
        super(TestArpLib, self).setUp()

        self.mod_arplib = importutils.import_module(_OFALIB_NAME)
        self.arplib = self.mod_arplib.ArpLib(self.ryuapp)
        self.packet_mod.get_protocol = self._fake_get_protocol
        self._fake_get_protocol_ethernet = True
        self._fake_get_protocol_vlan = True
        self._fake_get_protocol_arp = True

 def __init__(self, backend_mapping=None):
     if backend_mapping is None:
         backend_mapping = {}
     backend_name = CONF.database.backend
     # Import the untranslated name if we don't have a
     # mapping.
     backend_path = backend_mapping.get(backend_name, backend_name)
     backend_mod = importutils.import_module(backend_path)
     self.__backend = backend_mod.get_backend()

    def _import_ncclient(self):
        """Import the NETCONF client (ncclient) module.

        The ncclient module is not installed as part of the normal Neutron
        distributions. It is imported dynamically in this module so that
        the import can be mocked, allowing unit testing without requiring
        the installation of ncclient.
        """
        return importutils.import_module('ncclient.manager')

 def test_install_arp_responder(self):
     br = self.br
     with mock.patch.object(br, '_send_msg') as sendmsg:
         br.install_arp_responder(table_id=99)
     (dp, ofp, ofpp) = br._get_dp()
     arp = importutils.import_module('ryu.lib.packet.arp')
     ether = importutils.import_module('ryu.ofproto.ether')
     call = mock.call
     expected_calls = [
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS,
              [ofpp.OFPActionOutput(ofp.OFPP_CONTROLLER)])],
              match=ofpp.OFPMatch(arp_op=arp.ARP_REQUEST,
              eth_type=ether.ETH_TYPE_ARP), priority=1, table_id=99)),
         call(ofpp.OFPFlowMod(dp, instructions=[
              ofpp.OFPInstructionGotoTable(table_id=100)],
              priority=0, table_id=99)),
     ]
     sendmsg.assert_has_calls(expected_calls)

    def _service_admin_creds(api_version=2):
        # Import auth_token to have keystone_authtoken settings setup.
        importutils.import_module('keystoneclient.middleware.auth_token')

        creds = {
            'username': cfg.CONF.keystone_authtoken.admin_user,
            'password': cfg.CONF.keystone_authtoken.admin_password,
        }
        if api_version >= 3:
            creds['auth_url'] =\
                cfg.CONF.keystone_authtoken.auth_uri.replace('v2.0', 'v3')
            creds['project_name'] =\
                cfg.CONF.keystone_authtoken.admin_tenant_name
        else:
            creds['auth_url'] = cfg.CONF.keystone_authtoken.auth_uri
            creds['tenant_name'] =\
                cfg.CONF.keystone_authtoken.admin_tenant_name

        return creds

def main():
    cfg.CONF(default_config_files=(
        [NUAGE_CONFIG_FILE]))
    config.nuage_register_cfg_opts()
    server = cfg.CONF.RESTPROXY.server
    serverauth = cfg.CONF.RESTPROXY.serverauth
    serverssl = cfg.CONF.RESTPROXY.serverssl
    base_uri = cfg.CONF.RESTPROXY.base_uri
    auth_resource = cfg.CONF.RESTPROXY.auth_resource
    organization = cfg.CONF.RESTPROXY.organization
    fipquota = str(cfg.CONF.RESTPROXY.default_floatingip_quota)
    logging = importutils.import_module('logging')
    nuageclientinst = importutils.import_module('nuagenetlib.nuageclient')
    nuageclient = nuageclientinst.NuageClient(server, base_uri,
                                              serverssl, serverauth,
                                              auth_resource,
                                              organization)
    logging.basicConfig(level=logging.DEBUG)
    SyncManager(nuageclient).synchronize(fipquota)

    def setUp(self):
        super(OFAAgentTestCase, self).setUp()

        Net = collections.namedtuple('Net', 'net, mac, ip')
        self.nets = [Net(net=10, mac='11:11:11:44:55:66', ip='10.1.2.20'),
                     Net(net=10, mac='11:11:11:44:55:67', ip='10.1.2.21'),
                     Net(net=20, mac='22:22:22:44:55:66', ip='10.2.2.20')]

        self.packet_mod = mock.Mock()
        self.proto_ethernet_mod = mock.Mock()
        self.proto_vlan_mod = mock.Mock()
        self.proto_vlan_mod.vid = 999
        self.proto_arp_mod = mock.Mock()
        self.fake_get_protocol = mock.Mock(return_value=self.proto_vlan_mod)
        self.packet_mod.get_protocol = self.fake_get_protocol
        self.fake_add_protocol = mock.Mock()
        self.packet_mod.add_protocol = self.fake_add_protocol
        self.arp = importutils.import_module('ryu.lib.packet.arp')
        self.ethernet = importutils.import_module('ryu.lib.packet.ethernet')
        self.vlan = importutils.import_module('ryu.lib.packet.vlan')
        mock.patch('ryu.lib.packet.packet.Packet',
                   return_value=self.packet_mod).start()

        self.ryuapp = 'ryuapp'
        self.inport = '1'
        self.ev = mock.Mock()
        self.datapath = self._mk_test_dp('tun_br')
        self.ofproto = importutils.import_module('ryu.ofproto.ofproto_v1_3')
        self.ofpp = mock.Mock()
        self.datapath.ofproto = self.ofproto
        self.datapath.ofproto_parser = self.ofpp
        self.OFPActionOutput = mock.Mock()
        self.OFPActionOutput.return_value = 'OFPActionOutput'
        self.ofpp.OFPActionOutput = self.OFPActionOutput
        self.msg = mock.Mock()
        self.msg.datapath = self.datapath
        self.msg.buffer_id = self.ofproto.OFP_NO_BUFFER
        self.msg_data = 'test_message_data'
        self.msg.data = self.msg_data
        self.ev.msg = self.msg
        self.msg.match = {'in_port': self.inport,
                          'metadata': meta.LOCAL | self.nets[0].net}