github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add HVV integration (Hamburg public transportation) (#31564) 

Co-authored-by: springstan <46536646+springstan@users.noreply.github.com>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Tom Schneider 2020-06-15 00:15:20 +02:00 committed by GitHub
parent 0b7d2aa4d7
commit 0331ebdd47
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 1211 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -343,6 +343,8 @@ omit =
homeassistant/components/hunterdouglas_powerview/sensor.py
homeassistant/components/hunterdouglas_powerview/cover.py
homeassistant/components/hunterdouglas_powerview/entity.py
homeassistant/components/hvv_departures/sensor.py
homeassistant/components/hvv_departures/__init__.py
homeassistant/components/hydrawise/*
homeassistant/components/hyperion/light.py
homeassistant/components/ialarm/alarm_control_panel.py

1
CODEOWNERS
View File

@ -185,6 +185,7 @@ homeassistant/components/huawei_lte/* @scop @fphammerle
homeassistant/components/huawei_router/* @abmantis
homeassistant/components/hue/* @balloob
homeassistant/components/hunterdouglas_powerview/* @bdraco
homeassistant/components/hvv_departures/* @vigonotion
homeassistant/components/iammeter/* @lewei50
homeassistant/components/iaqualink/* @flz
homeassistant/components/icloud/* @Quentame

52
homeassistant/components/hvv_departures/__init__.py Normal file
View File

@ -0,0 +1,52 @@
"""The HVV integration."""
import asyncio
from homeassistant.components.sensor import DOMAIN as DOMAIN_SENSOR
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client
from .const import DOMAIN
from .hub import GTIHub
PLATFORMS = [DOMAIN_SENSOR]
async def async_setup(hass: HomeAssistant, config: dict):
"""Set up the HVV component."""
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up HVV from a config entry."""
hub = GTIHub(
entry.data[CONF_HOST],
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
aiohttp_client.async_get_clientsession(hass),
)
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = hub
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
return unload_ok

218
homeassistant/components/hvv_departures/config_flow.py Normal file
View File

@ -0,0 +1,218 @@
"""Config flow for HVV integration."""
import logging
from pygti.auth import GTI_DEFAULT_HOST
from pygti.exceptions import CannotConnect, InvalidAuth
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_OFFSET, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.helpers import aiohttp_client
import homeassistant.helpers.config_validation as cv
from .const import ( # pylint:disable=unused-import
CONF_FILTER,
CONF_REAL_TIME,
CONF_STATION,
DOMAIN,
)
from .hub import GTIHub
_LOGGER = logging.getLogger(__name__)
SCHEMA_STEP_USER = vol.Schema(
{
vol.Required(CONF_HOST, default=GTI_DEFAULT_HOST): str,
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
SCHEMA_STEP_STATION = vol.Schema({vol.Required(CONF_STATION): str})
SCHEMA_STEP_OPTIONS = vol.Schema(
{
vol.Required(CONF_FILTER): vol.In([]),
vol.Required(CONF_OFFSET, default=0): vol.All(int, vol.Range(min=0)),
vol.Optional(CONF_REAL_TIME, default=True): bool,
}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for HVV."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
def __init__(self):
"""Initialize component."""
self.hub = None
self.data = None
self.stations = {}
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
session = aiohttp_client.async_get_clientsession(self.hass)
self.hub = GTIHub(
user_input[CONF_HOST],
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
session,
)
try:
response = await self.hub.authenticate()
_LOGGER.debug("Init gti: %r", response)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
if not errors:
self.data = user_input
return await self.async_step_station()
return self.async_show_form(
step_id="user", data_schema=SCHEMA_STEP_USER, errors=errors
)
async def async_step_station(self, user_input=None):
"""Handle the step where the user inputs his/her station."""
if user_input is not None:
errors = {}
check_name = await self.hub.gti.checkName(
{"theName": {"name": user_input[CONF_STATION]}, "maxList": 20}
)
stations = check_name.get("results")
self.stations = {
f"{station.get('name')}": station
for station in stations
if station.get("type") == "STATION"
}
if not self.stations:
errors["base"] = "no_results"
return self.async_show_form(
step_id="station", data_schema=SCHEMA_STEP_STATION, errors=errors
)
# schema
return await self.async_step_station_select()
return self.async_show_form(step_id="station", data_schema=SCHEMA_STEP_STATION)
async def async_step_station_select(self, user_input=None):
"""Handle the step where the user inputs his/her station."""
schema = vol.Schema({vol.Required(CONF_STATION): vol.In(list(self.stations))})
if user_input is None:
return self.async_show_form(step_id="station_select", data_schema=schema)
self.data.update({"station": self.stations[user_input[CONF_STATION]]})
title = self.data[CONF_STATION]["name"]
return self.async_create_entry(title=title, data=self.data)
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Get options flow."""
return OptionsFlowHandler(config_entry)
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Options flow handler."""
def __init__(self, config_entry):
"""Initialize HVV Departures options flow."""
self.config_entry = config_entry
self.options = dict(config_entry.options)
self.departure_filters = {}
self.hub = None
async def async_step_init(self, user_input=None):
"""Manage the options."""
errors = {}
if not self.departure_filters:
departure_list = {}
self.hub = self.hass.data[DOMAIN][self.config_entry.entry_id]
try:
departure_list = await self.hub.gti.departureList(
{
"station": self.config_entry.data[CONF_STATION],
"time": {"date": "heute", "time": "jetzt"},
"maxList": 5,
"maxTimeOffset": 200,
"useRealtime": True,
"returnFilters": True,
}
)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
if not errors:
self.departure_filters = {
str(i): departure_filter
for i, departure_filter in enumerate(departure_list.get("filter"))
}
if user_input is not None and not errors:
options = {
CONF_FILTER: [
self.departure_filters[x] for x in user_input[CONF_FILTER]
],
CONF_OFFSET: user_input[CONF_OFFSET],
CONF_REAL_TIME: user_input[CONF_REAL_TIME],
}
return self.async_create_entry(title="", data=options)
if CONF_FILTER in self.config_entry.options:
old_filter = [
i
for (i, f) in self.departure_filters.items()
if f in self.config_entry.options.get(CONF_FILTER)
]
else:
old_filter = []
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Optional(CONF_FILTER, default=old_filter): cv.multi_select(
{
key: f"{departure_filter['serviceName']}, {departure_filter['label']}"
for key, departure_filter in self.departure_filters.items()
}
),
vol.Required(
CONF_OFFSET,
default=self.config_entry.options.get(CONF_OFFSET, 0),
): vol.All(int, vol.Range(min=0)),
vol.Optional(
CONF_REAL_TIME,
default=self.config_entry.options.get(CONF_REAL_TIME, True),
): bool,
}
),
errors=errors,
)

10
homeassistant/components/hvv_departures/const.py Normal file
View File

@ -0,0 +1,10 @@
"""Constants for the HVV Departure integration."""
DOMAIN = "hvv_departures"
DEFAULT_NAME = DOMAIN
MANUFACTURER = "HVV"
ATTRIBUTION = "Data provided by www.hvv.de"
CONF_STATION = "station"
CONF_REAL_TIME = "real_time"
CONF_FILTER = "filter"

20
homeassistant/components/hvv_departures/hub.py Normal file
View File

@ -0,0 +1,20 @@
"""Hub."""
from pygti.gti import GTI, Auth
class GTIHub:
"""GTI Hub."""
def __init__(self, host, username, password, session):
"""Initialize."""
self.host = host
self.username = username
self.password = password
self.gti = GTI(Auth(session, self.username, self.password, self.host))
async def authenticate(self):
"""Test if we can authenticate with the host."""
return await self.gti.init()

12
homeassistant/components/hvv_departures/manifest.json Normal file
View File

@ -0,0 +1,12 @@
{
"domain": "hvv_departures",
"name": "HVV Departures",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/hvv_departures",
"requirements": [
"pygti==0.6.0"
],
"codeowners": [
"@vigonotion"
]
}

201
homeassistant/components/hvv_departures/sensor.py Normal file
View File

@ -0,0 +1,201 @@
"""Sensor platform for hvv."""
from datetime import timedelta
import logging
from aiohttp import ClientConnectorError
from pygti.exceptions import InvalidAuth
from homeassistant.const import ATTR_ATTRIBUTION, ATTR_ID, DEVICE_CLASS_TIMESTAMP
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle
from homeassistant.util.dt import utcnow
from .const import ATTRIBUTION, CONF_STATION, DOMAIN, MANUFACTURER
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=1)
MAX_LIST = 20
MAX_TIME_OFFSET = 360
ICON = "mdi:bus"
UNIT_OF_MEASUREMENT = "min"
ATTR_DEPARTURE = "departure"
ATTR_LINE = "line"
ATTR_ORIGIN = "origin"
ATTR_DIRECTION = "direction"
ATTR_TYPE = "type"
ATTR_DELAY = "delay"
ATTR_NEXT = "next"
PARALLEL_UPDATES = 0
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_devices):
"""Set up the sensor platform."""
hub = hass.data[DOMAIN][config_entry.entry_id]
session = aiohttp_client.async_get_clientsession(hass)
sensor = HVVDepartureSensor(hass, config_entry, session, hub)
async_add_devices([sensor], True)
class HVVDepartureSensor(Entity):
"""HVVDepartureSensor class."""
def __init__(self, hass, config_entry, session, hub):
"""Initialize."""
self.config_entry = config_entry
self.station_name = self.config_entry.data[CONF_STATION]["name"]
self.attr = {ATTR_ATTRIBUTION: ATTRIBUTION}
self._available = False
self._state = None
self._name = f"Departures at {self.station_name}"
self._last_error = None
self.gti = hub.gti
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self, **kwargs):
"""Update the sensor."""
departure_time = utcnow() + timedelta(
minutes=self.config_entry.options.get("offset", 0)
)
payload = {
"station": self.config_entry.data[CONF_STATION],
"time": {
"date": departure_time.strftime("%d.%m.%Y"),
"time": departure_time.strftime("%H:%M"),
},
"maxList": MAX_LIST,
"maxTimeOffset": MAX_TIME_OFFSET,
"useRealtime": self.config_entry.options.get("realtime", False),
}
if "filter" in self.config_entry.options:
payload.update({"filter": self.config_entry.options["filter"]})
try:
data = await self.gti.departureList(payload)
except InvalidAuth as error:
if self._last_error != InvalidAuth:
_LOGGER.error("Authentication failed: %r", error)
self._last_error = InvalidAuth
self._available = False
except ClientConnectorError as error:
if self._last_error != ClientConnectorError:
_LOGGER.warning("Network unavailable: %r", error)
self._last_error = ClientConnectorError
self._available = False
except Exception as error: # pylint: disable=broad-except
if self._last_error != error:
_LOGGER.error("Error occurred while fetching data: %r", error)
self._last_error = error
self._available = False
if not (data["returnCode"] == "OK" and data.get("departures")):
self._available = False
return
if self._last_error == ClientConnectorError:
_LOGGER.debug("Network available again")
self._last_error = None
departure = data["departures"][0]
line = departure["line"]
delay = departure.get("delay", 0)
self._available = True
self._state = (
departure_time
+ timedelta(minutes=departure["timeOffset"])
+ timedelta(seconds=delay)
).isoformat()
self.attr.update(
{
ATTR_LINE: line["name"],
ATTR_ORIGIN: line["origin"],
ATTR_DIRECTION: line["direction"],
ATTR_TYPE: line["type"]["shortInfo"],
ATTR_ID: line["id"],
ATTR_DELAY: delay,
}
)
departures = []
for departure in data["departures"]:
line = departure["line"]
delay = departure.get("delay", 0)
departures.append(
{
ATTR_DEPARTURE: departure_time
+ timedelta(minutes=departure["timeOffset"])
+ timedelta(seconds=delay),
ATTR_LINE: line["name"],
ATTR_ORIGIN: line["origin"],
ATTR_DIRECTION: line["direction"],
ATTR_TYPE: line["type"]["shortInfo"],
ATTR_ID: line["id"],
ATTR_DELAY: delay,
}
)
self.attr[ATTR_NEXT] = departures
@property
def unique_id(self):
"""Return a unique ID to use for this sensor."""
station_id = self.config_entry.data[CONF_STATION]["id"]
station_type = self.config_entry.data[CONF_STATION]["type"]
return f"{self.config_entry.entry_id}-{station_id}-{station_type}"
@property
def device_info(self):
"""Return the device info for this sensor."""
return {
"identifiers": {
(
DOMAIN,
self.config_entry.entry_id,
self.config_entry.data[CONF_STATION]["id"],
self.config_entry.data[CONF_STATION]["type"],
)
},
"name": self._name,
"manufacturer": MANUFACTURER,
}
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def state(self):
"""Return the state of the sensor."""
return self._state
@property
def icon(self):
"""Return the icon of the sensor."""
return ICON
@property
def available(self):
"""Return True if entity is available."""
return self._available
@property
def device_class(self):
"""Return the class of this device, from component DEVICE_CLASSES."""
return DEVICE_CLASS_TIMESTAMP
@property
def device_state_attributes(self):
"""Return the state attributes."""
return self.attr

48
homeassistant/components/hvv_departures/strings.json Normal file
View File

@ -0,0 +1,48 @@
{
"title": "HVV Departures",
"config": {
"step": {
"user": {
"title": "Connect to the HVV API",
"data": {
"host": "Host",
"username": "Username",
"password": "Password"
}
},
"station": {
"title": "Enter Station/Address",
"data": {
"station": "Station/Address"
}
},
"station_select": {
"title": "Select Station/Address",
"data": {
"station": "Station/Address"
}
}
},
"error": {
"cannot_connect": "Failed to connect, please try again",
"invalid_auth": "Invalid authentication",
"no_results": "No results. Try with a different station/address"
},
"abort": {
"already_configured": "Device is already configured"
}
},
"options": {
"step": {
"init": {
"title": "Options",
"description": "Change options for this departure sensor",
"data": {
"filter": "Select lines",
"offset": "Offset (minutes)",
"real_time": "Use real time data"
}
}
}
}
}

48
homeassistant/components/hvv_departures/translations/en.json Normal file
View File

@ -0,0 +1,48 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect, please try again",
"invalid_auth": "Invalid authentication",
"no_results": "No results. Try with a different station/address"
},
"step": {
"station": {
"data": {
"station": "Station/Address"
},
"title": "Enter Station/Address"
},
"station_select": {
"data": {
"station": "Station/Address"
},
"title": "Select Station/Address"
},
"user": {
"data": {
"host": "Host",
"password": "Password",
"username": "Username"
},
"title": "Connect to the HVV API"
}
}
},
"options": {
"step": {
"init": {
"data": {
"filter": "Select lines",
"offset": "Offset (minutes)",
"real_time": "Use real time data"
},
"description": "Change options for this departure sensor",
"title": "Options"
}
}
},
"title": "HVV Departures"
}

1
homeassistant/generated/config_flows.py
View File

@ -70,6 +70,7 @@ FLOWS = [
"huawei_lte",
"hue",
"hunterdouglas_powerview",
"hvv_departures",
"iaqualink",
"icloud",
"ifttt",

3
requirements_all.txt
View File

@ -1359,6 +1359,9 @@ pygatt[GATTTOOL]==4.0.5
# homeassistant.components.gtfs
pygtfs==0.1.5
# homeassistant.components.hvv_departures
pygti==0.6.0
# homeassistant.components.version
pyhaversion==3.3.0

3
requirements_test_all.txt
View File

@ -587,6 +587,9 @@ pyfttt==0.3
# homeassistant.components.skybeacon
pygatt[GATTTOOL]==4.0.5
# homeassistant.components.hvv_departures
pygti==0.6.0
# homeassistant.components.version
pyhaversion==3.3.0

1
tests/components/hvv_departures/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the HVV Departures integration."""

344
tests/components/hvv_departures/test_config_flow.py Normal file
View File

@ -0,0 +1,344 @@
"""Test the HVV Departures config flow."""
import json
from pygti.exceptions import CannotConnect, InvalidAuth
from homeassistant import data_entry_flow
from homeassistant.components.hvv_departures.const import (
CONF_FILTER,
CONF_REAL_TIME,
CONF_STATION,
DOMAIN,
)
from homeassistant.config_entries import CONN_CLASS_CLOUD_POLL, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_OFFSET, CONF_PASSWORD, CONF_USERNAME
from tests.async_mock import patch
from tests.common import MockConfigEntry, load_fixture
FIXTURE_INIT = json.loads(load_fixture("hvv_departures/init.json"))
FIXTURE_CHECK_NAME = json.loads(load_fixture("hvv_departures/check_name.json"))
FIXTURE_STATION_INFORMATION = json.loads(
load_fixture("hvv_departures/station_information.json")
)
FIXTURE_CONFIG_ENTRY = json.loads(load_fixture("hvv_departures/config_entry.json"))
FIXTURE_OPTIONS = json.loads(load_fixture("hvv_departures/options.json"))
FIXTURE_DEPARTURE_LIST = json.loads(load_fixture("hvv_departures/departure_list.json"))
async def test_user_flow(hass):
"""Test that config flow works."""
with patch(
"homeassistant.components.hvv_departures.hub.GTI.init",
return_value=FIXTURE_INIT,
), patch("pygti.gti.GTI.checkName", return_value=FIXTURE_CHECK_NAME,), patch(
"pygti.gti.GTI.stationInformation", return_value=FIXTURE_STATION_INFORMATION,
), patch(
"homeassistant.components.hvv_departures.async_setup", return_value=True
), patch(
"homeassistant.components.hvv_departures.async_setup_entry", return_value=True,
):
# step: user
result_user = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={
CONF_HOST: "api-test.geofox.de",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
assert result_user["step_id"] == "station"
# step: station
result_station = await hass.config_entries.flow.async_configure(
result_user["flow_id"], {CONF_STATION: "Wartenau"},
)
assert result_station["step_id"] == "station_select"
# step: station_select
result_station_select = await hass.config_entries.flow.async_configure(
result_user["flow_id"], {CONF_STATION: "Wartenau"},
)
assert result_station_select["type"] == "create_entry"
assert result_station_select["title"] == "Wartenau"
assert result_station_select["data"] == {
CONF_HOST: "api-test.geofox.de",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_STATION: {
"name": "Wartenau",
"city": "Hamburg",
"combinedName": "Wartenau",
"id": "Master:10901",
"type": "STATION",
"coordinate": {"x": 10.035515, "y": 53.56478},
"serviceTypes": ["bus", "u"],
"hasStationInformation": True,
},
}
async def test_user_flow_no_results(hass):
"""Test that config flow works when there are no results."""
with patch(
"homeassistant.components.hvv_departures.hub.GTI.init",
return_value=FIXTURE_INIT,
), patch(
"pygti.gti.GTI.checkName", return_value={"returnCode": "OK", "results": []},
), patch(
"homeassistant.components.hvv_departures.async_setup", return_value=True
), patch(
"homeassistant.components.hvv_departures.async_setup_entry", return_value=True,
):
# step: user
result_user = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={
CONF_HOST: "api-test.geofox.de",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
assert result_user["step_id"] == "station"
# step: station
result_station = await hass.config_entries.flow.async_configure(
result_user["flow_id"], {CONF_STATION: "non_existing_station"},
)
assert result_station["step_id"] == "station"
assert result_station["errors"]["base"] == "no_results"
async def test_user_flow_invalid_auth(hass):
"""Test that config flow handles invalid auth."""
with patch(
"homeassistant.components.hvv_departures.hub.GTI.init",
side_effect=InvalidAuth(
"ERROR_TEXT",
"Bei der Verarbeitung der Anfrage ist ein technisches Problem aufgetreten.",
"Authentication failed!",
),
):
# step: user
result_user = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={
CONF_HOST: "api-test.geofox.de",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
assert result_user["type"] == "form"
assert result_user["errors"] == {"base": "invalid_auth"}
async def test_user_flow_cannot_connect(hass):
"""Test that config flow handles connection errors."""
with patch(
"homeassistant.components.hvv_departures.hub.GTI.init",
side_effect=CannotConnect(),
):
# step: user
result_user = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={
CONF_HOST: "api-test.geofox.de",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
assert result_user["type"] == "form"
assert result_user["errors"] == {"base": "cannot_connect"}
async def test_user_flow_station(hass):
"""Test that config flow handles empty data on step station."""
with patch(
"homeassistant.components.hvv_departures.hub.GTI.init", return_value=True,
), patch(
"pygti.gti.GTI.checkName", return_value={"returnCode": "OK", "results": []},
):
# step: user
result_user = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={
CONF_HOST: "api-test.geofox.de",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
assert result_user["step_id"] == "station"
# step: station
result_station = await hass.config_entries.flow.async_configure(
result_user["flow_id"], None,
)
assert result_station["type"] == "form"
assert result_station["step_id"] == "station"
async def test_user_flow_station_select(hass):
"""Test that config flow handles empty data on step station_select."""
with patch(
"homeassistant.components.hvv_departures.hub.GTI.init", return_value=True,
), patch(
"pygti.gti.GTI.checkName", return_value=FIXTURE_CHECK_NAME,
):
result_user = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={
CONF_HOST: "api-test.geofox.de",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
result_station = await hass.config_entries.flow.async_configure(
result_user["flow_id"], {CONF_STATION: "Wartenau"},
)
# step: station_select
result_station_select = await hass.config_entries.flow.async_configure(
result_station["flow_id"], None,
)
assert result_station_select["type"] == "form"
assert result_station_select["step_id"] == "station_select"
async def test_options_flow(hass):
"""Test that options flow works."""
config_entry = MockConfigEntry(
version=1,
domain=DOMAIN,
title="Wartenau",
data=FIXTURE_CONFIG_ENTRY,
source="user",
connection_class=CONN_CLASS_CLOUD_POLL,
system_options={"disable_new_entities": False},
options=FIXTURE_OPTIONS,
unique_id="1234",
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.hvv_departures.hub.GTI.init", return_value=True,
), patch(
"pygti.gti.GTI.departureList", return_value=FIXTURE_DEPARTURE_LIST,
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_FILTER: ["0"], CONF_OFFSET: 15, CONF_REAL_TIME: False},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert config_entry.options == {
CONF_FILTER: [
{
"serviceID": "HHA-U:U1_HHA-U",
"stationIDs": ["Master:10902"],
"label": "Fuhlsbüttel Nord / Ochsenzoll / Norderstedt Mitte / Kellinghusenstraße / Ohlsdorf / Garstedt",
"serviceName": "U1",
}
],
CONF_OFFSET: 15,
CONF_REAL_TIME: False,
}
async def test_options_flow_invalid_auth(hass):
"""Test that options flow works."""
config_entry = MockConfigEntry(
version=1,
domain=DOMAIN,
title="Wartenau",
data=FIXTURE_CONFIG_ENTRY,
source="user",
connection_class=CONN_CLASS_CLOUD_POLL,
system_options={"disable_new_entities": False},
options=FIXTURE_OPTIONS,
unique_id="1234",
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.hvv_departures.hub.GTI.init",
side_effect=InvalidAuth(
"ERROR_TEXT",
"Bei der Verarbeitung der Anfrage ist ein technisches Problem aufgetreten.",
"Authentication failed!",
),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
assert result["errors"] == {"base": "invalid_auth"}
async def test_options_flow_cannot_connect(hass):
"""Test that options flow works."""
config_entry = MockConfigEntry(
version=1,
domain=DOMAIN,
title="Wartenau",
data=FIXTURE_CONFIG_ENTRY,
source="user",
connection_class=CONN_CLASS_CLOUD_POLL,
system_options={"disable_new_entities": False},
options=FIXTURE_OPTIONS,
unique_id="1234",
)
config_entry.add_to_hass(hass)
with patch(
"pygti.gti.GTI.departureList", side_effect=CannotConnect(),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
assert result["errors"] == {"base": "cannot_connect"}

15
tests/fixtures/hvv_departures/check_name.json vendored Normal file
View File

@ -0,0 +1,15 @@
{
"returnCode": "OK",
"results": [
{
"name": "Wartenau",
"city": "Hamburg",
"combinedName": "Wartenau",
"id": "Master:10901",
"type": "STATION",
"coordinate": {"x": 10.035515, "y": 53.56478},
"serviceTypes": ["bus", "u"],
"hasStationInformation": true
}
]
}

16
tests/fixtures/hvv_departures/config_entry.json vendored Normal file
View File

@ -0,0 +1,16 @@
{
"host": "api-test.geofox.de",
"username": "test-username",
"password": "test-password",
"station": {
"city": "Schmalfeld",
"combinedName": "Schmalfeld, Holstenstra\u00dfe",
"coordinate": {"x": 9.986115, "y": 53.874122},
"hasStationInformation": false,
"id": "Master:75279",
"name": "Holstenstra\u00dfe",
"serviceTypes": ["bus"],
"type": "STATION"
},
"stationInformation": {"returnCode": "OK"}
}

162
tests/fixtures/hvv_departures/departure_list.json vendored Normal file
View File

@ -0,0 +1,162 @@
{
"returnCode": "OK",
"time": {"date": "26.01.2020", "time": "22:52"},
"departures": [
{
"line": {
"name": "U1",
"direction": "Großhansdorf",
"origin": "Norderstedt Mitte",
"type": {
"simpleType": "TRAIN",
"shortInfo": "U",
"longInfo": "U-Bahn",
"model": "DT4"
},
"id": "HHA-U:U1_HHA-U"
},
"timeOffset": 0,
"delay": 0,
"serviceId": 1482563187,
"station": {"combinedName": "Wartenau", "id": "Master:10901"},
"attributes": [{"isPlanned": true, "types": ["REALTIME", "ACCURATE"]}]
},
{
"line": {
"name": "25",
"direction": "Bf. Altona",
"origin": "U Burgstraße",
"type": {
"simpleType": "BUS",
"shortInfo": "Bus",
"longInfo": "Niederflur Metrobus",
"model": "Gelenkbus"
},
"id": "HHA-B:25_HHA-B"
},
"timeOffset": 1,
"delay": 0,
"serviceId": 74567,
"station": {"combinedName": "U Wartenau", "id": "Master:60015"},
"attributes": [{"isPlanned": true, "types": ["REALTIME", "ACCURATE"]}]
},
{
"line": {
"name": "25",
"direction": "U Burgstraße",
"origin": "Bf. Altona",
"type": {
"simpleType": "BUS",
"shortInfo": "Bus",
"longInfo": "Niederflur Metrobus",
"model": "Gelenkbus"
},
"id": "HHA-B:25_HHA-B"
},
"timeOffset": 5,
"delay": 0,
"serviceId": 74328,
"station": {"combinedName": "U Wartenau", "id": "Master:60015"},
"attributes": [{"isPlanned": true, "types": ["REALTIME", "ACCURATE"]}]
},
{
"line": {
"name": "U1",
"direction": "Norderstedt Mitte",
"origin": "Großhansdorf",
"type": {
"simpleType": "TRAIN",
"shortInfo": "U",
"longInfo": "U-Bahn",
"model": "DT4"
},
"id": "HHA-U:U1_HHA-U"
},
"timeOffset": 8,
"delay": 0,
"station": {"combinedName": "Wartenau", "id": "Master:10901"},
"attributes": [{"isPlanned": true, "types": ["REALTIME", "ACCURATE"]}]
},
{
"line": {
"name": "U1",
"direction": "Ohlstedt",
"origin": "Norderstedt Mitte",
"type": {
"simpleType": "TRAIN",
"shortInfo": "U",
"longInfo": "U-Bahn",
"model": "DT4"
},
"id": "HHA-U:U1_HHA-U"
},
"timeOffset": 10,
"delay": 0,
"station": {"combinedName": "Wartenau", "id": "Master:10901"},
"attributes": [{"isPlanned": true, "types": ["REALTIME", "ACCURATE"]}]
}
],
"filter": [
{
"serviceID": "HHA-U:U1_HHA-U",
"stationIDs": ["Master:10902"],
"label": "Fuhlsbüttel Nord / Ochsenzoll / Norderstedt Mitte / Kellinghusenstraße / Ohlsdorf / Garstedt",
"serviceName": "U1"
},
{
"serviceID": "HHA-U:U1_HHA-U",
"stationIDs": ["Master:60904"],
"label": "Volksdorf / Farmsen / Großhansdorf / Ohlstedt",
"serviceName": "U1"
},
{
"serviceID": "HHA-B:25_HHA-B",
"stationIDs": ["Master:10047"],
"label": "Sachsenstraße / U Burgstraße",
"serviceName": "25"
},
{
"serviceID": "HHA-B:25_HHA-B",
"stationIDs": ["Master:60029"],
"label": "Winterhuder Marktplatz / Bf. Altona",
"serviceName": "25"
},
{
"serviceID": "HHA-B:36_HHA-B",
"stationIDs": ["Master:10049"],
"label": "S Blankenese / Rathausmarkt",
"serviceName": "36"
},
{
"serviceID": "HHA-B:36_HHA-B",
"stationIDs": ["Master:60013"],
"label": "Berner Heerweg",
"serviceName": "36"
},
{
"serviceID": "HHA-B:606_HHA-B",
"stationIDs": ["Master:10047"],
"label": "S Landwehr (Ramazan-Avci-Platz) - Rathausmarkt",
"serviceName": "606"
},
{
"serviceID": "HHA-B:606_HHA-B",
"stationIDs": ["Master:60029"],
"label": "Uferstraße - Winterhuder Marktplatz / Uferstraße - S Hamburg Airport / Uferstraße - U Langenhorn Markt (Krohnstieg)",
"serviceName": "606"
},
{
"serviceID": "HHA-B:608_HHA-B",
"stationIDs": ["Master:10048"],
"label": "Rathausmarkt / S Reeperbahn",
"serviceName": "608"
},
{
"serviceID": "HHA-B:608_HHA-B",
"stationIDs": ["Master:60012"],
"label": "Bf. Rahlstedt (Amtsstraße) / Großlohe",
"serviceName": "608"
}
],
"serviceTypes": ["UBAHN", "BUS", "METROBUS", "SCHNELLBUS", "NACHTBUS"]
}

10
tests/fixtures/hvv_departures/init.json vendored Normal file
View File

@ -0,0 +1,10 @@
{
"returnCode": "OK",
"beginOfService": "04.06.2020",
"endOfService": "13.12.2020",
"id": "1.80.0",
"dataId": "32.55.01",
"buildDate": "04.06.2020",
"buildTime": "14:29:59",
"buildText": "Regelfahrplan 2020"
}

12
tests/fixtures/hvv_departures/options.json vendored Normal file
View File

@ -0,0 +1,12 @@
{
"filter": [
{
"label": "S Landwehr (Ramazan-Avci-Platz) - Rathausmarkt",
"serviceID": "HHA-B:606_HHA-B",
"serviceName": "606",
"stationIDs": ["Master:10047"]
}
],
"offset": 10,
"realtime": true
}

32
tests/fixtures/hvv_departures/station_information.json vendored Normal file
View File

@ -0,0 +1,32 @@
{
"returnCode": "OK",
"partialStations": [
{
"stationOutline": "http://www.geofox.de/images/mobi/stationDescriptions/U_Wartenau.ZM3.jpg",
"elevators": [
{
"label": "A",
"cabinWidth": 124,
"cabinLength": 147,
"doorWidth": 110,
"description": "Zugang Landwehr <-> Schalterhalle",
"elevatorType": "Durchlader",
"buttonType": "BRAILLE",
"state": "READY"
},
{
"lines": ["U1"],
"label": "B",
"cabinWidth": 123,
"cabinLength": 145,
"doorWidth": 90,
"description": "Schalterhalle <-> U1",
"elevatorType": "Durchlader",
"buttonType": "COMBI",
"state": "READY"
}
]
}
],
"lastUpdate": {"date": "26.01.2020", "time": "22:49"}
}