test: lwm2m: Refactor tests to work with fallback changes
In fallback refactoring to the LwM2M engine, some changes to the server object are visible in hard-coded test values. Also, add Endpoint wrapper class that ensures the registration state of the returned endpoint. Signed-off-by: Seppo Takalo <seppo.takalo@nordicsemi.no>
This commit is contained in:
parent
dc8f6da53c
commit
30dcf1b309
2 changed files with 33 additions and 10 deletions
|
@ -27,6 +27,25 @@ BOOTSTRAP_COAPS_PORT: int = 5784
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
class Endpoint:
|
||||||
|
def __init__(self, name: str, shell: Shell, registered: bool = False, bootstrap: bool = False):
|
||||||
|
self.name = name
|
||||||
|
self.registered = registered
|
||||||
|
self.bootstrap = bootstrap
|
||||||
|
self.shell = shell
|
||||||
|
self.last_update = 0.0
|
||||||
|
|
||||||
|
def check_update(self):
|
||||||
|
if not self.registered:
|
||||||
|
return
|
||||||
|
if self.last_update < time.time() - 5:
|
||||||
|
self.shell.exec_command('lwm2m update')
|
||||||
|
self.last_update = time.time()
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return self.name
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope='session')
|
@pytest.fixture(scope='session')
|
||||||
def leshan() -> Leshan:
|
def leshan() -> Leshan:
|
||||||
"""
|
"""
|
||||||
|
@ -89,9 +108,8 @@ def endpoint_nosec(shell: Shell, dut: DeviceAdapter, leshan: Leshan) -> str:
|
||||||
shell.exec_command('lwm2m write 1/0/0 -u16 1')
|
shell.exec_command('lwm2m write 1/0/0 -u16 1')
|
||||||
shell.exec_command('lwm2m write 1/0/1 -u32 86400')
|
shell.exec_command('lwm2m write 1/0/1 -u32 86400')
|
||||||
shell.exec_command(f'lwm2m start {ep} -b 0')
|
shell.exec_command(f'lwm2m start {ep} -b 0')
|
||||||
dut.readlines_until(regex=f"RD Client started with endpoint '{ep}'", timeout=10.0)
|
|
||||||
|
|
||||||
yield ep
|
yield Endpoint(ep, shell)
|
||||||
|
|
||||||
# All done
|
# All done
|
||||||
shell.exec_command('lwm2m stop')
|
shell.exec_command('lwm2m stop')
|
||||||
|
@ -125,7 +143,7 @@ def endpoint_bootstrap(shell: Shell, dut: DeviceAdapter, leshan: Leshan, leshan_
|
||||||
shell.exec_command(f'lwm2m write 0/0/5 -s {bs_passwd}')
|
shell.exec_command(f'lwm2m write 0/0/5 -s {bs_passwd}')
|
||||||
shell.exec_command(f'lwm2m start {ep} -b 1')
|
shell.exec_command(f'lwm2m start {ep} -b 1')
|
||||||
|
|
||||||
yield ep
|
yield Endpoint(ep, shell)
|
||||||
|
|
||||||
shell.exec_command('lwm2m stop')
|
shell.exec_command('lwm2m stop')
|
||||||
dut.readlines_until(regex=r'.*Deregistration success', timeout=10.0)
|
dut.readlines_until(regex=r'.*Deregistration success', timeout=10.0)
|
||||||
|
@ -137,12 +155,16 @@ def endpoint_bootstrap(shell: Shell, dut: DeviceAdapter, leshan: Leshan, leshan_
|
||||||
leshan_bootstrap.delete_bs_device(ep)
|
leshan_bootstrap.delete_bs_device(ep)
|
||||||
|
|
||||||
@pytest.fixture(scope='module')
|
@pytest.fixture(scope='module')
|
||||||
def endpoint_registered(endpoint_bootstrap, shell: Shell, dut: DeviceAdapter) -> str:
|
def endpoint_registered(endpoint_bootstrap, dut: DeviceAdapter) -> str:
|
||||||
"""Fixture that returns an endpoint that is registered."""
|
"""Fixture that returns an endpoint that is registered."""
|
||||||
|
if not endpoint_bootstrap.registered:
|
||||||
dut.readlines_until(regex='.*Registration Done', timeout=5.0)
|
dut.readlines_until(regex='.*Registration Done', timeout=5.0)
|
||||||
|
endpoint_bootstrap.bootstrap = True
|
||||||
|
endpoint_bootstrap.registered = True
|
||||||
return endpoint_bootstrap
|
return endpoint_bootstrap
|
||||||
|
|
||||||
@pytest.fixture(scope='module')
|
@pytest.fixture(scope='function')
|
||||||
def endpoint(endpoint_registered) -> str:
|
def endpoint(endpoint_registered) -> str:
|
||||||
"""Fixture that returns an endpoint that is registered."""
|
"""Fixture that returns an endpoint that is registered."""
|
||||||
|
endpoint_registered.check_update()
|
||||||
return endpoint_registered
|
return endpoint_registered
|
||||||
|
|
|
@ -171,6 +171,7 @@ def verify_setting_basic_in_format(shell, leshan, endpoint, format):
|
||||||
verify_server_object(server_obj)
|
verify_server_object(server_obj)
|
||||||
# Remove Read-Only resources, so we don't end up writing those
|
# Remove Read-Only resources, so we don't end up writing those
|
||||||
del server_obj[0][0]
|
del server_obj[0][0]
|
||||||
|
del server_obj[0][13]
|
||||||
data = {
|
data = {
|
||||||
2: 101,
|
2: 101,
|
||||||
3: 1010,
|
3: 1010,
|
||||||
|
@ -208,7 +209,7 @@ def test_LightweightM2M_1_1_int_222(shell: Shell, leshan: Leshan, endpoint: str)
|
||||||
"""LightweightM2M-1.1-int-222 - Read on Object"""
|
"""LightweightM2M-1.1-int-222 - Read on Object"""
|
||||||
resp = leshan.read(endpoint, '1')
|
resp = leshan.read(endpoint, '1')
|
||||||
assert len(resp) == 1
|
assert len(resp) == 1
|
||||||
assert len(resp[1][0]) == 9
|
assert len(resp[1][0]) == 11
|
||||||
resp = leshan.read(endpoint, '3')
|
resp = leshan.read(endpoint, '3')
|
||||||
assert len(resp) == 1
|
assert len(resp) == 1
|
||||||
assert len(resp[3]) == 1
|
assert len(resp[3]) == 1
|
||||||
|
@ -218,7 +219,7 @@ def test_LightweightM2M_1_1_int_222(shell: Shell, leshan: Leshan, endpoint: str)
|
||||||
def test_LightweightM2M_1_1_int_223(shell: Shell, leshan: Leshan, endpoint: str):
|
def test_LightweightM2M_1_1_int_223(shell: Shell, leshan: Leshan, endpoint: str):
|
||||||
"""LightweightM2M-1.1-int-223 - Read on Object Instance"""
|
"""LightweightM2M-1.1-int-223 - Read on Object Instance"""
|
||||||
resp = leshan.read(endpoint, '1/0')
|
resp = leshan.read(endpoint, '1/0')
|
||||||
assert len(resp[0]) == 9
|
assert len(resp[0]) == 11
|
||||||
resp = leshan.read(endpoint, '3/0')
|
resp = leshan.read(endpoint, '3/0')
|
||||||
assert len(resp[0]) == 15
|
assert len(resp[0]) == 15
|
||||||
assert resp[0][0] == 'Zephyr'
|
assert resp[0][0] == 'Zephyr'
|
||||||
|
@ -282,7 +283,7 @@ def test_LightweightM2M_1_1_int_229(shell: Shell, leshan: Leshan, endpoint: str)
|
||||||
assert resp[3] is not None
|
assert resp[3] is not None
|
||||||
assert resp[1][0] is not None
|
assert resp[1][0] is not None
|
||||||
assert len(resp[3][0]) == 15
|
assert len(resp[3][0]) == 15
|
||||||
assert len(resp[1][0]) == 9
|
assert len(resp[1][0]) == 11
|
||||||
|
|
||||||
resp = leshan.composite_read(endpoint, ['1/0/1', '/3/0/11/0'])
|
resp = leshan.composite_read(endpoint, ['1/0/1', '/3/0/11/0'])
|
||||||
logger.debug(resp)
|
logger.debug(resp)
|
||||||
|
@ -370,7 +371,7 @@ def test_LightweightM2M_1_1_int_234(shell: Shell, leshan: Leshan, endpoint: str)
|
||||||
def test_LightweightM2M_1_1_int_235(leshan: Leshan, endpoint: str):
|
def test_LightweightM2M_1_1_int_235(leshan: Leshan, endpoint: str):
|
||||||
"""LightweightM2M-1.1-int-235 - Read-Composite Operation on root path"""
|
"""LightweightM2M-1.1-int-235 - Read-Composite Operation on root path"""
|
||||||
resp = leshan.composite_read(endpoint, ['/'])
|
resp = leshan.composite_read(endpoint, ['/'])
|
||||||
expected_keys = [16, 1, 3, 5]
|
expected_keys = [1, 3, 5]
|
||||||
missing_keys = [key for key in expected_keys if key not in resp.keys()]
|
missing_keys = [key for key in expected_keys if key not in resp.keys()]
|
||||||
assert len(missing_keys) == 0
|
assert len(missing_keys) == 0
|
||||||
|
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue