evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Re-write tests for transmission (#76607)

Browse source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Rami Mosleh 2022-08-19 13:10:34 +03:00 • committed by GitHub
parent 039c071a80
commit 80c1c11b1a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 238 additions and 375 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
tests/components/transmission/__init__.py
View file

@ -1 +1,9 @@
"""Tests for Transmission."""
MOCK_CONFIG_DATA = {
"name": "Transmission",
"host": "0.0.0.0",
"username": "user",
"password": "pass",
"port": 9091,
}

459
tests/components/transmission/test_config_flow.py
View file

@ -1,334 +1,165 @@
"""Tests for Transmission config flow."""
from unittest.mock import patch
from unittest.mock import MagicMock, patch
import pytest
from transmissionrpc.error import TransmissionError
from homeassistant import config_entries, data_entry_flow
from homeassistant import config_entries
from homeassistant.components import transmission
from homeassistant.components.transmission import config_flow
from homeassistant.components.transmission.const import DEFAULT_SCAN_INTERVAL
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_SCAN_INTERVAL,
CONF_USERNAME,
)
from homeassistant.components.transmission.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import MOCK_CONFIG_DATA
from tests.common import MockConfigEntry
NAME = "Transmission"
HOST = "192.168.1.100"
USERNAME = "username"
PASSWORD = "password"
PORT = 9091
SCAN_INTERVAL = 10
MOCK_ENTRY = {
CONF_NAME: NAME,
CONF_HOST: HOST,
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_PORT: PORT,
}
@pytest.fixture(name="api")
def mock_transmission_api():
@pytest.fixture(autouse=True)
def mock_api():
"""Mock an api."""
with patch("transmissionrpc.Client"):
yield
with patch("transmissionrpc.Client") as api:
yield api
@pytest.fixture(name="auth_error")
def mock_api_authentication_error():
"""Mock an api."""
with patch(
"transmissionrpc.Client", side_effect=TransmissionError("401: Unauthorized")
):
yield
@pytest.fixture(name="conn_error")
def mock_api_connection_error():
"""Mock an api."""
with patch(
"transmissionrpc.Client",
side_effect=TransmissionError("111: Connection refused"),
):
yield
@pytest.fixture(name="unknown_error")
def mock_api_unknown_error():
"""Mock an api."""
with patch("transmissionrpc.Client", side_effect=TransmissionError):
yield
@pytest.fixture(name="transmission_setup", autouse=True)
def transmission_setup_fixture():
"""Mock transmission entry setup."""
with patch(
"homeassistant.components.transmission.async_setup_entry", return_value=True
):
yield
def init_config_flow(hass):
"""Init a configuration flow."""
flow = config_flow.TransmissionFlowHandler()
flow.hass = hass
return flow
async def test_flow_user_config(hass, api):
"""Test user config."""
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["type"] == FlowResultType.FORM
with patch(
"homeassistant.components.transmission.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG_DATA,
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Transmission"
assert result2["data"] == MOCK_CONFIG_DATA
assert len(mock_setup_entry.mock_calls) == 1
async def test_flow_required_fields(hass, api):
"""Test with required fields only."""
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_NAME: NAME, CONF_HOST: HOST, CONF_PORT: PORT},
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == NAME
assert result["data"][CONF_NAME] == NAME
assert result["data"][CONF_HOST] == HOST
assert result["data"][CONF_PORT] == PORT
async def test_flow_all_provided(hass, api):
"""Test with all provided."""
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=MOCK_ENTRY,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == NAME
assert result["data"][CONF_NAME] == NAME
assert result["data"][CONF_HOST] == HOST
assert result["data"][CONF_USERNAME] == USERNAME
assert result["data"][CONF_PASSWORD] == PASSWORD
assert result["data"][CONF_PORT] == PORT
async def test_options(hass):
"""Test updating options."""
entry = MockConfigEntry(
domain=transmission.DOMAIN,
title=CONF_NAME,
data=MOCK_ENTRY,
options={CONF_SCAN_INTERVAL: DEFAULT_SCAN_INTERVAL},
)
flow = init_config_flow(hass)
options_flow = flow.async_get_options_flow(entry)
result = await options_flow.async_step_init()
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
result = await options_flow.async_step_init({CONF_SCAN_INTERVAL: 10})
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"][CONF_SCAN_INTERVAL] == 10
async def test_host_already_configured(hass, api):
"""Test host is already configured."""
entry = MockConfigEntry(
domain=transmission.DOMAIN,
data=MOCK_ENTRY,
options={CONF_SCAN_INTERVAL: DEFAULT_SCAN_INTERVAL},
)
async def test_device_already_configured(
hass: HomeAssistant,
) -> None:
"""Test aborting if the device is already configured."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
mock_entry_unique_name = MOCK_ENTRY.copy()
mock_entry_unique_name[CONF_NAME] = "Transmission 1"
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=mock_entry_unique_name,
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
assert result["type"] == FlowResultType.FORM
mock_entry_unique_port = MOCK_ENTRY.copy()
mock_entry_unique_port[CONF_PORT] = 9092
mock_entry_unique_port[CONF_NAME] = "Transmission 2"
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=mock_entry_unique_port,
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG_DATA,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
await hass.async_block_till_done()
mock_entry_unique_host = MOCK_ENTRY.copy()
mock_entry_unique_host[CONF_HOST] = "192.168.1.101"
mock_entry_unique_host[CONF_NAME] = "Transmission 3"
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=mock_entry_unique_host,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "already_configured"
async def test_name_already_configured(hass, api):
async def test_name_already_configured(hass):
"""Test name is already configured."""
entry = MockConfigEntry(
domain=transmission.DOMAIN,
data=MOCK_ENTRY,
options={CONF_SCAN_INTERVAL: DEFAULT_SCAN_INTERVAL},
data=MOCK_CONFIG_DATA,
options={"scan_interval": 120},
)
entry.add_to_hass(hass)
mock_entry = MOCK_ENTRY.copy()
mock_entry[CONF_HOST] = "0.0.0.0"
mock_entry = MOCK_CONFIG_DATA.copy()
mock_entry["host"] = "1.1.1.1"
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=mock_entry,
)
assert result["type"] == "form"
assert result["errors"] == {CONF_NAME: "name_exists"}
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"name": "name_exists"}
async def test_error_on_wrong_credentials(hass, auth_error):
"""Test with wrong credentials."""
flow = init_config_flow(hass)
result = await flow.async_step_user(
{
CONF_NAME: NAME,
CONF_HOST: HOST,
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_PORT: PORT,
}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {
CONF_USERNAME: "invalid_auth",
CONF_PASSWORD: "invalid_auth",
}
async def test_error_on_connection_failure(hass, conn_error):
"""Test when connection to host fails."""
flow = init_config_flow(hass)
result = await flow.async_step_user(
{
CONF_NAME: NAME,
CONF_HOST: HOST,
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_PORT: PORT,
}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
async def test_error_on_unknown_error(hass, unknown_error):
"""Test when connection to host fails."""
flow = init_config_flow(hass)
result = await flow.async_step_user(
{
CONF_NAME: NAME,
CONF_HOST: HOST,
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_PORT: PORT,
}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
async def test_reauth_success(hass, api):
"""Test we can reauth."""
async def test_options(hass: HomeAssistant) -> None:
"""Test updating options."""
entry = MockConfigEntry(
domain=transmission.DOMAIN,
data=MOCK_ENTRY,
data=MOCK_CONFIG_DATA,
options={"scan_interval": 120},
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": entry.entry_id,
},
data=MOCK_ENTRY,
with patch(
"homeassistant.components.transmission.async_setup_entry",
return_value=True,
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={"scan_interval": 10}
)
assert result["type"] == "form"
assert result["step_id"] == "reauth_confirm"
assert result["description_placeholders"] == {CONF_USERNAME: USERNAME}
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["data"]["scan_interval"] == 10
async def test_error_on_wrong_credentials(
hass: HomeAssistant, mock_api: MagicMock
) -> None:
"""Test we handle invalid credentials."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_api.side_effect = TransmissionError("401: Unauthorized")
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: "test-password",
},
MOCK_CONFIG_DATA,
)
assert result2["type"] == "abort"
assert result2["reason"] == "reauth_successful"
async def test_reauth_failed(hass, auth_error):
"""Test we can reauth."""
entry = MockConfigEntry(
domain=transmission.DOMAIN,
data=MOCK_ENTRY,
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": entry.entry_id,
},
data=MOCK_ENTRY,
)
assert result["type"] == "form"
assert result["step_id"] == "reauth_confirm"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: "test-wrong-password",
},
)
assert result2["type"] == "form"
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {
CONF_PASSWORD: "invalid_auth",
"username": "invalid_auth",
"password": "invalid_auth",
}
async def test_reauth_failed_conn_error(hass, conn_error):
async def test_error_on_connection_failure(
hass: HomeAssistant, mock_api: MagicMock
) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_api.side_effect = TransmissionError("111: Connection refused")
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG_DATA,
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
async def test_reauth_success(hass: HomeAssistant) -> None:
"""Test we can reauth."""
entry = MockConfigEntry(
domain=transmission.DOMAIN,
data=MOCK_ENTRY,
data=MOCK_CONFIG_DATA,
)
entry.add_to_hass(hass)
@ -338,18 +169,92 @@ async def test_reauth_failed_conn_error(hass, conn_error):
"source": config_entries.SOURCE_REAUTH,
"entry_id": entry.entry_id,
},
data=MOCK_ENTRY,
data=MOCK_CONFIG_DATA,
)
assert result["type"] == "form"
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["description_placeholders"] == {"username": "user"}
with patch(
"homeassistant.components.transmission.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"password": "test-password",
},
)
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
assert len(mock_setup_entry.mock_calls) == 1
async def test_reauth_failed(hass: HomeAssistant, mock_api: MagicMock) -> None:
"""Test we can't reauth due to invalid password."""
entry = MockConfigEntry(
domain=transmission.DOMAIN,
data=MOCK_CONFIG_DATA,
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": entry.entry_id,
},
data=MOCK_CONFIG_DATA,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["description_placeholders"] == {"username": "user"}
mock_api.side_effect = TransmissionError("401: Unauthorized")
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: "test-wrong-password",
"password": "wrong-password",
},
)
assert result2["type"] == "form"
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"password": "invalid_auth"}
async def test_reauth_failed_connection_error(
hass: HomeAssistant, mock_api: MagicMock
) -> None:
"""Test we can't reauth due to connection error."""
entry = MockConfigEntry(
domain=transmission.DOMAIN,
data=MOCK_CONFIG_DATA,
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
transmission.DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": entry.entry_id,
},
data=MOCK_CONFIG_DATA,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["description_placeholders"] == {"username": "user"}
mock_api.side_effect = TransmissionError("111: Connection refused")
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"password": "test-password",
},
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}

