xs
70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
"""测试参数变化回调: 接受合法值 / 拒绝非法值 / 同步缓存。"""
|
|
from rclpy.parameter import Parameter
|
|
|
|
from py_params.param_node import ParamsTalker
|
|
|
|
|
|
def _make_double_param(name: str, value: float) -> Parameter:
|
|
"""构造 DOUBLE 类型 Parameter(测试样板)。"""
|
|
return Parameter(name, Parameter.Type.DOUBLE, value)
|
|
|
|
|
|
def _make_string_param(name: str, value: str) -> Parameter:
|
|
"""构造 STRING 类型 Parameter(测试样板)。"""
|
|
return Parameter(name, Parameter.Type.STRING, value)
|
|
|
|
|
|
def test_set_valid_rate(params_talker: ParamsTalker) -> None:
|
|
"""合法 rate 被接受,值更新。"""
|
|
results = params_talker.set_parameters([_make_double_param('publish_rate_hz', 5.0)])
|
|
assert results[0].successful is True
|
|
assert params_talker.get_parameter('publish_rate_hz').value == 5.0
|
|
|
|
|
|
def test_reject_zero_rate(params_talker: ParamsTalker) -> None:
|
|
"""rate == 0 被拒绝。"""
|
|
results = params_talker.set_parameters([_make_double_param('publish_rate_hz', 0.0)])
|
|
assert results[0].successful is False
|
|
assert '必须 > 0' in results[0].reason
|
|
assert params_talker.get_parameter('publish_rate_hz').value == 1.0
|
|
|
|
|
|
def test_reject_negative_rate(params_talker: ParamsTalker) -> None:
|
|
"""rate < 0 被拒绝。"""
|
|
results = params_talker.set_parameters([_make_double_param('publish_rate_hz', -2.5)])
|
|
assert results[0].successful is False
|
|
assert '必须 > 0' in results[0].reason
|
|
assert params_talker.get_parameter('publish_rate_hz').value == 1.0
|
|
|
|
|
|
def test_set_prefix_syncs_cache(params_talker: ParamsTalker) -> None:
|
|
"""message_prefix 改变时,内部缓存同步。"""
|
|
results = params_talker.set_parameters(
|
|
[_make_string_param('message_prefix', 'HelloWorld:')],
|
|
)
|
|
assert results[0].successful is True
|
|
assert params_talker._prefix == 'HelloWorld:'
|
|
|
|
|
|
def test_set_parameters_atomically_all_success(params_talker: ParamsTalker) -> None:
|
|
"""原子设置: 全部合法 → 全部生效。"""
|
|
new_params = [
|
|
_make_double_param('publish_rate_hz', 3.0),
|
|
_make_string_param('message_prefix', 'Atomic:'),
|
|
]
|
|
result = params_talker.set_parameters_atomically(new_params)
|
|
assert result.successful is True
|
|
assert params_talker.get_parameter('publish_rate_hz').value == 3.0
|
|
assert params_talker.get_parameter('message_prefix').value == 'Atomic:'
|
|
|
|
|
|
def test_set_parameters_atomically_rollback(params_talker: ParamsTalker) -> None:
|
|
"""原子设置: 任一非法 → 全部回滚。"""
|
|
new_params = [
|
|
_make_string_param('message_prefix', 'ShouldNotApply:'),
|
|
_make_double_param('publish_rate_hz', -1.0),
|
|
]
|
|
result = params_talker.set_parameters_atomically(new_params)
|
|
assert result.successful is False
|
|
assert params_talker.get_parameter('message_prefix').value == 'Params:'
|