tim
452a3bc207
The upstream River 3 schema spells field 355 `ups_alram`. Use `ups_alarm` in the decoder, the generated schema, and the field reference. The schema generator now applies a rename map, so a regenerated schema keeps the corrected name.
209 lines
7.9 KiB
Python
209 lines
7.9 KiB
Python
"""Schema-based diagnostic rendering with synthetic data only."""
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import re
|
|
import struct
|
|
|
|
import pytest
|
|
|
|
from bmspy.ecoflow_proto import (
|
|
decode_payload, decode_schema_protobuf, describe_protobuf_field,
|
|
)
|
|
from bmspy.ecoflow_river3 import decode_river3_payload
|
|
from bmspy.ecoflow_river3_schema import ENUMS, MESSAGES, SCHEMA_COMMIT
|
|
|
|
|
|
def varint(value):
|
|
value &= (1 << 64) - 1
|
|
result = bytearray()
|
|
while value > 127:
|
|
result.append((value & 127) | 128)
|
|
value >>= 7
|
|
result.append(value)
|
|
return bytes(result)
|
|
|
|
|
|
def integer(number, value):
|
|
return varint(number << 3) + varint(value)
|
|
|
|
|
|
def blob(number, data):
|
|
return varint(number << 3 | 2) + varint(len(data)) + data
|
|
|
|
|
|
def floating(number, value):
|
|
return varint(number << 3 | 5) + struct.pack('<f', value)
|
|
|
|
|
|
def display(data):
|
|
return '\n'.join(decode_payload(2, 0xFE, 0x15, data))
|
|
|
|
|
|
def sample(number, kind):
|
|
if kind in MESSAGES or kind == 'string':
|
|
return blob(number, b'')
|
|
if kind == 'float':
|
|
return floating(number, 1.5)
|
|
assert kind in ('bool', 'uint32', 'int32') or kind in ENUMS
|
|
return integer(number, 0)
|
|
|
|
|
|
@pytest.mark.parametrize('message,cmd_id,count', [
|
|
('DisplayPropertyUpload', 0x15, 296), ('RuntimePropertyUpload', 0x16, 98),
|
|
])
|
|
def test_every_pinned_top_level_field_has_a_label(message, cmd_id, count):
|
|
assert len(MESSAGES[message]) == count
|
|
for number, (name, kind, repeated) in MESSAGES[message].items():
|
|
text = '\n'.join(decode_payload(2, 0xFE, cmd_id, sample(number, kind)))
|
|
assert name + ':' in text
|
|
assert describe_protobuf_field(2, 0xFE, cmd_id, number) == name
|
|
assert not re.search(r'\bfield\s+\d+:', text)
|
|
assert not any(word in text for word in ('<invalid', '<malformed', '<wire type'))
|
|
|
|
|
|
def test_error_records_have_named_children_and_preserve_empty_records():
|
|
item = integer(1, 42) + integer(2, 1234)
|
|
text = display(blob(141, blob(1, item) + blob(1, b'')))
|
|
assert 'err_code_record_list: {' in text
|
|
assert 'list_info: {' in text
|
|
assert 'errcode: 42' in text
|
|
assert 'errcode_timestamp: 1234' in text
|
|
assert 'list_info: {}' in text
|
|
assert text.count('errcode:') == 1 # Empty records do not become invented errors.
|
|
assert not re.search(r'\bfield\s+\d+:', text)
|
|
|
|
|
|
def test_statistics_have_named_categories_without_invented_energy_units():
|
|
items = blob(1, integer(1, 2) + integer(2, 17))
|
|
items += blob(1, integer(1, 4)) # Content absent, not an explicit zero.
|
|
items += blob(1, b'')
|
|
text = display(blob(463, items))
|
|
assert 'display_statistics_sum:' in text
|
|
assert 'statistics_object: STATISTICS_OBJECT_AC_OUT_ENERGY (2)' in text
|
|
assert 'statistics_object: STATISTICS_OBJECT_TYPEC_OUT_ENERGY (4)' in text
|
|
assert 'statistics_content: 17' in text
|
|
assert text.count('statistics_content:') == 1
|
|
assert 'list_info: {}' in text
|
|
assert ' Wh' not in text
|
|
assert not re.search(r'\bfield\s+\d+:', text)
|
|
|
|
|
|
def test_strings_are_not_guessed_as_nested_messages():
|
|
text = display(blob(88, b'') + blob(433, b'TEST-ACCESSORY') + blob(91, b''))
|
|
assert "plug_in_info_dcp2_sn: ''" in text
|
|
assert "plug_in_info_dcp_sn: 'TEST-ACCESSORY'" in text
|
|
assert 'plug_in_info_dcp2_resv: {}' in text
|
|
invalid = display(blob(433, b'\xff'))
|
|
assert 'plug_in_info_dcp_sn: <invalid string:' in invalid
|
|
assert 'raw:' in invalid
|
|
|
|
|
|
def test_packed_and_unpacked_repeated_integers():
|
|
values = [0, 127, 128, 0xFFFFFFFF]
|
|
packed = display(blob(432, blob(1, b''.join(varint(n) for n in values))))
|
|
assert 'resv_info: [0, 127, 128, 4294967295]' in packed
|
|
unpacked = display(blob(432, b''.join(integer(1, n) for n in values)))
|
|
for value in values:
|
|
assert f'resv_info: {value}' in unpacked
|
|
assert 'resv_info: []' in display(blob(432, blob(1, b'')))
|
|
assert 'resv_info: <invalid uint32:' in display(blob(432, blob(1, b'\x80')))
|
|
|
|
|
|
def test_signed_boolean_enum_and_unknown_enum_values():
|
|
assert 'bms_min_cell_temp: -7 °C' in display(integer(258, -7))
|
|
assert 'xboost_en: true' in display(integer(25, 1))
|
|
assert 'xboost_en: false' in display(integer(25, 0))
|
|
assert 'statistics_object: 999 (unknown STATISTICS_OBJECT)' in display(blob(463, blob(1, integer(1, 999))))
|
|
|
|
|
|
def test_known_field_with_wrong_encoding_retains_label_and_raw_bytes():
|
|
text = display(integer(3, 10))
|
|
assert 'pow_in_sum_w: <wire type 0, expected 5, raw:' in text
|
|
assert '10 W' not in text
|
|
assert 'pow_get_bms: <wire type 2, expected 5' in display(blob(158, b'bad'))
|
|
|
|
|
|
@pytest.mark.parametrize('bad', [b'\x00', b'\x80', b'\x0a\x05\x01', b'\x0d\x00', b'\x09\x00'])
|
|
def test_malformed_reports_are_not_normal_measurements(bad):
|
|
text = display(floating(262, 89) + bad)
|
|
assert '<malformed DisplayPropertyUpload:' in text
|
|
assert 'raw:' in text
|
|
assert 'cms_batt_soc: 89' not in text
|
|
assert decode_river3_payload(2, 0xFE, 0x15, floating(262, 89) + bad) is None
|
|
|
|
|
|
def test_unknown_future_fields_remain_numeric_and_raw():
|
|
text = display(integer(999, 7) + blob(1000, b'opaque'))
|
|
assert 'field 999: 7' in text
|
|
assert 'field 1000:' in text
|
|
assert b'opaque'.hex() in text
|
|
assert 'unknown' in text
|
|
assert decode_payload(0x35, 0x35, 0x20, bytes(8)) is None
|
|
|
|
|
|
def test_runtime_routing_stays_explicitly_unconfirmed():
|
|
text = '\n'.join(decode_payload(2, 0xFE, 0x16, floating(244, 19.2)))
|
|
assert 'routing unconfirmed' in text
|
|
assert 'bms_batt_vol: 19.200 V' in text
|
|
assert decode_river3_payload(2, 0xFE, 0x16, floating(244, 19.2)) is None
|
|
|
|
|
|
def test_schema_nesting_has_a_depth_limit(monkeypatch):
|
|
monkeypatch.setitem(MESSAGES, 'SyntheticRecursive', {1: ('child', 'SyntheticRecursive', False)})
|
|
payload = b''
|
|
for _ in range(30):
|
|
payload = blob(1, payload)
|
|
text = '\n'.join(decode_schema_protobuf(payload, 'SyntheticRecursive'))
|
|
assert 'maximum nesting depth reached' in text
|
|
assert 'raw:' in text
|
|
|
|
|
|
@pytest.fixture
|
|
def generator():
|
|
path = Path(__file__).resolve().parents[1] / 'tools/generate_ecoflow_river3_schema.py'
|
|
spec = importlib.util.spec_from_file_location('schema_generator', path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_generator_is_deterministic_and_pinned(generator):
|
|
path = Path(__file__).resolve().parents[1] / 'bmspy/ecoflow_river3_schema.py'
|
|
assert re.fullmatch('[a-f0-9]{40}', SCHEMA_COMMIT)
|
|
assert generator.generate(MESSAGES, ENUMS, SCHEMA_COMMIT) == path.read_text()
|
|
|
|
|
|
def test_generator_keeps_reachable_types_and_ignores_unrelated_nested_types(generator):
|
|
schema = '''
|
|
message DisplayPropertyUpload { optional Item values = 1; }
|
|
message RuntimePropertyUpload { repeated uint32 raw = 2; }
|
|
message Item { optional Kind kind = 1; }
|
|
enum Kind { NONE = 0; ACTIVE = 1; }
|
|
message Unrelated { message Item { optional uint32 unrelated = 1; } }
|
|
'''
|
|
messages, enums = generator.parse_schema(schema)
|
|
assert messages['Item'] == {1: ('kind', 'Kind', False)}
|
|
assert messages['RuntimePropertyUpload'][2] == ('raw', 'uint32', True)
|
|
assert enums['Kind'][1] == 'ACTIVE'
|
|
assert 'Unrelated' not in messages
|
|
|
|
|
|
def test_generator_corrects_upstream_field_spelling(generator):
|
|
messages, _ = generator.parse_schema(
|
|
'message DisplayPropertyUpload { optional bool ups_alram = 355; } message RuntimePropertyUpload {}'
|
|
)
|
|
assert messages['DisplayPropertyUpload'][355] == ('ups_alarm', 'bool', False)
|
|
assert MESSAGES['DisplayPropertyUpload'][355] == ('ups_alarm', 'bool', False)
|
|
|
|
|
|
@pytest.mark.parametrize('body', [
|
|
'optional Missing value = 1;',
|
|
'optional uint32 a = 1; optional uint32 b = 1;',
|
|
'map<string, int32> unsupported = 1;',
|
|
'message Nested {} optional Nested value = 1;',
|
|
])
|
|
def test_generator_rejects_unsupported_or_incomplete_reachable_types(generator, body):
|
|
with pytest.raises(ValueError):
|
|
generator.parse_schema('message DisplayPropertyUpload {' + body + '} message RuntimePropertyUpload {}')
|