summaryrefslogtreecommitdiff
path: root/cesar/test_general/station
diff options
context:
space:
mode:
authorThierry Carré2011-11-22 12:14:19 +0100
committerNélio Laranjeiro2011-11-28 17:12:49 +0100
commit47503273b2f07b353b3fe1a3875db301b3042d91 (patch)
treeacdc4b5fc9ad6d052bd7d559ff72d2f09fd292bf /cesar/test_general/station
parentef22508e3fb57ed84ce3e944e47a4b984a6364a7 (diff)
cesar: update and move cco/s2 cm_nw_info test to scenario
Diffstat (limited to 'cesar/test_general/station')
-rw-r--r--cesar/test_general/station/cco0/s2/py/sc12_cm_nw_info.py167
-rw-r--r--cesar/test_general/station/scenario/Makefile3
-rw-r--r--cesar/test_general/station/scenario/py/sc11_cm_nw_info.py92
-rw-r--r--cesar/test_general/station/scenario/py/sc13_data_rate.py2
4 files changed, 94 insertions, 170 deletions
diff --git a/cesar/test_general/station/cco0/s2/py/sc12_cm_nw_info.py b/cesar/test_general/station/cco0/s2/py/sc12_cm_nw_info.py
deleted file mode 100644
index 0d8a6a1ad7..0000000000
--- a/cesar/test_general/station/cco0/s2/py/sc12_cm_nw_info.py
+++ /dev/null
@@ -1,167 +0,0 @@
-#!/usr/bin/python
-
-##############################################################################
-# CM_NW_INFO #
-# ----------------------------------------------------------------------- #
-# build /home/mercadie/sources/work/cesar/maximus/python/Makefile before #
-# executing this. #
-# Objective: #
-##############################################################################
-
-import unittest
-import sys
-import re
-import struct
-
-sys.path.append ('../../../../maximus/python/tools/csi/')
-sys.path.append ('../../../../maximus/python/obj/')
-sys.path.append ('../../../../maximus/python/')
-sys.path.append ('../../../../maximus/python/lib/cesar')
-
-from maximus.mme.mme import MME
-from maximus.mme.mmheader import MMHeader
-from maximus.simu.rx import recv
-from struct import pack, unpack
-from csicore import *
-
-# Csi call
-csi = csiCore (1234)
-csi.process_init (sys.argv + ['-e', 'obj/cco0s2.elf'])
-
-# Common def
-def data_msdu_filter (msdu):
- if msdu.get_type () is 'ETHERNET_TYPE_MME':
- # The received object is an MSDU
- return True
- else:
- return False
-
-def send_mme_CM_NW_INFO (self, sta):
- global csi
-
- # Create and send mme message
-
- dst_mac = sta.get_mac_addr()
- src_mac = "10:12:13:14:15:16"
- MMTYPE_CM_NW_INFO = 0x6038
- staRole = [ "STA" , "PCO" , "CCO" ]
-
- mme = MME (MMHeader=MMHeader (dst_mac, src_mac,
- MTYPE = 0x88E1,
- MMV = 1,
- MMTYPE = MMTYPE_CM_NW_INFO,
- FMI = 0),
- MMEntry="")
- mme.send (csi.get_maximus(), sta.get_sta_cesar ().get ())
-
- # Wait for response
- rsp = recv (csi.get_maximus (), count = 1, filter = data_msdu_filter, timeout = 50000000)
-
- # Fail in case of no response
- self.failUnless (rsp != None)
-
- rsp = rsp[0]
- entry = rsp.get_mmentry()
- head = rsp.get_mmheader()
-
- # display header
- oda = ':'.join ("%02x" % i for i in unpack ("6B", head[0:6]))
- osa = ':'.join ("%02x" % i for i in unpack ("6B", head[6:12]))
- mtype = unpack ("H", head[12:14])[0] # no VLAN Tag
- mmv = unpack ("B", head[14:15])[0]
- mmtype = unpack ("H", head[15:17])[0]
- fmi = unpack ("H", head[17:19])[0]
-
- # Check that mmtype is correct (expected 0x6039)
- self.failUnless (mmtype == MMTYPE_CM_NW_INFO + 1)
-
- CCo_MACAddr = ':'.join ("%02x" % i for i in unpack ("6B", entry[11:17]))
- StationRole = staRole[unpack ("B", entry[10:11])[0]]
- NumNws = unpack ("B", entry[0:1])[0]
-
- print "\nmme header(hex): |%s|%s|%04x|%02x|%04x|%04x|" % (oda, osa, mtype, mmv, mmtype, fmi)
- print "\n| NumNws:0x%02x |" % (NumNws),
- if NumNws:
- print "NID:%s |" % (hex(unpack ('Q', entry[1:8] + '\0')[0])),
- print "SNID:0x%02x |" % (unpack ("B", entry[8:9])[0]),
- print "TEI:0x%02x |" % (unpack ("B", entry[9:10])[0]),
- print "StationRole:%s |" % (StationRole),
- print "CCo_MACAddr:%s |" % (CCo_MACAddr),
- print "Access:0x%02x |" % (unpack ("B", entry[17:18])[0])
- else:
- print "\n"
-
- return (CCo_MACAddr, StationRole, NumNws)
-
-
-class TestInitFunctions (unittest.TestCase):
- def setUp (self):
- pass
-
- def tearDown (self):
- pass
-
- def test_CM_CW_INFO (self):
- global csi
-
- cco_mac = "20:22:23:24:25:26"
- sta_mac = "30:32:33:34:35:36"
-
- # Creating an AVLN.
- avln = csi.avln_add ("Homeplug_AVLN", "AVLN")
-
- # adding a station
- target_sta = avln.sta_add (sta_mac, False, False, "authenticated_avln",
- "spidcom_spc300_sta2", "station2", 1, False)
- cco_sta = avln.sta_add (cco_mac, True, True, "authenticated_avln",
- "spidcom_spc300_sta3", "station3", 1, False)
-
- csi.process_avln_launch (avln)
-
- ##############################################################################
- # Test 1 #
- # ----------------------------------------------------------------------- #
- # * Two station recorded but not yet authenticated #
- # Check that there is no network associated yet #
- ##############################################################################
- (mac, role, numNw) = send_mme_CM_NW_INFO (self, target_sta)
-
- # Check that there is no network
- self.failUnless (numNw == 0)
-
- csi.process_wait_association ()
- csi.process_wait_authentication ()
-
- ##############################################################################
- # Test 2 #
- # ----------------------------------------------------------------------- #
- # * Two station recorded #
- # The target station is not the CCo #
- # We check that we receive a correct answer (especially the CCo mac) #
- ##############################################################################
- (mac, role, numNw) = send_mme_CM_NW_INFO (self, target_sta)
- self.failUnless (role == "STA")
- self.failUnless (numNw == 1)
- self.failUnless (mac == cco_mac)
-
- ##############################################################################
- # Test 3 #
- # ----------------------------------------------------------------------- #
- # * Two station recorded #
- # The target station is the CCo #
- # We check that we receive a correct answer (especially the CCo mac) #
- ##############################################################################
- (mac, role, numNw) = send_mme_CM_NW_INFO (self, cco_sta)
- self.failUnless (mac == cco_mac)
- self.failUnless (role == "CCO")
- self.failUnless (numNw == 1)
-
- csi.process_avln_remove (avln)
-
-
-suite = unittest.TestLoader ().loadTestsFromTestCase (TestInitFunctions)
-
-testResult = unittest.TextTestRunner (verbosity=2).run (suite)
-
-# For nightly build errors
-sys.exit ((1, 0)[testResult.wasSuccessful ()])
diff --git a/cesar/test_general/station/scenario/Makefile b/cesar/test_general/station/scenario/Makefile
index 5e1de2cc8a..975d2af0ad 100644
--- a/cesar/test_general/station/scenario/Makefile
+++ b/cesar/test_general/station/scenario/Makefile
@@ -16,7 +16,8 @@ testbook: py/sc01_assoc_auth.py py/sc02_stas_communication.py \
py/sc05_cc_leave.py py/sc06_discover_procedure.py \
py/sc07_bridge.py py/sc08_bentry_change.py \
py/sc09_simple_connect.py py/sc10_short_messages.py \
- py/sc13_data_rate.py py/sc14_igmp.py
+ py/sc11_cm_nw_info.py \
+ py/sc13_data_rate.py py/sc14_igmp.py py/sc15_change_nmk.py
python testbook.py $^ > $@.rst
CLEAN_FILES += testbook.rst
diff --git a/cesar/test_general/station/scenario/py/sc11_cm_nw_info.py b/cesar/test_general/station/scenario/py/sc11_cm_nw_info.py
new file mode 100644
index 0000000000..7d407d84b1
--- /dev/null
+++ b/cesar/test_general/station/scenario/py/sc11_cm_nw_info.py
@@ -0,0 +1,92 @@
+#!/usr/bin/python
+
+#############################################################################
+# Copyright (C) 2011 Spidcom
+#############################################################################
+
+import sys
+sys.path.append ('py')
+from scenario_init import *
+
+class Test_CM_NW_INFO (unittest.TestCase):
+ """test the CM_NW_INFO.CNF received for different destination (sta, cco).
+ """
+
+ def setUp (self):
+ self.csi = csiCore (1234)
+ self.csi.process_init (args)
+ self.maximus = self.csi.get_maximus ()
+
+ def tearDown (self):
+ del self.maximus
+ self.csi.process_uninit ()
+ del self.csi
+
+ def __sendnrecv_cm_nw_info (self, sta):
+ """Send a CM_NW_INFO.REQ MME to the sta and read the confirmation."""
+ self.computer_mac = "10:10:10:10:10:10"
+ req = Ether (dst = sta.get_mac_addr (), src = self.computer_mac) \
+ / scammer.MME (mmtype = scammer.HPAV_MMTYPES ['CM_NW_INFO_REQ']) \
+ / scammer.CM_NW_INFO_REQ ()
+ rsp = sendnrecv_mme (self.csi.get_maximus (), sta, req)
+ self.failUnless (rsp)
+ return Ether (rsp[0].get ())
+
+ def test_cm_nw_info_to_cco (self):
+ """Two stations authenticated, the target is the CCo.
+
+ Excepted results: We check that we receive a correct answer
+ (especially the CCo mac)
+ """
+ self.csi.init_test_bed ([2])
+ avln = self.csi.avln_get(0)
+ cco = avln.get_cco (self.maximus)
+ cnf = self.__sendnrecv_cm_nw_info (cco)
+
+ nwinfo = cnf.getlayer (3)
+ self.failUnless (cnf.numnws == 1)
+ self.failUnless (nwinfo.stationrole == 2)
+ self.failUnless (nwinfo.mac == cco.get_mac_addr())
+ self.csi.avln_remove (avln)
+ self.csi.uninit_test_bed ()
+
+ def test_cm_nw_info_to_sta (self):
+ """Two stations authenticated, the target station is not the CCo.
+
+ Excepted results: We check that we receive a correct answer
+ (especially the CCo mac)
+ """
+ self.csi.init_test_bed ([2])
+ avln = self.csi.avln_get(0)
+ sta = avln.get_stas (self.maximus)[0]
+ cco = avln.get_cco (self.maximus)
+ cnf = self.__sendnrecv_cm_nw_info (sta)
+
+ nwinfo = cnf.getlayer (3)
+ self.failUnless (cnf.numnws == 1)
+ self.failUnless (nwinfo.stationrole == 0)
+ self.failUnless (nwinfo.mac == cco.get_mac_addr())
+ self.csi.avln_remove (avln)
+ self.csi.uninit_test_bed ()
+
+ def test_cm_nw_info_empty (self):
+ """Station not authenticated or associated.
+
+ Excepted results: numnws field must indicate 0
+ """
+ npw = str (random.getrandbits(64))
+ ahfid = str (random.getrandbits(32))
+ avln = self.csi.avln_add (npw, ahfid)
+ avln.add_stas (1)
+ self.csi.process_avlns_launch ()
+ cco = avln.get_cco (self.maximus)
+ sta = avln.get_stas (self.maximus)[0]
+ cnf = self.__sendnrecv_cm_nw_info (sta)
+
+ self.failUnless (cnf.numnws == 0)
+ self.csi.avln_remove (avln)
+
+if __name__ == '__main__':
+ suite = unittest.TestLoader().loadTestsFromTestCase(Test_CM_NW_INFO)
+ testResult = unittest.TextTestRunner(verbosity=2).run(suite)
+ sys.exit ((1, 0)[testResult.wasSuccessful ()])
diff --git a/cesar/test_general/station/scenario/py/sc13_data_rate.py b/cesar/test_general/station/scenario/py/sc13_data_rate.py
index b8fef97fcb..da24f8722c 100644
--- a/cesar/test_general/station/scenario/py/sc13_data_rate.py
+++ b/cesar/test_general/station/scenario/py/sc13_data_rate.py
@@ -48,7 +48,6 @@ class TestDataRate (unittest.TestCase):
rsp = self.cm_nw_stats (sta1)
self.failUnless (rsp)
cm_nw_stats_cnf2 = Ether (rsp[0].get ())
- cm_nw_stats_cnf2.show ()
self.failUnless (cm_nw_stats_cnf == cm_nw_stats_cnf2)
del cm_nw_stats_cnf2
# Send data from station 1 to station 2.
@@ -63,7 +62,6 @@ class TestDataRate (unittest.TestCase):
rsp = self.cm_nw_stats (sta1)
self.failUnless (rsp)
cm_nw_stats_cnf2 = Ether (rsp[0].get ())
- cm_nw_stats_cnf2.show ()
self.failUnless (cm_nw_stats_cnf != cm_nw_stats_cnf2)
del cm_nw_stats_cnf2