ronantakizawa/codereview-bench huggingface dataset:
Repository: home-assistant/core
File: tests/components/remote_calendar/test_config_flow.py (Python)

Code before:
            CONF_VERIFY_SSL: True,
        },
    )

    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"base": "invalid_ics_file"}
    respx.get(CALENDER_URL).mock(
        return_value=Response(
            status_code=200,
            text=ics_content,
        )
    )
    result3 = await hass.config_entries.flow.async_configure(
        result["flow_id"],
        {
            CONF_CALENDAR_NAME: CALENDAR_NAME,
            CONF_URL: CALENDER_URL,
            CONF_VERIFY_SSL: True,
        },
    )
    assert result3["type"] is FlowResultType.CREATE_ENTRY
    assert result3["title"] == CALENDAR_NAME
    assert result3["data"] == {
        CONF_CALENDAR_NAME: CALENDAR_NAME,
        CONF_URL: CALENDER_URL,
        CONF_VERIFY_SSL: True,
    }


async def test_duplicate_name(
    hass: HomeAssistant,
    config_entry: MockConfigEntry,
) -> None:
    """Test two calendars cannot be added with the same name."""

    await setup_integration(hass, config_entry)
    result = await hass.config_entries.flow.async_init(
        DOMAIN, context={"source": SOURCE_USER}
    )
    assert result["type"] is FlowResultType.FORM
    assert not result.get("errors")

    result2 = await hass.config_entries.flow.async_configure(
        result["flow_id"],
        {
            CONF_CALENDAR_NAME: CALENDAR_NAME,
            CONF_URL: "http://other-calendar.com",
            CONF_VERIFY_SSL: True,
        },
    )
    await hass.async_block_till_done()

Review comment (suggestion):
```suggestion
        CONF_VERIFY_SSL: False,
```

Code after:
            CONF_VERIFY_SSL: True,
        },
    )

    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"base": "invalid_ics_file"}
    respx.get(CALENDER_URL).mock(
        return_value=Response(
            status_code=200,
            text=ics_content,
        )
    )
    result3 = await hass.config_entries.flow.async_configure(
        result["flow_id"],
        {
            CONF_CALENDAR_NAME: CALENDAR_NAME,
            CONF_URL: CALENDER_URL,
            CONF_VERIFY_SSL: False,
        },
    )
    assert result3["type"] is FlowResultType.CREATE_ENTRY
    assert result3["title"] == CALENDAR_NAME
    assert result3["data"] == {
        CONF_CALENDAR_NAME: CALENDAR_NAME,
        CONF_URL: CALENDER_URL,
        CONF_VERIFY_SSL: False,
    }


async def test_duplicate_name(
    hass: HomeAssistant,
    config_entry: MockConfigEntry,
) -> None:
    """Test two calendars cannot be added with the same name."""

    await setup_integration(hass, config_entry)
    result = await hass.config_entries.flow.async_init(
        DOMAIN, context={"source": SOURCE_USER}
    )
    assert result["type"] is FlowResultType.FORM
    assert not result.get("errors")

    result2 = await hass.config_entries.flow.async_configure(
        result["flow_id"],
        {
            CONF_CALENDAR_NAME: CALENDAR_NAME,
            CONF_URL: "http://other-calendar.com",
            CONF_VERIFY_SSL: True,
        },
    )
    await hass.async_block_till_done()
Repository: home-assistant/core
File: homeassistant/components/tuya/type_information.py (Python)

Code before:


@dataclass(kw_only=True)
class TypeInformation[T]:
    """Type information.

    As provided by the SDK, from `device.function` / `device.status_range`.
    """

    _DPTYPE: ClassVar[DPType]
    dpcode: str
    type_data: str

    def process_raw_value(
        self, raw_value: Any | None, device: CustomerDevice
    ) -> T | None:
        """Read and process raw value against this type information.

        Base implementation does no validation, subclasses may override to provide
        specific validation.
        """
        return raw_value

    @classmethod
    def _from_json(
        cls, dpcode: str, type_data: str, *, report_type: str | None = None
    ) -> Self | None:
        """Load JSON string and return a TypeInformation object."""
        return cls(dpcode=dpcode, type_data=type_data)

    @classmethod
    def find_dpcode(
        cls,
        device: CustomerDevice,
        dpcodes: str | tuple[str, ...] | None,
        *,
        prefer_function: bool = False,
    ) -> Self | None:
        """Find type information for a matching DP code available for this device."""
        if dpcodes is None:
            return None

        if not isinstance(dpcodes, tuple):
            dpcodes = (dpcodes,)

        lookup_tuple = (
            (device.function, device.status_range)
            if prefer_function
            else (device.status_range, device.function)
        )


