Rewrite imap_email_content tests to use pytest (#41200)

This commit is contained in:
Edward Knight 2020-10-06 14:01:25 +01:00 committed by GitHub
parent 357a0e9d7e
commit 8bcd6c1880
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -4,14 +4,11 @@ import datetime
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."""
@ -31,207 +28,203 @@ class FakeEMailReader:
return self._messages.popleft()
class EmailContentSensor(unittest.TestCase):
"""Test the IMAP email content sensor."""
async def test_allowed_sender(hass):
"""Test emails from allowed sender."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
test_message["Subject"] = "Test"
test_message["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message.set_payload("Test Message")
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.addCleanup(self.hass.stop)
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["sender@test.com"],
None,
)
def test_allowed_sender(self):
"""Test emails from allowed sender."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
test_message["Subject"] = "Test"
test_message["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message.set_payload("Test Message")
sensor.entity_id = "sensor.emailtest"
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"]
assert "Test" == sensor.device_state_attributes["subject"]
assert (
datetime.datetime(2016, 1, 1, 12, 44, 57)
== sensor.device_state_attributes["date"]
)
sensor = imap_email_content.EmailContentSensor(
self.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()
assert "Test" == sensor.state
assert "Test Message" == sensor.device_state_attributes["body"]
assert "sender@test.com" == sensor.device_state_attributes["from"]
assert "Test" == sensor.device_state_attributes["subject"]
assert (
datetime.datetime(2016, 1, 1, 12, 44, 57)
== sensor.device_state_attributes["date"]
)
async def test_multi_part_with_text(hass):
"""Test multi part emails."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
msg["From"] = "sender@test.com"
def test_multi_part_with_text(self):
"""Test multi part emails."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
msg["From"] = "sender@test.com"
text = "Test Message"
html = "<html><head></head><body>Test Message</body></html>"
text = "Test Message"
html = "<html><head></head><body>Test Message</body></html>"
textPart = MIMEText(text, "plain")
htmlPart = MIMEText(html, "html")
textPart = MIMEText(text, "plain")
htmlPart = MIMEText(html, "html")
msg.attach(textPart)
msg.attach(htmlPart)
msg.attach(textPart)
msg.attach(htmlPart)
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)
sensor = imap_email_content.EmailContentSensor(
self.hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)
sensor.entity_id = "sensor.emailtest"
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"]
sensor.entity_id = "sensor.emailtest"
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
assert "Link" == sensor.state
assert "Test Message" == sensor.device_state_attributes["body"]
def test_multi_part_only_html(self):
"""Test multi part emails with only HTML."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
msg["From"] = "sender@test.com"
async def test_multi_part_only_html(hass):
"""Test multi part emails with only HTML."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
msg["From"] = "sender@test.com"
html = "<html><head></head><body>Test Message</body></html>"
html = "<html><head></head><body>Test Message</body></html>"
htmlPart = MIMEText(html, "html")
htmlPart = MIMEText(html, "html")
msg.attach(htmlPart)
msg.attach(htmlPart)
sensor = imap_email_content.EmailContentSensor(
self.hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)
sensor = imap_email_content.EmailContentSensor(
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()
assert "Link" == sensor.state
assert (
"<html><head></head><body>Test Message</body></html>"
== sensor.device_state_attributes["body"]
)
sensor.entity_id = "sensor.emailtest"
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):
"""Test multi part emails with only other text."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
msg["From"] = "sender@test.com"
other = "Test Message"
async def test_multi_part_only_other_text(hass):
"""Test multi part emails with only other text."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
msg["From"] = "sender@test.com"
htmlPart = MIMEText(other, "other")
other = "Test Message"
msg.attach(htmlPart)
htmlPart = MIMEText(other, "other")
sensor = imap_email_content.EmailContentSensor(
self.hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)
msg.attach(htmlPart)
sensor.entity_id = "sensor.emailtest"
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
assert "Link" == sensor.state
assert "Test Message" == sensor.device_state_attributes["body"]
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)
def test_multiple_emails(self):
"""Test multiple emails."""
states = []
sensor.entity_id = "sensor.emailtest"
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"]
test_message1 = email.message.Message()
test_message1["From"] = "sender@test.com"
test_message1["Subject"] = "Test"
test_message1["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message1.set_payload("Test Message")
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.set_payload("Test Message 2")
async def test_multiple_emails(hass):
"""Test multiple emails."""
states = []
def state_changed_listener(entity_id, from_s, to_s):
states.append(to_s)
test_message1 = email.message.Message()
test_message1["From"] = "sender@test.com"
test_message1["Subject"] = "Test"
test_message1["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message1.set_payload("Test Message")
track_state_change(self.hass, ["sensor.emailtest"], state_changed_listener)
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.set_payload("Test Message 2")
sensor = imap_email_content.EmailContentSensor(
self.hass,
FakeEMailReader(deque([test_message1, test_message2])),
"test_emails_sensor",
["sender@test.com"],
None,
)
def state_changed_listener(entity_id, from_s, to_s):
states.append(to_s)
sensor.entity_id = "sensor.emailtest"
async_track_state_change(hass, ["sensor.emailtest"], state_changed_listener)
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([test_message1, test_message2])),
"test_emails_sensor",
["sender@test.com"],
None,
)
assert "Test" == states[0].state
assert "Test 2" == states[1].state
sensor.entity_id = "sensor.emailtest"
assert "Test Message 2" == sensor.device_state_attributes["body"]
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()
def test_sender_not_allowed(self):
"""Test not whitelisted emails."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
test_message["Subject"] = "Test"
test_message["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message.set_payload("Test Message")
assert "Test" == states[0].state
assert "Test 2" == states[1].state
sensor = imap_email_content.EmailContentSensor(
self.hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["other@test.com"],
None,
)
assert "Test Message 2" == sensor.device_state_attributes["body"]
sensor.entity_id = "sensor.emailtest"
sensor.schedule_update_ha_state(True)
self.hass.block_till_done()
assert sensor.state is None
def test_template(self):
"""Test value template."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
test_message["Subject"] = "Test"
test_message["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message.set_payload("Test Message")
async def test_sender_not_allowed(hass):
"""Test not whitelisted emails."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
test_message["Subject"] = "Test"
test_message["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message.set_payload("Test Message")
sensor = imap_email_content.EmailContentSensor(
self.hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["sender@test.com"],
Template(
"{{ subject }} from {{ from }} with message {{ body }}", self.hass
),
)
sensor = imap_email_content.EmailContentSensor(
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()
assert "Test from sender@test.com with message Test Message" == sensor.state
sensor.entity_id = "sensor.emailtest"
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert sensor.state is None
async def test_template(hass):
"""Test value template."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
test_message["Subject"] = "Test"
test_message["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message.set_payload("Test Message")
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["sender@test.com"],
Template("{{ subject }} from {{ from }} with message {{ body }}", hass),
)
sensor.entity_id = "sensor.emailtest"
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