Merge pull request #6 from cgaonker/master
Adding DHCP test cases.
diff --git a/src/test/builder/buildFsm.sh b/src/test/builder/buildFsm.sh
index 51fd8d3..87bc14f 100644
--- a/src/test/builder/buildFsm.sh
+++ b/src/test/builder/buildFsm.sh
@@ -14,3 +14,7 @@
##Generate DNS test state machine
python yamlFsm.py -p DnsHolder -f noseDnsTest.yaml > ${odir}/noseDnsHolder.py
+#Generate EAP MD5 authentication state machine
+python yamlFsm.py -p Md5AuthHolder -f noseMd5AuthTest.yaml > ${odir}/noseMd5AuthHolder.py
+
+
diff --git a/src/test/dhcp/dhcpTest.py b/src/test/dhcp/dhcpTest.py
new file mode 100644
index 0000000..5be99b7
--- /dev/null
+++ b/src/test/dhcp/dhcpTest.py
@@ -0,0 +1,89 @@
+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, sys
+import copy
+CORD_TEST_UTILS = 'utils'
+test_root = os.getenv('CORD_TEST_ROOT') or './'
+sys.path.append(test_root + CORD_TEST_UTILS)
+from DHCP import DHCPTest
+from OnosCtrl import OnosCtrl
+
+log.setLevel('INFO')
+
+class dhcp_exchange(unittest.TestCase):
+
+ dhcp_server_config = {
+ "ip": "10.1.11.50",
+ "mac": "ca:fe:ca:fe:ca:fe",
+ "subnet": "255.255.252.0",
+ "broadcast": "10.1.11.255",
+ "router": "10.1.8.1",
+ "domain": "8.8.8.8",
+ "ttl": "63",
+ "delay": "2",
+ "startip": "10.1.11.51",
+ "endip": "10.1.11.100"
+ }
+
+ app = 'org.onosproject.dhcp'
+
+ def setUp(self):
+ ''' Activate the dhcp app'''
+ self.onos_ctrl = OnosCtrl(self.app)
+ status, _ = self.onos_ctrl.activate()
+ assert_equal(status, True)
+ time.sleep(3)
+
+ def teardown(self):
+ '''Deactivate the dhcp app'''
+ self.onos_ctrl.deactivate()
+
+ def onos_load_config(self, config):
+ status, code = self.onos_ctrl.config(config)
+ if status is False:
+ log.info('JSON request returned status %d' %code)
+ assert_equal(status, True)
+ time.sleep(2)
+
+ def onos_dhcp_table_load(self, config = None):
+ dhcp_dict = {'apps' : { 'org.onosproject.dhcp' : { 'dhcp' : copy.copy(self.dhcp_server_config) } } }
+ dhcp_config = dhcp_dict['apps']['org.onosproject.dhcp']['dhcp']
+ if config:
+ for k in config.keys():
+ if dhcp_config.has_key(k):
+ dhcp_config[k] = config[k]
+ self.onos_load_config(dhcp_dict)
+
+ def send_recv(self, update_seed = False):
+ cip, sip = self.dhcp.send(update_seed = update_seed)
+ 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'):
+ config = {'startip':'10.10.10.20', 'endip':'10.10.10.69',
+ 'ip':'10.10.10.2', 'mac': "ca:fe:ca:fe:ca:fe",
+ 'subnet': '255.255.255.0', 'broadcast':'10.10.10.255', 'router':'10.10.10.1'}
+ self.onos_dhcp_table_load(config)
+ self.dhcp = DHCPTest(seed_ip = '10.10.10.1', iface = iface)
+ self.send_recv()
+
+ def test_dhcp_Nrequest(self, iface = 'veth0'):
+ config = {'startip':'192.168.1.20', 'endip':'192.168.1.69',
+ 'ip':'192.168.1.2', 'mac': "ca:fe:ca:fe:cc:fe",
+ 'subnet': '255.255.255.0', 'broadcast':'192.168.1.255', 'router': '192.168.1.1'}
+ self.onos_dhcp_table_load(config)
+ self.dhcp = DHCPTest(seed_ip = '192.169.1.1', iface = iface)
+ ip_map = {}
+ for i in range(10):
+ cip, sip = self.send_recv(update_seed = True)
+ if ip_map.has_key(cip):
+ log.info('IP %s given out multiple times' %cip)
+ assert_equal(False, ip_map.has_key(cip))
+ ip_map[cip] = sip
diff --git a/src/test/igmp/igmpTest.py b/src/test/igmp/igmpTest.py
index 28c243a..bc3463b 100644
--- a/src/test/igmp/igmpTest.py
+++ b/src/test/igmp/igmpTest.py
@@ -6,7 +6,6 @@
import time, monotonic
import os, sys
import tempfile
-import json
import random
import threading
@@ -16,6 +15,7 @@
from IGMP import *
from McastTraffic import *
from Stats import Stats
+from OnosCtrl import OnosCtrl
log.setLevel('INFO')
IGMP_DST_MAC = "01:00:5e:00:01:01"
@@ -52,7 +52,26 @@
IGMP_TEST_TIMEOUT = 5
MCAST_TRAFFIC_TIMEOUT = 10
max_packets = 100
-
+ app = 'org.onosproject.igmp'
+
+ def setUp(self):
+ ''' Activate the dhcp app'''
+ self.onos_ctrl = OnosCtrl(self.app)
+ status, _ = self.onos_ctrl.activate()
+ assert_equal(status, True)
+ time.sleep(3)
+
+ def teardown(self):
+ '''Deactivate the dhcp app'''
+ self.onos_ctrl.deactivate()
+
+ def onos_load_config(self, config):
+ status, code = self.onos_ctrl.config(config)
+ if status is False:
+ log.info('JSON request returned status %d' %code)
+ assert_equal(status, True)
+ time.sleep(2)
+
def onos_ssm_table_load(self, groups, src_list):
ssm_dict = {'apps' : { 'org.onosproject.igmp' : { 'ssmTranslate' : [] } } }
ssm_xlate_list = ssm_dict['apps']['org.onosproject.igmp']['ssmTranslate']
@@ -62,15 +81,7 @@
d['source'] = s
d['group'] = g
ssm_xlate_list.append(d)
- json_dict = json.JSONEncoder().encode(ssm_dict)
- with tempfile.NamedTemporaryFile(delete=False) as temp:
- temp.write(json_dict)
- temp.flush()
- temp.close()
- log.debug('Loading SSM config in file %s to ONOS.' %temp.name)
- os.system('./igmp_ssm_load.sh %s' %temp.name)
- os.unlink(temp.name)
- ##Wait for ONOS to populate the SSM map before sending join.Huh
+ self.onos_load_config(ssm_dict)
time.sleep(2)
def igmp_verify_join(self, igmpStateList):
diff --git a/src/test/igmp/igmp_ssm_load.sh b/src/test/igmp/igmp_ssm_load.sh
deleted file mode 100755
index b2e9623..0000000
--- a/src/test/igmp/igmp_ssm_load.sh
+++ /dev/null
@@ -1,17 +0,0 @@
-#!/usr/bin/env bash
-json="$1"
-controller="$2"
-if [ x"$json" = "x" ]; then
- echo "No json file specified. Exiting"
- exit 127
-fi
-if [ x"$controller" = "x" ]; then
- controller=`ovs-vsctl show | egrep "Controller|tcp" | grep -v ptcp | sed 's,Controller,,g' | sed 's,\",,g' | tr -s ' '|awk -F":" '{print $2}'`
- #echo "Controller at $controller"
-fi
-#echo "Loading ssm translate json file $json to controller at $controller"
-curl --fail -sSL --user karaf:karaf \
- -X POST -H 'Content-Type:application/json' \
- http://$controller:8181/onos/v1/network/configuration/ -d@$json
-
-
diff --git a/src/test/utils/DHCP.py b/src/test/utils/DHCP.py
new file mode 100644
index 0000000..cafa95e
--- /dev/null
+++ b/src/test/utils/DHCP.py
@@ -0,0 +1,110 @@
+from scapy.all import *
+
+conf.verb = 0 # Disable Scapy verbosity
+conf.checkIPaddr = 0 # Don't check response packets for matching destination IPs
+
+class DHCPTest:
+
+ def __init__(self, seed_ip = '192.168.1.1', iface = 'veth0'):
+ self.seed_ip = seed_ip
+ self.seed_mac = self.ipToMac(self.seed_ip)
+ self.iface = iface
+ self.mac_map = {}
+ self.mac_inverse_map = {}
+
+ def is_mcast(self, ip):
+ mcast_octet = (atol(ip) >> 24) & 0xff
+ return True if mcast_octet >= 224 and mcast_octet <= 239 else False
+
+ def send(self, mac = None, update_seed = False):
+ '''Send a DHCP discover/offer'''
+
+ if mac is None:
+ mac = self.seed_mac
+ if update_seed:
+ self.seed_ip = self.incIP(self.seed_ip)
+ self.seed_mac = self.ipToMac(self.seed_ip)
+ mac = self.seed_mac
+
+ chmac = self.macToChaddr(mac)
+ L2 = Ether(dst="ff:ff:ff:ff:ff:ff", src=mac)
+ L3 = IP(src="0.0.0.0", dst="255.255.255.255")
+ L4 = UDP(sport=68, dport=67)
+ L5 = BOOTP(chaddr=chmac)
+ L6 = DHCP(options=[("message-type","discover"),"end"])
+ resp = srp1(L2/L3/L4/L5/L6, filter="udp and port 68", timeout=5, iface=self.iface)
+ try:
+ srcIP = resp.yiaddr
+ serverIP = resp.siaddr
+ except AttributeError:
+ print("Failed to acquire IP via DHCP for %s on interface %s" %(mac, self.iface))
+ return (None, None)
+
+ for x in resp.lastlayer().options:
+ if(x == 'end'):
+ break
+ op,val = x
+ if(op == "subnet_mask"):
+ subnet_mask = val
+ elif(op == 'server_id'):
+ server_id = val
+
+ L5 = BOOTP(chaddr=chmac, yiaddr=srcIP)
+ L6 = DHCP(options=[("message-type","request"), ("server_id",server_id),
+ ("subnet_mask",subnet_mask), ("requested_addr",srcIP), "end"])
+ srp1(L2/L3/L4/L5/L6, filter="udp and port 68", timeout=5, iface=self.iface)
+ self.mac_map[mac] = (srcIP, serverIP)
+ self.mac_inverse_map[srcIP] = (mac, serverIP)
+ return (srcIP, serverIP)
+
+ def send_next(self):
+ '''Send next dhcp discover/request with updated mac'''
+
+ return self.send(update_seed = True)
+
+ def macToChaddr(self, mac):
+ rv = []
+ mac = mac.split(":")
+ for x in mac:
+ rv.append(chr(int(x, 16)))
+ return reduce(lambda x,y: x + y, rv)
+
+ def get_ip(self, mac):
+ if self.mac_map.has_key(mac):
+ return self.mac_map[mac]
+ return (None, None)
+
+ def get_mac(self, ip):
+ if self.mac_inverse_map.has_key(ip):
+ return self.mac_inverse_map[ip]
+ return (None, None)
+
+ def ipToMac(self, ip):
+ '''Generate a mac from a ip'''
+
+ mcast = self.is_mcast(ip)
+ mac = "01:00:5e" if mcast == True else "00:00:00"
+ octets = ip.split(".")
+ for x in range(1,4):
+ num = str(hex(int(octets[x])))
+ num = num.split("x")[1]
+ if len(num) < 2:
+ num = "0" + str(num)
+ mac += ":" + num
+ return mac
+
+ def incIP(self, ip, n=1):
+ '''Increment an IP'''
+
+ if n < 1:
+ return ip
+ o = ip.split(".")
+ for ii in range(3,-1,-1):
+ if int(o[ii]) < 255:
+ o[ii] = str(int(o[ii]) + 1)
+ break
+ else:
+ o[ii] = str(0)
+
+ n -= 1
+ return self.incIP(".".join(o), n)
diff --git a/src/test/utils/EapPAP.py b/src/test/utils/EapPAP.py
index c936dc3..a40ae9d 100644
--- a/src/test/utils/EapPAP.py
+++ b/src/test/utils/EapPAP.py
@@ -65,20 +65,21 @@
self.nextEvent = self.PAPEventTable.EVT_EAP_PAP_USER_REQ
def _eapPAPUserReq(self):
- print 'Inside Challenge'
+ print 'UserReq Inside Challenge'
p = self.eapol_recv()
code, pkt_id, eaplen = unpack("!BBH", p[0:4])
print "Code %d, id %d, len %d" %(code, pkt_id, eaplen)
assert_equal(code, EAP_REQUEST)
reqtype = unpack("!B", p[4:5])[0]
reqdata = p[5:4+eaplen]
- assert_equal(reqtype, EAP_TYPE_MD5)
+ assert_equal(reqtype, EAP_TYPE_TLS)
print "<====== Send EAP Response with Password = %s ================>" % PAP_PASSWD
self.eapol_id_req(pkt_id, PAP_PASSWD)
- self.nextEvent = self.PAPEventTable.EVT_EAP_PAP_PASSWD_REQ
+ #self.nextEvent = self.PAPEventTable.EVT_EAP_PAP_PASSWD_REQ
+ self.nextEvent = self.PAPEventTable.EVT_EAP_PAP_DONE
def _eapPAPPassReq(self):
- print 'Inside Challenge'
+ print 'PassReq Inside Challenge'
p = self.eapol_recv()
code, pkt_id, eaplen = unpack("!BBH", p[0:4])
print "Code %d, id %d, len %d" %(code, pkt_id, eaplen)
diff --git a/src/test/utils/EapTLS.py b/src/test/utils/EapTLS.py
index 575fb20..d6b6b7e 100644
--- a/src/test/utils/EapTLS.py
+++ b/src/test/utils/EapTLS.py
@@ -11,6 +11,7 @@
import scapy
from nose.tools import *
from CordTestBase import CordTester
+import re
class TLSAuthTest(EapolPacket, CordTester):
@@ -89,4 +90,75 @@
print 'Inside EAP TLS Cert Req'
p = self.eapol_recv()
print 'Got TLS Cert Req with payload len: %d' %len(p)
+ code, pkt_id, eaplen = unpack("!BBH", p[0:4])
+ print "Code %d, id %d, len %d" %(code, pkt_id, eaplen)
+ assert_equal(code, EAP_REQUEST)
+ reqtype = unpack("!B", p[4:5])[0]
+ assert_equal(reqtype, EAP_TYPE_TLS)
+ rex_pem = re.compile(r'\-+BEGIN[^\-]+\-+(.*?)\-+END[^\-]+\-+', re.DOTALL)
+ self.pem_cert="""-----BEGIN CERTIFICATE-----
+MIIE4TCCA8mgAwIBAgIJANhJTS6x4B0iMA0GCSqGSIb3DQEBCwUAMIGTMQswCQYD
+VQQGEwJGUjEPMA0GA1UECBMGUmFkaXVzMRIwEAYDVQQHEwlTb21ld2hlcmUxFTAT
+BgNVBAoTDEV4YW1wbGUgSW5jLjEgMB4GCSqGSIb3DQEJARYRYWRtaW5AZXhhbXBs
+ZS5jb20xJjAkBgNVBAMTHUV4YW1wbGUgQ2VydGlmaWNhdGUgQXV0aG9yaXR5MB4X
+DTE0MDUyMDExNTkzNloXDTE0MDcxOTExNTkzNlowgZMxCzAJBgNVBAYTAkZSMQ8w
+DQYDVQQIEwZSYWRpdXMxEjAQBgNVBAcTCVNvbWV3aGVyZTEVMBMGA1UEChMMRXhh
+bXBsZSBJbmMuMSAwHgYJKoZIhvcNAQkBFhFhZG1pbkBleGFtcGxlLmNvbTEmMCQG
+A1UEAxMdRXhhbXBsZSBDZXJ0aWZpY2F0ZSBBdXRob3JpdHkwggEiMA0GCSqGSIb3
+DQEBAQUAA4IBDwAwggEKAoIBAQC/KUyltP0BS5A/sYg/XJOZMSHDIUiW+D8s1JgJ
+9Q/FIAnlMpevjPQtlmWi+hpgOUGgTryV+rTlzcUNw/gjmMs1Z4bAakFIc2vCPybw
+5hgKMU2E9SMgLr1aMVzwN3BH/njt1eWQ5Q9ajyu3JzmXZwOg/tV03L7BYpjLajhT
+iln4pvO/nq9YHVGurE5qCwyyrleYmtEXPi8MxrgudaKShrr7KgXbhlSwEaGGapSD
+JFhKvyQ4UZ56qiDFXD/AIXE9o8Soouv+8ufsCOyf/xKp1QkUaZ17Fe6YHqvQYdNM
+ovwnXnX+vRW0cZVui7ufxHncb9sJSAlovxzDy/GeL0SHtdH9AgMBAAGjggE0MIIB
+MDAdBgNVHQ4EFgQUHjtJ/Mjl+dcwmT5UI37N74qh2YUwgcgGA1UdIwSBwDCBvYAU
+HjtJ/Mjl+dcwmT5UI37N74qh2YWhgZmkgZYwgZMxCzAJBgNVBAYTAkZSMQ8wDQYD
+VQQIEwZSYWRpdXMxEjAQBgNVBAcTCVNvbWV3aGVyZTEVMBMGA1UEChMMRXhhbXBs
+ZSBJbmMuMSAwHgYJKoZIhvcNAQkBFhFhZG1pbkBleGFtcGxlLmNvbTEmMCQGA1UE
+AxMdRXhhbXBsZSBDZXJ0aWZpY2F0ZSBBdXRob3JpdHmCCQDYSU0useAdIjAMBgNV
+HRMEBTADAQH/MDYGA1UdHwQvMC0wK6ApoCeGJWh0dHA6Ly93d3cuZXhhbXBsZS5j
+b20vZXhhbXBsZV9jYS5jcmwwDQYJKoZIhvcNAQELBQADggEBAEbkq17kbT7X/oiy
+E2DOV7g1W8Au+TQR0GKzjXPgmYVGixN8l/9dQZ9WVDmCetBy71UHgTxPp20My2zr
+uA7hy9FYNGtZ2jmu0p019fH+CSCL7RHHgKsY63UsldT1qPYiWiyqbWy5GvJX778N
+GxVo7oN33se1c4KEmMOLVqQqX5dDWjN2r27l0GFh1ssx4RHqOc57G5Txq861i6UT
+KlrN0xpyu7LjcQGMwKbfzCXfwys5i4rrAVX1spILTIihUKpD6FYxp6oj+d4ELZOh
+br3zfhKrkbvPCG0gEziBLnwd11ZJELQfm89IYBhmoOYkk5+ZOszDsXKGWzfV6XSW
+ZRv+LU0=
+-----END CERTIFICATE-----"""
+ self.der_cert = rex_pem.findall(self.pem_cert)[0].decode("base64")
+ self.pem_priv_key = """-----BEGIN RSA PRIVATE KEY-----
+MIIEpAIBAAKCAQEA84TzkjbcskbKZnrlKcXzSSgi07n+4N7kOM7uIhzpkTuU0HIv
+h4VZS2axxfV6hV3CD9MuKVg2zEhroqK1Js5n4ke230nSP/qiELfCl0R+hzRtbfKL
+tFUr1iHeU0uQ6v3q+Tg1K/Tmmg72uxKrhyHDL7z0BriPjhAHJ5XlQsvR1RCMkqzu
+D9wjSInJxpMMIgLndOclAKv4D1wQtYU7ZpTw+01XBlUhIiXb86qpYL9NqnnRq5JI
+uhmOEuxo2ca63+xaHNhD/udSyc8C0Md/yX6wlONTRFgLLv0pdLUGm1xEjfsydaQ6
+qGd7hzIKUI3hohNKJa/mHLElv7SZolPTogK/EQIDAQABAoIBAADq9FwNtuE5IRQn
+zGtO4q7Y5uCzZ8GDNYr9RKp+P2cbuWDbvVAecYq2NV9QoIiWJOAYZKklOvekIju3
+r0UZLA0PRiIrTg6NrESx3JrjWDK8QNlUO7CPTZ39/K+FrmMkV9lem9yxjJjyC34D
+AQB+YRTx+l14HppjdxNwHjAVQpIx/uO2F5xAMuk32+3K+pq9CZUtrofe1q4Agj9R
+5s8mSy9pbRo9kW9wl5xdEotz1LivFOEiqPUJTUq5J5PeMKao3vdK726XI4Z455Nm
+W2/MA0YV0ug2FYinHcZdvKM6dimH8GLfa3X8xKRfzjGjTiMSwsdjgMa4awY3tEHH
+674jhAECgYEA/zqMrc0zsbNk83sjgaYIug5kzEpN4ic020rSZsmQxSCerJTgNhmg
+utKSCt0Re09Jt3LqG48msahX8ycqDsHNvlEGPQSbMu9IYeO3Wr3fAm75GEtFWePY
+BhM73I7gkRt4s8bUiUepMG/wY45c5tRF23xi8foReHFFe9MDzh8fJFECgYEA9EFX
+4qAik1pOJGNei9BMwmx0I0gfVEIgu0tzeVqT45vcxbxr7RkTEaDoAG6PlbWP6D9a
+WQNLp4gsgRM90ZXOJ4up5DsAWDluvaF4/omabMA+MJJ5kGZ0gCj5rbZbKqUws7x8
+bp+6iBfUPJUbcqNqFmi/08Yt7vrDnMnyMw2A/sECgYEAiiuRMxnuzVm34hQcsbhH
+6ymVqf7j0PW2qK0F4H1ocT9qhzWFd+RB3kHWrCjnqODQoI6GbGr/4JepHUpre1ex
+4UEN5oSS3G0ru0rC3U4C59dZ5KwDHFm7ffZ1pr52ljfQDUsrjjIMRtuiwNK2OoRa
+WSsqiaL+SDzSB+nBmpnAizECgYBdt/y6rerWUx4MhDwwtTnel7JwHyo2MDFS6/5g
+n8qC2Lj6/fMDRE22w+CA2esp7EJNQJGv+b27iFpbJEDh+/Lf5YzIT4MwVskQ5bYB
+JFcmRxUVmf4e09D7o705U/DjCgMH09iCsbLmqQ38ONIRSHZaJtMDtNTHD1yi+jF+
+OT43gQKBgQC/2OHZoko6iRlNOAQ/tMVFNq7fL81GivoQ9F1U0Qr+DH3ZfaH8eIkX
+xT0ToMPJUzWAn8pZv0snA0um6SIgvkCuxO84OkANCVbttzXImIsL7pFzfcwV/ERK
+UM6j0ZuSMFOCr/lGPAoOQU0fskidGEHi1/kW+suSr28TqsyYZpwBDQ==
+-----END RSA PRIVATE KEY-----
+ """
+ self.der_priv_key = rex_pem.findall(self.pem_priv_key)[0].decode("base64")
+ reqdata = TLSRecord(version="TLS_1_0")/TLSHandshake()/TLSCertificateList(
+ certificates=[TLSCertificate(data=x509.X509Cert(self.der_cert))])
+ #reqdata.show()
+ print "------> Sending Client Hello TLS Certificate payload of len %d ----------->" %len(reqdata)
+ eap_payload = self.eapTLS(EAP_RESPONSE, pkt_id, TLS_LENGTH_INCLUDED, str(reqdata))
+ self.eapol_send(EAPOL_EAPPACKET, eap_payload)
self.nextEvent = None
diff --git a/src/test/utils/EapolAAA.py b/src/test/utils/EapolAAA.py
index f7a8a2d..6c477f0 100644
--- a/src/test/utils/EapolAAA.py
+++ b/src/test/utils/EapolAAA.py
@@ -84,3 +84,26 @@
eap_payload = self.eap(EAP_RESPONSE, pkt_id, EAP_TYPE_ID, user)
return self.eapol_send(EAPOL_EAPPACKET, eap_payload)
+ def eap_md5_challenge_recv(self,rad_pwd):
+ PASS = rad_pwd
+ print 'Inside EAP MD5 Challenge Exchange'
+ p = self.s.recv(self.max_payload_size)[14:]
+ vers,pkt_type,eapollen = unpack("!BBH",p[:4])
+ print "EAPOL Version %d, type %d, len %d" %(vers, pkt_type, eapollen)
+ code, pkt_id, eaplen = unpack("!BBH", p[4:8])
+ print "EAP Code %d, id %d, len %d" %(code, pkt_id, eaplen)
+ assert_equal(code, EAP_REQUEST)
+ reqtype = unpack("!B", p[8:9])[0]
+ reqdata = p[9:4+eaplen]
+ print 'Request type is %d' %(reqtype)
+ assert_equal(reqtype, EAP_TYPE_MD5)
+ challenge=pack("!B",pkt_id)+PASS+reqdata[1:]
+ print "Generating md5 challenge for %s" % challenge
+ return (challenge,pkt_id)
+
+ def eap_Status(self):
+ print 'Inside EAP Status'
+ p = self.s.recv(self.max_payload_size)[14:]
+ code, id, eaplen = unpack("!BBH", p[4:8])
+ return code
+
diff --git a/src/test/utils/OnosCtrl.py b/src/test/utils/OnosCtrl.py
new file mode 100644
index 0000000..043438f
--- /dev/null
+++ b/src/test/utils/OnosCtrl.py
@@ -0,0 +1,32 @@
+import json
+import requests
+import os,sys,time
+
+class OnosCtrl:
+
+ def __init__(self, app, controller = None):
+ self.app = app
+ if controller is None:
+ self.controller = os.getenv('ONOS_CONTROLLER_IP') or 'localhost'
+ else:
+ self.controller = controller
+ self.app_url = 'http://%s:8181/onos/v1/applications/%s' %(self.controller, self.app)
+ self.cfg_url = 'http://%s:8181/onos/v1/network/configuration/' %(self.controller)
+ self.auth = ('karaf', 'karaf')
+
+ def config(self, config):
+ if config:
+ json_data = json.dumps(config)
+ resp = requests.post(self.cfg_url, auth = self.auth, data = json_data)
+ return resp.ok, resp.status_code
+ return False, 400
+
+ def activate(self):
+ resp = requests.post(self.app_url + '/active', auth = self.auth)
+ return resp.ok, resp.status_code
+
+ def deactivate(self):
+ resp = requests.delete(self.app_url + '/active', auth = self.auth)
+ return resp.ok, resp.status_code
+
+