9fc2048bcc
Add testcases: * LightweightM2M-1.1-int-1630 - Create Portfolio Object Instance * LightweightM2M-1.1-int-1635 - Delete Portfolio Object Instance Signed-off-by: Seppo Takalo <seppo.takalo@nordicsemi.no>
199 lines
7.1 KiB
Python
199 lines
7.1 KiB
Python
"""
|
|
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__)
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption('--leshan_addr', action='store', default=LESHAN_IP)
|
|
parser.addoption('--leshan_rest_api', action='store', default='http://localhost:8080/api')
|
|
parser.addoption('--leshan_bootstrap_rest_api', action='store', default='http://localhost:8081/api')
|
|
parser.addoption('--passwd', action='store', default='')
|
|
|
|
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')
|
|
def leshan(request) -> Leshan:
|
|
"""
|
|
Fixture that returns a Leshan object for interacting with the Leshan server.
|
|
|
|
:return: The Leshan object.
|
|
:rtype: Leshan
|
|
"""
|
|
try:
|
|
return Leshan(request.config.getoption('--leshan_rest_api'))
|
|
except RuntimeError:
|
|
pytest.skip('Leshan server not available')
|
|
|
|
@pytest.fixture(scope='session')
|
|
def leshan_bootstrap(request) -> Leshan:
|
|
"""
|
|
Fixture that returns a Leshan object for interacting with the Bootstrap Leshan server.
|
|
|
|
:return: The Leshan object.
|
|
:rtype: Leshan
|
|
"""
|
|
try:
|
|
return Leshan(request.config.getoption('--leshan_bootstrap_rest_api'))
|
|
except RuntimeError:
|
|
pytest.skip('Leshan Bootstrap server not available')
|
|
|
|
@pytest.fixture(scope='module')
|
|
def helperclient(request) -> 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=(request.config.getoption('--leshan_addr'), COAP_PORT))
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def endpoint_nosec(request, 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)
|
|
#
|
|
addr = request.config.getoption('--leshan_addr')
|
|
shell.exec_command(f'lwm2m write 0/0/0 -s coap://{addr}:{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='.*Registration Done', timeout=5.0)
|
|
yield Endpoint(ep, shell)
|
|
|
|
# All done
|
|
shell.exec_command('lwm2m stop')
|
|
dut.readlines_until(regex=r'.*Deregistration success', timeout=10.0)
|
|
|
|
@pytest.fixture(scope='module')
|
|
def endpoint_bootstrap(request, shell: Shell, dut: DeviceAdapter, leshan: Leshan, leshan_bootstrap: Leshan) -> str:
|
|
"""Fixture that returns an endpoint that starts the bootstrap."""
|
|
try:
|
|
static_passwd = request.config.getoption('--passwd')
|
|
# Generate randon device id and password (PSK key)
|
|
ep = 'client_' + binascii.b2a_hex(os.urandom(1)).decode()
|
|
if static_passwd:
|
|
bs_passwd = static_passwd
|
|
passwd = static_passwd
|
|
else:
|
|
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
|
|
addr = request.config.getoption('--leshan_addr')
|
|
leshan_bootstrap.create_bs_device(ep, f'coaps://{addr}:{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://{addr}:{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 Endpoint(ep, shell)
|
|
|
|
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, dut: DeviceAdapter) -> str:
|
|
"""Fixture that returns an endpoint that is registered."""
|
|
if not endpoint_bootstrap.registered:
|
|
dut.readlines_until(regex='.*Registration Done', timeout=5.0)
|
|
endpoint_bootstrap.bootstrap = True
|
|
endpoint_bootstrap.registered = True
|
|
return endpoint_bootstrap
|
|
|
|
@pytest.fixture(scope='function')
|
|
def endpoint(endpoint_registered) -> str:
|
|
"""Fixture that returns an endpoint that is registered."""
|
|
endpoint_registered.check_update()
|
|
return endpoint_registered
|
|
|
|
@pytest.fixture(scope='module')
|
|
def configuration_C13(endpoint_registered, shell: Shell) -> str:
|
|
"""Fixture that returns an endpoint that has C13 configuration."""
|
|
shell.exec_command('lwm2m create /16/0')
|
|
shell.exec_command('lwm2m create /16/0/0/0')
|
|
shell.exec_command('lwm2m create /16/0/0/1')
|
|
shell.exec_command('lwm2m create /16/0/0/2')
|
|
shell.exec_command('lwm2m create /16/0/0/3')
|
|
shell.exec_command('lwm2m write /16/0/0/0 "Host Device ID #1"')
|
|
shell.exec_command('lwm2m write /16/0/0/1 "Host Device Manufacturer #1"')
|
|
shell.exec_command('lwm2m write /16/0/0/2 "Host Device Model #1"')
|
|
shell.exec_command('lwm2m write /16/0/0/3 "Host Device Software Version #1"')
|
|
yield endpoint_registered
|
|
shell.exec_command('lwm2m delete /16/0')
|