summaryrefslogtreecommitdiff
path: root/cesar/test_general
diff options
context:
space:
mode:
authorNélio Laranjeiro2011-09-07 14:41:28 +0200
committerNélio Laranjeiro2011-11-28 17:12:19 +0100
commit332ed7203d6b1e72587536e654395d705e4e20b8 (patch)
tree30b0875efc400356c1826a10529939fa0af512cf /cesar/test_general
parenta89edab51fdb0a6b03d3d5f97e1207c650e82f61 (diff)
cesar/maximus/tools/csi: add snr function
Diffstat (limited to 'cesar/test_general')
-rw-r--r--cesar/test_general/station/cco0/s1/py/sc10_two_sta_with_snr.py206
1 files changed, 41 insertions, 165 deletions
diff --git a/cesar/test_general/station/cco0/s1/py/sc10_two_sta_with_snr.py b/cesar/test_general/station/cco0/s1/py/sc10_two_sta_with_snr.py
index 4f6666c74a..0c8dbd4d56 100644
--- a/cesar/test_general/station/cco0/s1/py/sc10_two_sta_with_snr.py
+++ b/cesar/test_general/station/cco0/s1/py/sc10_two_sta_with_snr.py
@@ -11,177 +11,53 @@
# Hint: The two station have the same NID. #
##############################################################################
-import sys
-sys.path.append('../../../../maximus/python/obj');
-sys.path.append('../../../../maximus/python');
-sys.path.append('../../../../maximus/python/lib/cesar');
-
-from interface import *
-from string import *
-from struct import *
-from maximus import *
-from sta_cesar import STACesar
-from own_data import Station_own_data
-from sniffer import *
-
+import os
import unittest
-
-def ms_to_tck (ms):
- return (ms * 25000)
-
-
-def my_mpdu_filter(mpdu):
- if mpdu.get_type() is 'PHY_TYPE_MPDU_PAYLOAD':
- # The received object is an MPDU
- return True
- else:
- return False
-
-def data_msdu_filter (msdu):
- if msdu.get_type () is 'ETHERNET_TYPE_DATA':
- # The received object is an MSDU
- return True
- else:
- return False
-
-# Initialise Maximus with the station.
-maximus = Maximus()
-maximus.init (sys.argv + ['-e', './obj/cco0s1.elf'])
-
-#Add noise on the medium.
-maximus.disturb_channel (True)
-set_snr (maximus, value = 1)
-
-realloc_buffer ( True )
-
-# Configure the first station.
-sta1_conf = Config ()
-sta1_conf.mac_address = (0x10, 0x11, 0x12, 0x13, 0x14, 0x15)
-sta1_conf.cco_preference = True
-sta1_conf.was_cco = True
-sta1_conf.npw = "HomePlugAV0123"
-sta1_conf.dpw = "STATION_DEVICE_PASSWORD"
-sta1_conf.m_sta_hfid = "HPAV_STATION"
-sta1_conf.u_sta_hfid = "STA1"
-sta1_conf.avln_hfid = "AVLN1"
-sta1_conf.tonemask = (85,139,167,214,225,282,302,409,419,569,591,736,748,856,882,1015,1027,1143,1535)
-sta1_conf.sl = 1
-sta1_conf.snid = None
-
-# Configure the second station.
-sta2_conf = Config ()
-sta2_conf.mac_address = (0x20, 0x21, 0x22, 0x23, 0x24, 0x25)
-sta2_conf.cco_preference = False
-sta2_conf.was_cco = False
-sta2_conf.npw = "HomePlugAV0123"
-sta2_conf.dpw = "STATION_DEVICE_PASSWORD"
-sta2_conf.m_sta_hfid = "HPAV_STATION"
-sta2_conf.u_sta_hfid = "STA2"
-sta2_conf.avln_hfid = "AVLN1"
-sta2_conf.tonemask = (85,139,167,214,225,282,302,409,419,569,591,736,748,856,882,1015,1027,1143,1535)
-sta2_conf.sl = 1
-sta2_conf.snid = None
-
-sta1_mac_addr = '10:11:12:13:14:15'
-sta2_mac_addr = '20:21:22:23:24:25'
-
-# Create the station
-print ""
-print "Starting the test ..."
-sta1 = STACesar (maximus, config=sta1_conf, mme_buffer_nb=1, debug=False, config_mode="MME", seed = 0x2345)
-sta2 = STACesar (maximus, config=sta2_conf, mme_buffer_nb=1, debug=False, config_mode="MME", seed = 0x1234)
-
-maximus.wait (50000000)
-
-class TestInitFunctions(unittest.TestCase):
+import sys
+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 csicore import *
+csi = csiCore (1234)
+MACSDEFAULT = "00:13:d7:00:0%x:%02x"
+# Creating an AVLN.
+avln1 = csi.avln_add ("Homeplug_AVLN1", "AVLN1")
+avln1_sta_nb = 2
+# Adding the stations.
+avln1_stas = list ()
+for i in range (0, avln1_sta_nb):
+ avln1_stas.append (avln1.sta_add (MACSDEFAULT % (1, i+1), False, False,
+ "HomePlug_AVLN1_Station%d" % (i+1),
+ "av1_spc300_sta%d" % (i+1),
+ "avln1_station%d" % (i+1), 1, False))
+csi.process_init (sys.argv + ['-e' , 'obj/cco0s1.elf'])
+csi.process_avlns_launch ()
+csi.process_wait_association (15)
+csi.process_wait_authentication (15)
+print "Association status on AVLN1"
+csi.authentication_status (avln1)
+
+class TestSNR(unittest.TestCase):
def setUp(self):
- pass
+ pass
def tearDown(self):
- pass
-
- def testAuthenticateSta1(self):
- sta_data = Station_own_data ()
- data = sta_data.get_data (maximus, sta1)
-
- self.failUnless (data.mac_address == sta1_conf.mac_address)
- self.failUnless (data.cco_prefered == sta1_conf.cco_preference)
- # The station become CCo so the was_cco status have evolved.
- self.failUnless (data.was_cco == True)
- self.failUnless (data.npw == sta1_conf.npw)
- self.failUnless (data.dpw == sta1_conf.dpw)
- self.failUnless (data.hfid_manufacturer == sta1_conf.m_sta_hfid)
- self.failUnless (data.hfid_user == sta1_conf.u_sta_hfid)
- self.failUnless (data.security_level == sta1_conf.sl)
- # The station is CCo and automatically authenticated.
- self.failUnless (data.tei != 0)
- self.failUnless (data.authenticated != 0)
- self.failUnless (data.is_cco != 0)
-
- def testAuthenticateSta2(self):
- sta_data = Station_own_data ()
- data = sta_data.get_data (maximus, sta2)
-
- self.failUnless (data.mac_address == sta2_conf.mac_address)
- self.failUnless (data.cco_prefered == sta2_conf.cco_preference)
- # The station become CCo so the was_cco status have evolved.
- self.failUnless (data.was_cco == sta2_conf.was_cco)
- self.failUnless (data.npw == sta2_conf.npw)
- self.failUnless (data.dpw == sta2_conf.dpw)
- self.failUnless (data.hfid_manufacturer == sta2_conf.m_sta_hfid)
- self.failUnless (data.hfid_user == sta2_conf.u_sta_hfid)
- self.failUnless (data.security_level == sta2_conf.sl)
- self.failUnless (data.tei != 0)
- self.failUnless (data.authenticated != 0)
- self.failUnless (data.is_cco == 0)
-
- def testDataTransmission1 (self):
- alloc_data_buffer (maximus, sta1)
- frame = Eth()
- frame.dst = sta1_mac_addr
- frame.src = sta2_mac_addr
- frame.type = 0x4F50
- frame.payload = "Hello world, this is only a test to send data over the PWL with two stations."
- frame.send (maximus, sta2)
-
- rsp = recv(maximus, count=1, filter=data_msdu_filter, timeout = 1000000000)
- self.failUnless (rsp != None)
- if rsp != None:
- rsp = rsp[0]
- self.failUnless (rsp.dst == frame.dst)
- self.failUnless (rsp.src == frame.src)
- self.failUnless (rsp.type == frame.type)
- self.failUnless (rsp.payload == frame.payload)
-
- def testDataTransmission2 (self):
- alloc_data_buffer (maximus, sta2)
- frame = Eth()
- frame.dst = sta2_mac_addr
- frame.src = sta1_mac_addr
- frame.type = 0x4F50
- frame.payload = "Hello I'm the station 1, this is only a test to send data over the PWL with two stations."
- frame.send (maximus, sta1)
-
- rsp = recv(maximus, count=1, filter=data_msdu_filter, timeout = 50000)
- self.failUnless (rsp != None)
- if rsp != None:
- rsp = rsp[0]
- self.failUnless (rsp.dst == frame.dst)
- self.failUnless (rsp.src == frame.src)
- self.failUnless (rsp.type == frame.type)
- self.failUnless (rsp.payload == frame.payload)
-
-suite = unittest.TestLoader().loadTestsFromTestCase(TestInitFunctions)
-
+ pass
+
+ def testSNR (self):
+ """Add SNR between the station and start a transmission."""
+ avln1.add_snr (csi.get_maximus(), 1.0)
+ packets = csi.avln_create_traffic (avln1, 10)
+ csi.process_data_send_traffic (packets)
+ csi.process_wait_sec (10)
+ csi.process_verify_transmission ()
+ self.failUnless (csi.process_end_get_result ())
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestSNR)
testResult = unittest.TextTestRunner(verbosity=2).run(suite)
-# Delete the station.
-print "Removing STA 1 ..."
-sta1.remove()
-print "Removing STA 2 ..."
-sta2.remove()
-
-
+csi.process_uninit ()
# For nightly build errors
sys.exit ((1, 0)[testResult.wasSuccessful ()])