evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix imap_email_content unknown status and replaying stale states (#89563)

Browse source
This commit is contained in:
Jan Bouwhuis 2023-03-20 04:35:45 +01:00 • committed by GitHub
parent 939fce4607
commit 2039955ef7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 70 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
homeassistant/components/imap_email_content/sensor.py
View file

@ -95,9 +95,25 @@ class EmailReader:
self._folder = folder
self._verify_ssl = verify_ssl
self._last_id = None
self._last_message = None
self._unread_ids = deque([])
self.connection = None
@property
def last_id(self) -> int | None:
"""Return last email uid that was processed."""
return self._last_id
@property
def last_unread_id(self) -> int | None:
"""Return last email uid received."""
# We assume the last id in the list is the last unread id
# We cannot know if that is the newest one, because it could arrive later
# https://stackoverflow.com/questions/12409862/python-imap-the-order-of-uids
if self._unread_ids:
return int(self._unread_ids[-1])
return self._last_id
def connect(self):
"""Login and setup the connection."""
ssl_context = client_context() if self._verify_ssl else None
@ -128,21 +144,21 @@ class EmailReader:
try:
self.connection.select(self._folder, readonly=True)
if not self._unread_ids:
search = f"SINCE {datetime.date.today():%d-%b-%Y}"
if self._last_id is not None:
search = f"UID {self._last_id}:*"
_, data = self.connection.uid("search", None, search)
self._unread_ids = deque(data[0].split())
if self._last_id is None:
# search for today and yesterday
time_from = datetime.datetime.now() - datetime.timedelta(days=1)
search = f"SINCE {time_from:%d-%b-%Y}"
else:
search = f"UID {self._last_id}:*"
_, data = self.connection.uid("search", None, search)
self._unread_ids = deque(data[0].split())
while self._unread_ids:
message_uid = self._unread_ids.popleft()
if self._last_id is None or int(message_uid) > self._last_id:
self._last_id = int(message_uid)
return self._fetch_message(message_uid)
return self._fetch_message(str(self._last_id))
self._last_message = self._fetch_message(message_uid)
return self._last_message
except imaplib.IMAP4.error:
_LOGGER.info("Connection to %s lost, attempting to reconnect", self._server)
@ -254,22 +270,30 @@ class EmailContentSensor(SensorEntity):
def update(self) -> None:
"""Read emails and publish state change."""
email_message = self._email_reader.read_next()
while (
self._last_id is None or self._last_id != self._email_reader.last_unread_id
):
if email_message is None:
self._message = None
self._state_attributes = {}
return
if email_message is None:
self._message = None
self._state_attributes = {}
return
self._last_id = self._email_reader.last_id
if self.sender_allowed(email_message):
message = EmailContentSensor.get_msg_subject(email_message)
if self.sender_allowed(email_message):
message = EmailContentSensor.get_msg_subject(email_message)
if self._value_template is not None:
message = self.render_template(email_message)
if self._value_template is not None:
message = self.render_template(email_message)
self._message = message
self._state_attributes = {
ATTR_FROM: EmailContentSensor.get_msg_sender(email_message),
ATTR_SUBJECT: EmailContentSensor.get_msg_subject(email_message),
ATTR_DATE: email_message["Date"],
ATTR_BODY: EmailContentSensor.get_msg_text(email_message),
}
self._message = message
self._state_attributes = {
ATTR_FROM: EmailContentSensor.get_msg_sender(email_message),
ATTR_SUBJECT: EmailContentSensor.get_msg_subject(email_message),
ATTR_DATE: email_message["Date"],
ATTR_BODY: EmailContentSensor.get_msg_text(email_message),
}
if self._last_id == self._email_reader.last_unread_id:
break
email_message = self._email_reader.read_next()

26
tests/components/imap_email_content/test_sensor.py
View file

@ -14,9 +14,16 @@ from homeassistant.helpers.template import Template
class FakeEMailReader:
"""A test class for sending test emails."""
def __init__(self, messages):
def __init__(self, messages) -> None:
"""Set up the fake email reader."""
self._messages = messages
self.last_id = 0
self.last_unread_id = len(messages)
def add_test_message(self, message):
"""Add a new message."""
self.last_unread_id += 1
self._messages.append(message)
def connect(self):
"""Stay always Connected."""
@ -26,6 +33,7 @@ class FakeEMailReader:
"""Get the next email."""
if len(self._messages) == 0:
return None
self.last_id += 1
return self._messages.popleft()
@ -146,7 +154,7 @@ async def test_multi_part_only_other_text(hass: HomeAssistant) -> None:
async def test_multiple_emails(hass: HomeAssistant) -> None:
"""Test multiple emails."""
"""Test multiple emails, discarding stale states."""
states = []
test_message1 = email.message.Message()
@ -158,9 +166,15 @@ async def test_multiple_emails(hass: HomeAssistant) -> None:
test_message2 = email.message.Message()
test_message2["From"] = "sender@test.com"
test_message2["Subject"] = "Test 2"
test_message2["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message2["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 58)
test_message2.set_payload("Test Message 2")
test_message3 = email.message.Message()
test_message3["From"] = "sender@test.com"
test_message3["Subject"] = "Test 3"
test_message3["Date"] = datetime.datetime(2016, 1, 1, 12, 50, 1)
test_message3.set_payload("Test Message 2")
def state_changed_listener(entity_id, from_s, to_s):
states.append(to_s)
@ -178,11 +192,13 @@ async def test_multiple_emails(hass: HomeAssistant) -> None:
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
# Fake a new received message
sensor._email_reader.add_test_message(test_message3)
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert states[0].state == "Test"
assert states[1].state == "Test 2"
assert states[0].state == "Test 2"
assert states[1].state == "Test 3"
assert sensor.extra_state_attributes["body"] == "Test Message 2"