Add advanced imap option to set custom event max message size (#93163)

tests/components/imap/test_config_flow.py

@@ -5,6 +5,7 @@ from unittest.mock import AsyncMock, patch
 
 from aioimaplib import AioImapException
 import pytest
+import voluptuous as vol
 
 from homeassistant import config_entries, data_entry_flow
 from homeassistant.components.imap.const import (
@@ -397,6 +398,54 @@ async def test_key_options_in_options_form(hass: HomeAssistant) -> None:
     assert result2["errors"] == {"base": "already_configured"}
 
 
+@pytest.mark.parametrize(
+    ("advanced_options", "assert_result"),
+    [
+        ({"max_message_size": "8192"}, data_entry_flow.FlowResultType.CREATE_ENTRY),
+        ({"max_message_size": "1024"}, data_entry_flow.FlowResultType.FORM),
+        ({"max_message_size": "65536"}, data_entry_flow.FlowResultType.FORM),
+    ],
+)
+async def test_advanced_options_form(
+    hass: HomeAssistant,
+    advanced_options: dict[str, str],
+    assert_result: data_entry_flow.FlowResultType,
+) -> None:
+    """Test we show the advanced options."""
+
+    entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG)
+    entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(entry.entry_id)
+
+    result = await hass.config_entries.options.async_init(
+        entry.entry_id,
+        context={"source": config_entries.SOURCE_USER, "show_advanced_options": True},
+    )
+
+    assert result["type"] == data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "init"
+
+    new_config = MOCK_OPTIONS.copy()
+    new_config.update(advanced_options)
+
+    try:
+        with patch(
+            "homeassistant.components.imap.config_flow.connect_to_server"
+        ) as mock_client:
+            mock_client.return_value.search.return_value = ("OK", [b""])
+            # Option update should fail if FlowResultType.FORM is expected
+            result2 = await hass.config_entries.options.async_configure(
+                result["flow_id"], new_config
+            )
+            assert result2["type"] == assert_result
+            # Check if entry was updated
+            for key, value in new_config.items():
+                assert str(entry.data[key]) == value
+    except vol.MultipleInvalid:
+        # Check if form was expected with these options
+        assert assert_result == data_entry_flow.FlowResultType.FORM
+
+
 async def test_import_flow_success(hass: HomeAssistant) -> None:
     """Test a successful import of yaml."""
     with patch(

tests/components/imap/test_init.py

@@ -446,3 +446,62 @@ async def test_reset_last_message(
 
     # One new event
     assert len(event_called) == 2
+
+
+@pytest.mark.parametrize("imap_search", [TEST_SEARCH_RESPONSE])
+@pytest.mark.parametrize(
+    "imap_fetch", [(TEST_FETCH_RESPONSE_TEXT_PLAIN)], ids=["plain"]
+)
+@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
+@patch("homeassistant.components.imap.coordinator.MAX_EVENT_DATA_BYTES", 500)
+async def test_event_skipped_message_too_large(
+    hass: HomeAssistant, mock_imap_protocol: MagicMock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test skipping event when message is to large."""
+    event_called = async_capture_events(hass, "imap_content")
+
+    config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG)
+    config_entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+    # Make sure we have had one update (when polling)
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=5))
+    await hass.async_block_till_done()
+    state = hass.states.get("sensor.imap_email_email_com")
+    # We should have received one message
+    assert state is not None
+    assert state.state == "1"
+    assert len(event_called) == 0
+    assert "Custom imap_content event skipped" in caplog.text
+
+
+@pytest.mark.parametrize("imap_search", [TEST_SEARCH_RESPONSE])
+@pytest.mark.parametrize(
+    "imap_fetch", [(TEST_FETCH_RESPONSE_TEXT_PLAIN)], ids=["plain"]
+)
+@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
+async def test_message_is_truncated(
+    hass: HomeAssistant, mock_imap_protocol: MagicMock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test truncating message text in event data."""
+    event_called = async_capture_events(hass, "imap_content")
+
+    config = MOCK_CONFIG.copy()
+
+    # Mock the max message size to test it is truncated
+    config["max_message_size"] = 3
+    config_entry = MockConfigEntry(domain=DOMAIN, data=config)
+    config_entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+    # Make sure we have had one update (when polling)
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=5))
+    await hass.async_block_till_done()
+    state = hass.states.get("sensor.imap_email_email_com")
+    # We should have received one message
+    assert state is not None
+    assert state.state == "1"
+    assert len(event_called) == 1
+
+    event_data = event_called[0].data
+    assert len(event_data["text"]) == 3