| # |
| # Copyright 2016-present Ciena Corporation |
| # |
| # 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 unittest |
| from nose.tools import * |
| from nose.twistedtools import reactor, deferred |
| from twisted.internet import defer |
| from scapy.all import * |
| import time |
| import os |
| from DHCP import DHCPTest |
| from OnosCtrl import OnosCtrl |
| from OnosFlowCtrl import get_mac |
| from portmaps import g_subscriber_port_map |
| log.setLevel('INFO') |
| |
| class dhcprelay_exchange(unittest.TestCase): |
| |
| app = 'org.onosproject.dhcprelay' |
| app_dhcp = 'org.onosproject.dhcp' |
| relay_device_id = 'of:' + get_mac('ovsbr0') |
| relay_interface_port = 100 |
| relay_interfaces = (g_subscriber_port_map[relay_interface_port],) |
| interface_to_mac_map = {} |
| dhcp_data_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', 'setup') |
| default_config = { 'default-lease-time' : 600, 'max-lease-time' : 7200, } |
| default_options = [ ('subnet-mask', '255.255.255.0'), |
| ('broadcast-address', '192.168.1.255'), |
| ('domain-name-servers', '192.168.1.1'), |
| ('domain-name', '"mydomain.cord-tester"'), |
| ] |
| ##specify the IP for the dhcp interface matching the subnet and subnet config |
| ##this is done for each interface dhcpd server would be listening on |
| default_subnet_config = [ ('192.168.1.2', |
| ''' |
| subnet 192.168.1.0 netmask 255.255.255.0 { |
| range 192.168.1.10 192.168.1.100; |
| } |
| '''), ] |
| |
| @classmethod |
| def setUpClass(cls): |
| ''' Activate the dhcprelay app''' |
| OnosCtrl(cls.app_dhcp).deactivate() |
| time.sleep(3) |
| cls.onos_ctrl = OnosCtrl(cls.app) |
| status, _ = cls.onos_ctrl.activate() |
| assert_equal(status, True) |
| time.sleep(3) |
| ##start dhcpd initially with default config |
| cls.dhcpd_start() |
| cls.onos_dhcp_relay_load() |
| |
| @classmethod |
| def tearDownClass(cls): |
| '''Deactivate the dhcp relay app''' |
| try: |
| os.unlink('{}/dhcpd.conf'.format(cls.dhcp_data_dir)) |
| os.unlink('{}/dhcpd.leases'.format(cls.dhcp_data_dir)) |
| except: pass |
| cls.onos_ctrl.deactivate() |
| cls.dhcpd_stop() |
| |
| @classmethod |
| def onos_load_config(cls, config): |
| status, code = OnosCtrl.config(config) |
| if status is False: |
| log.info('JSON request returned status %d' %code) |
| assert_equal(status, True) |
| time.sleep(3) |
| |
| @classmethod |
| def onos_dhcp_relay_load(cls): |
| relay_device_map = '{}/{}'.format(cls.relay_device_id, cls.relay_interface_port) |
| dhcp_dict = {'apps':{'org.onosproject.dhcp-relay':{'dhcprelay': |
| {'dhcpserverConnectPoint':relay_device_map}}}} |
| cls.onos_load_config(dhcp_dict) |
| |
| @classmethod |
| def host_load(cls, iface): |
| '''Have ONOS discover the hosts for dhcp-relay responses''' |
| port = g_subscriber_port_map[iface] |
| host = '173.17.1.{}'.format(port) |
| cmds = ( 'ifconfig {} 0'.format(iface), |
| 'ifconfig {0} {1}'.format(iface, host), |
| 'arping -I {0} {1} -c 2'.format(iface, host), |
| 'ifconfig {} 0'.format(iface), ) |
| for c in cmds: |
| os.system(c) |
| |
| @classmethod |
| def dhcpd_conf_generate(cls, config = default_config, options = default_options, |
| subnet = default_subnet_config): |
| conf = '' |
| for k, v in config.items(): |
| conf += '{} {};\n'.format(k, v) |
| |
| opts = '' |
| for k, v in options: |
| opts += 'option {} {};\n'.format(k, v) |
| |
| subnet_config = '' |
| for _, v in subnet: |
| subnet_config += '{}\n'.format(v) |
| |
| return '{}{}{}'.format(conf, opts, subnet_config) |
| |
| @classmethod |
| def dhcpd_start(cls, intf_list = relay_interfaces, |
| config = default_config, options = default_options, |
| subnet = default_subnet_config): |
| '''Start the dhcpd server by generating the conf file''' |
| ##stop dhcpd if already running |
| cls.dhcpd_stop() |
| dhcp_conf = cls.dhcpd_conf_generate(config = config, options = options, |
| subnet = subnet) |
| ##first touch dhcpd.leases if it doesn't exist |
| lease_file = '{}/dhcpd.leases'.format(cls.dhcp_data_dir) |
| if os.access(lease_file, os.F_OK) is False: |
| with open(lease_file, 'w') as fd: pass |
| |
| conf_file = '{}/dhcpd.conf'.format(cls.dhcp_data_dir) |
| with open(conf_file, 'w') as fd: |
| fd.write(dhcp_conf) |
| |
| #now configure the dhcpd interfaces for various subnets |
| index = 0 |
| for ip,_ in subnet: |
| intf = intf_list[index] |
| index += 1 |
| os.system('ifconfig {} {}'.format(intf, ip)) |
| |
| intf_str = ','.join(intf_list) |
| dhcpd_cmd = '/usr/sbin/dhcpd -4 --no-pid -cf {0} -lf {1} {2}'.format(conf_file, lease_file, intf_str) |
| log.info('Starting DHCPD server with command: %s' %dhcpd_cmd) |
| ret = os.system(dhcpd_cmd) |
| assert_equal(ret, 0) |
| time.sleep(3) |
| cls.relay_interfaces = intf_list |
| |
| @classmethod |
| def dhcpd_stop(cls): |
| os.system('pkill -9 dhcpd') |
| for intf in cls.relay_interfaces: |
| os.system('ifconfig {} 0'.format(intf)) |
| |
| def get_mac(self, iface): |
| if self.interface_to_mac_map.has_key(iface): |
| return self.interface_to_mac_map[iface] |
| mac = get_mac(iface, pad = 0) |
| self.interface_to_mac_map[iface] = mac |
| return mac |
| |
| def send_recv(self, mac, update_seed = False, validate = True): |
| cip, sip = self.dhcp.discover(mac = mac, update_seed = update_seed) |
| if validate: |
| assert_not_equal(cip, None) |
| assert_not_equal(sip, None) |
| log.info('Got dhcp client IP %s from server %s for mac %s' % |
| (cip, sip, self.dhcp.get_mac(cip)[0])) |
| return cip,sip |
| |
| def test_dhcp_1request(self, iface = 'veth0'): |
| mac = self.get_mac(iface) |
| self.host_load(iface) |
| ##we use the defaults for this test that serves as an example for others |
| ##You don't need to restart dhcpd server if retaining default config |
| config = self.default_config |
| options = self.default_options |
| subnet = self.default_subnet_config |
| dhcpd_interface_list = self.relay_interfaces |
| self.dhcpd_start(intf_list = dhcpd_interface_list, |
| config = config, |
| options = options, |
| subnet = subnet) |
| self.dhcp = DHCPTest(seed_ip = '10.10.10.1', iface = iface) |
| self.send_recv(mac) |