Jack Nagy
231e88a8c6
A notification carries only the first block of a large representation (RFC 7959 §2.6). _dispatch_coap handed that block straight to on_notification, so consumers decoded a truncated CBOR buffer. Reported twice on /mode/vs/0: #37 and mbillow/localthings#361. The recovery is a re-read from block 0 on a fresh 4-byte one-shot token, not a §2.6 continuation. §3.4 rules out reusing the observation's token, and this server drops a transfer that opens at NUM>0 under a token it has not seen, so a continuation is the one shape that cannot work here. A truncated notification is now withheld and queued to a worker thread that re-reads the resource and delivers the reassembled representation. When the re-read fails the notification is dropped at debug level and the poll tiers carry freshness, which is what they already did. The re-read has to run off the reader thread: _dispatch_coap runs there and the transfer waits on an event only that same thread can set. The worker is serialized and paces between transfers, so a notification storm stays under the firmware request ceiling. Also in the Block2 loop, now extracted and shared by both paths: - compare the response's Block2 NUM against the one requested, so a retransmitted block is no longer concatenated as if it were the next - compare ETags across blocks (§2.4) and restart once when the representation changes mid-transfer - recompute the next block number from the accumulated byte offset when the server negotiates the block size down - re-check reader liveness while waiting on a block, so a mid-transfer reader death fails fast instead of burning the whole timeout - guard the token counters and _pending with a lock, now that the session issues concurrent reads of its own Closes #39
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
import pytest
|
|
|
|
from smartthings_local.errors import MalformedMessageError
|
|
from smartthings_local.protocol.coap import (
|
|
build_coap, parse_coap, encode_options, block_value, block_fields,
|
|
fmt_code,
|
|
TYPE_CON, METHOD_GET, URI_PATH, ACCEPT, CF_CBOR, BLOCK2,
|
|
)
|
|
|
|
|
|
def test_build_then_parse_roundtrip_no_payload():
|
|
opts = [(URI_PATH, b'device'), (URI_PATH, b'0'), (ACCEPT, CF_CBOR)]
|
|
datagram = build_coap(TYPE_CON, METHOD_GET, 0xABCD, b'\x01\x02', opts)
|
|
mtype, code, mid, tok, parsed_opts, payload = parse_coap(datagram)
|
|
assert mtype == TYPE_CON
|
|
assert code == METHOD_GET
|
|
assert mid == 0xABCD
|
|
assert tok == b'\x01\x02'
|
|
assert payload == b''
|
|
assert sorted(parsed_opts) == sorted(opts)
|
|
|
|
|
|
def test_build_then_parse_roundtrip_with_payload():
|
|
datagram = build_coap(TYPE_CON, 0x45, 1, b'\xff', [], payload=b'\xa1\x01\x02')
|
|
_, code, _, _, _, payload = parse_coap(datagram)
|
|
assert code == 0x45
|
|
assert payload == b'\xa1\x01\x02'
|
|
|
|
|
|
def test_encode_options_orders_by_option_number():
|
|
# ACCEPT (17) must be encoded after URI_PATH (11) regardless of input order
|
|
encoded_in_order = encode_options([(URI_PATH, b'x'), (ACCEPT, CF_CBOR)])
|
|
encoded_reversed = encode_options([(ACCEPT, CF_CBOR), (URI_PATH, b'x')])
|
|
assert encoded_in_order == encoded_reversed
|
|
|
|
|
|
def test_block_value_encodes_num_more_szx():
|
|
# num=2, more=1, szx=6 -> (2<<4)|(1<<3)|6 = 0x2E
|
|
assert block_value(2, 1, 6) == bytes([0x2E])
|
|
|
|
|
|
def test_block_value_promotes_to_two_bytes_when_num_is_large():
|
|
v = block_value(num=0xFFF, more=0, szx=0)
|
|
assert len(v) == 2
|
|
|
|
|
|
@pytest.mark.parametrize('num, more, szx', [
|
|
(0, 0, 0),
|
|
(0, 1, 6),
|
|
(2, 1, 6),
|
|
(1, 0, 4),
|
|
(0xFFF, 0, 0),
|
|
(0xFFFFF, 1, 7),
|
|
])
|
|
def test_block_fields_inverts_block_value(num, more, szx):
|
|
assert block_fields(block_value(num, more, szx)) == (num, more, szx)
|
|
|
|
|
|
def test_block_fields_treats_empty_value_as_block_zero():
|
|
# RFC 7959 §2.2: a zero-length Block option means num=0, m=0, szx=0.
|
|
assert block_fields(b'') == (0, 0, 0)
|
|
|
|
|
|
def test_fmt_code_formats_class_dot_detail():
|
|
assert fmt_code(0x45) == '2.05'
|
|
assert fmt_code(0x84) == '4.04'
|
|
|
|
|
|
@pytest.mark.parametrize('option_header', (b'\xf0', b'\x0f'))
|
|
def test_reserved_option_nibbles_raise_classified_value_error(option_header):
|
|
datagram = b'\x40\x01\x00\x01' + option_header
|
|
|
|
with pytest.raises(MalformedMessageError) as exc:
|
|
parse_coap(datagram)
|
|
|
|
assert isinstance(exc.value, ValueError)
|