Use storage helper in feedreader (#98754)
This commit is contained in:
Martin Hjelmare
GitHub
parent
a42d975c49
commit
91df9434d0
2 changed files with 283 additions and 127 deletions
|
|
@ -1,7 +1,11 @@
|
|||
"""The tests for the feedreader component."""
|
||||
from datetime import timedelta
|
||||
from collections.abc import Generator
|
||||
from datetime import datetime, timedelta
|
||||
import pickle
|
||||
from time import gmtime
|
||||
from typing import Any
|
||||
from unittest import mock
|
||||
from unittest.mock import mock_open, patch
|
||||
from unittest.mock import MagicMock, mock_open, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
|
@ -13,7 +17,7 @@ from homeassistant.components.feedreader import (
|
|||
EVENT_FEEDREADER,
|
||||
)
|
||||
from homeassistant.const import CONF_SCAN_INTERVAL, EVENT_HOMEASSISTANT_START
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.core import Event, HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
|
|
@ -27,7 +31,7 @@ VALID_CONFIG_4 = {feedreader.DOMAIN: {CONF_URLS: [URL], CONF_MAX_ENTRIES: 5}}
|
|||
VALID_CONFIG_5 = {feedreader.DOMAIN: {CONF_URLS: [URL], CONF_MAX_ENTRIES: 1}}
|
||||
|
||||
|
||||
def load_fixture_bytes(src):
|
||||
def load_fixture_bytes(src: str) -> bytes:
|
||||
"""Return byte stream of fixture."""
|
||||
feed_data = load_fixture(src)
|
||||
raw = bytes(feed_data, "utf-8")
|
||||
|
|
@ -35,72 +39,198 @@ def load_fixture_bytes(src):
|
|||
|
||||
|
||||
@pytest.fixture(name="feed_one_event")
|
||||
def fixture_feed_one_event(hass):
|
||||
def fixture_feed_one_event(hass: HomeAssistant) -> bytes:
|
||||
"""Load test feed data for one event."""
|
||||
return load_fixture_bytes("feedreader.xml")
|
||||
|
||||
|
||||
@pytest.fixture(name="feed_two_event")
|
||||
def fixture_feed_two_events(hass):
|
||||
def fixture_feed_two_events(hass: HomeAssistant) -> bytes:
|
||||
"""Load test feed data for two event."""
|
||||
return load_fixture_bytes("feedreader1.xml")
|
||||
|
||||
|
||||
@pytest.fixture(name="feed_21_events")
|
||||
def fixture_feed_21_events(hass):
|
||||
def fixture_feed_21_events(hass: HomeAssistant) -> bytes:
|
||||
"""Load test feed data for twenty one events."""
|
||||
return load_fixture_bytes("feedreader2.xml")
|
||||
|
||||
|
||||
@pytest.fixture(name="feed_three_events")
|
||||
def fixture_feed_three_events(hass):
|
||||
def fixture_feed_three_events(hass: HomeAssistant) -> bytes:
|
||||
"""Load test feed data for three events."""
|
||||
return load_fixture_bytes("feedreader3.xml")
|
||||
|
||||
|
||||
@pytest.fixture(name="feed_atom_event")
|
||||
def fixture_feed_atom_event(hass):
|
||||
def fixture_feed_atom_event(hass: HomeAssistant) -> bytes:
|
||||
"""Load test feed data for atom event."""
|
||||
return load_fixture_bytes("feedreader5.xml")
|
||||
|
||||
|
||||
@pytest.fixture(name="events")
|
||||
async def fixture_events(hass):
|
||||
async def fixture_events(hass: HomeAssistant) -> list[Event]:
|
||||
"""Fixture that catches alexa events."""
|
||||
return async_capture_events(hass, EVENT_FEEDREADER)
|
||||
|
||||
|
||||
@pytest.fixture(name="feed_storage", autouse=True)
|
||||
def fixture_feed_storage():
|
||||
@pytest.fixture(name="storage")
|
||||
def fixture_storage(request: pytest.FixtureRequest) -> Generator[None, None, None]:
|
||||
"""Set up the test storage environment."""
|
||||
if request.param == "legacy_storage":
|
||||
with patch("os.path.exists", return_value=False):
|
||||
yield
|
||||
elif request.param == "json_storage":
|
||||
with patch("os.path.exists", return_value=True):
|
||||
yield
|
||||
else:
|
||||
raise RuntimeError("Invalid storage fixture")
|
||||
|
||||
|
||||
@pytest.fixture(name="legacy_storage_open")
|
||||
def fixture_legacy_storage_open() -> Generator[MagicMock, None, None]:
|
||||
"""Mock builtins.open for feedreader storage."""
|
||||
with patch("homeassistant.components.feedreader.open", mock_open(), create=True):
|
||||
yield
|
||||
|
||||
|
||||
async def test_setup_one_feed(hass: HomeAssistant) -> None:
|
||||
"""Test the general setup of this component."""
|
||||
with patch(
|
||||
"homeassistant.components.feedreader.track_time_interval"
|
||||
"homeassistant.components.feedreader.open",
|
||||
mock_open(),
|
||||
create=True,
|
||||
) as open_mock:
|
||||
yield open_mock
|
||||
|
||||
|
||||
@pytest.fixture(name="legacy_storage_load", autouse=True)
|
||||
def fixture_legacy_storage_load(
|
||||
legacy_storage_open,
|
||||
) -> Generator[MagicMock, None, None]:
|
||||
"""Mock builtins.open for feedreader storage."""
|
||||
with patch(
|
||||
"homeassistant.components.feedreader.pickle.load", return_value={}
|
||||
) as pickle_load:
|
||||
yield pickle_load
|
||||
|
||||
|
||||
async def test_setup_no_feeds(hass: HomeAssistant) -> None:
|
||||
"""Test config with no urls."""
|
||||
assert not await async_setup_component(
|
||||
hass, feedreader.DOMAIN, {feedreader.DOMAIN: {CONF_URLS: []}}
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("open_error", "load_error"),
|
||||
[
|
||||
(FileNotFoundError("No file"), None),
|
||||
(OSError("Boom"), None),
|
||||
(None, pickle.PickleError("Bad data")),
|
||||
],
|
||||
)
|
||||
async def test_legacy_storage_error(
|
||||
hass: HomeAssistant,
|
||||
legacy_storage_open: MagicMock,
|
||||
legacy_storage_load: MagicMock,
|
||||
open_error: Exception | None,
|
||||
load_error: Exception | None,
|
||||
) -> None:
|
||||
"""Test legacy storage error."""
|
||||
legacy_storage_open.side_effect = open_error
|
||||
legacy_storage_load.side_effect = load_error
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.feedreader.async_track_time_interval"
|
||||
) as track_method:
|
||||
assert await async_setup_component(hass, feedreader.DOMAIN, VALID_CONFIG_1)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
track_method.assert_called_once_with(
|
||||
hass, mock.ANY, DEFAULT_SCAN_INTERVAL, cancel_on_shutdown=True
|
||||
)
|
||||
track_method.assert_called_once_with(
|
||||
hass, mock.ANY, DEFAULT_SCAN_INTERVAL, cancel_on_shutdown=True
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("storage", ["legacy_storage", "json_storage"], indirect=True)
|
||||
async def test_storage_data_loading(
|
||||
hass: HomeAssistant,
|
||||
events: list[Event],
|
||||
feed_one_event: bytes,
|
||||
legacy_storage_load: MagicMock,
|
||||
hass_storage: dict[str, Any],
|
||||
storage: None,
|
||||
) -> None:
|
||||
"""Test loading existing storage data."""
|
||||
storage_data: dict[str, str] = {URL: "2018-04-30T05:10:00+00:00"}
|
||||
hass_storage[feedreader.DOMAIN] = {
|
||||
"version": 1,
|
||||
"minor_version": 1,
|
||||
"key": feedreader.DOMAIN,
|
||||
"data": storage_data,
|
||||
}
|
||||
legacy_storage_data = {
|
||||
URL: gmtime(datetime.fromisoformat(storage_data[URL]).timestamp())
|
||||
}
|
||||
legacy_storage_load.return_value = legacy_storage_data
|
||||
|
||||
with patch(
|
||||
"feedparser.http.get",
|
||||
return_value=feed_one_event,
|
||||
):
|
||||
assert await async_setup_component(hass, feedreader.DOMAIN, VALID_CONFIG_2)
|
||||
|
||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# no new events
|
||||
assert not events
|
||||
|
||||
|
||||
async def test_storage_data_writing(
|
||||
hass: HomeAssistant,
|
||||
events: list[Event],
|
||||
feed_one_event: bytes,
|
||||
hass_storage: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test writing to storage."""
|
||||
storage_data: dict[str, str] = {URL: "2018-04-30T05:10:00+00:00"}
|
||||
|
||||
with patch(
|
||||
"feedparser.http.get",
|
||||
return_value=feed_one_event,
|
||||
), patch("homeassistant.components.feedreader.DELAY_SAVE", new=0):
|
||||
assert await async_setup_component(hass, feedreader.DOMAIN, VALID_CONFIG_2)
|
||||
|
||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# one new event
|
||||
assert len(events) == 1
|
||||
|
||||
# storage data updated
|
||||
assert hass_storage[feedreader.DOMAIN]["data"] == storage_data
|
||||
|
||||
|
||||
@pytest.mark.parametrize("storage", ["legacy_storage", "json_storage"], indirect=True)
|
||||
async def test_setup_one_feed(hass: HomeAssistant, storage: None) -> None:
|
||||
"""Test the general setup of this component."""
|
||||
with patch(
|
||||
"homeassistant.components.feedreader.async_track_time_interval"
|
||||
) as track_method:
|
||||
assert await async_setup_component(hass, feedreader.DOMAIN, VALID_CONFIG_1)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
track_method.assert_called_once_with(
|
||||
hass, mock.ANY, DEFAULT_SCAN_INTERVAL, cancel_on_shutdown=True
|
||||
)
|
||||
|
||||
|
||||
async def test_setup_scan_interval(hass: HomeAssistant) -> None:
|
||||
"""Test the setup of this component with scan interval."""
|
||||
with patch(
|
||||
"homeassistant.components.feedreader.track_time_interval"
|
||||
"homeassistant.components.feedreader.async_track_time_interval"
|
||||
) as track_method:
|
||||
assert await async_setup_component(hass, feedreader.DOMAIN, VALID_CONFIG_2)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
track_method.assert_called_once_with(
|
||||
hass, mock.ANY, timedelta(seconds=60), cancel_on_shutdown=True
|
||||
)
|
||||
track_method.assert_called_once_with(
|
||||
hass, mock.ANY, timedelta(seconds=60), cancel_on_shutdown=True
|
||||
)
|
||||
|
||||
|
||||
async def test_setup_max_entries(hass: HomeAssistant) -> None:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue