a43505a0a3
* Create a new NWS Alerts integration * Create a new NWS Alerts integration * Create new PECO integration * Remove empty keys * Revert "Create a new NWS Alerts integration" This reverts commit38309c5a87. * Revert "Create a new NWS Alerts integration" This reverts commitaeabdd37b8. * Fix test with new mock data * Add init and sensor to .coveragerc and more tests for config flow * Small fixes and replacing patch with pytest.raises in testing invalid county * Add type defs and fix test_config_flow to use MultipleValid instead * Fix issues with 'typing.Dict' * Move API communication to seperate PyPI library * Switch PyPI library from httpx to aiohttp to allow for passing in websessions * Commit file changes requested by farmio as listed here:d267e4300a* Add suggestions requested by farmio as listed here:586d8ffa42* Move native_unit_of_measurement from prop to attr * Update PLATFORMS constant type annotation Co-authored-by: Matthias Alphart <farmio@alphart.net> * Add peco to .strict-typing I am from school so I can't run mypy atm * Forgot to import Final * Do as requested [here](https://github.com/home-assistant/core/runs/5070634928?check_suite_focus=true) * Updated mypy.ini, checks should pass now * Fix to conform to mypy restrictions https://github.com/home-assistant/core/runs/5072861837\?check_suite_focus\=true * Fix type annotations * Fix tests * Use cast in async_update_data * Add data type to CoordinatorEntity and DataUpdateCoordinator * More cleanup from suggestions here: https://github.com/home-assistant/core/pull/65194\#pullrequestreview-908183493 * Fix tests for new code * Cleaning up a speck of dust * Remove unused variable from the peco sensor * Add rounding to percentage, and extra clean-up * Final suggestions from @farmio * Update SCAN_INTERVAL to be a little bit faster * Change the SCAN_INTERVAL to be somewhat near the update interval of the outage map, as noted by farmio * New UpdateCoordinator typing
256 lines
9 KiB
Python
256 lines
9 KiB
Python
"""Test the PECO Outage Counter sensors."""
|
|
import asyncio
|
|
|
|
from homeassistant.components.peco.const import DOMAIN
|
|
from homeassistant.config_entries import ConfigEntryState
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
from tests.common import MockConfigEntry
|
|
from tests.test_util.aiohttp import AiohttpClientMocker
|
|
|
|
MOCK_ENTRY_DATA = {"county": "TOTAL"}
|
|
COUNTY_ENTRY_DATA = {"county": "BUCKS"}
|
|
INVALID_COUNTY_DATA = {"county": "INVALID"}
|
|
|
|
|
|
async def test_sensor_available(
|
|
aioclient_mock: AiohttpClientMocker, hass: HomeAssistant
|
|
) -> None:
|
|
"""Test that the sensors are working."""
|
|
# Totals Test
|
|
aioclient_mock.get(
|
|
"https://kubra.io/stormcenter/api/v1/stormcenters/39e6d9f3-fdea-4539-848f-b8631945da6f/views/74de8a50-3f45-4f6a-9483-fd618bb9165d/currentState?preview=false",
|
|
json={"data": {"interval_generation_data": "data/TEST"}},
|
|
)
|
|
aioclient_mock.get(
|
|
"https://kubra.io/data/TEST/public/reports/a36a6292-1c55-44de-a6a9-44fedf9482ee_report.json",
|
|
json={
|
|
"file_data": {
|
|
"totals": {
|
|
"cust_a": {
|
|
"val": 123,
|
|
},
|
|
"percent_cust_a": {
|
|
"val": 1.23,
|
|
},
|
|
"n_out": 456,
|
|
"cust_s": 789,
|
|
}
|
|
}
|
|
},
|
|
)
|
|
|
|
config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_ENTRY_DATA)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert hass.data[DOMAIN]
|
|
|
|
entries = hass.config_entries.async_entries(DOMAIN)
|
|
assert len(entries) == 1
|
|
assert config_entry.state == ConfigEntryState.LOADED
|
|
|
|
sensors_to_get = [
|
|
"total_customers_out",
|
|
"total_percent_customers_out",
|
|
"total_outage_count",
|
|
"total_customers_served",
|
|
]
|
|
|
|
for sensor in sensors_to_get:
|
|
sensor_entity = hass.states.get(f"sensor.{sensor}")
|
|
assert sensor_entity is not None
|
|
assert sensor_entity.state != "unavailable"
|
|
|
|
if sensor == "total_customers_out":
|
|
assert sensor_entity.state == "123"
|
|
elif sensor == "total_percent_customers_out":
|
|
assert sensor_entity.state == "15.589"
|
|
elif sensor == "total_outage_count":
|
|
assert sensor_entity.state == "456"
|
|
elif sensor == "total_customers_served":
|
|
assert sensor_entity.state == "789"
|
|
|
|
# County Test
|
|
aioclient_mock.clear_requests()
|
|
aioclient_mock.get(
|
|
"https://kubra.io/stormcenter/api/v1/stormcenters/39e6d9f3-fdea-4539-848f-b8631945da6f/views/74de8a50-3f45-4f6a-9483-fd618bb9165d/currentState?preview=false",
|
|
json={"data": {"interval_generation_data": "data/TEST"}},
|
|
)
|
|
aioclient_mock.get(
|
|
"https://kubra.io/data/TEST/public/reports/a36a6292-1c55-44de-a6a9-44fedf9482ee_report.json",
|
|
json={
|
|
"file_data": {
|
|
"areas": [
|
|
{
|
|
"name": "BUCKS",
|
|
"cust_a": {
|
|
"val": 123,
|
|
},
|
|
"percent_cust_a": {
|
|
"val": 1.23,
|
|
},
|
|
"n_out": 456,
|
|
"cust_s": 789,
|
|
}
|
|
]
|
|
}
|
|
},
|
|
)
|
|
|
|
config_entry = MockConfigEntry(domain=DOMAIN, data=COUNTY_ENTRY_DATA)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert hass.data[DOMAIN]
|
|
|
|
entries = hass.config_entries.async_entries(DOMAIN)
|
|
assert len(entries) == 2
|
|
assert config_entry.state == ConfigEntryState.LOADED
|
|
|
|
sensors_to_get = [
|
|
"bucks_customers_out",
|
|
"bucks_percent_customers_out",
|
|
"bucks_outage_count",
|
|
"bucks_customers_served",
|
|
]
|
|
|
|
for sensor in sensors_to_get:
|
|
sensor_entity = hass.states.get(f"sensor.{sensor}")
|
|
assert sensor_entity is not None
|
|
assert sensor_entity.state != "unavailable"
|
|
|
|
if sensor == "bucks_customers_out":
|
|
assert sensor_entity.state == "123"
|
|
elif sensor == "bucks_percent_customers_out":
|
|
assert sensor_entity.state == "15.589"
|
|
elif sensor == "bucks_outage_count":
|
|
assert sensor_entity.state == "456"
|
|
elif sensor == "bucks_customers_served":
|
|
assert sensor_entity.state == "789"
|
|
|
|
|
|
async def test_http_error(
|
|
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, caplog
|
|
):
|
|
"""Test if it raises an error when there is an HTTP error."""
|
|
aioclient_mock.get(
|
|
"https://kubra.io/stormcenter/api/v1/stormcenters/39e6d9f3-fdea-4539-848f-b8631945da6f/views/74de8a50-3f45-4f6a-9483-fd618bb9165d/currentState?preview=false",
|
|
json={"data": {"interval_generation_data": "data/TEST"}},
|
|
)
|
|
aioclient_mock.get(
|
|
"https://kubra.io/data/TEST/public/reports/a36a6292-1c55-44de-a6a9-44fedf9482ee_report.json",
|
|
status=500,
|
|
json={"error": "Internal Server Error"},
|
|
)
|
|
|
|
config_entry = MockConfigEntry(domain=DOMAIN, data=COUNTY_ENTRY_DATA)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert hass.data[DOMAIN]
|
|
|
|
assert "Error getting PECO outage counter data" in caplog.text
|
|
|
|
|
|
async def test_bad_json(
|
|
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, caplog
|
|
):
|
|
"""Test if it raises an error when there is bad JSON."""
|
|
aioclient_mock.get(
|
|
"https://kubra.io/stormcenter/api/v1/stormcenters/39e6d9f3-fdea-4539-848f-b8631945da6f/views/74de8a50-3f45-4f6a-9483-fd618bb9165d/currentState?preview=false",
|
|
json={"data": {}},
|
|
)
|
|
|
|
config_entry = MockConfigEntry(domain=DOMAIN, data=COUNTY_ENTRY_DATA)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert hass.data[DOMAIN]
|
|
|
|
assert "ConfigEntryNotReady" in caplog.text
|
|
|
|
|
|
async def test_total_http_error(
|
|
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, caplog
|
|
):
|
|
"""Test if it raises an error when there is an HTTP error."""
|
|
aioclient_mock.get(
|
|
"https://kubra.io/stormcenter/api/v1/stormcenters/39e6d9f3-fdea-4539-848f-b8631945da6f/views/74de8a50-3f45-4f6a-9483-fd618bb9165d/currentState?preview=false",
|
|
json={"data": {"interval_generation_data": "data/TEST"}},
|
|
)
|
|
aioclient_mock.get(
|
|
"https://kubra.io/data/TEST/public/reports/a36a6292-1c55-44de-a6a9-44fedf9482ee_report.json",
|
|
status=500,
|
|
json={"error": "Internal Server Error"},
|
|
)
|
|
|
|
config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_ENTRY_DATA)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert hass.data[DOMAIN]
|
|
|
|
assert "Error getting PECO outage counter data" in caplog.text
|
|
|
|
|
|
async def test_total_bad_json(
|
|
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, caplog
|
|
):
|
|
"""Test if it raises an error when there is bad JSON."""
|
|
aioclient_mock.get(
|
|
"https://kubra.io/stormcenter/api/v1/stormcenters/39e6d9f3-fdea-4539-848f-b8631945da6f/views/74de8a50-3f45-4f6a-9483-fd618bb9165d/currentState?preview=false",
|
|
json={"data": {}},
|
|
)
|
|
|
|
config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_ENTRY_DATA)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert hass.data[DOMAIN]
|
|
|
|
assert "ConfigEntryNotReady" in caplog.text
|
|
|
|
|
|
async def test_update_timeout(
|
|
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, caplog
|
|
):
|
|
"""Test if it raises an error when there is a timeout."""
|
|
aioclient_mock.get(
|
|
"https://kubra.io/stormcenter/api/v1/stormcenters/39e6d9f3-fdea-4539-848f-b8631945da6f/views/74de8a50-3f45-4f6a-9483-fd618bb9165d/currentState?preview=false",
|
|
exc=asyncio.TimeoutError(),
|
|
)
|
|
|
|
config_entry = MockConfigEntry(domain=DOMAIN, data=COUNTY_ENTRY_DATA)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert hass.data[DOMAIN]
|
|
|
|
assert "Timeout fetching data" in caplog.text
|
|
|
|
|
|
async def test_total_update_timeout(
|
|
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, caplog
|
|
):
|
|
"""Test if it raises an error when there is a timeout."""
|
|
aioclient_mock.get(
|
|
"https://kubra.io/stormcenter/api/v1/stormcenters/39e6d9f3-fdea-4539-848f-b8631945da6f/views/74de8a50-3f45-4f6a-9483-fd618bb9165d/currentState?preview=false",
|
|
exc=asyncio.TimeoutError(),
|
|
)
|
|
|
|
config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_ENTRY_DATA)
|
|
config_entry.add_to_hass(hass)
|
|
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert hass.data[DOMAIN]
|
|
|
|
assert "Timeout fetching data" in caplog.text
|