ronantakizawa/codereview-bench huggingface dataset:
Repository: home-assistant/core
File: homeassistant/components/enocean/__init__.py (Python)

Code before:
"""Support for EnOcean devices."""

import voluptuous as vol

from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_DEVICE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType

from .const import DOMAIN
from .dongle import EnOceanDongle

type EnOceanConfigEntry = ConfigEntry[EnOceanConfigRuntimeData]


class EnOceanConfigRuntimeData:
    """Runtime data for EnOcean config entries."""

    def __init__(self, dongle: EnOceanDongle) -> None:
        """Initialize runtime data."""
        self._dongle = dongle

    @property
    def dongle(self) -> EnOceanDongle:
        """Return the EnOcean dongle."""
        return self._dongle


CONFIG_SCHEMA = vol.Schema(
    {DOMAIN: vol.Schema({vol.Required(CONF_DEVICE): cv.string})}, extra=vol.ALLOW_EXTRA
)


async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
    """Set up the EnOcean component."""
    # support for text-based configuration (legacy)
    if DOMAIN not in config:
        return True

Review comment (suggestion):
```suggestion
type EnOceanConfigEntry = ConfigEntry[EnOceanDongle]
```

Code after:
"""Support for EnOcean devices."""

import voluptuous as vol

from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_DEVICE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType

from .const import DOMAIN
from .dongle import EnOceanDongle

type EnOceanConfigEntry = ConfigEntry[EnOceanDongle]

CONFIG_SCHEMA = vol.Schema(
    {DOMAIN: vol.Schema({vol.Required(CONF_DEVICE): cv.string})}, extra=vol.ALLOW_EXTRA
)


async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
    """Set up the EnOcean component."""
    # support for text-based configuration (legacy)
    if DOMAIN not in config:
        return True

    if hass.config_entries.async_entries(DOMAIN):
        # We can only have one dongle. If there is already one in the config,
        # there is no need to import the yaml based config.
        return True

    hass.async_create_task(
        hass.config_entries.flow.async_init(
            DOMAIN, context={"source": SOURCE_IMPORT}, data=config[DOMAIN]
        )
    )

    return True
Repository: home-assistant/core
File: tests/components/teslemetry/test_init.py (Python)

