test: lwm2m: Test cancellation using observe parameter
Implement test cases: LightweightM2M-1.1-int-303 - Cancel observations using Observe with Cancel parameter LightweightM2M-1.1-int-305 - Cancel Observation-Composite Operation Modify existing Leshan API to passive_cancel(). Signed-off-by: Seppo Takalo <seppo.takalo@nordicsemi.no>
This commit is contained in:
parent
2135e009e4
commit
31e9a56742
|
@ -172,9 +172,9 @@ Tests are written from test spec;
|
|||
|LightweightM2M-1.1-int-281 - Partially Successful Read-Composite Operation|:white_check_mark:| |
|
||||
|LightweightM2M-1.1-int-301 - Observation and Notification of parameter values|:white_check_mark:| |
|
||||
|LightweightM2M-1.1-int-302 - Cancel Observations using Reset Operation|:white_check_mark:| |
|
||||
|LightweightM2M-1.1-int-303 - Cancel observations using Observe with Cancel parameter|:large_orange_diamond:|Leshan only supports passive cancelling|
|
||||
|LightweightM2M-1.1-int-303 - Cancel observations using Observe with Cancel parameter|:white_check_mark:||
|
||||
|LightweightM2M-1.1-int-304 - Observe-Composite Operation|:white_check_mark:| |
|
||||
|LightweightM2M-1.1-int-305 - Cancel Observation-Composite Operation|:large_orange_diamond:|Leshan only supports passive cancelling|
|
||||
|LightweightM2M-1.1-int-305 - Cancel Observation-Composite Operation|:white_check_mark:| |
|
||||
|LightweightM2M-1.1-int-306 – Send Operation|:white_check_mark:|[~~#64290~~](https://github.com/zephyrproject-rtos/zephyr/issues/64290)|
|
||||
|LightweightM2M-1.1-int-307 – Muting Send|:white_check_mark:| |
|
||||
|LightweightM2M-1.1-int-308 - Observe-Composite and Creating Object Instance|:white_check_mark:|[~~#64634~~](https://github.com/zephyrproject-rtos/zephyr/issues/64634)|
|
||||
|
|
|
@ -389,6 +389,9 @@ class Leshan:
|
|||
return self.post(f'/clients/{endpoint}/{path}/observe', data="")
|
||||
|
||||
def cancel_observe(self, endpoint: str, path: str):
|
||||
return self.delete_raw(f'/clients/{endpoint}/{path}/observe?active')
|
||||
|
||||
def passive_cancel_observe(self, endpoint: str, path: str):
|
||||
return self.delete_raw(f'/clients/{endpoint}/{path}/observe')
|
||||
|
||||
def composite_observe(self, endpoint: str, paths: list[str]):
|
||||
|
@ -398,6 +401,10 @@ class Leshan:
|
|||
return self.parse_composite(payload)
|
||||
|
||||
def cancel_composite_observe(self, endpoint: str, paths: list[str]):
|
||||
paths = [path if path.startswith('/') else '/' + path for path in paths]
|
||||
return self.delete_raw(f'/clients/{endpoint}/composite/observe?paths=' + ','.join(paths) + '&active')
|
||||
|
||||
def passive_cancel_composite_observe(self, endpoint: str, paths: list[str]):
|
||||
paths = [path if path.startswith('/') else '/' + path for path in paths]
|
||||
return self.delete_raw(f'/clients/{endpoint}/composite/observe?paths=' + ','.join(paths))
|
||||
|
||||
|
|
|
@ -532,7 +532,6 @@ def test_LightweightM2M_1_1_int_301(shell: Shell, leshan: Leshan, endpoint: str)
|
|||
leshan.cancel_observe(endpoint, '3/0/7')
|
||||
leshan.remove_attributes(endpoint, '3/0/7', ['pmin', 'pmax'])
|
||||
|
||||
@pytest.mark.slow
|
||||
def test_LightweightM2M_1_1_int_302(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str):
|
||||
"""LightweightM2M-1.1-int-302 - Cancel Observations using Reset Operation"""
|
||||
leshan.observe(endpoint, '3/0/7')
|
||||
|
@ -541,17 +540,34 @@ def test_LightweightM2M_1_1_int_302(shell: Shell, dut: DeviceAdapter, leshan: Le
|
|||
shell.exec_command('lwm2m write /3/0/7/0 -u32 4000')
|
||||
data = events.next_event('NOTIFICATION')
|
||||
assert data[3][0][7][0] == 4000
|
||||
leshan.cancel_observe(endpoint, '3/0/7')
|
||||
leshan.passive_cancel_observe(endpoint, '3/0/7')
|
||||
shell.exec_command('lwm2m write /3/0/7/0 -u32 3000')
|
||||
dut.readlines_until(regex=r'.*Observer removed for 3/0/7')
|
||||
with leshan.get_event_stream(endpoint) as events:
|
||||
shell.exec_command('lwm2m write /3/0/8/0 -u32 100')
|
||||
data = events.next_event('NOTIFICATION')
|
||||
assert data[3][0][8][0] == 100
|
||||
leshan.cancel_observe(endpoint, '3/0/8')
|
||||
leshan.passive_cancel_observe(endpoint, '3/0/8')
|
||||
shell.exec_command('lwm2m write /3/0/8/0 -u32 50')
|
||||
dut.readlines_until(regex=r'.*Observer removed for 3/0/8')
|
||||
|
||||
def test_LightweightM2M_1_1_int_303(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str):
|
||||
"""LightweightM2M-1.1-int-303 - Cancel observations using Observe with Cancel parameter"""
|
||||
leshan.observe(endpoint, '3/0/7')
|
||||
leshan.observe(endpoint, '3/0/8')
|
||||
with leshan.get_event_stream(endpoint) as events:
|
||||
shell.exec_command('lwm2m write /3/0/7/0 -u32 4000')
|
||||
data = events.next_event('NOTIFICATION')
|
||||
assert data[3][0][7][0] == 4000
|
||||
leshan.cancel_observe(endpoint, '3/0/7')
|
||||
dut.readlines_until(regex=r'.*Observer removed for 3/0/7')
|
||||
with leshan.get_event_stream(endpoint) as events:
|
||||
shell.exec_command('lwm2m write /3/0/8/0 -u32 100')
|
||||
data = events.next_event('NOTIFICATION')
|
||||
assert data[3][0][8][0] == 100
|
||||
leshan.cancel_observe(endpoint, '3/0/8')
|
||||
dut.readlines_until(regex=r'.*Observer removed for 3/0/8')
|
||||
|
||||
@pytest.mark.slow
|
||||
def test_LightweightM2M_1_1_int_304(shell: Shell, leshan: Leshan, endpoint: str):
|
||||
"""LightweightM2M-1.1-int-304 - Observe-Composite Operation"""
|
||||
|
@ -584,6 +600,14 @@ def test_LightweightM2M_1_1_int_304(shell: Shell, leshan: Leshan, endpoint: str)
|
|||
shell.exec_command('lwm2m write 1/0/3 -u32 10')
|
||||
leshan.remove_attributes(endpoint, '1/0/1', ['pmin', 'pmax'])
|
||||
|
||||
def test_LightweightM2M_1_1_int_305(dut: DeviceAdapter, leshan: Leshan, endpoint: str):
|
||||
"""LightweightM2M-1.1-int-305 - Cancel Observation-Composite Operation"""
|
||||
leshan.composite_observe(endpoint, ['/1/0/1', '/3/0/11/0', '/3/0/16'])
|
||||
leshan.cancel_composite_observe(endpoint, ['/1/0/1', '/3/0/11/0', '/3/0/16'])
|
||||
dut.readlines_until(regex=r'.*Observer removed for 1/0/1')
|
||||
dut.readlines_until(regex=r'.*Observer removed for 3/0/11/0')
|
||||
dut.readlines_until(regex=r'.*Observer removed for 3/0/16')
|
||||
|
||||
def test_LightweightM2M_1_1_int_306(shell: Shell, dut: DeviceAdapter, leshan: Leshan, endpoint: str):
|
||||
"""LightweightM2M-1.1-int-306 - Send Operation"""
|
||||
with leshan.get_event_stream(endpoint) as events:
|
||||
|
|
Loading…
Reference in a new issue