1""" 2Various LwM2M interoperability tests 3#################################### 4 5Copyright (c) 2023 Nordic Semiconductor ASA 6 7SPDX-License-Identifier: Apache-2.0 8 9Test specification: 10=================== 11https://www.openmobilealliance.org/release/LightweightM2M/ETS/OMA-ETS-LightweightM2M-V1_1-20190912-D.pdf 12 13 14This module contains testcases for 15 * Registration Interface [100-199] 16 * Device management & Service Enablement Interface [200-299] 17 * Information Reporting Interface [300-399] 18 19""" 20 21import time 22import logging 23from datetime import datetime 24import pytest 25from leshan import Leshan 26 27from twister_harness import Shell 28from twister_harness import DeviceAdapter 29 30logger = logging.getLogger(__name__) 31 32 33def test_LightweightM2M_1_1_int_102(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 34 """LightweightM2M-1.1-int-102 - Registration Update""" 35 lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/1 -u32')) 36 lifetime = int(lines[0]) 37 lifetime = lifetime + 10 38 start_time = time.time() * 1000 39 leshan.write(endpoint, '1/0/1', lifetime) 40 dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) 41 latest = leshan.get(f'/clients/{endpoint}') 42 assert latest["lastUpdate"] > start_time 43 assert latest["lastUpdate"] <= time.time()*1000 44 assert latest["lifetime"] == lifetime 45 shell.exec_command('lwm2m write 1/0/1 -u32 86400') 46 47def test_LightweightM2M_1_1_int_103(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 48 """LightweightM2M-1.1-int-103 - Deregistration""" 49 leshan.execute(endpoint, '1/0/4') 50 dut.readlines_until(regex='LwM2M server disabled', timeout=5.0) 51 dut.readlines_until(regex='Deregistration success', timeout=5.0) 52 # Reset timers by restarting the client 53 shell.exec_command('lwm2m stop') 54 time.sleep(1) 55 shell.exec_command(f'lwm2m start {endpoint}') 56 dut.readlines_until(regex='.*Registration Done', timeout=5.0) 57 58def test_LightweightM2M_1_1_int_104(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 59 """LightweightM2M-1.1-int-104 - Registration Update Trigger""" 60 shell.exec_command('lwm2m update') 61 dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) 62 leshan.execute(endpoint, '1/0/8') 63 dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) 64 65@pytest.mark.slow 66def test_LightweightM2M_1_1_int_107(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 67 """LightweightM2M-1.1-int-107 - Extending the lifetime of a registration""" 68 leshan.write(endpoint, '1/0/1', 120) 69 dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) 70 lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/1 -u32')) 71 lifetime = int(lines[0]) 72 assert lifetime == 120 73 logger.debug(f'Wait for update, max {lifetime} s') 74 dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=lifetime) 75 assert leshan.get(f'/clients/{endpoint}') 76 77def test_LightweightM2M_1_1_int_108(leshan, endpoint): 78 """LightweightM2M-1.1-int-108 - Turn on Queue Mode""" 79 assert leshan.get(f'/clients/{endpoint}')["queuemode"] 80 81@pytest.mark.slow 82def test_LightweightM2M_1_1_int_109(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 83 """LightweightM2M-1.1-int-109 - Behavior in Queue Mode""" 84 logger.debug('Wait for Queue RX OFF') 85 dut.readlines_until(regex='.*Queue mode RX window closed', timeout=120) 86 # Restore previous value 87 shell.exec_command('lwm2m write 1/0/1 -u32 86400') 88 dut.readlines_until(regex='.*Registration update complete', timeout=10) 89 90def test_LightweightM2M_1_1_int_201(shell: Shell, leshan: Leshan, endpoint: str): 91 """LightweightM2M-1.1-int-201 - Querying basic information in Plain Text format""" 92 fmt = leshan.format 93 leshan.format = 'TEXT' 94 assert leshan.read(endpoint, '3/0/0') == 'Zephyr' 95 assert leshan.read(endpoint, '3/0/1') == 'client-1' 96 assert leshan.read(endpoint, '3/0/2') == 'serial-1' 97 leshan.format = fmt 98 99def verify_device_object(resp): 100 ''' Verify that Device object match Configuration 3 ''' 101 assert resp[0][0] == 'Zephyr' 102 assert resp[0][1] == 'client-1' 103 assert resp[0][2] == 'serial-1' 104 assert resp[0][3] == '1.2.3' 105 assert resp[0][11][0] == 0 106 assert resp[0][16] == 'U' 107 108def verify_server_object(obj): 109 ''' Verify that server object match Configuration 3 ''' 110 assert obj[0][0] == 1 111 assert obj[0][1] == 86400 112 assert obj[0][2] == 1 113 assert obj[0][3] == 10 114 assert obj[0][5] == 86400 115 assert obj[0][6] is False 116 assert obj[0][7] == 'U' 117 118def test_LightweightM2M_1_1_int_203(shell: Shell, leshan: Leshan, endpoint: str): 119 """LightweightM2M-1.1-int-203 - Querying basic information in TLV format""" 120 fmt = leshan.format 121 leshan.format = 'TLV' 122 resp = leshan.read(endpoint,'3/0') 123 verify_device_object(resp) 124 leshan.format = fmt 125 126def test_LightweightM2M_1_1_int_204(shell: Shell, leshan: Leshan, endpoint: str): 127 """LightweightM2M-1.1-int-204 - Querying basic information in JSON format""" 128 fmt = leshan.format 129 leshan.format = 'JSON' 130 resp = leshan.read(endpoint, '3/0') 131 verify_device_object(resp) 132 leshan.format = fmt 133 134def test_LightweightM2M_1_1_int_205(shell: Shell, leshan: Leshan, endpoint: str): 135 """LightweightM2M-1.1-int-205 - Setting basic information in Plain Text format""" 136 fmt = leshan.format 137 leshan.format = 'TEXT' 138 leshan.write(endpoint, '1/0/2', 101) 139 leshan.write(endpoint, '1/0/3', 1010) 140 leshan.write(endpoint, '1/0/5', 2000) 141 assert leshan.read(endpoint, '1/0/2') == 101 142 assert leshan.read(endpoint, '1/0/3') == 1010 143 assert leshan.read(endpoint, '1/0/5') == 2000 144 leshan.write(endpoint, '1/0/2', 1) 145 leshan.write(endpoint, '1/0/3', 10) 146 leshan.write(endpoint, '1/0/5', 86400) 147 assert leshan.read(endpoint, '1/0/2') == 1 148 assert leshan.read(endpoint, '1/0/3') == 10 149 assert leshan.read(endpoint, '1/0/5') == 86400 150 leshan.format = fmt 151 152def test_LightweightM2M_1_1_int_211(shell: Shell, leshan: Leshan, endpoint: str): 153 """LightweightM2M-1.1-int-211 - Querying basic information in CBOR format""" 154 fmt = leshan.format 155 leshan.format = 'CBOR' 156 lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/0 -u16')) 157 short_id = int(lines[0]) 158 assert leshan.read(endpoint, '1/0/0') == short_id 159 assert leshan.read(endpoint, '1/0/6') is False 160 assert leshan.read(endpoint, '1/0/7') == 'U' 161 leshan.format = fmt 162 163def test_LightweightM2M_1_1_int_212(shell: Shell, leshan: Leshan, endpoint: str): 164 """LightweightM2M-1.1-int-212 - Setting basic information in CBOR format""" 165 fmt = leshan.format 166 leshan.format = 'CBOR' 167 leshan.write(endpoint, '1/0/2', 101) 168 leshan.write(endpoint, '1/0/3', 1010) 169 leshan.write(endpoint, '1/0/6', True) 170 assert leshan.read(endpoint, '1/0/2') == 101 171 assert leshan.read(endpoint, '1/0/3') == 1010 172 assert leshan.read(endpoint, '1/0/6') is True 173 leshan.write(endpoint, '1/0/2', 1) 174 leshan.write(endpoint, '1/0/3', 10) 175 leshan.write(endpoint, '1/0/6', False) 176 leshan.format = fmt 177 178def verify_setting_basic_in_format(shell, leshan, endpoint, format): 179 fmt = leshan.format 180 leshan.format = format 181 server_obj = leshan.read(endpoint, '1/0') 182 verify_server_object(server_obj) 183 # Remove Read-Only resources, so we don't end up writing those 184 del server_obj[0][0] 185 del server_obj[0][13] 186 data = { 187 2: 101, 188 3: 1010, 189 5: 2000, 190 6: True, 191 7: 'U' 192 } 193 assert leshan.update_obj_instance(endpoint, '1/0', data)['status'] == 'CHANGED(204)' 194 resp = leshan.read(endpoint, '1/0') 195 assert resp[0][2] == 101 196 assert resp[0][3] == 1010 197 assert resp[0][5] == 2000 198 assert resp[0][6] is True 199 assert resp[0][7] == 'U' 200 assert leshan.replace_obj_instance(endpoint, '1/0', server_obj[0])['status'] == 'CHANGED(204)' 201 server_obj = leshan.read(endpoint, '1/0') 202 verify_server_object(server_obj) 203 leshan.format = fmt 204 205def test_LightweightM2M_1_1_int_215(shell: Shell, leshan: Leshan, endpoint: str): 206 """LightweightM2M-1.1-int-215 - Setting basic information in TLV format""" 207 verify_setting_basic_in_format(shell, leshan, endpoint, 'TLV') 208 209def test_LightweightM2M_1_1_int_220(shell: Shell, leshan: Leshan, endpoint: str): 210 """LightweightM2M-1.1-int-220 - Setting basic information in JSON format""" 211 verify_setting_basic_in_format(shell, leshan, endpoint, 'JSON') 212 213def test_LightweightM2M_1_1_int_221(shell: Shell, leshan: Leshan, endpoint: str): 214 """LightweightM2M-1.1-int-221 - Attempt to perform operations on Security""" 215 assert leshan.read(endpoint, '0/0')['status'] == 'UNAUTHORIZED(401)' 216 assert leshan.write(endpoint, '0/0/0', 'coap://localhost')['status'] == 'UNAUTHORIZED(401)' 217 assert leshan.write_attributes(endpoint, '0', {'pmin':10})['status'] == 'UNAUTHORIZED(401)' 218 219def test_LightweightM2M_1_1_int_222(shell: Shell, leshan: Leshan, endpoint: str): 220 """LightweightM2M-1.1-int-222 - Read on Object""" 221 resp = leshan.read(endpoint, '1') 222 assert len(resp) == 1 223 assert len(resp[1][0]) == 11 224 resp = leshan.read(endpoint, '3') 225 assert len(resp) == 1 226 assert len(resp[3]) == 1 227 assert len(resp[3][0]) == 15 228 assert resp[3][0][0] == 'Zephyr' 229 230def test_LightweightM2M_1_1_int_223(shell: Shell, leshan: Leshan, endpoint: str): 231 """LightweightM2M-1.1-int-223 - Read on Object Instance""" 232 resp = leshan.read(endpoint, '1/0') 233 assert len(resp[0]) == 11 234 resp = leshan.read(endpoint, '3/0') 235 assert len(resp[0]) == 15 236 assert resp[0][0] == 'Zephyr' 237 238def test_LightweightM2M_1_1_int_224(shell: Shell, leshan: Leshan, endpoint: str): 239 """LightweightM2M-1.1-int-224 - Read on Resource""" 240 assert leshan.read(endpoint, '1/0/0') == 1 241 assert leshan.read(endpoint, '1/0/1') == 86400 242 assert leshan.read(endpoint, '1/0/6') is False 243 assert leshan.read(endpoint, '1/0/7') == 'U' 244 245def test_LightweightM2M_1_1_int_225(shell: Shell, leshan: Leshan, endpoint: str): 246 """LightweightM2M-1.1-int-225 - Read on Resource Instance""" 247 assert leshan.read(endpoint, '3/0/11/0') == 0 248 249def test_LightweightM2M_1_1_int_226(shell: Shell, leshan: Leshan, endpoint: str): 250 """LightweightM2M-1.1-int-226 - Write (Partial Update) on Object Instance""" 251 lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/1 -u32')) 252 lifetime = int(lines[0]) 253 resources = { 254 1: 60, 255 6: True 256 } 257 assert leshan.update_obj_instance(endpoint, '1/0', resources)['status'] == 'CHANGED(204)' 258 assert leshan.read(endpoint, '1/0/1') == 60 259 assert leshan.read(endpoint, '1/0/6') is True 260 resources = { 261 1: lifetime, 262 6: False 263 } 264 assert leshan.update_obj_instance(endpoint, '1/0', resources)['status'] == 'CHANGED(204)' 265 266def test_LightweightM2M_1_1_int_227(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 267 """LightweightM2M-1.1-int-227 - Write (replace) on Resource""" 268 lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/1 -u32')) 269 lifetime = int(lines[0]) 270 assert leshan.write(endpoint, '1/0/1', int(63))['status'] == 'CHANGED(204)' 271 dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) 272 latest = leshan.get(f'/clients/{endpoint}') 273 assert latest["lifetime"] == 63 274 assert leshan.read(endpoint, '1/0/1') == 63 275 assert leshan.write(endpoint, '1/0/1', lifetime)['status'] == 'CHANGED(204)' 276 277def test_LightweightM2M_1_1_int_228(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 278 """LightweightM2M-1.1-int-228 - Write on Resource Instance""" 279 resources = { 280 0: {0: 'a', 1: 'b'} 281 } 282 assert leshan.create_obj_instance(endpoint, '16/0', resources)['status'] == 'CREATED(201)' 283 dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) 284 assert leshan.write(endpoint, '16/0/0/0', 'test')['status'] == 'CHANGED(204)' 285 assert leshan.read(endpoint, '16/0/0/0') == 'test' 286 287def test_LightweightM2M_1_1_int_229(shell: Shell, leshan: Leshan, endpoint: str): 288 """LightweightM2M-1.1-int-229 - Read-Composite Operation""" 289 old_fmt = leshan.format 290 for fmt in ['SENML_JSON', 'SENML_CBOR']: 291 leshan.format = fmt 292 resp = leshan.composite_read(endpoint, ['/3', '1/0']) 293 assert len(resp.keys()) == 2 294 assert resp[3] is not None 295 assert resp[1][0] is not None 296 assert len(resp[3][0]) == 15 297 assert len(resp[1][0]) == 11 298 299 resp = leshan.composite_read(endpoint, ['1/0/1', '/3/0/11/0']) 300 logger.debug(resp) 301 assert len(resp.keys()) == 2 302 assert resp[1][0][1] is not None 303 assert resp[3][0][11][0] is not None 304 leshan.format = old_fmt 305 306def test_LightweightM2M_1_1_int_230(shell: Shell, leshan: Leshan, endpoint: str): 307 """LightweightM2M-1.1-int-230 - Write-Composite Operation""" 308 resources = { 309 "/1/0/1": 60, 310 "/1/0/6": True, 311 "/16/0/0": { 312 "0": "aa", 313 "1": "bb", 314 "2": "cc", 315 "3": "dd" 316 } 317 } 318 old_fmt = leshan.format 319 for fmt in ['SENML_JSON', 'SENML_CBOR']: 320 leshan.format = fmt 321 assert leshan.composite_write(endpoint, resources)['status'] == 'CHANGED(204)' 322 resp = leshan.read(endpoint, '1/0') 323 assert resp[0][1] == 60 324 assert resp[0][6] is True 325 resp = leshan.read(endpoint, '16/0/0') 326 assert resp[0][0] == "aa" 327 assert resp[0][1] == "bb" 328 assert resp[0][2] == "cc" 329 assert resp[0][3] == "dd" 330 # Return to default 331 shell.exec_command('lwm2m write /1/0/1 -u32 86400') 332 shell.exec_command('lwm2m write /1/0/6 -u8 0') 333 leshan.format = old_fmt 334 335def query_basic_in_senml(leshan: Leshan, endpoint: str, fmt: str): 336 """Querying basic information in one of the SenML formats""" 337 old_fmt = leshan.format 338 leshan.format = fmt 339 verify_server_object(leshan.read(endpoint, '1')[1]) 340 verify_device_object(leshan.read(endpoint, '3/0')) 341 assert leshan.read(endpoint, '3/0/16') == 'U' 342 assert leshan.read(endpoint, '3/0/11/0') == 0 343 leshan.format = old_fmt 344 345def test_LightweightM2M_1_1_int_231(shell: Shell, leshan: Leshan, endpoint: str): 346 """LightweightM2M-1.1-int-231 - Querying basic information in SenML JSON format""" 347 query_basic_in_senml(leshan, endpoint, 'SENML_JSON') 348 349def test_LightweightM2M_1_1_int_232(shell: Shell, leshan: Leshan, endpoint: str): 350 """LightweightM2M-1.1-int-232 - Querying basic information in SenML CBOR format""" 351 query_basic_in_senml(leshan, endpoint, 'SENML_CBOR') 352 353def setting_basic_senml(shell: Shell, leshan: Leshan, endpoint: str, fmt: str): 354 """Setting basic information in one of the SenML formats""" 355 old_fmt = leshan.format 356 leshan.format = fmt 357 resources = { 358 1: 61, 359 6: True, 360 } 361 assert leshan.update_obj_instance(endpoint, '1/0', resources)['status'] == 'CHANGED(204)' 362 srv_obj = leshan.read(endpoint, '1/0') 363 assert srv_obj[0][1] == 61 364 assert srv_obj[0][6] is True 365 assert leshan.write(endpoint, '16/0/0/0', 'test_value')['status'] == 'CHANGED(204)' 366 portfolio = leshan.read(endpoint, '16') 367 assert portfolio[16][0][0][0] == 'test_value' 368 assert leshan.write(endpoint, '1/0/1', 63)['status'] == 'CHANGED(204)' 369 assert leshan.read(endpoint, '1/0/1') == 63 370 shell.exec_command('lwm2m write /1/0/1 -u32 86400') 371 shell.exec_command('lwm2m write /1/0/6 -u8 0') 372 leshan.format = old_fmt 373 374def test_LightweightM2M_1_1_int_233(shell: Shell, leshan: Leshan, endpoint: str): 375 """LightweightM2M-1.1-int-233 - Setting basic information in SenML CBOR format""" 376 setting_basic_senml(shell, leshan, endpoint, 'SENML_CBOR') 377 378def test_LightweightM2M_1_1_int_234(shell: Shell, leshan: Leshan, endpoint: str): 379 """LightweightM2M-1.1-int-234 - Setting basic information in SenML JSON format""" 380 setting_basic_senml(shell, leshan, endpoint, 'SENML_JSON') 381 382def test_LightweightM2M_1_1_int_235(leshan: Leshan, endpoint: str): 383 """LightweightM2M-1.1-int-235 - Read-Composite Operation on root path""" 384 resp = leshan.composite_read(endpoint, ['/']) 385 expected_keys = [1, 3, 5] 386 missing_keys = [key for key in expected_keys if key not in resp.keys()] 387 assert len(missing_keys) == 0 388 389def test_LightweightM2M_1_1_int_236(shell: Shell, leshan: Leshan, endpoint: str): 390 """LightweightM2M-1.1-int-236 - Read-Composite - Partial Presence""" 391 resp = leshan.composite_read(endpoint, ['1/0', '/3/0/11/0', '/3339/0/5522', '/3353/0/6030']) 392 assert resp[1][0][1] is not None 393 assert resp[3][0][11][0] is not None 394 assert len(resp) == 2 395 396def test_LightweightM2M_1_1_int_237(shell: Shell, leshan: Leshan, endpoint: str): 397 """LightweightM2M-1.1-int-237 - Read on Object without specifying Content-Type""" 398 old_fmt = leshan.format 399 leshan.format = None 400 assert leshan.read(endpoint, '1')[1][0][1] is not None 401 assert leshan.read(endpoint, '3')[3][0][0] == 'Zephyr' 402 leshan.format = old_fmt 403 404def test_LightweightM2M_1_1_int_241(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 405 """LightweightM2M-1.1-int-241 - Executable Resource: Rebooting the device""" 406 leshan.execute(endpoint, '3/0/4') 407 dut.readlines_until(regex='.*DEVICE: REBOOT', timeout=5.0) 408 dut.readlines_until(regex='.*rd_client_event: Disconnected', timeout=5.0) 409 shell.exec_command(f'lwm2m start {endpoint} -b 0') 410 dut.readlines_until(regex='.*Registration Done', timeout=5.0) 411 assert leshan.get(f'/clients/{endpoint}') 412 413def test_LightweightM2M_1_1_int_256(shell: Shell, leshan: Leshan, endpoint: str): 414 """LightweightM2M-1.1-int-256 - Write Operation Failure""" 415 lines = shell.get_filtered_output(shell.exec_command('lwm2m read 1/0/0 -u16')) 416 short_id = int(lines[0]) 417 assert leshan.write(endpoint, '1/0/0', 123)['status'] == 'METHOD_NOT_ALLOWED(405)' 418 assert leshan.read(endpoint, '1/0/0') == short_id 419 420def test_LightweightM2M_1_1_int_257(shell: Shell, leshan: Leshan, endpoint: str): 421 """LightweightM2M-1.1-int-257 - Write-Composite Operation""" 422 resources = { 423 "/1/0/2": 102, 424 "/1/0/6": True, 425 "/3/0/13": datetime.fromtimestamp(0) 426 } 427 old_fmt = leshan.format 428 for fmt in ['SENML_JSON', 'SENML_CBOR']: 429 leshan.format = fmt 430 assert leshan.composite_write(endpoint, resources)['status'] == 'CHANGED(204)' 431 assert leshan.read(endpoint, '1/0/2') == 102 432 assert leshan.read(endpoint, '1/0/6') is True 433 # Cannot verify the /3/0/13, it is a timestamp that moves forward. 434 435 # Return to default 436 shell.exec_command(f'lwm2m write /3/0/13 -u32 {int(datetime.now().timestamp())}') 437 shell.exec_command('lwm2m write /1/0/6 -u8 0') 438 shell.exec_command('lwm2m write /1/0/2 -u32 1') 439 leshan.format = old_fmt 440 441def test_LightweightM2M_1_1_int_260(shell: Shell, leshan: Leshan, endpoint: str): 442 """LightweightM2M-1.1-int-260 - Discover Command""" 443 resp = leshan.discover(endpoint, '3') 444 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'] 445 missing_keys = [key for key in expected_keys if key not in resp.keys()] 446 assert len(missing_keys) == 0 447 assert leshan.write_attributes(endpoint, '3', {'pmin': 10, 'pmax': 200})['status'] == 'CHANGED(204)' 448 resp = leshan.discover(endpoint, '3/0') 449 assert int(resp['/3/0/6']['dim']) == 2 450 assert int(resp['/3/0/7']['dim']) == 2 451 assert int(resp['/3/0/8']['dim']) == 2 452 assert leshan.write_attributes(endpoint, '3/0/7', {'lt': 1, 'gt': 6, 'st': 1})['status'] == 'CHANGED(204)' 453 resp = leshan.discover(endpoint, '3/0') 454 expected_keys = ['/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'] 455 missing_keys = [key for key in expected_keys if key not in resp.keys()] 456 assert len(missing_keys) == 0 457 assert int(resp['/3/0/7']['dim']) == 2 458 assert float(resp['/3/0/7']['lt']) == 1.0 459 assert float(resp['/3/0/7']['gt']) == 6.0 460 assert float(resp['/3/0/7']['st']) == 1.0 461 resp = leshan.discover(endpoint, '3/0/7') 462 expected_keys = ['/3/0/7', '/3/0/7/0', '/3/0/7/1'] 463 missing_keys = [key for key in expected_keys if key not in resp.keys()] 464 assert len(missing_keys) == 0 465 assert len(resp) == len(expected_keys) 466 # restore 467 leshan.remove_attributes(endpoint, '3', ['pmin', 'pmax']) 468 469def test_LightweightM2M_1_1_int_261(shell: Shell, leshan: Leshan, endpoint: str): 470 """LightweightM2M-1.1-int-261 - Write-Attribute Operation on a multiple resource""" 471 resp = leshan.discover(endpoint, '3/0/11') 472 logger.debug(resp) 473 expected_keys = ['/3/0/11', '/3/0/11/0'] 474 missing_keys = [key for key in expected_keys if key not in resp.keys()] 475 assert len(missing_keys) == 0 476 assert len(resp) == len(expected_keys) 477 assert int(resp['/3/0/11']['dim']) == 1 478 assert leshan.write_attributes(endpoint, '3', {'pmin':10, 'pmax':200})['status'] == 'CHANGED(204)' 479 assert leshan.write_attributes(endpoint, '3/0', {'pmax':320})['status'] == 'CHANGED(204)' 480 assert leshan.write_attributes(endpoint, '3/0/11/0', {'pmax':100, 'epmin':1, 'epmax':20})['status'] == 'CHANGED(204)' 481 resp = leshan.discover(endpoint, '3/0/11') 482 logger.debug(resp) 483 assert int(resp['/3/0/11']['pmin']) == 10 484 assert int(resp['/3/0/11']['pmax']) == 320 485 assert int(resp['/3/0/11/0']['pmax']) == 100 486 # Note: Zephyr does not support epmin&epmax. 487 # Restore 488 leshan.remove_attributes(endpoint, '3', ['pmin', 'pmax']) 489 leshan.remove_attributes(endpoint, '3/0', ['pmax']) 490 leshan.remove_attributes(endpoint, '3/0/11/0', ['pmax']) 491 492 493def test_LightweightM2M_1_1_int_280(shell: Shell, leshan: Leshan, endpoint: str): 494 """LightweightM2M-1.1-int-280 - Successful Read-Composite Operation""" 495 resp = leshan.composite_read(endpoint, ['/3/0/16', '/3/0/11/0', '/1/0']) 496 logger.debug(resp) 497 assert len(resp) == 2 498 assert len(resp[3]) == 1 499 assert len(resp[3][0]) == 2 # No extra resources 500 assert resp[3][0][11][0] == 0 501 assert resp[3][0][16] == 'U' 502 assert resp[1][0][0] == 1 503 assert resp[1][0][1] == 86400 504 assert resp[1][0][6] is False 505 assert resp[1][0][7] == 'U' 506 507def test_LightweightM2M_1_1_int_281(shell: Shell, leshan: Leshan, endpoint: str): 508 """LightweightM2M-1.1-int-281 - Partially Successful Read-Composite Operation""" 509 resp = leshan.composite_read(endpoint, ['/1/0/1', '/1/0/7', '/1/0/8']) 510 assert len(resp) == 1 511 assert len(resp[1][0]) == 2 # /1/0/8 should not be there 512 assert resp[1][0][1] == 86400 513 assert resp[1][0][7] == 'U' 514 515# 516# Information Reporting Interface [300-399] 517# 518 519@pytest.mark.slow 520def test_LightweightM2M_1_1_int_301(shell: Shell, leshan: Leshan, endpoint: str): 521 """LightweightM2M-1.1-int-301 - Observation and Notification of parameter values""" 522 pwr_src = leshan.read(endpoint, '3/0/6') 523 logger.debug(pwr_src) 524 assert pwr_src[6][0] == 1 525 assert pwr_src[6][1] == 5 526 assert leshan.write_attributes(endpoint, '3/0/7', {'pmin': 5, 'pmax': 10})['status'] == 'CHANGED(204)' 527 leshan.observe(endpoint, '3/0/7') 528 with leshan.get_event_stream(endpoint, timeout=30) as events: 529 shell.exec_command('lwm2m write /3/0/7/0 -u32 3000') 530 data = events.next_event('NOTIFICATION') 531 assert data is not None 532 assert data[3][0][7][0] == 3000 533 # Ensure that we don't get new data before pMin 534 start = time.time() 535 shell.exec_command('lwm2m write /3/0/7/0 -u32 3500') 536 data = events.next_event('NOTIFICATION') 537 assert data[3][0][7][0] == 3500 538 assert (start + 5) < time.time() + 0.5 # Allow 0.5 second diff 539 assert (start + 5) > time.time() - 0.5 540 # Ensure that we get update when pMax expires 541 data = events.next_event('NOTIFICATION') 542 assert data[3][0][7][0] == 3500 543 assert (start + 15) <= time.time() + 1 # Allow 1 second slack. (pMinx + pMax=15) 544 leshan.cancel_observe(endpoint, '3/0/7') 545 leshan.remove_attributes(endpoint, '3/0/7', ['pmin', 'pmax']) 546 547def test_LightweightM2M_1_1_int_302(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 548 """LightweightM2M-1.1-int-302 - Cancel Observations using Reset Operation""" 549 leshan.observe(endpoint, '3/0/7') 550 leshan.observe(endpoint, '3/0/8') 551 with leshan.get_event_stream(endpoint) as events: 552 shell.exec_command('lwm2m write /3/0/7/0 -u32 4000') 553 data = events.next_event('NOTIFICATION') 554 assert data[3][0][7][0] == 4000 555 leshan.passive_cancel_observe(endpoint, '3/0/7') 556 shell.exec_command('lwm2m write /3/0/7/0 -u32 3000') 557 dut.readlines_until(regex=r'.*Observer removed for 3/0/7') 558 with leshan.get_event_stream(endpoint) as events: 559 shell.exec_command('lwm2m write /3/0/8/0 -u32 100') 560 data = events.next_event('NOTIFICATION') 561 assert data[3][0][8][0] == 100 562 leshan.passive_cancel_observe(endpoint, '3/0/8') 563 shell.exec_command('lwm2m write /3/0/8/0 -u32 50') 564 dut.readlines_until(regex=r'.*Observer removed for 3/0/8') 565 566def test_LightweightM2M_1_1_int_303(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 567 """LightweightM2M-1.1-int-303 - Cancel observations using Observe with Cancel parameter""" 568 leshan.observe(endpoint, '3/0/7') 569 leshan.observe(endpoint, '3/0/8') 570 with leshan.get_event_stream(endpoint) as events: 571 shell.exec_command('lwm2m write /3/0/7/0 -u32 4000') 572 data = events.next_event('NOTIFICATION') 573 assert data[3][0][7][0] == 4000 574 leshan.cancel_observe(endpoint, '3/0/7') 575 dut.readlines_until(regex=r'.*Observer removed for 3/0/7') 576 with leshan.get_event_stream(endpoint) as events: 577 shell.exec_command('lwm2m write /3/0/8/0 -u32 100') 578 data = events.next_event('NOTIFICATION') 579 assert data[3][0][8][0] == 100 580 leshan.cancel_observe(endpoint, '3/0/8') 581 dut.readlines_until(regex=r'.*Observer removed for 3/0/8') 582 583@pytest.mark.slow 584def test_LightweightM2M_1_1_int_304(shell: Shell, leshan: Leshan, endpoint: str): 585 """LightweightM2M-1.1-int-304 - Observe-Composite Operation""" 586 # Need to use Configuration C.1 587 shell.exec_command('lwm2m write 1/0/2 -u32 0') 588 shell.exec_command('lwm2m write 1/0/3 -u32 0') 589 assert leshan.write_attributes(endpoint, '1/0/1', {'pmin': 30, 'pmax': 45})['status'] == 'CHANGED(204)' 590 data = leshan.composite_observe(endpoint, ['/1/0/1', '/3/0/11/0', '/3/0/16']) 591 assert data[1][0][1] is not None 592 assert data[3][0][11][0] is not None 593 assert data[3][0][16] == 'U' 594 assert len(data) == 2 595 assert len(data[1]) == 1 596 assert len(data[3][0]) == 2 597 start = time.time() 598 with leshan.get_event_stream(endpoint, timeout=50) as events: 599 data = events.next_event('NOTIFICATION') 600 logger.debug(data) 601 assert data[1][0][1] is not None 602 assert data[3][0][11][0] is not None 603 assert data[3][0][16] == 'U' 604 assert len(data) == 2 605 assert len(data[1]) == 1 606 assert len(data[3][0]) == 2 607 assert (start + 30) < time.time() 608 assert (start + 45) > time.time() - 1 609 leshan.cancel_composite_observe(endpoint, ['/1/0/1', '/3/0/11/0', '/3/0/16']) 610 # Restore configuration C.3 611 shell.exec_command('lwm2m write 1/0/2 -u32 1') 612 shell.exec_command('lwm2m write 1/0/3 -u32 10') 613 leshan.remove_attributes(endpoint, '1/0/1', ['pmin', 'pmax']) 614 615def test_LightweightM2M_1_1_int_305(dut: DeviceAdapter, leshan: Leshan, endpoint: str): 616 """LightweightM2M-1.1-int-305 - Cancel Observation-Composite Operation""" 617 leshan.composite_observe(endpoint, ['/1/0/1', '/3/0/11/0', '/3/0/16']) 618 leshan.cancel_composite_observe(endpoint, ['/1/0/1', '/3/0/11/0', '/3/0/16']) 619 dut.readlines_until(regex=r'.*Observer removed for 1/0/1') 620 dut.readlines_until(regex=r'.*Observer removed for 3/0/11/0') 621 dut.readlines_until(regex=r'.*Observer removed for 3/0/16') 622 623def test_LightweightM2M_1_1_int_306(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 624 """LightweightM2M-1.1-int-306 - Send Operation""" 625 with leshan.get_event_stream(endpoint) as events: 626 shell.exec_command('lwm2m send /1 /3') 627 dut.readlines_until(regex=r'.*SEND status: 0', timeout=5.0) 628 data = events.next_event('SEND') 629 assert data is not None 630 verify_server_object(data[1]) 631 verify_device_object(data[3]) 632 633def test_LightweightM2M_1_1_int_307(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 634 """LightweightM2M-1.1-int-307 - Muting Send""" 635 leshan.write(endpoint, '1/0/23', True) 636 lines = shell.get_filtered_output(shell.exec_command('lwm2m send /3/0')) 637 assert any("can't do send operation" in line for line in lines) 638 leshan.write(endpoint, '1/0/23', False) 639 shell.exec_command('lwm2m send /3/0') 640 dut.readlines_until(regex=r'.*SEND status: 0', timeout=5.0) 641 642 643@pytest.mark.slow 644def test_LightweightM2M_1_1_int_308(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 645 """LightweightM2M-1.1-int-308 - Observe-Composite and Creating Object Instance""" 646 shell.exec_command('lwm2m delete /16/0') 647 shell.exec_command('lwm2m delete /16/1') 648 # Need to use Configuration C.1 649 shell.exec_command('lwm2m write 1/0/2 -u32 0') 650 shell.exec_command('lwm2m write 1/0/3 -u32 0') 651 resources_a = { 652 0: {0: 'aa', 653 1: 'bb', 654 2: 'cc', 655 3: 'dd'} 656 } 657 content_one = {16: {0: resources_a}} 658 resources_b = { 659 0: {0: '11', 660 1: '22', 661 2: '33', 662 3: '44'} 663 } 664 content_both = {16: {0: resources_a, 1: resources_b}} 665 assert leshan.create_obj_instance(endpoint, '16/0', resources_a)['status'] == 'CREATED(201)' 666 dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) 667 assert leshan.write_attributes(endpoint, '16/0', {'pmin': 30, 'pmax': 45})['status'] == 'CHANGED(204)' 668 data = leshan.composite_observe(endpoint, ['/16/0', '/16/1']) 669 assert data == content_one 670 with leshan.get_event_stream(endpoint, timeout=50) as events: 671 data = events.next_event('NOTIFICATION') 672 start = time.time() 673 assert data == content_one 674 assert leshan.create_obj_instance(endpoint, '16/1', resources_b)['status'] == 'CREATED(201)' 675 data = events.next_event('NOTIFICATION') 676 assert (start + 30) < time.time() + 2 677 assert (start + 45) > time.time() - 2 678 assert data == content_both 679 leshan.cancel_composite_observe(endpoint, ['/16/0', '/16/1']) 680 # Restore configuration C.3 681 shell.exec_command('lwm2m write 1/0/2 -u32 1') 682 shell.exec_command('lwm2m write 1/0/3 -u32 10') 683 leshan.remove_attributes(endpoint, '16/0', ['pmin','pmax']) 684 685@pytest.mark.slow 686def test_LightweightM2M_1_1_int_309(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str): 687 """LightweightM2M-1.1-int-309 - Observe-Composite and Deleting Object Instance""" 688 shell.exec_command('lwm2m delete /16/0') 689 shell.exec_command('lwm2m delete /16/1') 690 # Need to use Configuration C.1 691 shell.exec_command('lwm2m write 1/0/2 -u32 0') 692 shell.exec_command('lwm2m write 1/0/3 -u32 0') 693 resources_a = { 694 0: {0: 'aa', 695 1: 'bb', 696 2: 'cc', 697 3: 'dd'} 698 } 699 content_one = {16: {0: resources_a}} 700 resources_b = { 701 0: {0: '11', 702 1: '22', 703 2: '33', 704 3: '44'} 705 } 706 content_both = {16: {0: resources_a, 1: resources_b}} 707 assert leshan.create_obj_instance(endpoint, '16/0', resources_a)['status'] == 'CREATED(201)' 708 assert leshan.create_obj_instance(endpoint, '16/1', resources_b)['status'] == 'CREATED(201)' 709 dut.readlines_until(regex='.*net_lwm2m_rd_client: Update Done', timeout=5.0) 710 assert leshan.write_attributes(endpoint, '16/0', {'pmin': 30, 'pmax': 45})['status'] == 'CHANGED(204)' 711 data = leshan.composite_observe(endpoint, ['/16/0', '/16/1']) 712 assert data == content_both 713 with leshan.get_event_stream(endpoint, timeout=50) as events: 714 data = events.next_event('NOTIFICATION') 715 start = time.time() 716 assert data == content_both 717 assert leshan.delete(endpoint, '16/1')['status'] == 'DELETED(202)' 718 data = events.next_event('NOTIFICATION') 719 assert (start + 30) < time.time() + 2 720 assert (start + 45) > time.time() - 2 721 assert data == content_one 722 leshan.cancel_composite_observe(endpoint, ['/16/0', '/16/1']) 723 # Restore configuration C.3 724 shell.exec_command('lwm2m write 1/0/2 -u32 1') 725 shell.exec_command('lwm2m write 1/0/3 -u32 10') 726 leshan.remove_attributes(endpoint, '16/0', ['pmin', 'pmax']) 727 728@pytest.mark.slow 729def test_LightweightM2M_1_1_int_310(shell: Shell, leshan: Leshan, endpoint: str): 730 """LightweightM2M-1.1-int-310 - Observe-Composite and modification of parameter values""" 731 # Need to use Configuration C.1 732 shell.exec_command('lwm2m write 1/0/2 -u32 0') 733 shell.exec_command('lwm2m write 1/0/3 -u32 0') 734 leshan.composite_observe(endpoint, ['/1/0/1', '/3/0']) 735 with leshan.get_event_stream(endpoint, timeout=50) as events: 736 assert leshan.write_attributes(endpoint, '3', {'pmax': 5})['status'] == 'CHANGED(204)' 737 start = time.time() 738 data = events.next_event('NOTIFICATION') 739 assert data[3][0][0] == 'Zephyr' 740 assert data[1] == {0: {1: 86400}} 741 assert (start + 5) > time.time() - 1 742 start = time.time() 743 data = events.next_event('NOTIFICATION') 744 assert (start + 5) > time.time() - 1 745 leshan.cancel_composite_observe(endpoint, ['/1/0/1', '/3/0']) 746 # Restore configuration C.3 747 shell.exec_command('lwm2m write 1/0/2 -u32 1') 748 shell.exec_command('lwm2m write 1/0/3 -u32 10') 749 leshan.remove_attributes(endpoint, '3', ['pmax']) 750 751def test_LightweightM2M_1_1_int_311(shell: Shell, leshan: Leshan, endpoint: str): 752 """LightweightM2M-1.1-int-311 - Send command""" 753 with leshan.get_event_stream(endpoint, timeout=50) as events: 754 shell.exec_command('lwm2m send /1/0/1 /3/0/11') 755 data = events.next_event('SEND') 756 assert data == {3: {0: {11: {0: 0}}}, 1: {0: {1: 86400}}} 757