Review comment (suggestion):
```suggestion
        cls, dpcode: str, type_data: str, *, report_type: str | None
```

Code after:


@dataclass(kw_only=True)
class TypeInformation[T]:
    """Type information.

    As provided by the SDK, from `device.function` / `device.status_range`.
    """

    _DPTYPE: ClassVar[DPType]
    dpcode: str
    type_data: str

    def process_raw_value(
        self, raw_value: Any | None, device: CustomerDevice
    ) -> T | None:
        """Read and process raw value against this type information.

        Base implementation does no validation, subclasses may override to provide
        specific validation.
        """
        return raw_value

    @classmethod
    def _from_json(
        cls, dpcode: str, type_data: str, *, report_type: str | None
    ) -> Self | None:
        """Load JSON string and return a TypeInformation object."""
        return cls(dpcode=dpcode, type_data=type_data)

    @classmethod
    def find_dpcode(
        cls,
        device: CustomerDevice,
        dpcodes: str | tuple[str, ...] | None,
        *,
        prefer_function: bool = False,
    ) -> Self | None:
        """Find type information for a matching DP code available for this device."""
        if dpcodes is None:
            return None

        if not isinstance(dpcodes, tuple):
            dpcodes = (dpcodes,)

        lookup_tuple = (
            (device.function, device.status_range)
            if prefer_function
            else (device.status_range, device.function)
        )
Repository: home-assistant/core
File: homeassistant/components/tuya/models.py (Python)

Code before:

class DeviceWrapper[T]:
    """Base device wrapper."""

    native_unit: str | None = None
    suggested_unit: str | None = None
    state_class: SensorStateClass | None = None

    max_value: float
    min_value: float
    value_step: float

    options: list[str]

    def initialize(self, device: CustomerDevice) -> None:
        """Initialize the wrapper with device data.

        Called when the entity is added to Home Assistant.
        Override in subclasses to perform initialization logic.
        """

    def skip_update(
        self,
        device: CustomerDevice,
        updated_status_properties: list[str] | None,
        dp_timestamps: dict | None = None,
    ) -> bool:
        """Determine if the wrapper should skip an update.

        The default is to always skip, unless overridden in subclasses.
        """
        return True

    def read_device_status(self, device: CustomerDevice) -> T | None:
        """Read device status and convert to a Home Assistant value."""
        raise NotImplementedError

    def get_update_commands(
        self, device: CustomerDevice, value: T
    ) -> list[dict[str, Any]]:
        """Generate update commands for a Home Assistant action."""
        raise NotImplementedError


class DPCodeWrapper(DeviceWrapper):
    """Base device wrapper for a single DPCode.

    Used as a common interface for referring to a DPCode, and
    access read conversion routines.
    """


Review comment (suggestion):
You do not need a default value, and it would be better to improve the type hints.
```suggestion
        dp_timestamps: dict[str, int] | None,
```

Code after:

class DeviceWrapper[T]:
    """Base device wrapper."""

    native_unit: str | None = None
    suggested_unit: str | None = None
    state_class: SensorStateClass | None = None

    max_value: float
    min_value: float
    value_step: float

    options: list[str]

    def initialize(self, device: CustomerDevice) -> None:
        """Initialize the wrapper with device data.

        Called when the entity is added to Home Assistant.
        Override in subclasses to perform initialization logic.
        """

    def skip_update(
        self,
        device: CustomerDevice,
        updated_status_properties: list[str] | None,
        dp_timestamps: dict[str, int] | None,
    ) -> bool:
        """Determine if the wrapper should skip an update.

        The default is to always skip, unless overridden in subclasses.
        """
        return True

    def read_device_status(self, device: CustomerDevice) -> T | None:
        """Read device status and convert to a Home Assistant value."""
        raise NotImplementedError

    def get_update_commands(
        self, device: CustomerDevice, value: T
    ) -> list[dict[str, Any]]:
        """Generate update commands for a Home Assistant action."""
        raise NotImplementedError


