mirror of https://github.com/home-assistant/core
979 lines
37 KiB
Python
979 lines
37 KiB
Python
"""Test the imap entry initialization."""
|
|
|
|
import asyncio
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, call, patch
|
|
|
|
from aioimaplib import AUTH, NONAUTH, SELECTED, AioImapException, Response
|
|
import pytest
|
|
|
|
from homeassistant.components.imap import DOMAIN
|
|
from homeassistant.components.imap.const import CONF_CHARSET
|
|
from homeassistant.components.imap.errors import InvalidAuth, InvalidFolder
|
|
from homeassistant.components.sensor import SensorStateClass
|
|
from homeassistant.const import STATE_UNAVAILABLE
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
from homeassistant.util.dt import utcnow
|
|
|
|
from .const import (
|
|
BAD_RESPONSE,
|
|
EMPTY_SEARCH_RESPONSE,
|
|
TEST_BADLY_ENCODED_CONTENT,
|
|
TEST_FETCH_RESPONSE_BINARY,
|
|
TEST_FETCH_RESPONSE_HTML,
|
|
TEST_FETCH_RESPONSE_INVALID_DATE1,
|
|
TEST_FETCH_RESPONSE_INVALID_DATE2,
|
|
TEST_FETCH_RESPONSE_INVALID_DATE3,
|
|
TEST_FETCH_RESPONSE_MULTIPART,
|
|
TEST_FETCH_RESPONSE_MULTIPART_BASE64,
|
|
TEST_FETCH_RESPONSE_MULTIPART_BASE64_INVALID,
|
|
TEST_FETCH_RESPONSE_MULTIPART_EMPTY_PLAIN,
|
|
TEST_FETCH_RESPONSE_NO_SUBJECT_TO_FROM,
|
|
TEST_FETCH_RESPONSE_TEXT_BARE,
|
|
TEST_FETCH_RESPONSE_TEXT_OTHER,
|
|
TEST_FETCH_RESPONSE_TEXT_PLAIN,
|
|
TEST_FETCH_RESPONSE_TEXT_PLAIN_ALT,
|
|
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(
|
|
("cipher_list", "verify_ssl", "enable_push"),
|
|
[
|
|
(None, None, None),
|
|
("python_default", True, None),
|
|
("python_default", False, None),
|
|
("modern", True, None),
|
|
("intermediate", True, None),
|
|
(None, None, False),
|
|
(None, None, True),
|
|
("python_default", True, False),
|
|
("python_default", False, True),
|
|
],
|
|
)
|
|
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
|
async def test_entry_startup_and_unload(
|
|
hass: HomeAssistant,
|
|
mock_imap_protocol: MagicMock,
|
|
cipher_list: str | None,
|
|
verify_ssl: bool | None,
|
|
enable_push: bool | None,
|
|
) -> None:
|
|
"""Test imap entry startup and unload with push and polling coordinator and alternate ciphers."""
|
|
config = MOCK_CONFIG.copy()
|
|
if cipher_list is not None:
|
|
config["ssl_cipher_list"] = cipher_list
|
|
if verify_ssl is not None:
|
|
config["verify_ssl"] = verify_ssl
|
|
if enable_push is not None:
|
|
config["enable_push"] = enable_push
|
|
|
|
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()
|
|
assert await hass.config_entries.async_unload(config_entry.entry_id)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"effect",
|
|
[
|
|
InvalidAuth,
|
|
InvalidFolder,
|
|
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", "valid_date"),
|
|
[
|
|
(TEST_FETCH_RESPONSE_TEXT_BARE, True),
|
|
(TEST_FETCH_RESPONSE_TEXT_PLAIN, True),
|
|
(TEST_FETCH_RESPONSE_TEXT_PLAIN_ALT, True),
|
|
(TEST_FETCH_RESPONSE_INVALID_DATE1, False),
|
|
(TEST_FETCH_RESPONSE_INVALID_DATE2, False),
|
|
(TEST_FETCH_RESPONSE_INVALID_DATE3, False),
|
|
(TEST_FETCH_RESPONSE_TEXT_OTHER, True),
|
|
(TEST_FETCH_RESPONSE_HTML, True),
|
|
(TEST_FETCH_RESPONSE_MULTIPART, True),
|
|
(TEST_FETCH_RESPONSE_MULTIPART_EMPTY_PLAIN, True),
|
|
(TEST_FETCH_RESPONSE_MULTIPART_BASE64, True),
|
|
(TEST_FETCH_RESPONSE_BINARY, True),
|
|
],
|
|
ids=[
|
|
"bare",
|
|
"plain",
|
|
"plain_alt",
|
|
"invalid_date1",
|
|
"invalid_date2",
|
|
"invalid_date3",
|
|
"other",
|
|
"html",
|
|
"multipart",
|
|
"multipart_empty_plain",
|
|
"multipart_base64",
|
|
"binary",
|
|
],
|
|
)
|
|
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
|
@pytest.mark.parametrize("charset", ["utf-8", "us-ascii"], ids=["utf-8", "us-ascii"])
|
|
async def test_receiving_message_successfully(
|
|
hass: HomeAssistant, mock_imap_protocol: MagicMock, valid_date: bool, charset: str
|
|
) -> None:
|
|
"""Test receiving a message successfully."""
|
|
event_called = async_capture_events(hass, "imap_content")
|
|
|
|
config = MOCK_CONFIG.copy()
|
|
config[CONF_CHARSET] = charset
|
|
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_messages")
|
|
# we should have received one message
|
|
assert state is not None
|
|
assert state.state == "1"
|
|
assert state.attributes["state_class"] == SensorStateClass.MEASUREMENT
|
|
|
|
# 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["uid"] == "1"
|
|
assert "Test body" in data["text"]
|
|
assert (
|
|
valid_date
|
|
and isinstance(data["date"], datetime)
|
|
or not valid_date
|
|
and data["date"] is None
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("imap_search", [TEST_SEARCH_RESPONSE])
|
|
@pytest.mark.parametrize(
|
|
("imap_fetch"),
|
|
[
|
|
TEST_FETCH_RESPONSE_MULTIPART_BASE64_INVALID,
|
|
],
|
|
ids=[
|
|
"multipart_base64_invalid",
|
|
],
|
|
)
|
|
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
|
async def test_receiving_message_with_invalid_encoding(
|
|
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_messages")
|
|
# we should have received one message
|
|
assert state is not None
|
|
assert state.state == "1"
|
|
assert state.attributes["state_class"] == SensorStateClass.MEASUREMENT
|
|
|
|
# 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"] == TEST_BADLY_ENCODED_CONTENT
|
|
assert data["uid"] == "1"
|
|
|
|
|
|
@pytest.mark.parametrize("imap_search", [TEST_SEARCH_RESPONSE])
|
|
@pytest.mark.parametrize("imap_fetch", [TEST_FETCH_RESPONSE_NO_SUBJECT_TO_FROM])
|
|
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
|
async def test_receiving_message_no_subject_to_from(
|
|
hass: HomeAssistant, mock_imap_protocol: MagicMock
|
|
) -> None:
|
|
"""Test receiving a message successfully without subject, to and from in body."""
|
|
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_messages")
|
|
# 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"] == ""
|
|
assert data["subject"] == ""
|
|
assert data["date"] == datetime(
|
|
2023, 3, 24, 13, 52, tzinfo=timezone(timedelta(seconds=3600))
|
|
)
|
|
assert data["text"] == "Test body\r\n"
|
|
assert data["headers"]["Return-Path"] == ("<john.doe@example.com>",)
|
|
assert data["headers"]["Delivered-To"] == ("notify@example.com",)
|
|
assert data["uid"] == "1"
|
|
|
|
|
|
@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_messages")
|
|
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_messages")
|
|
assert (state is not None) == success
|
|
|
|
|
|
@patch("homeassistant.components.imap.coordinator.MAX_ERRORS", 1)
|
|
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
|
async def test_late_authentication_retry(
|
|
hass: HomeAssistant,
|
|
caplog: pytest.LogCaptureFixture,
|
|
mock_imap_protocol: MagicMock,
|
|
) -> None:
|
|
"""Test retrying authentication 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 "Authentication failed, retrying" in caplog.text
|
|
|
|
# we still should have an entity with an unavailable state
|
|
state = hass.states.get("sensor.imap_email_email_com_messages")
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
|
|
@patch("homeassistant.components.imap.coordinator.MAX_ERRORS", 0)
|
|
@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_messages")
|
|
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_messages")
|
|
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=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_messages")
|
|
# 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 hass.config_entries.async_unload(config_entry.entry_id)
|
|
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_messages")
|
|
|
|
# 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"),
|
|
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 | 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_messages")
|
|
# Our entity should keep its current state as this
|
|
assert state is not None
|
|
assert state.state == "0"
|
|
|
|
|
|
@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_messages")
|
|
# we should have an entity with an unavailable state
|
|
assert state is not None
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
|
|
@pytest.mark.parametrize("imap_search", [TEST_SEARCH_RESPONSE])
|
|
@pytest.mark.parametrize(
|
|
("imap_fetch", "valid_date"),
|
|
[(TEST_FETCH_RESPONSE_TEXT_PLAIN, True)],
|
|
ids=["plain"],
|
|
)
|
|
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
|
async def test_reset_last_message(
|
|
hass: HomeAssistant, mock_imap_protocol: MagicMock, valid_date: bool
|
|
) -> None:
|
|
"""Test receiving a message successfully."""
|
|
event = asyncio.Event() # needed for pushed coordinator to make a new loop
|
|
idle_start_future = asyncio.Future()
|
|
idle_start_future.set_result(None)
|
|
|
|
async def _sleep_till_event() -> None:
|
|
"""Simulate imap server waiting for pushes message and keep the push loop going.
|
|
|
|
Needed for pushed coordinator only.
|
|
"""
|
|
nonlocal event
|
|
await event.wait()
|
|
event.clear()
|
|
mock_imap_protocol.idle_start = AsyncMock(return_value=idle_start_future)
|
|
|
|
# Make sure we make another cycle (needed for pushed coordinator)
|
|
mock_imap_protocol.idle_start = AsyncMock(return_value=idle_start_future)
|
|
# Mock we wait till we push an update (needed for pushed coordinator)
|
|
mock_imap_protocol.wait_server_push.side_effect = _sleep_till_event
|
|
|
|
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_messages")
|
|
# 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"]
|
|
assert data["initial"]
|
|
assert (
|
|
valid_date
|
|
and isinstance(data["date"], datetime)
|
|
or not valid_date
|
|
and data["date"] is None
|
|
)
|
|
|
|
# Simulate an update where no messages are found (needed for pushed coordinator)
|
|
mock_imap_protocol.search.return_value = Response(*EMPTY_SEARCH_RESPONSE)
|
|
|
|
# Make sure we have an update
|
|
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
|
|
|
|
# Awake loop (needed for pushed coordinator)
|
|
event.set()
|
|
|
|
await hass.async_block_till_done()
|
|
|
|
state = hass.states.get("sensor.imap_email_email_com_messages")
|
|
# We should have message
|
|
assert state is not None
|
|
assert state.state == "0"
|
|
# No new events should be called
|
|
assert len(event_called) == 1
|
|
|
|
# Simulate an update where with the original message
|
|
mock_imap_protocol.search.return_value = Response(*TEST_SEARCH_RESPONSE)
|
|
# Make sure we have an update again with the same UID
|
|
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
|
|
|
|
# Awake loop (needed for pushed coordinator)
|
|
event.set()
|
|
|
|
await hass.async_block_till_done()
|
|
|
|
state = hass.states.get("sensor.imap_email_email_com_messages")
|
|
# We should have received one message
|
|
assert state is not None
|
|
assert state.state == "1"
|
|
await hass.async_block_till_done()
|
|
await hass.async_block_till_done()
|
|
|
|
# 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_messages")
|
|
# 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_messages")
|
|
# 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
|
|
|
|
|
|
@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"])
|
|
@pytest.mark.parametrize("event_message_data", [[], ["text"], ["text", "headers"]])
|
|
async def test_message_data(
|
|
hass: HomeAssistant,
|
|
mock_imap_protocol: MagicMock,
|
|
caplog: pytest.LogCaptureFixture,
|
|
event_message_data: list,
|
|
) -> None:
|
|
"""Test with different message data."""
|
|
event_called = async_capture_events(hass, "imap_content")
|
|
|
|
config = MOCK_CONFIG.copy()
|
|
# Mock different message data
|
|
config["event_message_data"] = event_message_data
|
|
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_messages")
|
|
# 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 set(event_message_data).issubset(set(event_data))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("imap_search", "imap_fetch"),
|
|
[(TEST_SEARCH_RESPONSE, TEST_FETCH_RESPONSE_TEXT_PLAIN)],
|
|
ids=["plain"],
|
|
)
|
|
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
|
@pytest.mark.parametrize(
|
|
("custom_template", "result", "error"),
|
|
[
|
|
("{{ subject }}", "Test subject", None),
|
|
('{{ "@example.com" in sender }}', True, None),
|
|
("{% bad template }}", None, "Error rendering IMAP custom template"),
|
|
],
|
|
ids=["subject_test", "sender_filter", "template_error"],
|
|
)
|
|
async def test_custom_template(
|
|
hass: HomeAssistant,
|
|
mock_imap_protocol: MagicMock,
|
|
caplog: pytest.LogCaptureFixture,
|
|
custom_template: str,
|
|
result: str | bool | None,
|
|
error: str | None,
|
|
) -> None:
|
|
"""Test the custom template event data."""
|
|
event_called = async_capture_events(hass, "imap_content")
|
|
|
|
config = MOCK_CONFIG.copy()
|
|
config["custom_event_data_template"] = custom_template
|
|
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_messages")
|
|
# 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"]
|
|
assert data["custom"] == result
|
|
assert error in caplog.text if error is not None else True
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("imap_search", "imap_fetch"),
|
|
[(TEST_SEARCH_RESPONSE, TEST_FETCH_RESPONSE_TEXT_PLAIN)],
|
|
)
|
|
@pytest.mark.parametrize(
|
|
("imap_has_capability", "enable_push", "should_poll"),
|
|
[
|
|
(True, False, True),
|
|
(False, False, True),
|
|
(True, True, False),
|
|
(False, True, True),
|
|
],
|
|
ids=["enforce_poll", "poll", "auto_push", "auto_poll"],
|
|
)
|
|
async def test_enforce_polling(
|
|
hass: HomeAssistant,
|
|
mock_imap_protocol: MagicMock,
|
|
enable_push: bool,
|
|
should_poll: True,
|
|
) -> None:
|
|
"""Test enforce polling."""
|
|
event_called = async_capture_events(hass, "imap_content")
|
|
config = MOCK_CONFIG.copy()
|
|
config["enable_push"] = enable_push
|
|
|
|
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_messages")
|
|
# we should have received one message
|
|
assert state is not None
|
|
assert state.state == "1"
|
|
assert state.attributes["state_class"] == SensorStateClass.MEASUREMENT
|
|
|
|
# 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"]
|
|
|
|
if should_poll:
|
|
mock_imap_protocol.wait_server_push.assert_not_called()
|
|
else:
|
|
mock_imap_protocol.assert_has_calls([call.wait_server_push])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("imap_search", "imap_fetch"),
|
|
[(TEST_SEARCH_RESPONSE, TEST_FETCH_RESPONSE_TEXT_PLAIN)],
|
|
)
|
|
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
|
|
async def test_services(hass: HomeAssistant, mock_imap_protocol: MagicMock) -> None:
|
|
"""Test receiving a message successfully."""
|
|
event_called = async_capture_events(hass, "imap_content")
|
|
|
|
config = MOCK_CONFIG.copy()
|
|
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_messages")
|
|
# we should have received one message
|
|
assert state is not None
|
|
assert state.state == "1"
|
|
assert state.attributes["state_class"] == SensorStateClass.MEASUREMENT
|
|
|
|
# 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["uid"] == "1"
|
|
assert data["entry_id"] == config_entry.entry_id
|
|
|
|
# Test seen service
|
|
data = {"entry": config_entry.entry_id, "uid": "1"}
|
|
await hass.services.async_call(DOMAIN, "seen", data, blocking=True)
|
|
mock_imap_protocol.store.assert_called_with("1", "+FLAGS (\\Seen)")
|
|
mock_imap_protocol.store.reset_mock()
|
|
|
|
# Test move service
|
|
data = {
|
|
"entry": config_entry.entry_id,
|
|
"uid": "1",
|
|
"seen": True,
|
|
"target_folder": "Trash",
|
|
}
|
|
await hass.services.async_call(DOMAIN, "move", data, blocking=True)
|
|
mock_imap_protocol.store.assert_has_calls(
|
|
[call("1", "+FLAGS (\\Seen)"), call("1", "+FLAGS (\\Deleted)")]
|
|
)
|
|
mock_imap_protocol.copy.assert_called_with("1", "Trash")
|
|
mock_imap_protocol.protocol.expunge.assert_called_once()
|
|
mock_imap_protocol.store.reset_mock()
|
|
mock_imap_protocol.copy.reset_mock()
|
|
mock_imap_protocol.protocol.expunge.reset_mock()
|
|
|
|
# Test delete service
|
|
data = {"entry": config_entry.entry_id, "uid": "1"}
|
|
await hass.services.async_call(DOMAIN, "delete", data, blocking=True)
|
|
mock_imap_protocol.store.assert_called_with("1", "+FLAGS (\\Deleted)")
|
|
mock_imap_protocol.protocol.expunge.assert_called_once()
|
|
|
|
# Test fetch service
|
|
data = {"entry": config_entry.entry_id, "uid": "1"}
|
|
response = await hass.services.async_call(
|
|
DOMAIN, "fetch", data, blocking=True, return_response=True
|
|
)
|
|
mock_imap_protocol.fetch.assert_called_with("1", "BODY.PEEK[]")
|
|
assert response["text"] == "Test body\r\n"
|
|
assert response["sender"] == "john.doe@example.com"
|
|
assert response["subject"] == "Test subject"
|
|
assert response["uid"] == "1"
|
|
|
|
# Test with invalid entry_id
|
|
data = {"entry": "invalid", "uid": "1"}
|
|
with pytest.raises(ServiceValidationError) as exc:
|
|
await hass.services.async_call(DOMAIN, "seen", data, blocking=True)
|
|
assert exc.value.translation_domain == DOMAIN
|
|
assert exc.value.translation_key == "invalid_entry"
|
|
|
|
# Test processing imap client failures
|
|
exceptions = {
|
|
"invalid_auth": {"exc": InvalidAuth(), "translation_placeholders": None},
|
|
"invalid_folder": {"exc": InvalidFolder(), "translation_placeholders": None},
|
|
"imap_server_fail": {
|
|
"exc": AioImapException("Bla"),
|
|
"translation_placeholders": {"error": "Bla"},
|
|
},
|
|
}
|
|
for translation_key, attrs in exceptions.items():
|
|
with patch(
|
|
"homeassistant.components.imap.connect_to_server", side_effect=attrs["exc"]
|
|
):
|
|
data = {"entry": config_entry.entry_id, "uid": "1"}
|
|
with pytest.raises(ServiceValidationError) as exc:
|
|
await hass.services.async_call(DOMAIN, "seen", data, blocking=True)
|
|
assert exc.value.translation_domain == DOMAIN
|
|
assert exc.value.translation_key == translation_key
|
|
assert (
|
|
exc.value.translation_placeholders == attrs["translation_placeholders"]
|
|
)
|
|
|
|
# Test unexpected errors with storing a flag during a service call
|
|
service_calls_response = {
|
|
"seen": ({"entry": config_entry.entry_id, "uid": "1"}, False),
|
|
"move": (
|
|
{
|
|
"entry": config_entry.entry_id,
|
|
"uid": "1",
|
|
"seen": False,
|
|
"target_folder": "Trash",
|
|
},
|
|
False,
|
|
),
|
|
"delete": ({"entry": config_entry.entry_id, "uid": "1"}, False),
|
|
"fetch": ({"entry": config_entry.entry_id, "uid": "1"}, True),
|
|
}
|
|
patch_error_translation_key = {
|
|
"seen": ("store", "seen_failed"),
|
|
"move": ("copy", "copy_failed"),
|
|
"delete": ("store", "delete_failed"),
|
|
"fetch": ("fetch", "fetch_failed"),
|
|
}
|
|
for service, (data, response) in service_calls_response.items():
|
|
with (
|
|
pytest.raises(ServiceValidationError) as exc,
|
|
patch.object(
|
|
mock_imap_protocol,
|
|
patch_error_translation_key[service][0],
|
|
side_effect=AioImapException("Bla"),
|
|
),
|
|
):
|
|
await hass.services.async_call(
|
|
DOMAIN, service, data, blocking=True, return_response=response
|
|
)
|
|
assert exc.value.translation_domain == DOMAIN
|
|
assert exc.value.translation_key == "imap_server_fail"
|
|
assert exc.value.translation_placeholders == {"error": "Bla"}
|
|
# Test with bad responses
|
|
with (
|
|
pytest.raises(ServiceValidationError) as exc,
|
|
patch.object(
|
|
mock_imap_protocol,
|
|
patch_error_translation_key[service][0],
|
|
return_value=Response("BAD", [b"Bla"]),
|
|
),
|
|
):
|
|
await hass.services.async_call(
|
|
DOMAIN, service, data, blocking=True, return_response=response
|
|
)
|
|
assert exc.value.translation_domain == DOMAIN
|
|
assert exc.value.translation_key == patch_error_translation_key[service][1]
|
|
assert exc.value.translation_placeholders == {"error": "Bla"}
|