146
tests/components/transmission/test_init.py
View file

@ -1,125 +1,75 @@
"""Tests for Transmission init."""
from unittest.mock import patch
from unittest.mock import MagicMock, patch
import pytest
from transmissionrpc.error import TransmissionError
from homeassistant.components import transmission
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.setup import async_setup_component
from homeassistant.components.transmission.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, mock_coro
from . import MOCK_CONFIG_DATA
MOCK_ENTRY = MockConfigEntry(
domain=transmission.DOMAIN,
data={
transmission.CONF_NAME: "Transmission",
transmission.CONF_HOST: "0.0.0.0",
transmission.CONF_USERNAME: "user",
transmission.CONF_PASSWORD: "pass",
transmission.CONF_PORT: 9091,
},
)
from tests.common import MockConfigEntry
@pytest.fixture(name="api")
def mock_transmission_api():
@pytest.fixture(autouse=True)
def mock_api():
"""Mock an api."""
with patch("transmissionrpc.Client"):
yield
with patch("transmissionrpc.Client") as api:
yield api
@pytest.fixture(name="auth_error")
def mock_api_authentication_error():
"""Mock an api."""
with patch(
"transmissionrpc.Client", side_effect=TransmissionError("401: Unauthorized")
):
yield
async def test_successful_config_entry(hass: HomeAssistant) -> None:
"""Test settings up integration from config entry."""
@pytest.fixture(name="unknown_error")
def mock_api_unknown_error():
"""Mock an api."""
with patch("transmissionrpc.Client", side_effect=TransmissionError):
yield
async def test_setup_with_no_config(hass):
"""Test that we do not discover anything or try to set up a Transmission client."""
assert await async_setup_component(hass, transmission.DOMAIN, {}) is True
assert transmission.DOMAIN not in hass.data
async def test_setup_with_config(hass, api):
"""Test that we import the config and setup the client."""
config = {
transmission.DOMAIN: {
transmission.CONF_NAME: "Transmission",
transmission.CONF_HOST: "0.0.0.0",
transmission.CONF_USERNAME: "user",
transmission.CONF_PASSWORD: "pass",
transmission.CONF_PORT: 9091,
},
transmission.DOMAIN: {
transmission.CONF_NAME: "Transmission2",
transmission.CONF_HOST: "0.0.0.1",
transmission.CONF_USERNAME: "user",
transmission.CONF_PASSWORD: "pass",
transmission.CONF_PORT: 9091,
},
}
assert await async_setup_component(hass, transmission.DOMAIN, config) is True
async def test_successful_config_entry(hass, api):
"""Test that configured transmission is configured successfully."""
entry = MOCK_ENTRY
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
assert await transmission.async_setup_entry(hass, entry) is True
assert entry.options == {
transmission.CONF_SCAN_INTERVAL: transmission.DEFAULT_SCAN_INTERVAL,
transmission.CONF_LIMIT: transmission.DEFAULT_LIMIT,
transmission.CONF_ORDER: transmission.DEFAULT_ORDER,
}
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state == ConfigEntryState.LOADED
async def test_setup_failed(hass):
"""Test transmission failed due to an error."""
async def test_setup_failed_connection_error(
hass: HomeAssistant, mock_api: MagicMock
) -> None:
"""Test integration failed due to connection error."""
entry = MOCK_ENTRY
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
# test connection error raising ConfigEntryNotReady
with patch(
"transmissionrpc.Client",
side_effect=TransmissionError("111: Connection refused"),
), pytest.raises(ConfigEntryNotReady):
mock_api.side_effect = TransmissionError("111: Connection refused")
await transmission.async_setup_entry(hass, entry)
# test Authentication error returning false
with patch(
"transmissionrpc.Client", side_effect=TransmissionError("401: Unauthorized")
), pytest.raises(ConfigEntryAuthFailed):
assert await transmission.async_setup_entry(hass, entry) is False
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state == ConfigEntryState.SETUP_RETRY
async def test_unload_entry(hass, api):
"""Test removing transmission client."""
entry = MOCK_ENTRY
async def test_setup_failed_auth_error(
hass: HomeAssistant, mock_api: MagicMock
) -> None:
"""Test integration failed due to invalid credentials error."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
with patch.object(
hass.config_entries, "async_forward_entry_unload", return_value=mock_coro(True)
) as unload_entry:
assert await transmission.async_setup_entry(hass, entry)
mock_api.side_effect = TransmissionError("401: Unauthorized")
assert await transmission.async_unload_entry(hass, entry)
assert unload_entry.call_count == 2
assert entry.entry_id not in hass.data[transmission.DOMAIN]
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state == ConfigEntryState.SETUP_ERROR
async def test_unload_entry(hass: HomeAssistant) -> None:
"""Test removing integration."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
assert not hass.data[DOMAIN]