diff --git a/tests/net/lib/lwm2m/interop/pytest/conftest.py b/tests/net/lib/lwm2m/interop/pytest/conftest.py new file mode 100644 index 00000000000000..9ca971b85c7c8d --- /dev/null +++ b/tests/net/lib/lwm2m/interop/pytest/conftest.py @@ -0,0 +1,148 @@ +""" +Common test fixtures +#################### + +Copyright (c) 2023 Nordic Semiconductor ASA + +SPDX-License-Identifier: Apache-2.0 + +""" + +import time +import logging +import os +import binascii +import random +import string +import pytest +from leshan import Leshan + +from twister_harness import Shell +from twister_harness import DeviceAdapter + +LESHAN_IP: str = '192.0.2.2' +COAP_PORT: int = 5683 +COAPS_PORT: int = 5684 +BOOTSTRAP_COAPS_PORT: int = 5784 + +logger = logging.getLogger(__name__) + +@pytest.fixture(scope='session') +def leshan() -> Leshan: + """ + Fixture that returns a Leshan object for interacting with the Leshan server. + + :return: The Leshan object. + :rtype: Leshan + """ + try: + return Leshan("http://localhost:8080/api") + except RuntimeError: + pytest.skip('Leshan server not available') + +@pytest.fixture(scope='session') +def leshan_bootstrap() -> Leshan: + """ + Fixture that returns a Leshan object for interacting with the Bootstrap Leshan server. + + :return: The Leshan object. + :rtype: Leshan + """ + try: + return Leshan("http://localhost:8081/api") + except RuntimeError: + pytest.skip('Leshan Bootstrap server not available') + +@pytest.fixture(scope='module') +def helperclient() -> object: + """ + Fixture that returns a helper client object for testing. + + :return: The helper client object. + :rtype: object + """ + try: + from coapthon.client.helperclient import HelperClient + except ModuleNotFoundError: + pytest.skip('CoAPthon3 package not installed') + return HelperClient(server=('127.0.0.1', COAP_PORT)) + + +@pytest.fixture(scope='module') +def endpoint_nosec(shell: Shell, dut: DeviceAdapter, leshan: Leshan) -> str: + """Fixture that returns an endpoint that starts on no-secure mode""" + # Allow engine to start & stop once. + time.sleep(2) + + # Generate randon device id and password (PSK key) + ep = 'client_' + binascii.b2a_hex(os.urandom(1)).decode() + + # + # Registration Interface test cases (using Non-secure mode) + # + shell.exec_command(f'lwm2m write 0/0/0 -s coap://{LESHAN_IP}:{COAP_PORT}') + shell.exec_command('lwm2m write 0/0/1 -b 0') + shell.exec_command('lwm2m write 0/0/2 -u8 3') + shell.exec_command(f'lwm2m write 0/0/3 -s {ep}') + shell.exec_command('lwm2m create 1/0') + shell.exec_command('lwm2m write 0/0/10 -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(f'lwm2m start {ep} -b 0') + dut.readlines_until(regex=f"RD Client started with endpoint '{ep}'", timeout=10.0) + + yield ep + + # All done + shell.exec_command('lwm2m stop') + dut.readlines_until(regex=r'.*Deregistration success', timeout=10.0) + +@pytest.fixture(scope='module') +def endpoint_bootstrap(shell: Shell, dut: DeviceAdapter, leshan: Leshan, leshan_bootstrap: Leshan) -> str: + """Fixture that returns an endpoint that starts the bootstrap.""" + try: + # Generate randon device id and password (PSK key) + ep = 'client_' + binascii.b2a_hex(os.urandom(1)).decode() + bs_passwd = ''.join(random.choice(string.ascii_lowercase) for i in range(16)) + passwd = ''.join(random.choice(string.ascii_lowercase) for i in range(16)) + + logger.debug('Endpoint: %s', ep) + logger.debug('Boostrap PSK: %s', binascii.b2a_hex(bs_passwd.encode()).decode()) + logger.debug('PSK: %s', binascii.b2a_hex(passwd.encode()).decode()) + + # Create device entries in Leshan and Bootstrap server + leshan_bootstrap.create_bs_device(ep, f'coaps://{LESHAN_IP}:{COAPS_PORT}', bs_passwd, passwd) + leshan.create_psk_device(ep, passwd) + + # Allow engine to start & stop once. + time.sleep(2) + + # Write bootsrap server information and PSK keys + shell.exec_command(f'lwm2m write 0/0/0 -s coaps://{LESHAN_IP}:{BOOTSTRAP_COAPS_PORT}') + shell.exec_command('lwm2m write 0/0/1 -b 1') + shell.exec_command('lwm2m write 0/0/2 -u8 0') + shell.exec_command(f'lwm2m write 0/0/3 -s {ep}') + shell.exec_command(f'lwm2m write 0/0/5 -s {bs_passwd}') + shell.exec_command(f'lwm2m start {ep} -b 1') + + yield ep + + shell.exec_command('lwm2m stop') + dut.readlines_until(regex=r'.*Deregistration success', timeout=10.0) + + finally: + # Remove device and bootstrap information + # Leshan does not accept non-secure connection if device information is provided with PSK + leshan.delete_device(ep) + leshan_bootstrap.delete_bs_device(ep) + +@pytest.fixture(scope='module') +def endpoint_registered(endpoint_bootstrap, shell: Shell, dut: DeviceAdapter) -> str: + """Fixture that returns an endpoint that is registered.""" + dut.readlines_until(regex='.*Registration Done', timeout=5.0) + return endpoint_bootstrap + +@pytest.fixture(scope='module') +def endpoint(endpoint_registered) -> str: + """Fixture that returns an endpoint that is registered.""" + return endpoint_registered diff --git a/tests/net/lib/lwm2m/interop/pytest/leshan.py b/tests/net/lib/lwm2m/interop/pytest/leshan.py index 307a390bf9b639..88030139d97564 100644 --- a/tests/net/lib/lwm2m/interop/pytest/leshan.py +++ b/tests/net/lib/lwm2m/interop/pytest/leshan.py @@ -1,6 +1,12 @@ -# Copyright (c) 2023 Nordic Semiconductor ASA -# -# SPDX-License-Identifier: Apache-2.0 +""" +REST client for Leshan demo server +################################## + +Copyright (c) 2023 Nordic Semiconductor ASA + +SPDX-License-Identifier: Apache-2.0 + +""" from __future__ import annotations @@ -12,7 +18,9 @@ from contextlib import contextmanager class Leshan: + """This class represents a Leshan client that interacts with demo server's REAT API""" def __init__(self, url: str): + """Initialize Leshan client and check if server is available""" self.api_url = url self.timeout = 10 #self.format = 'TLV' @@ -22,8 +30,8 @@ def __init__(self, url: str): resp = self.get('/security/clients') if not isinstance(resp, list): raise RuntimeError('Did not receive list of endpoints') - except requests.exceptions.ConnectionError: - raise RuntimeError('Leshan not responding') + except requests.exceptions.ConnectionError as exc: + raise RuntimeError('Leshan not responding') from exc @staticmethod def handle_response(resp: requests.models.Response): @@ -47,7 +55,7 @@ def handle_response(resp: requests.models.Response): return None def get(self, path: str): - """Send HTTP GET query""" + """Send HTTP GET query with typical parameters""" params = {'timeout': self.timeout} if self.format is not None: params['format'] = self.format @@ -55,15 +63,18 @@ def get(self, path: str): return Leshan.handle_response(resp) def put_raw(self, path: str, data: str | dict | None = None, headers: dict | None = None): + """Send HTTP PUT query without any default parameters""" resp = self._s.put(f'{self.api_url}{path}', data=data, headers=headers, timeout=self.timeout) return Leshan.handle_response(resp) def put(self, path: str, data: str | dict, uri_options: str = ''): + """Send HTTP PUT query with typical parameters""" if isinstance(data, dict): data = json.dumps(data) return self.put_raw(f'{path}?timeout={self.timeout}&format={self.format}' + uri_options, data=data, headers={'content-type': 'application/json'}) def post(self, path: str, data: str | dict | None = None): + """Send HTTP POST query""" if isinstance(data, dict): data = json.dumps(data) if data is not None: @@ -76,13 +87,16 @@ def post(self, path: str, data: str | dict | None = None): return Leshan.handle_response(resp) def delete(self, path: str): + """Send HTTP DELETE query""" resp = self._s.delete(f'{self.api_url}{path}', timeout=self.timeout) return Leshan.handle_response(resp) def execute(self, endpoint: str, path: str): + """Send LwM2M EXECUTE command""" return self.post(f'/clients/{endpoint}/{path}') def write(self, endpoint: str, path: str, value: bool | int | str): + """Send LwM2M WRITE command to a single resource or resource instance""" if len(path.split('/')) == 3: kind = 'singleResource' else: @@ -91,14 +105,17 @@ def write(self, endpoint: str, path: str, value: bool | int | str): return self.put(f'/clients/{endpoint}/{path}', self._define_resource(rid, value, kind)) def update_obj_instance(self, endpoint: str, path: str, resources: dict): + """Update object instance""" data = self._define_obj_inst(path, resources) return self.put(f'/clients/{endpoint}/{path}', data, uri_options='&replace=false') def replace_obj_instance(self, endpoint: str, path: str, resources: dict): + """Replace object instance""" data = self._define_obj_inst(path, resources) return self.put(f'/clients/{endpoint}/{path}', data, uri_options='&replace=true') def create_obj_instance(self, endpoint: str, path: str, resources: dict): + """Send LwM2M CREATE command""" data = self._define_obj_inst(path, resources) path = '/'.join(path.split('/')[:-1]) # Create call should not have instance ID in path return self.post(f'/clients/{endpoint}/{path}', data) @@ -124,6 +141,7 @@ def _type_to_string(cls, value): @classmethod def _convert_type(cls, value): + """Wrapper for special types that are not understood by Json""" if isinstance(value, datetime): return int(value.timestamp()) else: @@ -131,6 +149,7 @@ def _convert_type(cls, value): @classmethod def _define_obj_inst(cls, path: str, resources: dict): + """Define an object instance for Leshan""" data = { "kind": "instance", "id": int(path.split('/')[-1]), # ID is last element of path @@ -146,6 +165,7 @@ def _define_obj_inst(cls, path: str, resources: dict): @classmethod def _define_resource(cls, rid, value, kind='singleResource'): + """Define a resource for Leshan""" if kind in ('singleResource', 'resourceInstance'): return { "id": rid, @@ -208,6 +228,7 @@ def _decode_obj(cls, content): return {content['id']: instances} def read(self, endpoint: str, path: str): + """Send LwM2M READ command and decode the response to a Python dictionary""" resp = self.get(f'/clients/{endpoint}/{path}') if not resp['success']: return resp @@ -223,9 +244,10 @@ def read(self, endpoint: str, path: str): raise RuntimeError(f'Unhandled type {content["kind"]}') @classmethod - def parse_composite(cls, content: dict): + def parse_composite(cls, payload: dict): + """Decode the Leshan's response to composite query back to a Python dictionary""" data = {} - for path, content in content.items(): + for path, content in payload.items(): keys = [int(key) for key in path.lstrip("/").split('/')] if len(keys) == 1: data.update(cls._decode_obj(content)) @@ -251,14 +273,22 @@ def parse_composite(cls, content: dict): raise RuntimeError(f'Unhandled path {path}') return data - def composite_read(self, endpoint: str, paths: list[str]): - paths = [path if path.startswith('/') else '/' + path for path in paths] + def _composite_params(self, paths: list[str] | None = None): + """Common URI parameters for composite query""" parameters = { 'pathformat': self.format, 'nodeformat': self.format, - 'timeout': self.timeout, - 'paths': ','.join(paths) + 'timeout': self.timeout } + if paths is not None: + paths = [path if path.startswith('/') else '/' + path for path in paths] + parameters['paths'] = ','.join(paths) + + return parameters + + def composite_read(self, endpoint: str, paths: list[str]): + """Send LwM2M Composite-Read command and decode the response to a Python dictionary""" + parameters = self._composite_params(paths) resp = self._s.get(f'{self.api_url}/clients/{endpoint}/composite', params=parameters, timeout=self.timeout) payload = Leshan.handle_response(resp) if not payload['status'] == 'CONTENT(205)': @@ -267,7 +297,7 @@ def composite_read(self, endpoint: str, paths: list[str]): def composite_write(self, endpoint: str, resources: dict): """ - Do LwM2m composite write operation. + Send LwM2m Composite-Write operation. Targeted resources are defined as a dictionary with the following structure: { @@ -356,11 +386,18 @@ def get_event_stream(self, endpoint: str): r.close() class LeshanEventsIterator: + """Iterator for Leshan event stream""" def __init__(self, req: requests.Response, timeout: int): + """Initialize the iterator in line mode""" self._it = req.iter_lines(chunk_size=1, decode_unicode=True) self._timeout = timeout def next_event(self, event: str): + """ + Finds the next occurrence of a specific event in the stream. + + If timeout happens, the returns None. + """ timeout = time.time() + self._timeout try: for line in self._it: diff --git a/tests/net/lib/lwm2m/interop/pytest/test_bootstrap.py b/tests/net/lib/lwm2m/interop/pytest/test_bootstrap.py new file mode 100644 index 00000000000000..65561db067670a --- /dev/null +++ b/tests/net/lib/lwm2m/interop/pytest/test_bootstrap.py @@ -0,0 +1,58 @@ +""" +LwM2M Bootstrap interface tests +############################### + +Copyright (c) 2023 Nordic Semiconductor ASA + +SPDX-License-Identifier: Apache-2.0 + +Test specification: +=================== +https://www.openmobilealliance.org/release/LightweightM2M/ETS/OMA-ETS-LightweightM2M-V1_1-20190912-D.pdf + + +This module contains only testcases that verify the bootstrap. +""" + +import logging +from leshan import Leshan +from twister_harness import Shell +from twister_harness import DeviceAdapter + +logger = logging.getLogger(__name__) + + +# +# Test specification: +# https://www.openmobilealliance.org/release/LightweightM2M/ETS/OMA-ETS-LightweightM2M-V1_1-20190912-D.pdf +# +# Bootstrap Interface: [0-99] +# + +def verify_LightweightM2M_1_1_int_0(shell: Shell, dut: DeviceAdapter): + """LightweightM2M-1.1-int-0 - Client Initiated Bootstrap""" + dut.readlines_until(regex='.*Bootstrap started with endpoint', timeout=5.0) + dut.readlines_until(regex='.*Bootstrap registration done', timeout=5.0) + dut.readlines_until(regex='.*Bootstrap data transfer done', timeout=5.0) + +def test_LightweightM2M_1_1_int_1(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint_bootstrap: str): + """LightweightM2M-1.1-int-1 - Client Initiated Bootstrap Full (PSK)""" + verify_LightweightM2M_1_1_int_0(shell, dut) + verify_LightweightM2M_1_1_int_101(shell, dut, leshan, endpoint_bootstrap) + verify_LightweightM2M_1_1_int_401(shell, leshan, endpoint_bootstrap) + +def verify_LightweightM2M_1_1_int_101(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-101 - Initial Registration""" + dut.readlines_until(regex='.*Registration Done', timeout=5.0) + assert leshan.get(f'/clients/{endpoint}') + +def verify_LightweightM2M_1_1_int_401(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-401 - UDP Channel Security - Pre-shared Key Mode""" + lines = shell.get_filtered_output(shell.exec_command('lwm2m read 0/0/0 -s')) + host = lines[0] + assert 'coaps://' in host + lines = shell.get_filtered_output(shell.exec_command('lwm2m read 0/0/2 -u8')) + mode = int(lines[0]) + assert mode == 0 + resp = leshan.get(f'/clients/{endpoint}') + assert resp["secure"] diff --git a/tests/net/lib/lwm2m/interop/pytest/test_lwm2m.py b/tests/net/lib/lwm2m/interop/pytest/test_lwm2m.py index 384e3cadda5c71..bdeb87838474d9 100644 --- a/tests/net/lib/lwm2m/interop/pytest/test_lwm2m.py +++ b/tests/net/lib/lwm2m/interop/pytest/test_lwm2m.py @@ -1,136 +1,81 @@ -# Copyright (c) 2023 Nordic Semiconductor ASA -# -# SPDX-License-Identifier: Apache-2.0 +""" +Various LwM2M interoperability tests +#################################### + +Copyright (c) 2023 Nordic Semiconductor ASA + +SPDX-License-Identifier: Apache-2.0 + +Test specification: +=================== +https://www.openmobilealliance.org/release/LightweightM2M/ETS/OMA-ETS-LightweightM2M-V1_1-20190912-D.pdf + + +This module contains testcases for + * Registration Interface [100-199] + * Device management & Service Enablement Interface [200-299] + * Information Reporting Interface [300-399] + +""" import time import logging +from datetime import datetime import pytest from leshan import Leshan -import os -import binascii -import random -import string from twister_harness import Shell -from datetime import datetime - -LESHAN_IP: str = '192.0.2.2' -COAP_PORT: int = 5683 -COAPS_PORT: int = 5684 -BOOTSTRAP_COAPS_PORT: int = 5784 +from twister_harness import DeviceAdapter logger = logging.getLogger(__name__) -@pytest.fixture(scope='module') -def helperclient() -> object: - try: - from coapthon.client.helperclient import HelperClient - except ModuleNotFoundError: - pytest.skip('CoAPthon3 package not installed') - return HelperClient(server=('127.0.0.1', COAP_PORT)) - -@pytest.fixture(scope='session') -def leshan() -> Leshan: - try: - return Leshan("http://localhost:8080/api") - except RuntimeError: - pytest.skip('Leshan server not available') - -@pytest.fixture(scope='session') -def leshan_bootstrap() -> Leshan: - try: - return Leshan("http://localhost:8081/api") - except RuntimeError: - pytest.skip('Leshan Bootstrap server not available') - -# -# Test specification: -# https://www.openmobilealliance.org/release/LightweightM2M/ETS/OMA-ETS-LightweightM2M-V1_1-20190912-D.pdf -# - -def verify_LightweightM2M_1_1_int_0(shell: Shell): - logger.info("LightweightM2M-1.1-int-0 - Client Initiated Bootstrap") - shell._device.readlines_until(regex='.*Bootstrap started with endpoint', timeout=5.0) - shell._device.readlines_until(regex='.*Bootstrap registration done', timeout=5.0) - shell._device.readlines_until(regex='.*Bootstrap data transfer done', timeout=5.0) - -def verify_LightweightM2M_1_1_int_1(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-1 - Client Initiated Bootstrap Full (PSK)") - verify_LightweightM2M_1_1_int_0(shell) - verify_LightweightM2M_1_1_int_101(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_401(shell, leshan, endpoint) - -def verify_LightweightM2M_1_1_int_101(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-101 - Initial Registration") - shell._device.readlines_until(regex='.*Registration Done', timeout=5.0) - assert leshan.get(f'/clients/{endpoint}') -def verify_LightweightM2M_1_1_int_102(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-102 - Registration Update") +def test_LightweightM2M_1_1_int_102(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-102 - Registration Update""" lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/1 -u32')) lifetime = int(lines[0]) lifetime = lifetime + 10 start_time = time.time() * 1000 leshan.write(endpoint, '1/0/1', lifetime) - shell._device.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) + dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) latest = leshan.get(f'/clients/{endpoint}') assert latest["lastUpdate"] > start_time assert latest["lastUpdate"] <= time.time()*1000 assert latest["lifetime"] == lifetime shell.exec_command('lwm2m write 1/0/1 -u32 86400') -def verify_LightweightM2M_1_1_int_103(): - """LightweightM2M-1.1-int-103 - Deregistration""" - # Unsupported. We don't have "disabled" functionality in server object - -def verify_LightweightM2M_1_1_int_104(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-104 - Registration Update Trigger") +def test_LightweightM2M_1_1_int_104(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-104 - Registration Update Trigger""" shell.exec_command('lwm2m update') - shell._device.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) + dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) leshan.execute(endpoint, '1/0/8') - shell._device.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) - -def verify_LightweightM2M_1_1_int_105(shell: Shell, leshan: Leshan, endpoint: str, helperclient: object): - logger.info("LightweightM2M-1.1-int-105 - Discarded Register Update") - status = leshan.get(f'/clients/{endpoint}') - if status["secure"]: - logger.debug("Skip, requires non-secure connection") - return - regid = status["registrationId"] - assert regid - # Fake unregister message - helperclient.delete(f'rd/{regid}', timeout=0.1) - helperclient.stop() - time.sleep(1) - shell.exec_command('lwm2m update') - shell._device.readlines_until(regex=r'.*Failed with code 4\.4', timeout=5.0) - shell._device.readlines_until(regex='.*Registration Done', timeout=10.0) + dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) -def verify_LightweightM2M_1_1_int_107(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-107 - Extending the lifetime of a registration") +def test_LightweightM2M_1_1_int_107(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-107 - Extending the lifetime of a registration""" leshan.write(endpoint, '1/0/1', 120) - shell._device.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) + dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/1 -u32')) lifetime = int(lines[0]) assert lifetime == 120 logger.debug(f'Wait for update, max {lifetime} s') - shell._device.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=lifetime) + dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=lifetime) assert leshan.get(f'/clients/{endpoint}') -def verify_LightweightM2M_1_1_int_108(leshan, endpoint): - logger.info("LightweightM2M-1.1-int-108 - Turn on Queue Mode") +def test_LightweightM2M_1_1_int_108(leshan, endpoint): + """LightweightM2M-1.1-int-108 - Turn on Queue Mode""" assert leshan.get(f'/clients/{endpoint}')["queuemode"] -def verify_LightweightM2M_1_1_int_109(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-109 - Behavior in Queue Mode") +def test_LightweightM2M_1_1_int_109(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-109 - Behavior in Queue Mode""" logger.debug('Wait for Queue RX OFF') - shell._device.readlines_until(regex='.*Queue mode RX window closed', timeout=120) + dut.readlines_until(regex='.*Queue mode RX window closed', timeout=120) # Restore previous value shell.exec_command('lwm2m write 1/0/1 -u32 86400') - shell._device.readlines_until(regex='.*Registration update complete', timeout=10) + dut.readlines_until(regex='.*Registration update complete', timeout=10) -def verify_LightweightM2M_1_1_int_201(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-201 - Querying basic information in Plain Text format") +def test_LightweightM2M_1_1_int_201(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-201 - Querying basic information in Plain Text format""" fmt = leshan.format leshan.format = 'TEXT' assert leshan.read(endpoint, '3/0/0') == 'Zephyr' @@ -157,26 +102,24 @@ def verify_server_object(obj): assert obj[0][6] is False assert obj[0][7] == 'U' -def verify_LightweightM2M_1_1_int_203(shell: Shell, leshan: Leshan, endpoint: str): - shell.exec_command('lwm2m update') - logger.info('LightweightM2M-1.1-int-203 - Querying basic information in TLV format') +def test_LightweightM2M_1_1_int_203(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-203 - Querying basic information in TLV format""" fmt = leshan.format leshan.format = 'TLV' resp = leshan.read(endpoint,'3/0') verify_device_object(resp) leshan.format = fmt -def verify_LightweightM2M_1_1_int_204(shell: Shell, leshan: Leshan, endpoint: str): - shell.exec_command('lwm2m update') - logger.info('LightweightM2M-1.1-int-204 - Querying basic information in JSON format') +def test_LightweightM2M_1_1_int_204(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-204 - Querying basic information in JSON format""" fmt = leshan.format leshan.format = 'JSON' resp = leshan.read(endpoint, '3/0') verify_device_object(resp) leshan.format = fmt -def verify_LightweightM2M_1_1_int_205(shell: Shell, leshan: Leshan, endpoint: str): - logger.info('LightweightM2M-1.1-int-205 - Setting basic information in Plain Text format') +def test_LightweightM2M_1_1_int_205(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-205 - Setting basic information in Plain Text format""" fmt = leshan.format leshan.format = 'TEXT' leshan.write(endpoint, '1/0/2', 101) @@ -193,8 +136,8 @@ def verify_LightweightM2M_1_1_int_205(shell: Shell, leshan: Leshan, endpoint: st assert leshan.read(endpoint, '1/0/5') == 86400 leshan.format = fmt -def verify_LightweightM2M_1_1_int_211(shell: Shell, leshan: Leshan, endpoint: str): - logger.info('LightweightM2M-1.1-int-211 - Querying basic information in CBOR format') +def test_LightweightM2M_1_1_int_211(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-211 - Querying basic information in CBOR format""" fmt = leshan.format leshan.format = 'CBOR' lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/0 -u16')) @@ -204,8 +147,8 @@ def verify_LightweightM2M_1_1_int_211(shell: Shell, leshan: Leshan, endpoint: st assert leshan.read(endpoint, '1/0/7') == 'U' leshan.format = fmt -def verify_LightweightM2M_1_1_int_212(shell: Shell, leshan: Leshan, endpoint: str): - logger.info('LightweightM2M-1.1-int-212 - Setting basic information in CBOR format') +def test_LightweightM2M_1_1_int_212(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-212 - Setting basic information in CBOR format""" fmt = leshan.format leshan.format = 'CBOR' leshan.write(endpoint, '1/0/2', 101) @@ -245,22 +188,22 @@ def verify_setting_basic_in_format(shell, leshan, endpoint, format): verify_server_object(server_obj) leshan.format = fmt -def verify_LightweightM2M_1_1_int_215(shell: Shell, leshan: Leshan, endpoint: str): - logger.info('LightweightM2M-1.1-int-215 - Setting basic information in TLV format') +def test_LightweightM2M_1_1_int_215(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-215 - Setting basic information in TLV format""" verify_setting_basic_in_format(shell, leshan, endpoint, 'TLV') -def verify_LightweightM2M_1_1_int_220(shell: Shell, leshan: Leshan, endpoint: str): - logger.info('LightweightM2M-1.1-int-220 - Setting basic information in JSON format') +def test_LightweightM2M_1_1_int_220(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-220 - Setting basic information in JSON format""" verify_setting_basic_in_format(shell, leshan, endpoint, 'JSON') -def verify_LightweightM2M_1_1_int_221(shell: Shell, leshan: Leshan, endpoint: str): - logger.info('LightweightM2M-1.1-int-221 - Attempt to perform operations on Security') +def test_LightweightM2M_1_1_int_221(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-221 - Attempt to perform operations on Security""" assert leshan.read(endpoint, '0/0')['status'] == 'UNAUTHORIZED(401)' assert leshan.write(endpoint, '0/0/0', 'coap://localhost')['status'] == 'UNAUTHORIZED(401)' assert leshan.put_raw(f'/clients/{endpoint}/0/attributes?pmin=10')['status'] == 'UNAUTHORIZED(401)' -def verify_LightweightM2M_1_1_int_222(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-222 - Read on Object") +def test_LightweightM2M_1_1_int_222(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-222 - Read on Object""" resp = leshan.read(endpoint, '1') assert len(resp) == 1 assert len(resp[1][0]) == 9 @@ -270,27 +213,27 @@ def verify_LightweightM2M_1_1_int_222(shell: Shell, leshan: Leshan, endpoint: st assert len(resp[3][0]) == 15 assert resp[3][0][0] == 'Zephyr' -def verify_LightweightM2M_1_1_int_223(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-223 - Read on Object Instance") +def test_LightweightM2M_1_1_int_223(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-223 - Read on Object Instance""" resp = leshan.read(endpoint, '1/0') assert len(resp[0]) == 9 resp = leshan.read(endpoint, '3/0') assert len(resp[0]) == 15 assert resp[0][0] == 'Zephyr' -def verify_LightweightM2M_1_1_int_224(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-224 - Read on Resource") +def test_LightweightM2M_1_1_int_224(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-224 - Read on Resource""" assert leshan.read(endpoint, '1/0/0') == 1 assert leshan.read(endpoint, '1/0/1') == 86400 assert leshan.read(endpoint, '1/0/6') is False assert leshan.read(endpoint, '1/0/7') == 'U' -def verify_LightweightM2M_1_1_int_225(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-225 - Read on Resource Instance") +def test_LightweightM2M_1_1_int_225(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-225 - Read on Resource Instance""" assert leshan.read(endpoint, '3/0/11/0') == 0 -def verify_LightweightM2M_1_1_int_226(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-226 - Write (Partial Update) on Object Instance") +def test_LightweightM2M_1_1_int_226(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-226 - Write (Partial Update) on Object Instance""" lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/1 -u32')) lifetime = int(lines[0]) resources = { @@ -306,29 +249,29 @@ def verify_LightweightM2M_1_1_int_226(shell: Shell, leshan: Leshan, endpoint: st } assert leshan.update_obj_instance(endpoint, '1/0', resources)['status'] == 'CHANGED(204)' -def verify_LightweightM2M_1_1_int_227(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-227 - Write (replace) on Resource") +def test_LightweightM2M_1_1_int_227(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-227 - Write (replace) on Resource""" lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/1 -u32')) lifetime = int(lines[0]) assert leshan.write(endpoint, '1/0/1', int(63))['status'] == 'CHANGED(204)' - shell._device.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) + dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) latest = leshan.get(f'/clients/{endpoint}') assert latest["lifetime"] == 63 assert leshan.read(endpoint, '1/0/1') == 63 assert leshan.write(endpoint, '1/0/1', lifetime)['status'] == 'CHANGED(204)' -def verify_LightweightM2M_1_1_int_228(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-228 - Write on Resource Instance") +def test_LightweightM2M_1_1_int_228(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-228 - Write on Resource Instance""" resources = { 0: {0: 'a', 1: 'b'} } assert leshan.create_obj_instance(endpoint, '16/0', resources)['status'] == 'CREATED(201)' - shell._device.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) + dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) assert leshan.write(endpoint, '16/0/0/0', 'test')['status'] == 'CHANGED(204)' assert leshan.read(endpoint, '16/0/0/0') == 'test' -def verify_LightweightM2M_1_1_int_229(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-229 - Read-Composite Operation") +def test_LightweightM2M_1_1_int_229(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-229 - Read-Composite Operation""" old_fmt = leshan.format for fmt in ['SENML_JSON', 'SENML_CBOR']: leshan.format = fmt @@ -346,8 +289,8 @@ def verify_LightweightM2M_1_1_int_229(shell: Shell, leshan: Leshan, endpoint: st assert resp[3][0][11][0] is not None leshan.format = old_fmt -def verify_LightweightM2M_1_1_int_230(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-230 - Write-Composite Operation") +def test_LightweightM2M_1_1_int_230(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-230 - Write-Composite Operation""" resources = { "/1/0/1": 60, "/1/0/6": True, @@ -376,6 +319,7 @@ def verify_LightweightM2M_1_1_int_230(shell: Shell, leshan: Leshan, endpoint: st leshan.format = old_fmt def query_basic_in_senml(leshan: Leshan, endpoint: str, fmt: str): + """Querying basic information in one of the SenML formats""" old_fmt = leshan.format leshan.format = fmt verify_server_object(leshan.read(endpoint, '1')[1]) @@ -384,15 +328,16 @@ def query_basic_in_senml(leshan: Leshan, endpoint: str, fmt: str): assert leshan.read(endpoint, '3/0/11/0') == 0 leshan.format = old_fmt -def verify_LightweightM2M_1_1_int_231(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-231 - Querying basic information in SenML JSON format") +def test_LightweightM2M_1_1_int_231(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-231 - Querying basic information in SenML JSON format""" query_basic_in_senml(leshan, endpoint, 'SENML_JSON') -def verify_LightweightM2M_1_1_int_232(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-232 - Querying basic information in SenML CBOR format") +def test_LightweightM2M_1_1_int_232(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-232 - Querying basic information in SenML CBOR format""" query_basic_in_senml(leshan, endpoint, 'SENML_CBOR') def setting_basic_senml(shell: Shell, leshan: Leshan, endpoint: str, fmt: str): + """Setting basic information in one of the SenML formats""" old_fmt = leshan.format leshan.format = fmt resources = { @@ -412,51 +357,51 @@ def setting_basic_senml(shell: Shell, leshan: Leshan, endpoint: str, fmt: str): shell.exec_command('lwm2m write /1/0/6 -u8 0') leshan.format = old_fmt -def verify_LightweightM2M_1_1_int_233(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-233 - Setting basic information in SenML CBOR format") +def test_LightweightM2M_1_1_int_233(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-233 - Setting basic information in SenML CBOR format""" setting_basic_senml(shell, leshan, endpoint, 'SENML_CBOR') -def verify_LightweightM2M_1_1_int_234(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-234 - Setting basic information in SenML JSON format") +def test_LightweightM2M_1_1_int_234(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-234 - Setting basic information in SenML JSON format""" setting_basic_senml(shell, leshan, endpoint, 'SENML_JSON') -def verify_LightweightM2M_1_1_int_235(): +@pytest.mark.skip("Leshan does not allow reading root path") +def test_LightweightM2M_1_1_int_235(): """LightweightM2M-1.1-int-235 - Read-Composite Operation on root path""" - # Unsupported. Leshan does not allow this. -def verify_LightweightM2M_1_1_int_236(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-236 - Read-Composite - Partial Presence") +def test_LightweightM2M_1_1_int_236(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-236 - Read-Composite - Partial Presence""" resp = leshan.composite_read(endpoint, ['1/0', '/3/0/11/0', '/3339/0/5522', '/3353/0/6030']) assert resp[1][0][1] is not None assert resp[3][0][11][0] is not None assert len(resp) == 2 -def verify_LightweightM2M_1_1_int_237(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-237 - Read on Object without specifying Content-Type") +def test_LightweightM2M_1_1_int_237(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-237 - Read on Object without specifying Content-Type""" old_fmt = leshan.format leshan.format = None assert leshan.read(endpoint, '1')[1][0][1] is not None assert leshan.read(endpoint, '3')[3][0][0] == 'Zephyr' leshan.format = old_fmt -def verify_LightweightM2M_1_1_int_241(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-241 - Executable Resource: Rebooting the device") +def test_LightweightM2M_1_1_int_241(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-241 - Executable Resource: Rebooting the device""" leshan.execute(endpoint, '3/0/4') - shell._device.readlines_until(regex='.*DEVICE: REBOOT', timeout=5.0) - shell._device.readlines_until(regex='.*rd_client_event: Disconnected', timeout=5.0) + dut.readlines_until(regex='.*DEVICE: REBOOT', timeout=5.0) + dut.readlines_until(regex='.*rd_client_event: Disconnected', timeout=5.0) shell.exec_command(f'lwm2m start {endpoint} -b 0') - shell._device.readlines_until(regex='.*Registration Done', timeout=5.0) + dut.readlines_until(regex='.*Registration Done', timeout=5.0) assert leshan.get(f'/clients/{endpoint}') -def verify_LightweightM2M_1_1_int_256(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-256 - Write Operation Failure") +def test_LightweightM2M_1_1_int_256(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-256 - Write Operation Failure""" lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/0 -u16')) short_id = int(lines[0]) assert leshan.write(endpoint, '1/0/0', 123)['status'] == 'METHOD_NOT_ALLOWED(405)' assert leshan.read(endpoint, '1/0/0') == short_id -def verify_LightweightM2M_1_1_int_257(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-257 - Write-Composite Operation") +def test_LightweightM2M_1_1_int_257(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-257 - Write-Composite Operation""" resources = { "/1/0/2": 102, "/1/0/6": True, @@ -476,8 +421,8 @@ def verify_LightweightM2M_1_1_int_257(shell: Shell, leshan: Leshan, endpoint: st shell.exec_command('lwm2m write /1/0/2 -u32 1') leshan.format = old_fmt -def verify_LightweightM2M_1_1_int_260(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-260 - Discover Command") +def test_LightweightM2M_1_1_int_260(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-260 - Discover Command""" resp = leshan.discover(endpoint, '3') expected_keys = ['/3', '/3/0', '/3/0/1', '/3/0/2', '/3/0/3', '/3/0/4', '/3/0/6', '/3/0/7', '/3/0/8', '/3/0/9', '/3/0/11', '/3/0/16'] missing_keys = [key for key in expected_keys if key not in resp.keys()] @@ -505,8 +450,9 @@ def verify_LightweightM2M_1_1_int_260(shell: Shell, leshan: Leshan, endpoint: st assert len(missing_keys) == 0 assert len(resp) == len(expected_keys) -def verify_LightweightM2M_1_1_int_261(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-261 - Write-Attribute Operation on a multiple resource") +@pytest.mark.skip(reason="Leshan don't allow writing attributes to resource instance") +def test_LightweightM2M_1_1_int_261(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-261 - Write-Attribute Operation on a multiple resource""" resp = leshan.discover(endpoint, '3/0/11') logger.debug(resp) expected_keys = ['/3/0/11', '/3/0/11/0'] @@ -528,8 +474,8 @@ def verify_LightweightM2M_1_1_int_261(shell: Shell, leshan: Leshan, endpoint: st assert int(resp['/3/0/11/0']['epmin']) == 1 assert int(resp['/3/0/11/0']['epmax']) == 20 -def verify_LightweightM2M_1_1_int_280(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-280 - Successful Read-Composite Operation") +def test_LightweightM2M_1_1_int_280(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-280 - Successful Read-Composite Operation""" resp = leshan.composite_read(endpoint, ['/3/0/16', '/3/0/11/0', '/1/0']) logger.debug(resp) assert len(resp) == 2 @@ -542,141 +488,10 @@ def verify_LightweightM2M_1_1_int_280(shell: Shell, leshan: Leshan, endpoint: st assert resp[1][0][6] is False assert resp[1][0][7] == 'U' -def verify_LightweightM2M_1_1_int_281(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-281 - Partially Successful Read-Composite Operation") +def test_LightweightM2M_1_1_int_281(shell: Shell, leshan: Leshan, endpoint: str): + """LightweightM2M-1.1-int-281 - Partially Successful Read-Composite Operation""" resp = leshan.composite_read(endpoint, ['/1/0/1', '/1/0/7', '/1/0/8']) assert len(resp) == 1 assert len(resp[1][0]) == 2 # /1/0/8 should not be there assert resp[1][0][1] == 86400 assert resp[1][0][7] == 'U' - -def verify_LightweightM2M_1_1_int_401(shell: Shell, leshan: Leshan, endpoint: str): - logger.info("LightweightM2M-1.1-int-401 - UDP Channel Security - Pre-shared Key Mode") - lines = shell.get_filtered_output(shell.exec_command('lwm2m read 0/0/0 -s')) - host = lines[0] - assert 'coaps://' in host - lines = shell.get_filtered_output(shell.exec_command('lwm2m read 0/0/2 -u8')) - mode = int(lines[0]) - assert mode == 0 - resp = leshan.get(f'/clients/{endpoint}') - assert resp["secure"] - -def test_lwm2m_bootstrap_psk(shell: Shell, leshan, leshan_bootstrap): - try: - # Generate randon device id and password (PSK key) - endpoint = 'client_' + binascii.b2a_hex(os.urandom(1)).decode() - bs_passwd = ''.join(random.choice(string.ascii_lowercase) for i in range(16)) - passwd = ''.join(random.choice(string.ascii_lowercase) for i in range(16)) - - logger.debug('Endpoint: %s', endpoint) - logger.debug('Boostrap PSK: %s', binascii.b2a_hex(bs_passwd.encode()).decode()) - logger.debug('PSK: %s', binascii.b2a_hex(passwd.encode()).decode()) - - # Create device entries in Leshan and Bootstrap server - leshan_bootstrap.create_bs_device(endpoint, f'coaps://{LESHAN_IP}:{COAPS_PORT}', bs_passwd, passwd) - leshan.create_psk_device(endpoint, passwd) - - # Allow engine to start & stop once. - time.sleep(2) - - # - # Verify PSK security using Bootstrap - # - - # Write bootsrap server information and PSK keys - shell.exec_command(f'lwm2m write 0/0/0 -s coaps://{LESHAN_IP}:{BOOTSTRAP_COAPS_PORT}') - shell.exec_command('lwm2m write 0/0/1 -b 1') - shell.exec_command('lwm2m write 0/0/2 -u8 0') - shell.exec_command(f'lwm2m write 0/0/3 -s {endpoint}') - shell.exec_command(f'lwm2m write 0/0/5 -s {bs_passwd}') - shell.exec_command(f'lwm2m start {endpoint} -b 1') - - - # - # Bootstrap Interface test cases - # LightweightM2M-1.1-int-0 (included) - # LightweightM2M-1.1-int-401 (included) - verify_LightweightM2M_1_1_int_1(shell, leshan, endpoint) - - # - # Registration Interface test cases (using PSK security) - # - verify_LightweightM2M_1_1_int_102(shell, leshan, endpoint) - # skip, not implemented verify_LightweightM2M_1_1_int_103() - verify_LightweightM2M_1_1_int_104(shell, leshan, endpoint) - # skip, included in 109: verify_LightweightM2M_1_1_int_107(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_108(leshan, endpoint) - verify_LightweightM2M_1_1_int_109(shell, leshan, endpoint) - - # - # Device management & Service Enablement Interface test cases - # - verify_LightweightM2M_1_1_int_201(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_203(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_204(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_205(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_211(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_212(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_215(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_220(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_221(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_222(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_223(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_224(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_225(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_226(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_227(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_228(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_229(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_230(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_231(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_232(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_233(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_234(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_236(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_237(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_241(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_256(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_257(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_260(shell, leshan, endpoint) - # skip, not supported in Leshan, verify_LightweightM2M_1_1_int_261(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_280(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_281(shell, leshan, endpoint) - - shell.exec_command('lwm2m stop') - shell._device.readlines_until(regex=r'.*Deregistration success', timeout=10.0) - - finally: - # Remove device and bootstrap information - # Leshan does not accept non-secure connection if device information is provided with PSK - leshan.delete_device(endpoint) - leshan_bootstrap.delete_bs_device(endpoint) - -def test_lwm2m_nosecure(shell: Shell, leshan, helperclient): - - # Allow engine to start & stop once. - time.sleep(2) - - # Generate randon device id and password (PSK key) - endpoint = 'client_' + binascii.b2a_hex(os.urandom(1)).decode() - - # - # Registration Interface test cases (using Non-secure mode) - # - shell.exec_command(f'lwm2m write 0/0/0 -s coap://{LESHAN_IP}:{COAP_PORT}') - shell.exec_command('lwm2m write 0/0/1 -b 0') - shell.exec_command('lwm2m write 0/0/2 -u8 3') - shell.exec_command(f'lwm2m write 0/0/3 -s {endpoint}') - shell.exec_command('lwm2m create 1/0') - shell.exec_command('lwm2m write 0/0/10 -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(f'lwm2m start {endpoint} -b 0') - shell._device.readlines_until(regex=f"RD Client started with endpoint '{endpoint}'", timeout=10.0) - - verify_LightweightM2M_1_1_int_101(shell, leshan, endpoint) - verify_LightweightM2M_1_1_int_105(shell, leshan, endpoint, helperclient) # needs no-security - - # All done - shell.exec_command('lwm2m stop') - shell._device.readlines_until(regex=r'.*Deregistration success', timeout=10.0) diff --git a/tests/net/lib/lwm2m/interop/pytest/test_nosec.py b/tests/net/lib/lwm2m/interop/pytest/test_nosec.py new file mode 100644 index 00000000000000..15ac98f2d1a7d6 --- /dev/null +++ b/tests/net/lib/lwm2m/interop/pytest/test_nosec.py @@ -0,0 +1,53 @@ +""" +Tests for No-security mode +########################## + +Copyright (c) 2023 Nordic Semiconductor ASA + +SPDX-License-Identifier: Apache-2.0 + +Test specification: +=================== +https://www.openmobilealliance.org/release/LightweightM2M/ETS/OMA-ETS-LightweightM2M-V1_1-20190912-D.pdf + + +This module contains only testcases that are able to run on non-secure mode. +""" + +import time +import logging +from leshan import Leshan + +from twister_harness import Shell +from twister_harness import DeviceAdapter + +logger = logging.getLogger(__name__) + +def test_LightweightM2M_1_1_int_101(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint_nosec: str): + """ + Verify that the client is registered. + Note that this MUST be the first testcase executed, otherwise it will fail to get the + correct log output. + """ + logger.info("LightweightM2M-1.1-int-101 - Initial Registration") + dut.readlines_until(regex='.*Registration Done', timeout=5.0) + assert leshan.get(f'/clients/{endpoint_nosec}') + +def test_LightweightM2M_1_1_int_105(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint_nosec: str, helperclient: object): + """ + Run testcase LightweightM2M-1.1-int-105 - Discarded Register Update + """ + logger.info("LightweightM2M-1.1-int-105 - Discarded Register Update") + status = leshan.get(f'/clients/{endpoint_nosec}') + if status["secure"]: + logger.debug("Skip, requires non-secure connection") + return + regid = status["registrationId"] + assert regid + # Fake unregister message + helperclient.delete(f'rd/{regid}', timeout=0.1) + helperclient.stop() + time.sleep(1) + shell.exec_command('lwm2m update') + dut.readlines_until(regex=r'.*Failed with code 4\.4', timeout=5.0) + dut.readlines_until(regex='.*Registration Done', timeout=10.0) diff --git a/tests/net/lib/lwm2m/interop/testcase.yaml b/tests/net/lib/lwm2m/interop/testcase.yaml index 93f3fc13b1cf32..1b9818e777e4b6 100644 --- a/tests/net/lib/lwm2m/interop/testcase.yaml +++ b/tests/net/lib/lwm2m/interop/testcase.yaml @@ -3,6 +3,8 @@ tests: harness: pytest timeout: 300 slow: true + harness_config: + pytest_dut_scope: module integration_platforms: - native_posix platform_allow: