From 057d7ed8d4a82d7b65948c226447b65baaa2087e Mon Sep 17 00:00:00 2001 From: changzhi1990 Date: Thu, 21 Jan 2016 11:12:38 +0800 Subject: [PATCH] dhcp check and packet driver Add send and sniff packet driver: * Scapy * pypcap Complete agent driver unittest Modify test-requirements mock==1.0.1 VXLAN support waiting for scapy 3.2.2 upload to pypi Change-Id: I5e6f707fff08a60135655217e72660b301a4071d --- requirements.txt | 2 + steth/agent/api.py | 62 ++++++++++ steth/agent/common/constants.py | 32 ++++++ steth/agent/common/utils.py | 6 + steth/agent/drivers/pcap_driver.py | 40 +++++++ steth/agent/drivers/scapy_driver.py | 55 +++++++++ steth/tests/unit/agent/common/test_utils.py | 11 +- steth/tests/unit/agent/drivers/test_pcap.py | 54 +++++++++ steth/tests/unit/agent/drivers/test_scapy.py | 38 +++++++ steth/tests/unit/agent/test_api.py | 113 +++++++++++-------- test-requirements.txt | 2 +- 11 files changed, 359 insertions(+), 56 deletions(-) create mode 100644 steth/agent/common/constants.py create mode 100644 steth/agent/drivers/pcap_driver.py create mode 100644 steth/agent/drivers/scapy_driver.py create mode 100644 steth/tests/unit/agent/drivers/test_pcap.py create mode 100644 steth/tests/unit/agent/drivers/test_scapy.py diff --git a/requirements.txt b/requirements.txt index 3a42a8f..7855bdf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,3 +3,5 @@ jsonrpclib netaddr cliff oslo.config>=1.6.0 +scapy>=2.3.1 +pypcap diff --git a/steth/agent/api.py b/steth/agent/api.py index 5548988..9f9c224 100644 --- a/steth/agent/api.py +++ b/steth/agent/api.py @@ -14,10 +14,14 @@ # under the License. import re +import time from netaddr import IPNetwork from steth.agent.common import utils as agent_utils from steth.agent.drivers import iperf as iperf_driver +from steth.agent.drivers import scapy_driver +from steth.agent.drivers import pcap_driver from steth.agent.common import log +from steth.agent.common import constants LOG = log.get_logger() @@ -178,3 +182,61 @@ class AgentApi(object): except Exception as e: message = e.message return agent_utils.make_response(code=1, message=message) + + def check_dhcp_on_comp(self, port_id, port_mac, + phy_iface, net_type='vlan'): + try: + pcap = pcap_driver.PcapDriver() + filter = '(udp and (port 68 or 67) and ether host %s)' % port_mac + listeners = pcap.setup_listener_on_comp(port_id, filter) + if not cmp(net_type, 'vlan'): + phy_listener = pcap.setup_listener(phy_iface, filter) + else: + # TODO(yaowei) vxlan subinterface + raise Exception("network type %s not supported." % net_type) + scapy = scapy_driver.ScapyDriver() + scapy.send_dhcp_over_qvb(port_id, port_mac) + # NOTE(yaowei) thread sleep 2 seconds wait for dhcp reply. + time.sleep(2) + map(pcap.set_nonblock, listeners) + pcap.set_nonblock(phy_listener) + data = dict() + for listener in listeners: + vif_pre = listener.name[:constants.VIF_PREFIX_LEN] + data[vif_pre] = [] + for packet in listener.readpkts(): + data[vif_pre].extend(scapy.get_dhcp_mt(str(packet[1]))) + data[phy_listener.name] = [] + for packet in phy_listener.readpkts(): + data[phy_listener.name].append( + scapy.get_dhcp_mt(str(packet[1]))) + return agent_utils.make_response(code=0, data=data) + except Exception as e: + return agent_utils.make_response(code=1, message=e.message) + + def check_dhcp_on_net(self, net_id, port_ip, phy_iface, net_type='vlan'): + if not cmp(net_type, 'vxlan'): + raise Exception("network type %s not supported." % net_type) + dhcp_ns = constants.DHCP_NS_PREFIX + net_id + # get tap interface in dhcp namespace + cmd = ['ip', 'netns', 'exec', dhcp_ns] + route_cmd = cmd + ['ip', 'r', 'show', 'default', '0.0.0.0/0'] + stdcode, stdout = agent_utils.execute(route_cmd, root=True) + if stdcode != 0: + raise Exception(stdout.pop()) + tap_iface = stdout.pop().split().pop() + arp_cmd = cmd + ['arping', '-I', tap_iface, '-c', '1', port_ip] + pcap = pcap_driver.PcapDriver() + filter = '(arp and host %s)' % port_ip + ifaces = ['br-int', 'ovsbr3', phy_iface] + listeners = map(lambda i: pcap.setup_listener(i, filter), ifaces) + agent_utils.execute(arp_cmd, root=True) + map(pcap.set_nonblock, listeners) + # unpack arp + data = dict() + scapy = scapy_driver.ScapyDriver() + for listener in listeners: + data[listener.name] = [] + for packet in listener.readpkts(): + data[listener.name].append(scapy.get_arp_op(str(packet[1]))) + return agent_utils.make_response(code=0, data=data) diff --git a/steth/agent/common/constants.py b/steth/agent/common/constants.py new file mode 100644 index 0000000..6784092 --- /dev/null +++ b/steth/agent/common/constants.py @@ -0,0 +1,32 @@ +# Copyright 2016 UnitedStack, Inc. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +# Virtuel Interface Prefix +TAP_DEVICE_PREFIX = 'tap' +QBR_DEVICE_PREFIX = 'qbr' +QVB_DEVICE_PREFIX = 'qvb' +QVO_DEVICE_PREFIX = 'qvo' + +VIF_PREFIX_LEN = 3 +DEVICE_NAME_LEN = 14 + +# DHCP Message Type +# Reference: http://www.networksorcery.com/enp/rfc/rfc1533.txt +DHCP_MESSATE_TYPE = ['', 'DHCPDISCOVER', 'DHCPOFFER', 'DHCPREQUEST', + 'DHCPDECLINE', 'DHCPACK', 'DHCPNAK', 'DHCPRELEASE'] +DHCP_NS_PREFIX = 'qdhcp-' + +# Reference: http://www.networksorcery.com/enp/rfc/rfc826.txt +ARP_OP_TYPE = ['', 'REQUEST', 'REPLY'] diff --git a/steth/agent/common/utils.py b/steth/agent/common/utils.py index d2a559c..0a299b0 100644 --- a/steth/agent/common/utils.py +++ b/steth/agent/common/utils.py @@ -22,6 +22,7 @@ import platform from threading import Timer from steth.agent.common import resource from steth.agent.common import log +from steth.agent.common import constants LOG = log.get_logger() @@ -165,3 +166,8 @@ def replace_file(file_name, mode=0o644): tmp_file = tempfile.NamedTemporaryFile('w+', dir=base_dir, delete=False) os.chmod(tmp_file.name, mode) os.rename(tmp_file.name, file_name) + + +def get_vif_name(prefix, port_id): + requested_name = prefix + port_id + return requested_name[:constants.DEVICE_NAME_LEN] diff --git a/steth/agent/drivers/pcap_driver.py b/steth/agent/drivers/pcap_driver.py new file mode 100644 index 0000000..1b6a620 --- /dev/null +++ b/steth/agent/drivers/pcap_driver.py @@ -0,0 +1,40 @@ +# Copyright 2016 UnitedStack, Inc. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import pcap +from steth.agent.common import log +from steth.agent.common import utils +from steth.agent.common import constants + +LOG = log.get_logger() + + +class PcapDriver(object): + + def setup_listener(self, iface, filter): + listener = pcap.pcap(iface) + listener.setfilter(filter) + return listener + + def setup_listener_on_comp(self, port_id, filter): + tap_device = utils.get_vif_name(constants.TAP_DEVICE_PREFIX, port_id) + qvb_device = utils.get_vif_name(constants.QVB_DEVICE_PREFIX, port_id) + qbr_device = utils.get_vif_name(constants.QBR_DEVICE_PREFIX, port_id) + qvo_device = utils.get_vif_name(constants.QVO_DEVICE_PREFIX, port_id) + vif_devices = [tap_device, qvb_device, qbr_device, qvo_device] + return map(lambda vif: self.setup_listener(vif, filter), vif_devices) + + def set_nonblock(self, listener): + listener.setnonblock(True) diff --git a/steth/agent/drivers/scapy_driver.py b/steth/agent/drivers/scapy_driver.py new file mode 100644 index 0000000..2ceec2b --- /dev/null +++ b/steth/agent/drivers/scapy_driver.py @@ -0,0 +1,55 @@ +# Copyright 2016 UnitedStack, Inc. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import struct +from scapy import all as scapy +from steth.agent.common import log +from steth.agent.common import utils +from steth.agent.common import constants + +LOG = log.get_logger() +scapy.conf.checkIPaddr = False + + +class ScapyDriver(object): + + def send_dhcp_over_qvb(self, port_id, port_mac): + """Send DHCP Discovery over qvb device. + """ + qvb_device = utils.get_vif_name(constants.QVB_DEVICE_PREFIX, port_id) + ethernet = scapy.Ether(dst='ff:ff:ff:ff:ff:ff', + src=port_mac, type=0x800) + ip = scapy.IP(src='0.0.0.0', dst='255.255.255.255') + udp = scapy.UDP(sport=68, dport=67) + port_mac_t = tuple(map(lambda x: int(x, 16), port_mac.split(':'))) + hw = struct.pack('6B', *port_mac_t) + bootp = scapy.BOOTP(chaddr=hw, flags=1) + dhcp = scapy.DHCP(options=[("message-type", "discover"), "end"]) + packet = ethernet / ip / udp / bootp / dhcp + scapy.sendp(packet, iface=qvb_device) + + def get_dhcp_mt(self, buff): + """Pick out DHCP Message Type from buffer. + """ + ether_packet = scapy.Ether(buff) + dhcp_packet = ether_packet[scapy.DHCP] + # ('message-type', 1) + message = dhcp_packet.options[0] + return constants.DHCP_MESSATE_TYPE[message[1]] + + def get_arp_op(self, buff): + ether_packet = scapy.Ether(buff) + arp_packet = ether_packet[scapy.ARP] + return constants.ARP_OP_TYPE[arp_packet.op] diff --git a/steth/tests/unit/agent/common/test_utils.py b/steth/tests/unit/agent/common/test_utils.py index 3cec41d..1c3da50 100644 --- a/steth/tests/unit/agent/common/test_utils.py +++ b/steth/tests/unit/agent/common/test_utils.py @@ -28,10 +28,6 @@ class TestUtils(unittest.TestCase): open(self.test_file, 'w+').close() self.pids = list() - def tearDown(self): - for pid in self.pids: - utils.kill_process_by_id(pid) - def test_execute(self): expected = "%s\n" % self.test_file code, result = utils.execute(["ls", self.test_file]) @@ -51,7 +47,8 @@ class TestUtils(unittest.TestCase): para['data']) self.assertEqual(para, result) - def test_get_interface(self): + @mock.patch('steth.agent.common.utils.execute') + def test_get_interface(self, execute): # test centos 6.5 platform.linux_distribution = mock.Mock(return_value=['', '6.5', '']) out = ['eth0 Link encap:Ethernet HWaddr FA:16:3E:61:F2:CF', @@ -62,7 +59,7 @@ class TestUtils(unittest.TestCase): 'TX packets:10163 errors:0 dropped:0 overruns:0 carrier:0', 'collisions:0 txqueuelen:1000', 'RX bytes:19492218 (18.5 MiB) TX bytes:1173768 (1.1 MiB)'] - utils.execute = mock.Mock(return_value=(0, out)) + execute.return_value = (0, out) self.assertEqual(utils.get_interface('eth0')[0], 0) # test centos 7.0 platform.linux_distribution = mock.Mock(return_value=['', '7.0', '']) @@ -74,7 +71,7 @@ class TestUtils(unittest.TestCase): 'RX errors 0 dropped 0 overruns 0 frame 0', 'TX packets 275332 bytes 91891644 (87.6 MiB)', 'TX errors 0 dropped 0 overruns 0 carrier 0 collisions 0'] - utils.execute = mock.Mock(return_value=(0, out)) + execute.return_value = (0, out) self.assertEqual(utils.get_interface('eth0')[0], 0) # test other distribution platform.linux_distribution = mock.Mock(return_value=['', '6.6', '']) diff --git a/steth/tests/unit/agent/drivers/test_pcap.py b/steth/tests/unit/agent/drivers/test_pcap.py new file mode 100644 index 0000000..6624d32 --- /dev/null +++ b/steth/tests/unit/agent/drivers/test_pcap.py @@ -0,0 +1,54 @@ +# Copyright 2016 UnitedStack, Inc. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import mock +import unittest +from steth.agent.drivers import pcap_driver +from steth.agent.common import utils +from steth.agent.common import constants + + +class TestPcapDriver(unittest.TestCase): + def setUp(self): + """Get interfaces on host for check. + """ + self.iface = 'eth0' + self.filter = '(tcp and port 80)' + self.pcap_dri = pcap_driver.PcapDriver() + + @mock.patch('pcap.pcap') + def test_setup_listener(self, pcap): + self.pcap_dri.setup_listener(self.iface, self.filter) + pcap.assert_called_with(self.iface) + pcap(self.iface).setfilter.assert_called_with(self.filter) + + @mock.patch('steth.agent.drivers.pcap_driver.PcapDriver.setup_listener') + def test_setup_listener_on_comp(self, setup_listener): + port_id = '27a9a962-8049-48c3-b77f-0653f8ee34df' + listeners = self.pcap_dri.setup_listener_on_comp(port_id, self.filter) + tap_device = utils.get_vif_name(constants.TAP_DEVICE_PREFIX, port_id) + qvb_device = utils.get_vif_name(constants.QVB_DEVICE_PREFIX, port_id) + qbr_device = utils.get_vif_name(constants.QBR_DEVICE_PREFIX, port_id) + qvo_device = utils.get_vif_name(constants.QVO_DEVICE_PREFIX, port_id) + vif_devices = [tap_device, qvb_device, qbr_device, qvo_device] + map(lambda vif: setup_listener.assert_any_call(vif, self.filter), + vif_devices) + self.assertEqual(len(listeners), 4) + + @mock.patch('pcap.pcap') + def test_set_nonblock(self, pcap): + listener = self.pcap_dri.setup_listener(self.iface, self.filter) + self.pcap_dri.set_nonblock(listener) + pcap(self.iface).setnonblock.assert_called_with(True) diff --git a/steth/tests/unit/agent/drivers/test_scapy.py b/steth/tests/unit/agent/drivers/test_scapy.py new file mode 100644 index 0000000..e347359 --- /dev/null +++ b/steth/tests/unit/agent/drivers/test_scapy.py @@ -0,0 +1,38 @@ +# Copyright 2016 UnitedStack, Inc. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import mock +import unittest +from scapy import all as scapy +from steth.agent.drivers import scapy_driver +from steth.agent.common import constants + + +class TestScapyDriver(unittest.TestCase): + def setUp(self): + self.scapy_dri = scapy_driver.ScapyDriver() + + @mock.patch('scapy.all.sendp') + def test_send_dhcp_over_qvb(self, sendp): + port_id = '27a9a962-8049-48c3-b77f-0653f8ee34df' + port_mac = 'fa:16:3e:18:fd:f7' + self.scapy_dri.send_dhcp_over_qvb(port_id, port_mac) + self.assertTrue(sendp.called, True) + + def test_get_dhcp_mt(self): + dhcp = scapy.DHCP(options=[("message-type", "discover"), "end"]) + pkt = scapy.Ether() / scapy.IP() / scapy.UDP() / scapy.BOOTP() / dhcp + message = self.scapy_dri.get_dhcp_mt(str(pkt)) + self.assertIn(message, constants.DHCP_MESSATE_TYPE) diff --git a/steth/tests/unit/agent/test_api.py b/steth/tests/unit/agent/test_api.py index 050d1ee..fa7ca89 100644 --- a/steth/tests/unit/agent/test_api.py +++ b/steth/tests/unit/agent/test_api.py @@ -23,26 +23,19 @@ class TestApi(unittest.TestCase): def setUp(self): self.agent_api = api.AgentApi() - def test_check_ports_on_br(self): - agent_utils.execute = mock.Mock(return_value=(0, ['execute'])) - agent_utils.make_response = mock.Mock(return_value=dict()) - self.agent_api.check_ports_on_br() - self.assertEqual(agent_utils.execute.called, True) - self.assertEqual(agent_utils.make_response.called, True) - agent_utils.execute = mock.Mock(return_value=(1, ['execute'])) - self.agent_api.check_ports_on_br() - self.assertEqual(agent_utils.make_response.called, True) + @mock.patch('steth.agent.common.utils.execute') + def test_check_ports_on_br(self, execute): + execute.return_value = (0, ['']) + result = self.agent_api.check_ports_on_br('br-ex', 'eth3') + self.assertEqual(execute.called, True) + self.assertEqual(result['code'], 0) - def test_ping(self): + @mock.patch('steth.agent.common.utils.execute') + def test_ping(self, execute): stdout = ['', '2 packets transmitted, 2 received, 0% packet loss', ''] - agent_utils.execute = mock.Mock(return_value=(0, stdout)) - agent_utils.make_response = mock.Mock(return_value=dict()) - self.agent_api.ping(['1.2.4.8', '1.2.4.9']) - self.assertEqual(agent_utils.make_response.called, True) - stdout = 'stdout' - agent_utils.execute = mock.Mock(return_value=(0, stdout)) - self.agent_api.ping(['1.2.4.8', '1.2.4.9']) - self.assertEqual(agent_utils.make_response.called, True) + execute.return_value = (0, stdout) + result = self.agent_api.ping(['1.2.4.8', '1.2.4.9']) + self.assertEqual(result['code'], 0) def test_get_interface(self): get_interface = mock.Mock(return_value=(0, '', dict())) @@ -50,42 +43,66 @@ class TestApi(unittest.TestCase): self.agent_api.get_interface() self.assertEqual(agent_utils.get_interface.called, True) - def test_set_link(self): + @mock.patch('steth.agent.common.utils.execute') + def test_set_link(self, execute): stdout = ['', ''] - agent_utils.execute = mock.Mock(return_value=(0, stdout)) - self.agent_api.setup_link('eth0', '10.0.0.100/24') - self.assertEqual(agent_utils.make_response.called, True) - agent_utils.execute = mock.Mock(return_value=(1, stdout)) - self.agent_api.setup_link('eth0', '10.0.0.100/24') - self.assertEqual(agent_utils.make_response.called, True) + execute.return_value = (0, stdout) + result = self.agent_api.setup_link('eth0', '10.0.0.100/24') + self.assertEqual(result['code'], 0) - def test_teardown_link(self): + @mock.patch('steth.agent.common.utils.execute') + def test_teardown_link(self, execute): stdout = ['', ''] - agent_utils.execute = mock.Mock(return_value=(0, stdout)) - self.agent_api.teardown_link('eth0') - self.assertEqual(agent_utils.make_response.called, True) - agent_utils.execute = mock.Mock(return_value=(1, stdout)) - self.agent_api.teardown_link('eth0') - self.assertEqual(agent_utils.make_response.called, True) + execute.return_value = (0, stdout) + result = self.agent_api.teardown_link('eth0') + self.assertEqual(result['code'], 0) + execute.return_value = (1, stdout) + result = self.agent_api.teardown_link('eth0') + self.assertEqual(result['code'], 1) - def test_start_iperf_client(self): - agent_utils.create_deamon = mock.Mock(return_value=100) - self.agent_api.setup_iperf_server('UDP') - self.assertEqual(agent_utils.make_response.called, True) + @mock.patch('steth.agent.common.utils.create_deamon') + def test_start_iperf_server(self, create_deamon): + create_deamon.return_value = 100 + result = self.agent_api.setup_iperf_server('UDP') + self.assertEqual(result['code'], 0) - def test_teardown_iperf_server(self): - agent_utils.kill_process_by_id = mock.Mock() - self.agent_api.setup_iperf_server(100) - self.assertEqual(agent_utils.make_response.called, True) + @mock.patch('steth.agent.common.utils.kill_process_by_id') + def test_teardown_iperf_server(self, kill_process_by_id): + result = self.agent_api.setup_iperf_server(100) + self.assertEqual(result['code'], 0) - def test_start_client(self): + @mock.patch('steth.agent.common.utils.execute_wait') + def test_start_iperf_client(self, execute_wait): stdout = '[ 3] 0.0- 3.0 sec 497 MBytes 1.39 Gbits/sec' - agent_utils.execute_wait = mock.Mock(return_value=(0, stdout, '')) - self.agent_api.start_iperf_client(host='127.0.0.1') - self.assertEqual(agent_utils.make_response.called, True) + execute_wait.return_value = (0, stdout, '') + result = self.agent_api.start_iperf_client(host='127.0.0.1') + self.assertEqual(result['code'], 0) - def test_validate_ip(self): + @mock.patch('steth.agent.common.utils.execute') + def test_validate_ip(self, execute): stdout = ['', ''] - agent_utils.execute = mock.Mock(return_value=(0, stdout)) - self.agent_api.validate_ip('1.2.3.4') - self.assertEqual(agent_utils.make_response.called, True) + execute.return_value = (0, stdout) + result = self.agent_api.validate_ip('1.2.3.4') + self.assertEqual(result['code'], 0) + + @mock.patch('steth.agent.drivers.pcap_driver.PcapDriver') + @mock.patch('steth.agent.drivers.scapy_driver.ScapyDriver') + def test_check_dhcp_on_comp(self, PcapDriver, ScapyDriver): + port_id = '27a9a962-8049-48c3-b77f-0653f8ee34df' + port_mac = 'fa:16:3e:18:fd:f7' + phy_iface = 'eth3' + net_type = 'vlan' + result = self.agent_api.check_dhcp_on_comp(port_id, port_mac, + phy_iface, net_type) + self.assertEqual(result['code'], 0) + + @mock.patch('steth.agent.drivers.pcap_driver.PcapDriver') + @mock.patch('steth.agent.drivers.scapy_driver.ScapyDriver') + def test_check_dhcp_on_comp_vxlan(self, PcapDriver, ScapyDriver): + port_id = '27a9a962-8049-48c3-b77f-0653f8ee34df' + port_mac = 'fa:16:3e:18:fd:f7' + phy_iface = 'eth3' + net_type = 'vxlan' + self.agent_api.check_dhcp_on_comp(port_id, port_mac, + phy_iface, net_type) + self.assertRaises(Exception()) diff --git a/test-requirements.txt b/test-requirements.txt index 3a4bb01..51910ce 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1,4 +1,4 @@ -mock +mock==1.0.1 flake8 unittest2 nose