1
0
mirror of https://github.com/esphome/esphome.git synced 2025-09-05 21:02:20 +01:00

Merge branch 'update_no_batch' into integration

This commit is contained in:
J. Nick Koston
2025-08-03 13:55:19 -10:00
3 changed files with 92 additions and 4 deletions

View File

@@ -703,10 +703,16 @@ class APIConnection : public APIServerConnection {
bool send_message_smart_(EntityBase *entity, MessageCreatorPtr creator, uint8_t message_type,
uint8_t estimated_size) {
// Try to send immediately if:
// 1. We should try to send immediately (should_try_send_immediately = true)
// 2. Batch delay is 0 (user has opted in to immediate sending)
// 3. Buffer has space available
if (this->flags_.should_try_send_immediately && this->get_batch_delay_ms_() == 0 &&
// 1. It's an UpdateStateResponse (always send immediately to handle cases where
// the main loop is blocked, e.g., during OTA updates)
// 2. OR: We should try to send immediately (should_try_send_immediately = true)
// AND Batch delay is 0 (user has opted in to immediate sending)
// 3. AND: Buffer has space available
if ((
#ifdef USE_UPDATE
message_type == UpdateStateResponse::MESSAGE_TYPE ||
#endif
(this->flags_.should_try_send_immediately && this->get_batch_delay_ms_() == 0)) &&
this->helper_->can_write_without_blocking()) {
// Now actually encode and send
if (creator(entity, this, MAX_BATCH_PACKET_SIZE, true) &&

View File

@@ -37,6 +37,8 @@ class AnsiStyle(Enum):
def color(col: AnsiFore, msg: str, reset: bool = True) -> str:
if col == AnsiFore.KEEP:
return msg
s = col.value + msg
if reset and col:
s += AnsiStyle.RESET_ALL.value

View File

@@ -0,0 +1,80 @@
import pytest
from esphome.log import AnsiFore, AnsiStyle, color
def test_color_keep_returns_unchanged_message() -> None:
"""Test that AnsiFore.KEEP returns the message unchanged."""
msg = "test message"
result = color(AnsiFore.KEEP, msg)
assert result == msg
def test_color_keep_ignores_reset_parameter() -> None:
"""Test that reset parameter is ignored when using AnsiFore.KEEP."""
msg = "test message"
result_with_reset = color(AnsiFore.KEEP, msg, reset=True)
result_without_reset = color(AnsiFore.KEEP, msg, reset=False)
assert result_with_reset == msg
assert result_without_reset == msg
def test_color_applies_color_code() -> None:
"""Test that color codes are properly applied to messages."""
msg = "test message"
result = color(AnsiFore.RED, msg, reset=False)
assert result == f"{AnsiFore.RED.value}{msg}"
def test_color_applies_reset_when_requested() -> None:
"""Test that RESET_ALL is added when reset=True."""
msg = "test message"
result = color(AnsiFore.GREEN, msg, reset=True)
expected = f"{AnsiFore.GREEN.value}{msg}{AnsiStyle.RESET_ALL.value}"
assert result == expected
def test_color_no_reset_when_not_requested() -> None:
"""Test that RESET_ALL is not added when reset=False."""
msg = "test message"
result = color(AnsiFore.BLUE, msg, reset=False)
expected = f"{AnsiFore.BLUE.value}{msg}"
assert result == expected
def test_color_with_empty_message() -> None:
"""Test color function with empty message."""
result = color(AnsiFore.YELLOW, "", reset=True)
expected = f"{AnsiFore.YELLOW.value}{AnsiStyle.RESET_ALL.value}"
assert result == expected
@pytest.mark.parametrize(
"col",
[
AnsiFore.BLACK,
AnsiFore.RED,
AnsiFore.GREEN,
AnsiFore.YELLOW,
AnsiFore.BLUE,
AnsiFore.MAGENTA,
AnsiFore.CYAN,
AnsiFore.WHITE,
AnsiFore.RESET,
],
)
def test_all_ansi_colors(col: AnsiFore) -> None:
"""Test that all AnsiFore colors work correctly."""
msg = "test"
result = color(col, msg, reset=True)
expected = f"{col.value}{msg}{AnsiStyle.RESET_ALL.value}"
assert result == expected
def test_ansi_fore_keep_is_enum_member() -> None:
"""Ensure AnsiFore.KEEP is an Enum member and evaluates to truthy."""
assert isinstance(AnsiFore.KEEP, AnsiFore)
# Enum members are truthy, even with empty string values
assert bool(AnsiFore.KEEP) is True
# But the value itself is still an empty string
assert AnsiFore.KEEP.value == ""