Implement imap_content event for imap integration (#90242)
This commit is contained in:
Jan Bouwhuis
GitHub
parent
d21433b6af
commit
24d0d15f38
5 changed files with 671 additions and 9 deletions
323
tests/components/imap/test_init.py
Normal file
323
tests/components/imap/test_init.py
Normal file
|
|
@ -0,0 +1,323 @@
|
|||
"""Test the imap entry initialization."""
|
||||
import asyncio
|
||||
from datetime import timedelta
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from aioimaplib import AUTH, NONAUTH, SELECTED, AioImapException, Response
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.imap import DOMAIN
|
||||
from homeassistant.components.imap.errors import InvalidAuth, InvalidFolder
|
||||
from homeassistant.const import STATE_UNAVAILABLE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.util.dt import utcnow
|
||||
|
||||
from .const import (
|
||||
BAD_RESPONSE,
|
||||
TEST_FETCH_RESPONSE_BINARY,
|
||||
TEST_FETCH_RESPONSE_HTML,
|
||||
TEST_FETCH_RESPONSE_MULTIPART,
|
||||
TEST_FETCH_RESPONSE_TEXT_BARE,
|
||||
TEST_FETCH_RESPONSE_TEXT_OTHER,
|
||||
TEST_FETCH_RESPONSE_TEXT_PLAIN,
|
||||
TEST_SEARCH_RESPONSE,
|
||||
)
|
||||
from .test_config_flow import MOCK_CONFIG
|
||||
|
||||
from tests.common import MockConfigEntry, async_capture_events, async_fire_time_changed
|
||||
|
||||
|
||||
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
||||
async def test_entry_startup_and_unload(
|
||||
hass: HomeAssistant, mock_imap_protocol: MagicMock
|
||||
) -> None:
|
||||
"""Test imap entry startup and unload with push and polling coordinator."""
|
||||
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()
|
||||
assert await config_entry.async_unload(hass)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"effect",
|
||||
[
|
||||
InvalidAuth,
|
||||
InvalidFolder,
|
||||
asyncio.TimeoutError,
|
||||
],
|
||||
)
|
||||
async def test_entry_startup_fails(
|
||||
hass: HomeAssistant,
|
||||
mock_imap_protocol: MagicMock,
|
||||
effect: Exception,
|
||||
) -> None:
|
||||
"""Test imap entry startup fails on invalid auth or folder."""
|
||||
config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.imap.connect_to_server",
|
||||
side_effect=effect,
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("imap_search", [TEST_SEARCH_RESPONSE])
|
||||
@pytest.mark.parametrize(
|
||||
"imap_fetch",
|
||||
[
|
||||
TEST_FETCH_RESPONSE_TEXT_BARE,
|
||||
TEST_FETCH_RESPONSE_TEXT_PLAIN,
|
||||
TEST_FETCH_RESPONSE_TEXT_OTHER,
|
||||
TEST_FETCH_RESPONSE_HTML,
|
||||
TEST_FETCH_RESPONSE_MULTIPART,
|
||||
TEST_FETCH_RESPONSE_BINARY,
|
||||
],
|
||||
ids=["bare", "plain", "other", "html", "multipart", "binary"],
|
||||
)
|
||||
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
||||
async def test_receiving_message_successfully(
|
||||
hass: HomeAssistant, mock_imap_protocol: MagicMock
|
||||
) -> None:
|
||||
"""Test receiving a message successfully."""
|
||||
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"
|
||||
|
||||
# we should have received one event
|
||||
assert len(event_called) == 1
|
||||
data: dict[str, Any] = event_called[0].data
|
||||
assert data["server"] == "imap.server.com"
|
||||
assert data["username"] == "email@email.com"
|
||||
assert data["search"] == "UnSeen UnDeleted"
|
||||
assert data["folder"] == "INBOX"
|
||||
assert data["sender"] == "john.doe@example.com"
|
||||
assert data["subject"] == "Test subject"
|
||||
assert data["text"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
||||
@pytest.mark.parametrize(
|
||||
("imap_login_state", "success"), [(AUTH, True), (NONAUTH, False)]
|
||||
)
|
||||
async def test_initial_authentication_error(
|
||||
hass: HomeAssistant, mock_imap_protocol: MagicMock, success: bool
|
||||
) -> None:
|
||||
"""Test authentication error when starting the entry."""
|
||||
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) == success
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.imap_email_email_com")
|
||||
assert (state is not None) == success
|
||||
|
||||
|
||||
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
||||
@pytest.mark.parametrize(
|
||||
("imap_select_state", "success"), [(AUTH, False), (SELECTED, True)]
|
||||
)
|
||||
async def test_initial_invalid_folder_error(
|
||||
hass: HomeAssistant, mock_imap_protocol: MagicMock, success: bool
|
||||
) -> None:
|
||||
"""Test invalid folder error when starting the entry."""
|
||||
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) == success
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.imap_email_email_com")
|
||||
assert (state is not None) == success
|
||||
|
||||
|
||||
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
||||
async def test_late_authentication_error(
|
||||
hass: HomeAssistant,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mock_imap_protocol: MagicMock,
|
||||
) -> None:
|
||||
"""Test authentication error handling after a search was failed."""
|
||||
|
||||
# Mock an error in waiting for a pushed update
|
||||
mock_imap_protocol.wait_server_push.side_effect = AioImapException(
|
||||
"Something went wrong"
|
||||
)
|
||||
|
||||
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)
|
||||
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=60))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Mock that the search fails, this will trigger
|
||||
# that the connection will be restarted
|
||||
# Then fail selecting the folder
|
||||
mock_imap_protocol.search.return_value = Response(*BAD_RESPONSE)
|
||||
mock_imap_protocol.login.side_effect = Response(*BAD_RESPONSE)
|
||||
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=60))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=60))
|
||||
await hass.async_block_till_done()
|
||||
assert "Username or password incorrect, starting reauthentication" in caplog.text
|
||||
|
||||
# we still should have an entity with an unavailable state
|
||||
state = hass.states.get("sensor.imap_email_email_com")
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
||||
async def test_late_folder_error(
|
||||
hass: HomeAssistant,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mock_imap_protocol: MagicMock,
|
||||
) -> None:
|
||||
"""Test invalid folder error handling after a search was failed.
|
||||
|
||||
Asserting the IMAP push coordinator.
|
||||
"""
|
||||
# Mock an error in waiting for a pushed update
|
||||
mock_imap_protocol.wait_server_push.side_effect = AioImapException(
|
||||
"Something went wrong"
|
||||
)
|
||||
|
||||
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 at least one update (when polling)
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=60))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Mock that the search fails, this will trigger
|
||||
# that the connection will be restarted
|
||||
# Then fail selecting the folder
|
||||
mock_imap_protocol.search.return_value = Response(*BAD_RESPONSE)
|
||||
mock_imap_protocol.select.side_effect = Response(*BAD_RESPONSE)
|
||||
|
||||
# Make sure we have had at least one update (when polling)
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=60))
|
||||
await hass.async_block_till_done()
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=60))
|
||||
await hass.async_block_till_done()
|
||||
assert "Selected mailbox folder is invalid" in caplog.text
|
||||
|
||||
# we still should have an entity with an unavailable state
|
||||
state = hass.states.get("sensor.imap_email_email_com")
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
||||
@pytest.mark.parametrize(
|
||||
"imap_close",
|
||||
[
|
||||
AsyncMock(side_effect=AioImapException("Something went wrong")),
|
||||
AsyncMock(side_effect=asyncio.TimeoutError),
|
||||
],
|
||||
ids=["AioImapException", "TimeoutError"],
|
||||
)
|
||||
async def test_handle_cleanup_exception(
|
||||
hass: HomeAssistant,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mock_imap_protocol: MagicMock,
|
||||
imap_close: Exception,
|
||||
) -> None:
|
||||
"""Test handling an excepton during cleaning up."""
|
||||
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 an entity
|
||||
assert state is not None
|
||||
assert state.state == "0"
|
||||
|
||||
# Fail cleaning up
|
||||
mock_imap_protocol.close.side_effect = imap_close
|
||||
|
||||
assert await config_entry.async_unload(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert "Error while cleaning up imap connection" in caplog.text
|
||||
|
||||
state = hass.states.get("sensor.imap_email_email_com")
|
||||
|
||||
# we should have an entity with an unavailable state
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
@pytest.mark.parametrize("imap_has_capability", [True], ids=["push"])
|
||||
@pytest.mark.parametrize(
|
||||
"imap_wait_server_push_exception",
|
||||
[
|
||||
AioImapException("Something went wrong"),
|
||||
asyncio.TimeoutError,
|
||||
],
|
||||
ids=["AioImapException", "TimeoutError"],
|
||||
)
|
||||
async def test_lost_connection_with_imap_push(
|
||||
hass: HomeAssistant,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mock_imap_protocol: MagicMock,
|
||||
imap_wait_server_push_exception: AioImapException | asyncio.TimeoutError,
|
||||
) -> None:
|
||||
"""Test error handling when the connection is lost."""
|
||||
# Mock an error in waiting for a pushed update
|
||||
mock_imap_protocol.wait_server_push.side_effect = imap_wait_server_push_exception
|
||||
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()
|
||||
assert "Lost imap.server.com (will attempt to reconnect after 10 s)" in caplog.text
|
||||
|
||||
state = hass.states.get("sensor.imap_email_email_com")
|
||||
# we should have an entity with an unavailable state
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
@pytest.mark.parametrize("imap_has_capability", [True], ids=["push"])
|
||||
async def test_fetch_number_of_messages(
|
||||
hass: HomeAssistant,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mock_imap_protocol: MagicMock,
|
||||
) -> None:
|
||||
"""Test _async_fetch_number_of_messages fails with push coordinator."""
|
||||
# Mock an error in waiting for a pushed update
|
||||
mock_imap_protocol.search.return_value = Response(*BAD_RESPONSE)
|
||||
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 wait for the backoff time
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
|
||||
await hass.async_block_till_done()
|
||||
assert "Invalid response for search" in caplog.text
|
||||
|
||||
state = hass.states.get("sensor.imap_email_email_com")
|
||||
# we should have an entity with an unavailable state
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
Loading…
Add table
Add a link
Reference in a new issue