espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-07-28 22:57:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

[srp-client] add lease changing test (#7280)

Browse Source
This commit is contained in:
Simon Lin
2022-01-18 08:52:03 -08:00
committed by GitHub
parent 395d502576
commit 167ffe9393
7 changed files with 197 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/node.py
+12 -1
View File
@@ -38,6 +38,7 @@ import subprocess
import sys
import time
import traceback
import typing
import unittest
from ipaddress import IPv6Address, IPv6Network
from typing import Union, Dict, Optional, List, Any
@@ -654,7 +655,7 @@ class NodeImpl:
if isinstance(pattern, str):
pattern = re.compile(pattern)
if isinstance(pattern, re.Pattern):
if isinstance(pattern, typing.Pattern):
return pattern.match(line)
else:
return any(NodeImpl._match_pattern(line, p) for p in pattern)
@@ -1130,6 +1131,16 @@ class NodeImpl:
service_lines = self._expect_command_output()
return [self._parse_srp_client_service(line) for line in service_lines]
def srp_client_set_lease_interval(self, leaseinterval: int):
cmd = f'srp client leaseinterval {leaseinterval}'
self.send_command(cmd)
self._expect_done()
def srp_client_get_lease_interval(self) -> int:
cmd = 'srp client leaseinterval'
self.send_command(cmd)
return int(self._expect_result('\d+'))
def _encode_txt_entry(self, entry):
"""Encodes the TXT entry to the DNS-SD TXT record format as a HEX string.
tests/scripts/thread-cert/pktverify/consts.py
+22
View File
@@ -310,6 +310,7 @@ REAL_LAYER_NAMES = {
'dns',
'igmp',
'mdns',
'dns',
}
FAKE_LAYER_NAMES = {'thread_nwd', 'thread_meshcop', 'ipv6inner'}
@@ -337,6 +338,27 @@ WIRESHARK_OVERRIDE_PREFS = {
WIRESHARK_DECODE_AS_ENTRIES = {
'udp.port==61631': 'coap',
# SRP server ports for dissecting SRP requests and responses
'udp.port==53535': 'dns',
'udp.port==53536': 'dns',
'udp.port==53537': 'dns',
'udp.port==53538': 'dns',
'udp.port==53539': 'dns',
'udp.port==53540': 'dns',
'udp.port==53541': 'dns',
'udp.port==53542': 'dns',
'udp.port==53543': 'dns',
'udp.port==53544': 'dns',
'udp.port==53545': 'dns',
'udp.port==53546': 'dns',
'udp.port==53547': 'dns',
'udp.port==53548': 'dns',
'udp.port==53549': 'dns',
'udp.port==53550': 'dns',
'udp.port==53551': 'dns',
'udp.port==53552': 'dns',
'udp.port==53553': 'dns',
'udp.port==53554': 'dns',
}
TIMEOUT_JOIN_NETWORK = 10
tests/scripts/thread-cert/pktverify/layer_fields.py
+5 -1
View File
@@ -638,7 +638,11 @@ _LAYER_FIELDS = {
# Thread Diagnostic
'thread_diagnostic.tlv.type': _list(_auto),
'thread_diagnostic.tlv.len8': _list(_auto),
'thread_diagnostic.tlv.general': _list(_str)
'thread_diagnostic.tlv.general': _list(_str),
# DNS
'dns.resp.ttl': _auto,
'dns.flags.response': _auto,
}
_layer_containers = set()
tests/scripts/thread-cert/pktverify/layers.py
+7
View File
@@ -179,3 +179,10 @@ class WpanLayer(Layer):
@property
def is_ack(self) -> bool:
return self.frame_type == 0x2
class DnsLayer(Layer):
"""
Represents the DNS layer of a packet.
"""
pass
tests/scripts/thread-cert/pktverify/packet.py
+5 -1
View File
@@ -38,7 +38,7 @@ from pktverify.addrs import EthAddr
from pktverify.coap import CoapLayer
from pktverify.consts import VALID_LAYER_NAMES
from pktverify.decorators import cached_property
from pktverify.layers import Layer, ThreadMeshcopLayer, Icmpv6Layer, WpanLayer, ThreadNetworkDataLayer
from pktverify.layers import Layer, ThreadMeshcopLayer, Icmpv6Layer, WpanLayer, ThreadNetworkDataLayer, DnsLayer
from pktverify.utils import make_filter_func
@@ -102,6 +102,10 @@ class Packet(object):
def thread_nwd(self) -> ThreadNetworkDataLayer:
return ThreadNetworkDataLayer(self._packet, 'thread_nwd')
@cached_property
def dns(self) -> DnsLayer:
return DnsLayer(self._packet, 'dns')
def __getattr__(self, layer_name: str) -> Layer:
real_layer_name = layer_name
tests/scripts/thread-cert/pktverify/utils.py
+1
View File
@@ -72,6 +72,7 @@ def make_filter_func(func: Union[str, Callable], **vars) -> Callable:
'thread_nwd': p.thread_nwd,
'thread_address': p.thread_address,
'thread_bcn': p.thread_bcn,
'dns': p.dns,
'null': nullField,
})
else:
tests/scripts/thread-cert/test_srp_client_change_lease.py
Executable
+145
View File
@@ -0,0 +1,145 @@
#!/usr/bin/env python3
#
# Copyright (c) 2021, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import unittest
import thread_cert
# Test description:
# This test verifies the SRP client lease changing works as expected.
#
# Topology:
# LEADER (SRP server)
# |
# |
# ROUTER (SRP client)
#
from pktverify.packet_filter import PacketFilter
SERVER = 1
CLIENT = 2
DEFAULT_LEASE_TIME = 7200
LEASE_TIME = 60
NEW_LEASE_TIME = 120
KEY_LEASE_TIME = 240
assert LEASE_TIME < KEY_LEASE_TIME
assert NEW_LEASE_TIME < KEY_LEASE_TIME
class SrpClientChangeLeaseTime(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
SUPPORT_NCP = False
TOPOLOGY = {
SERVER: {
'name': 'SRP_SERVER',
'networkkey': '00112233445566778899aabbccddeeff',
'mode': 'rdn',
},
CLIENT: {
'name': 'SRP_CLIENT',
'networkkey': '00112233445566778899aabbccddeeff',
'mode': 'rdn',
},
}
def test(self):
server = self.nodes[SERVER]
client = self.nodes[CLIENT]
#
# 0. Start the server and client devices.
#
server.srp_server_set_enabled(True)
server.srp_server_set_lease_range(LEASE_TIME, LEASE_TIME, KEY_LEASE_TIME, KEY_LEASE_TIME)
server.start()
self.simulator.go(5)
self.assertEqual(server.get_state(), 'leader')
self.simulator.go(5)
client.srp_server_set_enabled(False)
client.start()
self.simulator.go(10)
self.assertEqual(client.get_state(), 'router')
#
# 1. Register a single service and verify that it works.
#
client.srp_client_set_host_name('my-host')
client.srp_client_set_host_address('2001::1')
client.srp_client_start(server.get_addrs()[0], client.get_srp_server_port())
client.srp_client_add_service('my-service', '_ipps._tcp', 12345)
self.simulator.go(2)
self.check_host_and_service(server, client)
server.srp_server_set_lease_range(NEW_LEASE_TIME, NEW_LEASE_TIME, KEY_LEASE_TIME, KEY_LEASE_TIME)
client.srp_client_set_lease_interval(NEW_LEASE_TIME)
self.simulator.go(NEW_LEASE_TIME)
self.assertEqual(client.srp_client_get_host_state(), 'Registered')
self.simulator.go(KEY_LEASE_TIME * 2)
self.assertEqual(client.srp_client_get_host_state(), 'Registered')
def verify(self, pv):
pkts: PacketFilter = pv.pkts
pv.summary.show()
CLIENT_SRC64 = pv.vars['SRP_CLIENT']
pkts.filter_wpan_src64(CLIENT_SRC64).filter('dns.flags.response == 0 and dns.resp.ttl == {DEFAULT_LEASE_TIME}',
DEFAULT_LEASE_TIME=DEFAULT_LEASE_TIME).must_next()
pkts.filter_wpan_src64(CLIENT_SRC64).filter('dns.flags.response == 0 and dns.resp.ttl == {NEW_LEASE_TIME}',
NEW_LEASE_TIME=NEW_LEASE_TIME).must_next()
def check_host_and_service(self, server, client):
"""Check that we have properly registered host and service instance.
"""
client_services = client.srp_client_get_services()
print(client_services)
self.assertEqual(len(client_services), 1)
client_service = client_services[0]
# Verify that the client possesses correct service resources.
self.assertEqual(client_service['instance'], 'my-service')
self.assertEqual(client_service['name'], '_ipps._tcp')
self.assertEqual(int(client_service['port']), 12345)
self.assertEqual(int(client_service['priority']), 0)
self.assertEqual(int(client_service['weight']), 0)
# Verify that the client received a SUCCESS response for the server.
self.assertEqual(client_service['state'], 'Registered')
if __name__ == '__main__':
unittest.main()