mirror of
https://github.com/espressif/openthread.git
synced 2026-08-29 21:39:54 +00:00
[thread-cert] re-add test_mac_scan test (#7112)
This commit is contained in:
@@ -170,6 +170,7 @@ EXTRA_DIST = \
|
||||
test_ipv6_source_selection.py \
|
||||
test_lowpan.py \
|
||||
test_mac802154.py \
|
||||
test_mac_scan.py \
|
||||
test_mle.py \
|
||||
test_netdata_publisher.py \
|
||||
test_network_data.py \
|
||||
@@ -241,6 +242,7 @@ check_SCRIPTS = \
|
||||
test_ipv6_source_selection.py \
|
||||
test_lowpan.py \
|
||||
test_mac802154.py \
|
||||
test_mac_scan.py \
|
||||
test_mle.py \
|
||||
test_netdata_publisher.py \
|
||||
test_network_data.py \
|
||||
|
||||
@@ -1289,6 +1289,10 @@ class NodeImpl:
|
||||
self.send_command('extpanid %s' % extpanid)
|
||||
self._expect_done()
|
||||
|
||||
def get_extpanid(self):
|
||||
self.send_command('extpanid')
|
||||
return self._expect_result('[0-9a-fA-F]{16}')
|
||||
|
||||
def get_joiner_id(self):
|
||||
self.send_command('joiner id')
|
||||
return self._expect_result('[0-9a-fA-F]{16}')
|
||||
@@ -1922,12 +1926,30 @@ class NodeImpl:
|
||||
|
||||
self._expect('Conflict:', timeout=timeout)
|
||||
|
||||
def scan(self, result=1):
|
||||
def scan(self, result=1, timeout=10):
|
||||
self.send_command('scan')
|
||||
|
||||
self.simulator.go(timeout)
|
||||
|
||||
if result == 1:
|
||||
return self._expect_results(
|
||||
r'\|\s(\S+)\s+\|\s(\S+)\s+\|\s([0-9a-fA-F]{4})\s\|\s([0-9a-fA-F]{16})\s\|\s(\d+)')
|
||||
networks = []
|
||||
for line in self._expect_command_output()[2:]:
|
||||
_, J, networkname, extpanid, panid, extaddr, channel, dbm, lqi, _ = map(str.strip, line.split('|'))
|
||||
J = bool(int(J))
|
||||
panid = int(panid, 16)
|
||||
channel, dbm, lqi = map(int, (channel, dbm, lqi))
|
||||
|
||||
networks.append({
|
||||
'joinable': J,
|
||||
'networkname': networkname,
|
||||
'extpanid': extpanid,
|
||||
'panid': panid,
|
||||
'extaddr': extaddr,
|
||||
'channel': channel,
|
||||
'dbm': dbm,
|
||||
'lqi': lqi,
|
||||
})
|
||||
return networks
|
||||
|
||||
def ping(self, ipaddr, num_responses=1, size=8, timeout=5, count=1, interval=1, hoplimit=64, interface=None):
|
||||
args = f'{ipaddr} {size} {count} {interval} {hoplimit} {timeout}'
|
||||
|
||||
+17
-7
@@ -26,8 +26,7 @@
|
||||
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
||||
# POSSIBILITY OF SUCH DAMAGE.
|
||||
#
|
||||
|
||||
import time
|
||||
import logging
|
||||
import unittest
|
||||
|
||||
import thread_cert
|
||||
@@ -35,17 +34,22 @@ import thread_cert
|
||||
LEADER = 1
|
||||
ROUTER = 2
|
||||
|
||||
CHANNEL = 12
|
||||
|
||||
|
||||
class Test_MacScan(thread_cert.TestCase):
|
||||
USE_MESSAGE_FACTORY = False
|
||||
SUPPORT_NCP = False
|
||||
|
||||
TOPOLOGY = {
|
||||
LEADER: {
|
||||
'channel': 12,
|
||||
'channel': CHANNEL,
|
||||
'mode': 'rdn',
|
||||
'network_name': 'OpenThread',
|
||||
'allowlist': [ROUTER]
|
||||
},
|
||||
ROUTER: {
|
||||
'channel': 12,
|
||||
'channel': CHANNEL,
|
||||
'mode': 'rdn',
|
||||
'network_name': 'OpenThread',
|
||||
'allowlist': [LEADER]
|
||||
@@ -58,11 +62,17 @@ class Test_MacScan(thread_cert.TestCase):
|
||||
self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
|
||||
|
||||
self.nodes[ROUTER].start()
|
||||
time.sleep(5)
|
||||
self.simulator.go(10)
|
||||
self.assertEqual(self.nodes[ROUTER].get_state(), 'router')
|
||||
|
||||
results = self.nodes[LEADER].scan()
|
||||
self.assertEqual(len(results), 16)
|
||||
results = self.nodes[LEADER].scan(result=True)
|
||||
logging.info('scan result: %r', results)
|
||||
self.assertEqual(len(results), 1)
|
||||
network = results[0]
|
||||
self.assertEqual(network['extaddr'], self.nodes[ROUTER].get_addr64())
|
||||
self.assertEqual(network['extpanid'], self.nodes[ROUTER].get_extpanid())
|
||||
self.assertEqual(network['networkname'], self.nodes[ROUTER].get_network_name())
|
||||
self.assertEqual(network['channel'], CHANNEL)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
Reference in New Issue
Block a user