Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 39 additions & 46 deletions tests/components/imap_email_content/test_sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,11 @@
import email
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import unittest

from homeassistant.components.imap_email_content import sensor as imap_email_content
from homeassistant.helpers.event import track_state_change
from homeassistant.helpers.event import async_track_state_change
from homeassistant.helpers.template import Template

from tests.common import get_test_home_assistant


class FakeEMailReader:
"""A test class for sending test emails."""
Expand All @@ -31,15 +28,7 @@ def read_next(self):
return self._messages.popleft()


class EmailContentSensor(unittest.TestCase):
"""Test the IMAP email content sensor."""

def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.addCleanup(self.hass.stop)

def test_allowed_sender(self):
async def test_allowed_sender(hass):
"""Test emails from allowed sender."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
Expand All @@ -48,16 +37,16 @@ def test_allowed_sender(self):
test_message.set_payload("Test Message")

sensor = imap_email_content.EmailContentSensor(
self.hass,
hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["sender@test.com"],
None,
)

sensor.entity_id = "sensor.emailtest"
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert "Test" == sensor.state
assert "Test Message" == sensor.device_state_attributes["body"]
assert "sender@test.com" == sensor.device_state_attributes["from"]
Expand All @@ -67,7 +56,8 @@ def test_allowed_sender(self):
== sensor.device_state_attributes["date"]
)

def test_multi_part_with_text(self):

async def test_multi_part_with_text(hass):
"""Test multi part emails."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
Expand All @@ -83,20 +73,21 @@ def test_multi_part_with_text(self):
msg.attach(htmlPart)

sensor = imap_email_content.EmailContentSensor(
self.hass,
hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)

sensor.entity_id = "sensor.emailtest"
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert "Link" == sensor.state
assert "Test Message" == sensor.device_state_attributes["body"]

def test_multi_part_only_html(self):

async def test_multi_part_only_html(hass):
"""Test multi part emails with only HTML."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
Expand All @@ -109,23 +100,24 @@ def test_multi_part_only_html(self):
msg.attach(htmlPart)

sensor = imap_email_content.EmailContentSensor(
self.hass,
hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)

sensor.entity_id = "sensor.emailtest"
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert "Link" == sensor.state
assert (
"<html><head></head><body>Test Message</body></html>"
== sensor.device_state_attributes["body"]
)

def test_multi_part_only_other_text(self):

async def test_multi_part_only_other_text(hass):
"""Test multi part emails with only other text."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
Expand All @@ -138,20 +130,21 @@ def test_multi_part_only_other_text(self):
msg.attach(htmlPart)

sensor = imap_email_content.EmailContentSensor(
self.hass,
hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)

sensor.entity_id = "sensor.emailtest"
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert "Link" == sensor.state
assert "Test Message" == sensor.device_state_attributes["body"]

def test_multiple_emails(self):

async def test_multiple_emails(hass):
"""Test multiple emails."""
states = []

Expand All @@ -170,10 +163,10 @@ def test_multiple_emails(self):
def state_changed_listener(entity_id, from_s, to_s):
states.append(to_s)

track_state_change(self.hass, ["sensor.emailtest"], state_changed_listener)
async_track_state_change(hass, ["sensor.emailtest"], state_changed_listener)

sensor = imap_email_content.EmailContentSensor(
self.hass,
hass,
FakeEMailReader(deque([test_message1, test_message2])),
"test_emails_sensor",
["sender@test.com"],
Expand All @@ -182,17 +175,18 @@ def state_changed_listener(entity_id, from_s, to_s):

sensor.entity_id = "sensor.emailtest"

sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()

assert "Test" == states[0].state
assert "Test 2" == states[1].state

assert "Test Message 2" == sensor.device_state_attributes["body"]

def test_sender_not_allowed(self):

async def test_sender_not_allowed(hass):
"""Test not whitelisted emails."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
Expand All @@ -201,19 +195,20 @@ def test_sender_not_allowed(self):
test_message.set_payload("Test Message")

sensor = imap_email_content.EmailContentSensor(
self.hass,
hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["other@test.com"],
None,
)

sensor.entity_id = "sensor.emailtest"
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert sensor.state is None

def test_template(self):

async def test_template(hass):
"""Test value template."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
Expand All @@ -222,16 +217,14 @@ def test_template(self):
test_message.set_payload("Test Message")

sensor = imap_email_content.EmailContentSensor(
self.hass,
hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["sender@test.com"],
Template(
"{{ subject }} from {{ from }} with message {{ body }}", self.hass
),
Template("{{ subject }} from {{ from }} with message {{ body }}", hass),
)

sensor.entity_id = "sensor.emailtest"
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert "Test from sender@test.com with message Test Message" == sensor.state