class DPCodeWrapper(DeviceWrapper):
    """Base device wrapper for a single DPCode.

    Used as a common interface for referring to a DPCode, and
    access read conversion routines.
    """
Repository: home-assistant/core
File: homeassistant/components/tuya/models.py (Python)

Code before:

        delta = float(raw_value)

        if self._accumulated_value is None:
            self._accumulated_value = delta
            _LOGGER.debug(
                "Initialized accumulated value for %s: %s",
                self.dpcode,
                self._accumulated_value,
            )
        else:
            self._accumulated_value += delta
            _LOGGER.debug(
                "Delta update for %s: +%s, total: %s",
                self.dpcode,
                delta,
                self._accumulated_value,
            )

        self._last_dp_timestamp = current_timestamp

    def skip_update(
        self,
        device: CustomerDevice,
        updated_status_properties: list[str] | None,
        dp_timestamps: dict | None = None,
    ) -> bool:
        """Override skip_update to process delta updates.

        Processes delta accumulation before determining if update should be skipped.
        """
        if super().skip_update(device, updated_status_properties, dp_timestamps):
            return True
        # Process delta update when not skipping
        self.process_delta_update(device, dp_timestamps)
        return False

    def read_device_status(self, device: CustomerDevice) -> float | None:
        """Read device status, returning accumulated value for delta reports."""
        return self._accumulated_value


class DPCodeRawWrapper(DPCodeTypeInformationWrapper[RawTypeInformation]):
    """Wrapper to extract information from a RAW/binary value."""

    _DPTYPE = RawTypeInformation


class DPCodeStringWrapper(DPCodeTypeInformationWrapper[StringTypeInformation]):
    """Wrapper to extract information from a STRING value."""


Review comment (suggestion):
```suggestion
        dp_timestamps: dict[str, int] | None,
```

Code after:
    """Simple wrapper for a specific bit in bitmap values."""

    def __init__(self, dpcode: str, mask: int) -> None:
        """Init DPCodeBitmapWrapper."""
        super().__init__(dpcode)
        self._mask = mask

    def read_device_status(self, device: CustomerDevice) -> bool | None:
        """Read the device value for the dpcode."""
        if (raw_value := device.status.get(self.dpcode)) is None:
            return None
        return (raw_value & (1 << self._mask)) != 0

    @classmethod
    def find_dpcode(
        cls,
        device: CustomerDevice,
        dpcodes: str | tuple[str, ...],
        *,
        bitmap_key: str,
    ) -> Self | None:
        """Find and return a DPCodeBitmapBitWrapper for the given DP codes."""
        if (
            type_information := BitmapTypeInformation.find_dpcode(device, dpcodes)
        ) and bitmap_key in type_information.label:
            return cls(
                type_information.dpcode, type_information.label.index(bitmap_key)
            )
        return None
Repository: home-assistant/core
File: homeassistant/components/netatmo/binary_sensor.py (Python)

Code before:
    """Implementation of a Netatmo weather binary sensor."""

    entity_description: NetatmoBinarySensorEntityDescription

    def __init__(
        self,
        netatmo_device: NetatmoDevice,
        description: NetatmoBinarySensorEntityDescription,
    ) -> None:
        """Initialize a Netatmo weather binary sensor."""

        if netatmo_device.device.device_category is not None:
            if (
                DEVICE_CATEGORY_BINARY_URLS.get(netatmo_device.device.device_category)
                is not None
            ):
                self._attr_configuration_url = DEVICE_CATEGORY_BINARY_URLS[
                    netatmo_device.device.device_category
                ]
            else:
                _LOGGER.warning(
                    "Missing configuration URL for weather binary_sensor of %s device as category %s. Please report this issue",
                    netatmo_device.device.name,
                    netatmo_device.device.device_category,
                )
                return

        super().__init__(netatmo_device)

        self.entity_description = description
        self._attr_unique_id = f"{self.device.entity_id}-{description.key}"

    @callback
    def async_update_callback(self) -> None:
        """Update the entity's state."""

        value: StateType | None = None

        value = getattr(self.device, self.entity_description.netatmo_name, None)

        if value is None:
            self._attr_available = False
            self._attr_is_on = False
        else:
            self._attr_available = True
            self._attr_is_on = cast(bool, value)

        self.async_write_ha_state()


