Add tests to Transmission (#157355)

This commit is contained in:
Andrew Jackson
2025-11-27 11:15:10 +00:00
committed by GitHub
parent 7fe26223ac
commit 3a65d3c0dc
10 changed files with 1159 additions and 91 deletions

View File

@@ -32,10 +32,7 @@ rules:
log-when-unavailable: done
parallel-updates: todo
reauthentication-flow: done
test-coverage:
status: todo
comment: |
Change to mock_setup_entry to avoid repetition when expanding tests.
test-coverage: done
# Gold
devices:

View File

@@ -1,5 +1,18 @@
"""Tests for Transmission."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Fixture for setting up the component."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
OLD_MOCK_CONFIG_DATA = {
"name": "Transmission",
"host": "0.0.0.0",

View File

@@ -0,0 +1,101 @@
"""Transmission tests configuration."""
from collections.abc import Generator
from datetime import UTC, datetime
from unittest.mock import AsyncMock, patch
import pytest
from transmission_rpc.session import Session, SessionStats
from transmission_rpc.torrent import Torrent
from homeassistant.components.transmission.const import DOMAIN
from . import MOCK_CONFIG_DATA
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.transmission.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock a config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="Transmission",
data=MOCK_CONFIG_DATA,
entry_id="01J0BC4QM2YBRP6H5G933AETT7",
)
@pytest.fixture
def mock_transmission_client() -> Generator[AsyncMock]:
"""Mock a Transmission client."""
with (
patch(
"homeassistant.components.transmission.transmission_rpc.Client",
autospec=False,
) as mock_client_class,
):
client = mock_client_class.return_value
session_stats_data = {
"uploadSpeed": 1,
"downloadSpeed": 1,
"activeTorrentCount": 0,
"pausedTorrentCount": 0,
"torrentCount": 0,
}
client.session_stats.return_value = SessionStats(fields=session_stats_data)
session_data = {"alt-speed-enabled": False}
client.get_session.return_value = Session(fields=session_data)
client.get_torrents.return_value = []
yield mock_client_class
@pytest.fixture
def mock_torrent():
"""Fixture that returns a factory function to create mock torrents."""
def _create_mock_torrent(
torrent_id: int = 1,
name: str = "Test Torrent",
percent_done: float = 0.5,
status: int = 4,
download_dir: str = "/downloads",
eta: int = 3600,
added_date: datetime | None = None,
ratio: float = 1.5,
) -> Torrent:
"""Create a mock torrent with all required attributes."""
if added_date is None:
added_date = datetime(2025, 11, 26, 14, 18, 0, tzinfo=UTC)
torrent_data = {
"id": torrent_id,
"name": name,
"percentDone": percent_done,
"status": status,
"rateDownload": 0,
"rateUpload": 0,
"downloadDir": download_dir,
"eta": eta,
"addedDate": int(added_date.timestamp()),
"uploadRatio": ratio,
"error": 0,
"errorString": "",
}
return Torrent(fields=torrent_data)
return _create_mock_torrent

View File

@@ -0,0 +1,430 @@
# serializer version: 1
# name: test_sensors[sensor.transmission_active_torrents-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.transmission_active_torrents',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Active torrents',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'active_torrents',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-active_torrents',
'unit_of_measurement': 'torrents',
})
# ---
# name: test_sensors[sensor.transmission_active_torrents-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Transmission Active torrents',
'torrent_info': dict({
}),
'unit_of_measurement': 'torrents',
}),
'context': <ANY>,
'entity_id': 'sensor.transmission_active_torrents',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[sensor.transmission_completed_torrents-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.transmission_completed_torrents',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Completed torrents',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'completed_torrents',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-completed_torrents',
'unit_of_measurement': 'torrents',
})
# ---
# name: test_sensors[sensor.transmission_completed_torrents-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Transmission Completed torrents',
'torrent_info': dict({
}),
'unit_of_measurement': 'torrents',
}),
'context': <ANY>,
'entity_id': 'sensor.transmission_completed_torrents',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[sensor.transmission_download_speed-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.transmission_download_speed',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfDataRate.MEGABYTES_PER_SECOND: 'MB/s'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_RATE: 'data_rate'>,
'original_icon': None,
'original_name': 'Download speed',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'download_speed',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-download',
'unit_of_measurement': <UnitOfDataRate.MEGABYTES_PER_SECOND: 'MB/s'>,
})
# ---
# name: test_sensors[sensor.transmission_download_speed-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_rate',
'friendly_name': 'Transmission Download speed',
'unit_of_measurement': <UnitOfDataRate.MEGABYTES_PER_SECOND: 'MB/s'>,
}),
'context': <ANY>,
'entity_id': 'sensor.transmission_download_speed',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '1e-06',
})
# ---
# name: test_sensors[sensor.transmission_paused_torrents-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.transmission_paused_torrents',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Paused torrents',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'paused_torrents',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-paused_torrents',
'unit_of_measurement': 'torrents',
})
# ---
# name: test_sensors[sensor.transmission_paused_torrents-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Transmission Paused torrents',
'torrent_info': dict({
}),
'unit_of_measurement': 'torrents',
}),
'context': <ANY>,
'entity_id': 'sensor.transmission_paused_torrents',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[sensor.transmission_started_torrents-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.transmission_started_torrents',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Started torrents',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'started_torrents',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-started_torrents',
'unit_of_measurement': 'torrents',
})
# ---
# name: test_sensors[sensor.transmission_started_torrents-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Transmission Started torrents',
'torrent_info': dict({
}),
'unit_of_measurement': 'torrents',
}),
'context': <ANY>,
'entity_id': 'sensor.transmission_started_torrents',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[sensor.transmission_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'idle',
'up_down',
'seeding',
'downloading',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.transmission_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Status',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'transmission_status',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-status',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.transmission_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Transmission Status',
'options': list([
'idle',
'up_down',
'seeding',
'downloading',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.transmission_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'up_down',
})
# ---
# name: test_sensors[sensor.transmission_total_torrents-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.transmission_total_torrents',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Total torrents',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'total_torrents',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-total_torrents',
'unit_of_measurement': 'torrents',
})
# ---
# name: test_sensors[sensor.transmission_total_torrents-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Transmission Total torrents',
'torrent_info': dict({
}),
'unit_of_measurement': 'torrents',
}),
'context': <ANY>,
'entity_id': 'sensor.transmission_total_torrents',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[sensor.transmission_upload_speed-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.transmission_upload_speed',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfDataRate.MEGABYTES_PER_SECOND: 'MB/s'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_RATE: 'data_rate'>,
'original_icon': None,
'original_name': 'Upload speed',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'upload_speed',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-upload',
'unit_of_measurement': <UnitOfDataRate.MEGABYTES_PER_SECOND: 'MB/s'>,
})
# ---
# name: test_sensors[sensor.transmission_upload_speed-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_rate',
'friendly_name': 'Transmission Upload speed',
'unit_of_measurement': <UnitOfDataRate.MEGABYTES_PER_SECOND: 'MB/s'>,
}),
'context': <ANY>,
'entity_id': 'sensor.transmission_upload_speed',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '1e-06',
})
# ---

View File

@@ -0,0 +1,97 @@
# serializer version: 1
# name: test_switches[switch.transmission_switch-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.transmission_switch',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Switch',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'on_off',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-on_off',
'unit_of_measurement': None,
})
# ---
# name: test_switches[switch.transmission_switch-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Transmission Switch',
}),
'context': <ANY>,
'entity_id': 'switch.transmission_switch',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_switches[switch.transmission_turtle_mode-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.transmission_turtle_mode',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Turtle mode',
'platform': 'transmission',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'turtle_mode',
'unique_id': '01J0BC4QM2YBRP6H5G933AETT7-turtle_mode',
'unit_of_measurement': None,
})
# ---
# name: test_switches[switch.transmission_turtle_mode-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Transmission Turtle mode',
}),
'context': <ANY>,
'entity_id': 'switch.transmission_turtle_mode',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---

View File

@@ -1,6 +1,6 @@
"""Tests for Transmission config flow."""
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, patch
import pytest
from transmission_rpc.error import (
@@ -15,34 +15,26 @@ 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 . import MOCK_CONFIG_DATA, setup_integration
from tests.common import MockConfigEntry
@pytest.fixture(autouse=True)
def mock_api():
"""Mock an api."""
with patch("transmission_rpc.Client") as api:
yield api
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
async def test_full_flow(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test full flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
with patch(
"homeassistant.components.transmission.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG_DATA,
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG_DATA,
)
assert len(mock_setup_entry.mock_calls) == 1
assert result["title"] == "Transmission"
@@ -52,10 +44,10 @@ async def test_form(hass: HomeAssistant) -> None:
async def test_device_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test aborting if the device is already configured."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@@ -72,7 +64,10 @@ async def test_device_already_configured(
assert result["type"] is FlowResultType.ABORT
async def test_options(hass: HomeAssistant) -> None:
async def test_options(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test updating options."""
entry = MockConfigEntry(
domain=transmission.DOMAIN,
@@ -103,14 +98,15 @@ async def test_options(hass: HomeAssistant) -> None:
async def test_error_on_wrong_credentials(
hass: HomeAssistant, mock_api: MagicMock
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
) -> 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 = TransmissionAuthError()
mock_transmission_client.side_effect = TransmissionAuthError()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG_DATA,
@@ -121,7 +117,7 @@ async def test_error_on_wrong_credentials(
"password": "invalid_auth",
}
mock_api.side_effect = None
mock_transmission_client.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG_DATA,
@@ -133,12 +129,13 @@ async def test_error_on_wrong_credentials(
("exception", "error"),
[
(TransmissionError, "cannot_connect"),
(TransmissionConnectError, "invalid_auth"),
(TransmissionConnectError, "cannot_connect"),
],
)
async def test_flow_errors(
hass: HomeAssistant,
mock_api: MagicMock,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
exception: Exception,
error: str,
) -> None:
@@ -147,15 +144,15 @@ async def test_flow_errors(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_api.side_effect = exception
mock_transmission_client.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG_DATA,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
assert result["errors"] == {"base": error}
mock_api.side_effect = None
mock_transmission_client.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCK_CONFIG_DATA,
@@ -163,18 +160,21 @@ async def test_flow_errors(
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_reauth_success(hass: HomeAssistant) -> None:
async def test_reauth_success(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test we can reauth."""
entry = MockConfigEntry(domain=transmission.DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
await setup_integration(hass, mock_config_entry)
result = await entry.start_reauth_flow(hass)
result = await mock_config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["description_placeholders"] == {
"username": "user",
"name": "Mock Title",
"name": "Transmission",
}
with patch(
@@ -203,7 +203,8 @@ async def test_reauth_success(hass: HomeAssistant) -> None:
)
async def test_reauth_flow_errors(
hass: HomeAssistant,
mock_api: MagicMock,
mock_config_entry: MockConfigEntry,
mock_transmission_client: AsyncMock,
exception: Exception,
field: str,
error: str,
@@ -224,7 +225,7 @@ async def test_reauth_flow_errors(
"name": "Mock Title",
}
mock_api.side_effect = exception
mock_transmission_client.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
@@ -235,7 +236,7 @@ async def test_reauth_flow_errors(
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {field: error}
mock_api.side_effect = None
mock_transmission_client.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{

View File

@@ -1,7 +1,8 @@
"""Tests for Transmission init."""
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock
from freezegun.api import FrozenDateTimeFactory
import pytest
from transmission_rpc.error import (
TransmissionAuthError,
@@ -13,6 +14,7 @@ from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.components.transmission.const import (
DEFAULT_PATH,
DEFAULT_SCAN_INTERVAL,
DEFAULT_SSL,
DOMAIN,
)
@@ -21,30 +23,14 @@ from homeassistant.const import CONF_PATH, CONF_SSL
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import MOCK_CONFIG_DATA, MOCK_CONFIG_DATA_VERSION_1_1, OLD_MOCK_CONFIG_DATA
from . import MOCK_CONFIG_DATA_VERSION_1_1, OLD_MOCK_CONFIG_DATA
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.fixture(autouse=True)
def mock_api():
"""Mock an api."""
with patch("transmission_rpc.Client") as api:
yield api
async def test_successful_config_entry(hass: HomeAssistant) -> None:
"""Test settings up integration from config entry."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.LOADED
async def test_config_flow_entry_migrate_1_1_to_1_2(hass: HomeAssistant) -> None:
async def test_config_flow_entry_migrate_1_1_to_1_2(
hass: HomeAssistant,
) -> None:
"""Test that config flow entry is migrated correctly from v1.1 to v1.2."""
entry = MockConfigEntry(
domain=DOMAIN,
@@ -66,59 +52,65 @@ async def test_config_flow_entry_migrate_1_1_to_1_2(hass: HomeAssistant) -> None
async def test_setup_failed_connection_error(
hass: HomeAssistant, mock_api: MagicMock
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test integration failed due to connection error."""
mock_config_entry.add_to_hass(hass)
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
mock_transmission_client.side_effect = TransmissionConnectError()
mock_api.side_effect = TransmissionConnectError()
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.SETUP_RETRY
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_setup_failed_auth_error(
hass: HomeAssistant, mock_api: MagicMock
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test integration failed due to invalid credentials error."""
mock_config_entry.add_to_hass(hass)
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
mock_transmission_client.side_effect = TransmissionAuthError()
mock_api.side_effect = TransmissionAuthError()
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.SETUP_ERROR
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
async def test_setup_failed_unexpected_error(
hass: HomeAssistant, mock_api: MagicMock
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test integration failed due to unexpected error."""
mock_config_entry.add_to_hass(hass)
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
mock_transmission_client.side_effect = TransmissionError()
mock_api.side_effect = TransmissionError()
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.SETUP_ERROR
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
async def test_unload_entry(hass: HomeAssistant) -> None:
async def test_unload_entry(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test removing integration."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG_DATA)
entry.add_to_hass(hass)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_unload(entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
@@ -184,3 +176,28 @@ async def test_migrate_unique_id(
assert migrated_entity
assert migrated_entity.unique_id == new_unique_id
async def test_coordinator_update_error(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the sensors go unavailable."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
# Make the coordinator fail on next update
client = mock_transmission_client.return_value
client.session_stats.side_effect = TransmissionError("Connection failed")
# Trigger an update to make entities unavailable
freezer.tick(DEFAULT_SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
# Verify entities are unavailable
state = hass.states.get("sensor.transmission_status")
assert state is not None
assert state.state == "unavailable"

View File

@@ -0,0 +1,27 @@
"""Tests for the Transmission sensor platform."""
from unittest.mock import AsyncMock, patch
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
async def test_sensors(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the sensor entities."""
with patch("homeassistant.components.transmission.PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)

View File

@@ -0,0 +1,254 @@
"""Tests for the Transmission services."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.transmission.const import (
ATTR_DELETE_DATA,
ATTR_DOWNLOAD_PATH,
ATTR_TORRENT,
CONF_ENTRY_ID,
DOMAIN,
SERVICE_ADD_TORRENT,
SERVICE_REMOVE_TORRENT,
SERVICE_START_TORRENT,
SERVICE_STOP_TORRENT,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from tests.common import MockConfigEntry
async def test_service_config_entry_not_loaded_state(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test service call when config entry is in failed state."""
mock_config_entry.add_to_hass(hass)
assert mock_config_entry.state == ConfigEntryState.NOT_LOADED
with pytest.raises(ServiceValidationError, match="service_not_found"):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_TORRENT,
{
CONF_ENTRY_ID: mock_config_entry.entry_id,
ATTR_TORRENT: "magnet:?xt=urn:btih:test",
},
blocking=True,
)
async def test_service_integration_not_found(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test service call with non-existent config entry."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(
ServiceValidationError, match='Integration "transmission" not found'
):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_TORRENT,
{
CONF_ENTRY_ID: "non_existent_entry_id",
ATTR_TORRENT: "magnet:?xt=urn:btih:test",
},
blocking=True,
)
@pytest.mark.parametrize(
("payload", "expected_torrent", "kwargs"),
[
(
{ATTR_TORRENT: "magnet:?xt=urn:btih:test"},
"magnet:?xt=urn:btih:test",
{},
),
(
{
ATTR_TORRENT: "magnet:?xt=urn:btih:test",
ATTR_DOWNLOAD_PATH: "/custom/path",
},
"magnet:?xt=urn:btih:test",
{"download_dir": "/custom/path"},
),
(
{ATTR_TORRENT: "http://example.com/test.torrent"},
"http://example.com/test.torrent",
{},
),
(
{ATTR_TORRENT: "ftp://example.com/test.torrent"},
"ftp://example.com/test.torrent",
{},
),
(
{ATTR_TORRENT: "/config/test.torrent"},
"/config/test.torrent",
{},
),
],
)
async def test_add_torrent_service_success(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
payload: dict[str, str],
expected_torrent: str,
kwargs: dict[str, str | None],
) -> None:
"""Test successful torrent addition with url and path sources."""
client = mock_transmission_client.return_value
client.add_torrent.return_value = MagicMock(id=123, name="test_torrent")
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
full_service_data = {CONF_ENTRY_ID: mock_config_entry.entry_id} | payload
with patch.object(hass.config, "is_allowed_path", return_value=True):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_TORRENT,
full_service_data,
blocking=True,
)
client.add_torrent.assert_called_once_with(expected_torrent, **kwargs)
async def test_add_torrent_service_invalid_path(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test torrent addition with invalid path."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(ServiceValidationError, match="Could not add torrent"):
await hass.services.async_call(
DOMAIN,
SERVICE_ADD_TORRENT,
{
CONF_ENTRY_ID: mock_config_entry.entry_id,
ATTR_TORRENT: "/etc/bad.torrent",
},
blocking=True,
)
async def test_start_torrent_service_success(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test successful torrent start."""
client = mock_transmission_client.return_value
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN,
SERVICE_START_TORRENT,
{
CONF_ENTRY_ID: mock_config_entry.entry_id,
CONF_ID: 123,
},
blocking=True,
)
client.start_torrent.assert_called_once_with(123)
async def test_stop_torrent_service_success(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test successful torrent stop."""
client = mock_transmission_client.return_value
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN,
SERVICE_STOP_TORRENT,
{
CONF_ENTRY_ID: mock_config_entry.entry_id,
CONF_ID: 456,
},
blocking=True,
)
client.stop_torrent.assert_called_once_with(456)
async def test_remove_torrent_service_success(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test successful torrent removal without deleting data."""
client = mock_transmission_client.return_value
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN,
SERVICE_REMOVE_TORRENT,
{
CONF_ENTRY_ID: mock_config_entry.entry_id,
CONF_ID: 789,
},
blocking=True,
)
client.remove_torrent.assert_called_once_with(789, delete_data=False)
async def test_remove_torrent_service_with_delete_data(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test successful torrent removal with deleting data."""
client = mock_transmission_client.return_value
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
DOMAIN,
SERVICE_REMOVE_TORRENT,
{
CONF_ENTRY_ID: mock_config_entry.entry_id,
CONF_ID: 789,
ATTR_DELETE_DATA: True,
},
blocking=True,
)
client.remove_torrent.assert_called_once_with(789, delete_data=True)

View File

@@ -0,0 +1,131 @@
"""Tests for the Transmission switch platform."""
from unittest.mock import AsyncMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
async def test_switches(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the switch entities."""
with patch("homeassistant.components.transmission.PLATFORMS", [Platform.SWITCH]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("service", "api_method"),
[
(SERVICE_TURN_ON, "start_all"),
(SERVICE_TURN_OFF, "stop_torrent"),
],
)
async def test_on_off_switch_without_torrents(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_torrent,
service: str,
api_method: str,
) -> None:
"""Test on/off switch."""
client = mock_transmission_client.return_value
client.get_torrents.return_value = []
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: "switch.transmission_switch"},
blocking=True,
)
getattr(client, api_method).assert_not_called()
@pytest.mark.parametrize(
("service", "api_method"),
[
(SERVICE_TURN_ON, "start_all"),
(SERVICE_TURN_OFF, "stop_torrent"),
],
)
async def test_on_off_switch_with_torrents(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_torrent,
service: str,
api_method: str,
) -> None:
"""Test on/off switch."""
client = mock_transmission_client.return_value
client.get_torrents.return_value = [mock_torrent()]
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: "switch.transmission_switch"},
blocking=True,
)
getattr(client, api_method).assert_called_once()
@pytest.mark.parametrize(
("service", "alt_speed_enabled"),
[
(SERVICE_TURN_ON, True),
(SERVICE_TURN_OFF, False),
],
)
async def test_turtle_mode_switch(
hass: HomeAssistant,
mock_transmission_client: AsyncMock,
mock_config_entry: MockConfigEntry,
service: str,
alt_speed_enabled: bool,
) -> None:
"""Test turtle mode switch."""
client = mock_transmission_client.return_value
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: "switch.transmission_turtle_mode"},
blocking=True,
)
client.set_session.assert_called_once_with(alt_speed_enabled=alt_speed_enabled)