Code before:
    freezer: FrozenDateTimeFactory,
    mock_live_status: AsyncMock,
    mock_energy_history: AsyncMock,
    mock_name: str,
    interval: timedelta,
    side_effect: type[TeslaFleetError],
    initial_return: dict | None,
) -> None:
    """Test coordinator handles errors during refresh."""
    mock = mock_live_status if mock_name == "mock_live_status" else mock_energy_history

    if initial_return is not None:
        # For coordinators that do first_refresh during setup, succeed first then fail
        call_count = 0

        def mock_side_effect(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                return deepcopy(initial_return)
            raise side_effect

        mock.side_effect = mock_side_effect
    else:
        # For energy_history which doesn't do first_refresh during setup
        mock.side_effect = side_effect

    entry = await setup_platform(hass)
    assert entry.state is ConfigEntryState.LOADED

    # Trigger coordinator refresh
    freezer.tick(interval)
    async_fire_time_changed(hass)
    await hass.async_block_till_done()

    # Entry stays loaded - errors during refresh don't break the entry
    assert entry.state is ConfigEntryState.LOADED


async def test_energy_history_invalid_data(
    hass: HomeAssistant,
    freezer: FrozenDateTimeFactory,
    mock_energy_history: AsyncMock,
) -> None:
    """Test invalid data from energy history raises UpdateFailed."""
    call_count = 0

    def energy_history_side_effect(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        if call_count == 1:

Review comment (suggestion):
We don't want if statements in the tests.

So with side_effect, if you pass a list of things, it will return that. So if I do `side_effect = [True, True, Exception]` it will return True twice before raising an exception

Code after:
    assert entry.state is ConfigEntryState.LOADED

    freezer.tick(ENERGY_LIVE_INTERVAL)
    async_fire_time_changed(hass)
    await hass.async_block_till_done()

    assert entry.state is ConfigEntryState.LOADED


@pytest.mark.parametrize(
    "side_effect",
    [
        [InvalidToken],
        [TeslaFleetError],
        [ENERGY_HISTORY, {"response": {}}],
    ],
)
async def test_energy_history_coordinator_refresh_errors(
    hass: HomeAssistant,
    freezer: FrozenDateTimeFactory,
    mock_energy_history: AsyncMock,
    side_effect: list,
) -> None:
    """Test energy history coordinator handles errors during refresh."""
    mock_energy_history.side_effect = side_effect

    entry = await setup_platform(hass)
    assert entry.state is ConfigEntryState.LOADED

    freezer.tick(ENERGY_HISTORY_INTERVAL)
    async_fire_time_changed(hass)
    await hass.async_block_till_done()

    assert entry.state is ConfigEntryState.LOADED
Repository: home-assistant/core
File: homeassistant/components/saunum/services.py (Python)

Code before:
    if coordinator.data.door_open:
        raise ServiceValidationError(
            translation_domain=DOMAIN,
            translation_key="door_open",
        )

    try:
        # Set all parameters before starting the session
        await coordinator.client.async_set_sauna_duration(duration)
        await coordinator.client.async_set_target_temperature(target_temperature)
        await coordinator.client.async_set_fan_duration(fan_duration)
        await coordinator.client.async_start_session()
    except SaunumException as err:
        raise HomeAssistantError(
            translation_domain=DOMAIN,
            translation_key="start_session_failed",
            translation_placeholders={"error": str(err)},
        ) from err

    await coordinator.async_request_refresh()


@callback
def async_setup_services(hass: HomeAssistant) -> None:
    """Register services for the Saunum integration."""
    hass.services.async_register(
        DOMAIN,
        SERVICE_START_SESSION,
        _async_start_session,
        schema=SERVICE_START_SESSION_SCHEMA,
    )


Review comment (suggestion):
I'm wondering if this wouldn't make more sense as an entity service, rather than a generic service.

Not compulsory, but might make more sense
See https://developers.home-assistant.io/docs/dev_101_services/#entity-service-actions
(and an example registering it in `services.py` https://github.com/home-assistant/core/pull/162146)

Code after:
"""Define services for the Saunum integration."""

from __future__ import annotations

from pysaunum import MAX_DURATION, MAX_FAN_DURATION, MAX_TEMPERATURE, MIN_TEMPERATURE
import voluptuous as vol

from homeassistant.components.climate import DOMAIN as CLIMATE_DOMAIN
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, service

from .const import DOMAIN

ATTR_DURATION = "duration"
ATTR_TARGET_TEMPERATURE = "target_temperature"
ATTR_FAN_DURATION = "fan_duration"

SERVICE_START_SESSION = "start_session"


@callback
def async_setup_services(hass: HomeAssistant) -> None:
    """Register services for the Saunum integration."""
    service.async_register_platform_entity_service(
        hass,
        DOMAIN,
        SERVICE_START_SESSION,
        entity_domain=CLIMATE_DOMAIN,
        schema={
            vol.Optional(ATTR_DURATION, default=120): vol.All(
                cv.positive_int, vol.Range(min=1, max=MAX_DURATION)
            ),
            vol.Optional(ATTR_TARGET_TEMPERATURE, default=80): vol.All(
                cv.positive_int, vol.Range(min=MIN_TEMPERATURE, max=MAX_TEMPERATURE)
            ),
            vol.Optional(ATTR_FAN_DURATION, default=10): vol.All(
                cv.positive_int, vol.Range(min=1, max=MAX_FAN_DURATION)
            ),
        },
        func="async_start_session",
    )
Repository: home-assistant/core
File: tests/components/syncthing/test_config_flow.py (Python)

Code before:

    with patch("aiosyncthing.system.System.status", return_value={"myID": "server-id"}):
        result = await hass.config_entries.flow.async_init(
            DOMAIN,
            context={"source": "user"},
            data=MOCK_ENTRY,
        )

    assert result["type"] is FlowResultType.ABORT
    assert result["reason"] == "already_configured"


async def test_flow_invalid_auth(hass: HomeAssistant) -> None:
    """Test invalid auth."""

    with patch("aiosyncthing.system.System.status", side_effect=UnauthorizedError):
        result = await hass.config_entries.flow.async_init(
            DOMAIN,
            context={"source": "user"},
            data=MOCK_ENTRY,
        )

        assert result["type"] is FlowResultType.FORM
        assert (
            result["errors"] is not None and result["errors"]["token"] == "invalid_auth"
        )


async def test_flow_cannot_connect(hass: HomeAssistant) -> None:
    """Test cannot connect."""

    with patch("aiosyncthing.system.System.status", side_effect=Exception):
        result = await hass.config_entries.flow.async_init(
            DOMAIN,
            context={"source": "user"},
            data=MOCK_ENTRY,
        )

        assert result["type"] is FlowResultType.FORM
        assert (
            result["errors"] is not None
            and result["errors"]["base"] == "cannot_connect"
        )


Review comment (security):
This isn't needed


```suggestion
        assert result["errors"]["token"] == "invalid_auth"
```

Code after:

    with patch("aiosyncthing.system.System.status", return_value={"myID": "server-id"}):
        result = await hass.config_entries.flow.async_init(
            DOMAIN,
            context={"source": "user"},
            data=MOCK_ENTRY,
        )

    assert result["type"] is FlowResultType.ABORT
    assert result["reason"] == "already_configured"


async def test_flow_invalid_auth(hass: HomeAssistant) -> None:
    """Test invalid auth."""

    with patch("aiosyncthing.system.System.status", side_effect=UnauthorizedError):
        result = await hass.config_entries.flow.async_init(
            DOMAIN,
            context={"source": "user"},
            data=MOCK_ENTRY,
        )

        assert result["type"] is FlowResultType.FORM
        assert result["errors"]["token"] == "invalid_auth"


async def test_flow_cannot_connect(hass: HomeAssistant) -> None:
    """Test cannot connect."""

    with patch("aiosyncthing.system.System.status", side_effect=Exception):
        result = await hass.config_entries.flow.async_init(
            DOMAIN,
            context={"source": "user"},
            data=MOCK_ENTRY,
        )

        assert result["type"] is FlowResultType.FORM
        assert result["errors"]["base"] == "cannot_connect"
Repository: home-assistant/core
File: homeassistant/util/package.py (Python)

Code before:
        close_fds=False,  # required for posix_spawn
    )
    stdout, _ = await process.communicate()
    return stdout.decode().strip()


async def async_get_installed_packages() -> list[InstalledPackage]:
    """Return a list of installed packages and versions.

    Returns a list of InstalledPackage dicts with 'name' and 'version' keys.
    """
    args = [sys.executable, "-m", "uv", "pip", "list", "--format", "json"]
    process = await asyncio.create_subprocess_exec(
        *args,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.DEVNULL,
        close_fds=False,  # required for posix_spawn
    )
    stdout, _ = await process.communicate()
    if process.returncode != 0:
        return []

    try:
        return cast(list[InstalledPackage], json_loads_array(stdout.decode()))
    except (*JSON_DECODE_EXCEPTIONS, ValueError):
        return []


Review comment (bug):
The formatter wants to make
```
try:
   return cast(list[InstalledPackage], json_loads_array(stdout.decode()))
except* JSON_DECODE_EXCEPTIONS, ValueError:
   return []
```

out of it, which seems to be wrong

Code after:
        close_fds=False,  # required for posix_spawn
    )
    stdout, _ = await process.communicate()
    return stdout.decode().strip()


async def async_get_installed_packages() -> list[InstalledPackage]:
    """Return a list of installed packages and versions.

    Returns a list of InstalledPackage dicts with 'name' and 'version' keys.
    """
    args = [sys.executable, "-m", "uv", "pip", "list", "--format", "json"]
    process = await asyncio.create_subprocess_exec(
        *args,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.DEVNULL,
        close_fds=False,  # required for posix_spawn
    )
    stdout, _ = await process.communicate()
    if process.returncode != 0:
        return []

    try:
        return cast(list[InstalledPackage], json_loads_array(stdout.decode()))
    except (*JSON_DECODE_EXCEPTIONS, ValueError):  # fmt: off
        return []
Repository: home-assistant/core
File: homeassistant/components/liebherr/diagnostics.py (Python)

Code before:
"""Diagnostics support for Liebherr."""

from __future__ import annotations

from dataclasses import asdict
from typing import Any

from homeassistant.components.diagnostics import async_redact_data
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant

from .coordinator import LiebherrConfigEntry

TO_REDACT = {CONF_API_KEY}


async def async_get_config_entry_diagnostics(
    hass: HomeAssistant, entry: LiebherrConfigEntry
) -> dict[str, Any]:
    """Return diagnostics for a config entry."""
    return {
        "entry_data": async_redact_data(entry.data, TO_REDACT),
        "devices": {
            device_id: {
                "coordinator": {
                    "last_update_success": coordinator.last_update_success,
                    "update_interval": coordinator.update_interval.total_seconds()
                    if coordinator.update_interval
                    else None,
                },
                "data": asdict(coordinator.data),
            }
            for device_id, coordinator in entry.runtime_data.items()
        },
    }


Review comment (suggestion):
```suggestion
                "last_update_success": coordinator.last_update_success,
```

Code after:
"""Diagnostics support for Liebherr."""

from __future__ import annotations

from dataclasses import asdict
from typing import Any

from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant

from .coordinator import LiebherrConfigEntry

TO_REDACT = {CONF_API_KEY}


async def async_get_config_entry_diagnostics(
    hass: HomeAssistant, entry: LiebherrConfigEntry
) -> dict[str, Any]:
    """Return diagnostics for a config entry."""
    return {
        "devices": {
            device_id: {
                "coordinator": {
                    "last_update_success": coordinator.last_update_success,
                    "update_interval": str(coordinator.update_interval),
                    "last_exception": str(coordinator.last_exception)
                    if coordinator.last_exception
                    else None,
                },
                "data": asdict(coordinator.data),
            }
            for device_id, coordinator in entry.runtime_data.items()
        },
    }
Repository: home-assistant/core
File: homeassistant/components/liebherr/diagnostics.py (Python)

Code before:
"""Diagnostics support for Liebherr."""

from __future__ import annotations

from dataclasses import asdict
from typing import Any

from homeassistant.components.diagnostics import async_redact_data
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant

from .coordinator import LiebherrConfigEntry

TO_REDACT = {CONF_API_KEY}


async def async_get_config_entry_diagnostics(
    hass: HomeAssistant, entry: LiebherrConfigEntry
) -> dict[str, Any]:
    """Return diagnostics for a config entry."""
    return {
        "entry_data": async_redact_data(entry.data, TO_REDACT),
        "devices": {
            device_id: {
                "coordinator": {
                    "last_update_success": coordinator.last_update_success,
                    "update_interval": coordinator.update_interval.total_seconds()
                    if coordinator.update_interval
                    else None,
                },
                "data": asdict(coordinator.data),
            }
            for device_id, coordinator in entry.runtime_data.items()
        },
    }


Review comment (bug):
Side note (please consider for a separate PR): is it correct that `coordinator._async_setup` raises `ConfigEntryError` and not `ConfigEntryAuthFailed` on `LiebherrAuthenticationError` ?

Code after:
"""Diagnostics support for Liebherr."""

from __future__ import annotations

from dataclasses import asdict
from typing import Any

from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant

from .coordinator import LiebherrConfigEntry

TO_REDACT = {CONF_API_KEY}


async def async_get_config_entry_diagnostics(
    hass: HomeAssistant, entry: LiebherrConfigEntry
) -> dict[str, Any]:
    """Return diagnostics for a config entry."""
    return {
        "devices": {
            device_id: {
                "coordinator": {
                    "last_update_success": coordinator.last_update_success,
                    "update_interval": str(coordinator.update_interval),
                    "last_exception": str(coordinator.last_exception)
                    if coordinator.last_exception
                    else None,
                },
                "data": asdict(coordinator.data),
            }
            for device_id, coordinator in entry.runtime_data.items()
        },
    }
Repository: home-assistant/core
File: tests/components/axis/test_config_flow.py (Python)

Code before:
    assert result["reason"] == "already_configured"
    assert config_entry_setup.data == {
        CONF_HOST: "2.3.4.5",
        CONF_PORT: expected_port,
        CONF_USERNAME: "root",
        CONF_PASSWORD: "pass",
        CONF_MODEL: MODEL,
        CONF_NAME: NAME,
    }


@pytest.mark.parametrize(
    ("source", "discovery_info"),
    [
        (
            SOURCE_DHCP,
            DhcpServiceInfo(
                hostname="",
                ip="",
                macaddress=dr.format_mac("01234567890").replace(":", ""),
            ),
        ),
        (
            SOURCE_SSDP,
            SsdpServiceInfo(
                ssdp_usn="mock_usn",
                ssdp_st="mock_st",
                upnp={
                    "friendlyName": "",
                    "serialNumber": "01234567890",
                    "presentationURL": "",
                },
            ),
        ),
        (
            SOURCE_ZEROCONF,
            ZeroconfServiceInfo(
                ip_address=None,
                ip_addresses=[],
                hostname="mock_hostname",
                name="",
                port=0,
                properties={"macaddress": "01234567890"},
                type="mock_type",
            ),
        ),
    ],
)
async def test_discovery_flow_ignore_non_axis_device(
    hass: HomeAssistant, source: str, discovery_info: BaseServiceInfo
) -> None:

Review comment (suggestion):
Since you are not mocking 2.3.4.5 like you are in the other tests, it might be better to remove it. If not, perhaps change to localhost.

```suggestion
                "presentationURL": "",
```

Code after:
    assert result["reason"] == "already_configured"
    assert config_entry_setup.data == {
        CONF_HOST: "2.3.4.5",
        CONF_PORT: expected_port,
        CONF_USERNAME: "root",
        CONF_PASSWORD: "pass",
        CONF_MODEL: MODEL,
        CONF_NAME: NAME,
    }


@pytest.mark.parametrize(
    "axis_oui",
    ["00408C", "ACCC8E", "B8A44F", "E82725"],
)
async def test_discovery_flow_allowed_oui(hass: HomeAssistant, axis_oui: str) -> None:
    """Test that discovery flow accepts Axis OUI."""
    result = await hass.config_entries.flow.async_init(
        DOMAIN,
        data=SsdpServiceInfo(
            ssdp_usn="mock_usn",
            ssdp_st="mock_st",
            upnp={
                "friendlyName": f"AXIS M1065-LW - {axis_oui}",
                "serialNumber": f"{axis_oui}123456",
                "presentationURL": "http://2.3.4.5:8080/",
            },
        ),
        context={"source": SOURCE_SSDP},
    )

    assert result["type"] is FlowResultType.FORM
    assert result["step_id"] == "user"


@pytest.mark.parametrize(
    ("source", "discovery_info"),
    [
        (
            SOURCE_DHCP,
            DhcpServiceInfo(
                hostname="",
                ip="",
                macaddress=dr.format_mac("01234567890").replace(":", ""),
            ),
        ),
        (
            SOURCE_SSDP,
            SsdpServiceInfo(
                ssdp_usn="mock_usn",
                ssdp_st="mock_st",
Repository: home-assistant/core
File: homeassistant/components/onkyo/__init__.py (Python)

Code before:
"""The onkyo component."""

from dataclasses import dataclass
import logging

from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType

from .const import (
    DOMAIN,
    OPTION_INPUT_SOURCES,
    OPTION_LISTENING_MODES,
    InputSource,
    ListeningMode,
)
from .receiver import ReceiverManager, async_interview
from .services import DATA_MP_ENTITIES, async_setup_services

_LOGGER = logging.getLogger(__name__)

PLATFORMS = [Platform.MEDIA_PLAYER]

CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)


@dataclass
class OnkyoData:
    """Config Entry data."""

    manager: ReceiverManager
    sources: dict[InputSource, str]
    sound_modes: dict[ListeningMode, str]


type OnkyoConfigEntry = ConfigEntry[OnkyoData]


async def async_setup(hass: HomeAssistant, _: ConfigType) -> bool:
    """Set up Onkyo component."""
    async_setup_services(hass)
    return True


Review comment (suggestion):
I agree that it shouldn't be in services.py, but would it not make sense to move it `const.py`?
I think that is the most frequent pattern when the hasskey is shared between modules.

Also, I think it feels weird to have `hass.data.setdefault(DATA_MP_ENTITIES, {})` inside `services.py`
`hass.data` keyrs is usually initialised in `__init__.py`

Code after:
"""The onkyo component."""

from dataclasses import dataclass
import logging

from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType

from .const import (
    DOMAIN,
    OPTION_INPUT_SOURCES,
    OPTION_LISTENING_MODES,
    InputSource,
    ListeningMode,
)
from .receiver import ReceiverManager, async_interview
from .services import async_setup_services

_LOGGER = logging.getLogger(__name__)

PLATFORMS = [Platform.MEDIA_PLAYER]

CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)


@dataclass
class OnkyoData:
    """Config Entry data."""

    manager: ReceiverManager
    sources: dict[InputSource, str]
    sound_modes: dict[ListeningMode, str]


type OnkyoConfigEntry = ConfigEntry[OnkyoData]


async def async_setup(hass: HomeAssistant, _: ConfigType) -> bool:
    """Set up Onkyo component."""
    async_setup_services(hass)
    return True