Review comment (bug):
The error handling here is not correct, the entity is created but will be half broken because we exit `__init__` early.
Can the check be moved to `_create_weather_binary_sensor_entity` instead? Why do we actually need to check it?

Code after:
        description: NetatmoBinarySensorEntityDescription,
    ) -> None:
        """Initialize a Netatmo weather binary sensor."""

        super().__init__(netatmo_device)

        self.entity_description = description
        self._attr_unique_id = f"{self.device.entity_id}-{description.key}"

    @callback
    def async_update_callback(self) -> None:
        """Update the entity's state."""

        value: StateType | None = None

        value = getattr(self.device, self.entity_description.netatmo_name, None)

        if value is None:
            self._attr_available = False
            self._attr_is_on = False
        else:
            self._attr_available = True
            self._attr_is_on = cast(bool, value)

        self.async_write_ha_state()
Repository: home-assistant/core
File: homeassistant/components/tibber/binary_sensor.py (Python)

Code before:
        entity_description: BinarySensorEntityDescription,
    ) -> None:
        """Initialize the binary sensor."""
        super().__init__(coordinator)

        self._device_id: str = device.id
        self.entity_description = entity_description
        self._attr_translation_key = entity_description.translation_key

        self._attr_unique_id = f"{device.external_id}_{self.entity_description.key}"

        self._attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, device.external_id)},
            name=device.name,
            manufacturer=device.brand,
            model=device.model,
        )

    @property
    def is_on(self) -> bool | None:
        """Return the state of the binary sensor."""
        sensors = self.coordinator.sensors_by_device.get(self._device_id, {})
        sensor = sensors.get(self.entity_description.key)
        if sensor is None:
            return None
        if sensor.value in ("connected", "charging", "on"):
            return True
        if sensor.value in ("disconnected", "idle", "off"):
            return False
        return None


Review comment (suggestion):
I think adding a `is_on_fn` to the entity descriptions would be nicer, so we better map the values to the corresponding entities

Code after:
):
    """Representation of a Tibber Data API binary sensor."""

    _attr_has_entity_name = True
    entity_description: TibberBinarySensorEntityDescription

    def __init__(
        self,
        coordinator: TibberDataAPICoordinator,
        device: TibberDevice,
        entity_description: TibberBinarySensorEntityDescription,
    ) -> None:
        """Initialize the binary sensor."""
        super().__init__(coordinator)

        self._device_id: str = device.id
        self.entity_description = entity_description

        self._attr_unique_id = f"{device.id}_{entity_description.key}"

        self._attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, device.external_id)},
            name=device.name,
            manufacturer=device.brand,
            model=device.model,
        )

    @property
    def available(self) -> bool:
        """Return if entity is available."""
        return (
            super().available and self._device_id in self.coordinator.sensors_by_device
        )

    @property
    def device(self) -> dict[str, tibber.data_api.Sensor]:
        """Return the device sensors."""
        return self.coordinator.sensors_by_device[self._device_id]

    @property
    def is_on(self) -> bool | None:
        """Return the state of the binary sensor."""
        return self.entity_description.is_on_fn(
            str(self.device[self.entity_description.key].value)
        )
Repository: home-assistant/core
File: homeassistant/components/tibber/binary_sensor.py (Python)

Code before:
                continue
            entities.append(TibberDataAPIBinarySensor(coordinator, device, description))
    async_add_entities(entities)


class TibberDataAPIBinarySensor(
    CoordinatorEntity[TibberDataAPICoordinator], BinarySensorEntity
):
    """Representation of a Tibber Data API binary sensor."""

    _attr_has_entity_name = True
    entity_description: TibberBinarySensorEntityDescription

    def __init__(
        self,
        coordinator: TibberDataAPICoordinator,
        device: TibberDevice,
        entity_description: TibberBinarySensorEntityDescription,
    ) -> None:
        """Initialize the binary sensor."""
        super().__init__(coordinator)

        self._device_id: str = device.id
        self.entity_description = entity_description

        self._attr_unique_id = f"{device.external_id}_{self.entity_description.key}"

        self._attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, device.external_id)},
            name=device.name,
            manufacturer=device.brand,
            model=device.model,
        )

    @property
    def is_on(self) -> bool | None:
        """Return the state of the binary sensor."""
        sensors = self.coordinator.sensors_by_device.get(self._device_id, {})
        sensor = sensors[self.entity_description.key]
        value: str | None = str(sensor.value) if sensor.value is not None else None
        return self.entity_description.is_on_fn(value)


Review comment (suggestion):
```suggestion
        self._attr_unique_id = f"{device.external_id}_{entity_description.key}"
```

Code after:

        self._attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, device.external_id)},
            name=device.name,
            manufacturer=device.brand,
            model=device.model,
        )

    @property
    def available(self) -> bool:
        """Return if entity is available."""
        return (
            super().available and self._device_id in self.coordinator.sensors_by_device
        )

    @property
    def device(self) -> dict[str, tibber.data_api.Sensor]:
        """Return the device sensors."""
        return self.coordinator.sensors_by_device[self._device_id]

    @property
    def is_on(self) -> bool | None:
        """Return the state of the binary sensor."""
        return self.entity_description.is_on_fn(
            str(self.device[self.entity_description.key].value)
        )
Repository: home-assistant/core
File: tests/components/bang_olufsen/test_binary_sensor.py (Python)

Code before:

from .conftest import mock_websocket_connection
from .const import TEST_BATTERY, TEST_BATTERY_CHARGING_BINARY_SENSOR_ENTITY_ID

from tests.common import MockConfigEntry


async def test_battery_charging(
    hass: HomeAssistant,
    entity_registry: EntityRegistry,
    mock_mozart_client: AsyncMock,
    mock_config_entry_a5: MockConfigEntry,
) -> None:
    """Test the battery charging time entity."""
    # Ensure battery entities are created
    mock_mozart_client.get_battery_state.return_value = TEST_BATTERY

    # Load entry
    mock_config_entry_a5.add_to_hass(hass)
    await hass.config_entries.async_setup(mock_config_entry_a5.entry_id)
    await mock_websocket_connection(hass, mock_mozart_client)
    await hass.async_block_till_done()

    # Initial state is False
    assert (states := hass.states.get(TEST_BATTERY_CHARGING_BINARY_SENSOR_ENTITY_ID))
    assert states.state == "off"

    # Check binary sensor reacts as expected to WebSocket events
    battery_callback = mock_mozart_client.get_battery_notifications.call_args[0][0]

    battery_callback(BatteryState(is_charging=True))
    await hass.async_block_till_done()

    assert (states := hass.states.get(TEST_BATTERY_CHARGING_BINARY_SENSOR_ENTITY_ID))
    assert states.state == "on"


Review comment (suggestion):
```suggestion
    assert states.state == STATE_OFF
```

Code after:
from homeassistant.helpers.entity_registry import EntityRegistry

from .conftest import mock_websocket_connection
from .const import TEST_BATTERY, TEST_BATTERY_CHARGING_BINARY_SENSOR_ENTITY_ID

from tests.common import MockConfigEntry


async def test_battery_charging(
    hass: HomeAssistant,
    entity_registry: EntityRegistry,
    mock_mozart_client: AsyncMock,
    mock_config_entry_a5: MockConfigEntry,
) -> None:
    """Test the battery charging time entity."""
    # Ensure battery entities are created
    mock_mozart_client.get_battery_state.return_value = TEST_BATTERY

    # Load entry
    mock_config_entry_a5.add_to_hass(hass)
    await hass.config_entries.async_setup(mock_config_entry_a5.entry_id)
    await mock_websocket_connection(hass, mock_mozart_client)
    await hass.async_block_till_done()

    # Initial state is False
    assert (states := hass.states.get(TEST_BATTERY_CHARGING_BINARY_SENSOR_ENTITY_ID))
    assert states.state == STATE_OFF

    # Check binary sensor reacts as expected to WebSocket events
    battery_callback = mock_mozart_client.get_battery_notifications.call_args[0][0]

    battery_callback(BatteryState(is_charging=True))
    await hass.async_block_till_done()

    assert (states := hass.states.get(TEST_BATTERY_CHARGING_BINARY_SENSOR_ENTITY_ID))
    assert states.state == STATE_ON