diff --git a/.coverage b/.coverage index 6b898bb..3b3afea 100644 Binary files a/.coverage and b/.coverage differ diff --git a/README.md b/README.md index 40bfd41..7ea41a4 100644 --- a/README.md +++ b/README.md @@ -188,7 +188,7 @@ Dispense water with specified amount and carbonation level. **Parameters:** - `device_id` (required): The Blanco Unit device to dispense water from -- `amount_ml` (required): Amount of water in milliliters (100-1500, must be multiple of 100) +- `amount_ml` (required): Amount of water in milliliters, min 50ml - `co2_intensity` (required): Carbonation level - `1` = Still water - `2` = Medium carbonation diff --git a/custom_components/blanco_unit/client.py b/custom_components/blanco_unit/client.py index 8bb0b87..800a3c0 100644 --- a/custom_components/blanco_unit/client.py +++ b/custom_components/blanco_unit/client.py @@ -739,10 +739,8 @@ async def dispense_water(self, amount_ml: int, co2_intensity: int) -> bool: Raises: ValueError: If amount or intensity is invalid. """ - if not (100 <= amount_ml <= 1500): - raise ValueError("Amount must be between 100ml and 1500ml") - if amount_ml % 100 != 0: - raise ValueError("Amount must be a multiple of 100ml") + if amount_ml < 50: + raise ValueError("Amount must be at least 50ml") if co2_intensity not in (1, 2, 3): raise ValueError("CO2 intensity must be 1 (still), 2 (medium), or 3 (high)") diff --git a/custom_components/blanco_unit/services.py b/custom_components/blanco_unit/services.py index d5835c6..09cbbcd 100644 --- a/custom_components/blanco_unit/services.py +++ b/custom_components/blanco_unit/services.py @@ -1,326 +1,326 @@ -"""Home Assistant services provided by the Blanco Unit integration.""" - -import json -import logging - -import voluptuous as vol - -from homeassistant.core import HomeAssistant, ServiceCall, SupportsResponse -from homeassistant.exceptions import ServiceValidationError -from homeassistant.helpers import config_validation as cv, device_registry as dr - -from .const import ( - CONF_PIN, - DOMAIN, - HA_SERVICE_ALLOW_CLOUD, - HA_SERVICE_ATTR_AMOUNT_ML, - HA_SERVICE_ATTR_CO2_INTENSITY, - HA_SERVICE_ATTR_DATA, - HA_SERVICE_ATTR_DEVICE_ID, - HA_SERVICE_ATTR_NEW_PIN, - HA_SERVICE_ATTR_PASSWORD, - HA_SERVICE_ATTR_RCA_ID, - HA_SERVICE_ATTR_SSID, - HA_SERVICE_ATTR_UPDATE_CONFIG, - HA_SERVICE_CHANGE_PIN, - HA_SERVICE_CONNECT_WIFI, - HA_SERVICE_DISCONNECT_WIFI, - HA_SERVICE_DISPENSE_WATER, - HA_SERVICE_FACTORY_RESET, - HA_SERVICE_SCAN_PROTOCOL, - HA_SERVICE_SCAN_WIFI, -) -from .coordinator import BlancoUnitCoordinator - -_LOGGER = logging.getLogger(__name__) - -# Service schemas -def _validate_amount_ml(value: int) -> int: - """Validate amount is a multiple of 100.""" - if value % 100 != 0: - raise vol.Invalid("Amount must be a multiple of 100ml") - return value - - -SERVICE_DISPENSE_WATER_SCHEMA = vol.Schema( - { - vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, - vol.Required(HA_SERVICE_ATTR_AMOUNT_ML): vol.All( - vol.Coerce(int), - vol.Range(min=100, max=1500), - _validate_amount_ml, - ), - vol.Required(HA_SERVICE_ATTR_CO2_INTENSITY): vol.All( - vol.Coerce(int), vol.In([1, 2, 3]) - ), - } -) - -SERVICE_CHANGE_PIN_SCHEMA = vol.Schema( - { - vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, - vol.Required(HA_SERVICE_ATTR_NEW_PIN): vol.All( - cv.string, - vol.Length(min=5, max=5), - vol.Match(r"^\d{5}$"), - ), - vol.Optional(HA_SERVICE_ATTR_UPDATE_CONFIG, default=False): cv.boolean, - } -) - -SERVICE_SCAN_PROTOCOL_SCHEMA = vol.Schema( - { - vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, - vol.Required(HA_SERVICE_ATTR_DATA): dict, - } -) - -SERVICE_DEVICE_ONLY_SCHEMA = vol.Schema( - { - vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, - } -) - -SERVICE_CONNECT_WIFI_SCHEMA = vol.Schema( - { - vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, - vol.Required(HA_SERVICE_ATTR_SSID): cv.string, - vol.Required(HA_SERVICE_ATTR_PASSWORD): cv.string, - } -) - -SERVICE_ALLOW_CLOUD_SCHEMA = vol.Schema( - { - vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, - vol.Optional(HA_SERVICE_ATTR_RCA_ID, default=""): cv.string, - } -) - - -def async_setup_services(hass: HomeAssistant) -> None: - """Set up Blanco Unit integration services.""" - _LOGGER.debug("async_setup_services called") - - # Only register services once - if hass.services.has_service(DOMAIN, HA_SERVICE_DISPENSE_WATER): - _LOGGER.debug("Services already registered, skipping") - return - - async def handle_dispense_water(call: ServiceCall) -> None: - """Handle the dispense_water service call.""" - _LOGGER.debug("Dispense water service called with data: %s", call.data) - coordinator = _get_coordinator(hass, call) - amount_ml = call.data[HA_SERVICE_ATTR_AMOUNT_ML] - co2_intensity = call.data[HA_SERVICE_ATTR_CO2_INTENSITY] - - await coordinator.dispense_water(amount_ml, co2_intensity) - - async def handle_change_pin(call: ServiceCall) -> None: - """Handle the change_pin service call.""" - _LOGGER.debug("Change PIN service called with data: %s", call.data) - coordinator = _get_coordinator(hass, call) - new_pin = call.data[HA_SERVICE_ATTR_NEW_PIN] - update_config = call.data[HA_SERVICE_ATTR_UPDATE_CONFIG] - - # Change the PIN on the device - await coordinator.change_pin(new_pin) - - # If update_config is True, update the config entry with the new PIN - if update_config: - device_id = call.data[HA_SERVICE_ATTR_DEVICE_ID] - registry = dr.async_get(hass) - device = registry.async_get(device_id) - if device: - entry_id = next(iter(device.config_entries)) - entry = hass.config_entries.async_get_entry(entry_id) - if entry: - # Update the config entry with the new PIN - hass.config_entries.async_update_entry( - entry, - data={**entry.data, CONF_PIN: int(new_pin)}, - ) - _LOGGER.info("Updated config entry with new PIN") - # Reload the config entry to reconnect with new PIN - await hass.config_entries.async_reload(entry_id) - - async def handle_scan_protocol(call: ServiceCall) -> dict: - """Handle the scan_protocol_parameters service call.""" - - _LOGGER.debug("Scan protocol service called with data: %s", call.data) - coordinator = _get_coordinator(hass, call) - - # Extract parameters from the data object - data = call.data[HA_SERVICE_ATTR_DATA] - evt_type = data.get("evt_type", 7) - ctrl = data.get("ctrl") - pars = data.get("pars") - - _LOGGER.info( - "Testing protocol parameters: evt_type=%d, ctrl=%s, pars=%s", - evt_type, - ctrl, - pars, - ) - - # Test the specific parameter combination - response = await coordinator.test_protocol_parameters( - evt_type, ctrl, pars - ) - - result = { - "evt_type": evt_type, - "ctrl": ctrl, - "pars": pars, - "success": response is not None, - "response": response if response else None, - } - - _LOGGER.info("Response: %s", json.dumps(response, indent=2)) - return result - - async def handle_scan_wifi_networks(call: ServiceCall) -> dict: - """Handle the scan_wifi_networks service call.""" - _LOGGER.debug("Scan WiFi networks service called with data: %s", call.data) - coordinator = _get_coordinator(hass, call) - networks = await coordinator.scan_wifi_networks() - return { - "networks": [ - { - "ssid": n.ssid, - "signal": n.signal, - "auth_mode": n.auth_mode, - } - for n in networks - ] - } - - async def handle_connect_wifi(call: ServiceCall) -> None: - """Handle the connect_wifi service call.""" - _LOGGER.debug("Connect WiFi service called with data: %s", call.data) - coordinator = _get_coordinator(hass, call) - ssid = call.data[HA_SERVICE_ATTR_SSID] - password = call.data[HA_SERVICE_ATTR_PASSWORD] - await coordinator.connect_wifi(ssid, password) - - async def handle_disconnect_wifi(call: ServiceCall) -> None: - """Handle the disconnect_wifi service call.""" - _LOGGER.debug("Disconnect WiFi service called") - coordinator = _get_coordinator(hass, call) - await coordinator.disconnect_wifi() - - async def handle_allow_cloud_services(call: ServiceCall) -> None: - """Handle the allow_cloud_services service call.""" - _LOGGER.debug( - "Allow cloud services called with data: %s", call.data - ) - coordinator = _get_coordinator(hass, call) - rca_id = call.data[HA_SERVICE_ATTR_RCA_ID] - await coordinator.allow_cloud_services(rca_id) - - async def handle_factory_reset(call: ServiceCall) -> None: - """Handle the factory_reset service call.""" - _LOGGER.debug("Factory reset service called") - coordinator = _get_coordinator(hass, call) - await coordinator.factory_reset() - - hass.services.async_register( - DOMAIN, - HA_SERVICE_DISPENSE_WATER, - handle_dispense_water, - schema=SERVICE_DISPENSE_WATER_SCHEMA, - ) - - hass.services.async_register( - DOMAIN, - HA_SERVICE_CHANGE_PIN, - handle_change_pin, - schema=SERVICE_CHANGE_PIN_SCHEMA, - ) - - hass.services.async_register( - DOMAIN, - HA_SERVICE_SCAN_PROTOCOL, - handle_scan_protocol, - schema=SERVICE_SCAN_PROTOCOL_SCHEMA, - supports_response=SupportsResponse.ONLY, - ) - - hass.services.async_register( - DOMAIN, - HA_SERVICE_SCAN_WIFI, - handle_scan_wifi_networks, - schema=SERVICE_DEVICE_ONLY_SCHEMA, - supports_response=SupportsResponse.ONLY, - ) - - hass.services.async_register( - DOMAIN, - HA_SERVICE_CONNECT_WIFI, - handle_connect_wifi, - schema=SERVICE_CONNECT_WIFI_SCHEMA, - ) - - hass.services.async_register( - DOMAIN, - HA_SERVICE_DISCONNECT_WIFI, - handle_disconnect_wifi, - schema=SERVICE_DEVICE_ONLY_SCHEMA, - ) - - hass.services.async_register( - DOMAIN, - HA_SERVICE_ALLOW_CLOUD, - handle_allow_cloud_services, - schema=SERVICE_ALLOW_CLOUD_SCHEMA, - ) - - hass.services.async_register( - DOMAIN, - HA_SERVICE_FACTORY_RESET, - handle_factory_reset, - schema=SERVICE_DEVICE_ONLY_SCHEMA, - ) - - -def _get_coordinator(hass: HomeAssistant, call: ServiceCall) -> BlancoUnitCoordinator: - """Extract device_id from service call and return the coordinator.""" - device_id = call.data.get(HA_SERVICE_ATTR_DEVICE_ID) - if not device_id: - raise ServiceValidationError( - translation_domain=DOMAIN, - translation_key="device_id_not_specified", - ) - - registry = dr.async_get(hass) - device = registry.async_get(device_id) - if not device: - raise ServiceValidationError( - translation_domain=DOMAIN, - translation_key="device_not_found", - translation_placeholders={ - "device_id": str(device_id), - }, - ) - - entry_id = next(iter(device.config_entries)) - entry = hass.config_entries.async_get_entry(entry_id) - if entry is None: - raise ServiceValidationError( - translation_domain=DOMAIN, - translation_key="config_entry_not_found", - translation_placeholders={ - "device_id": str(device_id), - }, - ) - - runtime_data = entry.runtime_data - if not isinstance(runtime_data, BlancoUnitCoordinator): - raise ServiceValidationError( - translation_domain=DOMAIN, - translation_key="invalid_runtime_data", - translation_placeholders={ - "device_id": str(device_id), - }, - ) - - return runtime_data +"""Home Assistant services provided by the Blanco Unit integration.""" + +import json +import logging + +import voluptuous as vol + +from homeassistant.core import HomeAssistant, ServiceCall, SupportsResponse +from homeassistant.exceptions import ServiceValidationError +from homeassistant.helpers import config_validation as cv, device_registry as dr + +from .const import ( + CONF_PIN, + DOMAIN, + HA_SERVICE_ALLOW_CLOUD, + HA_SERVICE_ATTR_AMOUNT_ML, + HA_SERVICE_ATTR_CO2_INTENSITY, + HA_SERVICE_ATTR_DATA, + HA_SERVICE_ATTR_DEVICE_ID, + HA_SERVICE_ATTR_NEW_PIN, + HA_SERVICE_ATTR_PASSWORD, + HA_SERVICE_ATTR_RCA_ID, + HA_SERVICE_ATTR_SSID, + HA_SERVICE_ATTR_UPDATE_CONFIG, + HA_SERVICE_CHANGE_PIN, + HA_SERVICE_CONNECT_WIFI, + HA_SERVICE_DISCONNECT_WIFI, + HA_SERVICE_DISPENSE_WATER, + HA_SERVICE_FACTORY_RESET, + HA_SERVICE_SCAN_PROTOCOL, + HA_SERVICE_SCAN_WIFI, +) +from .coordinator import BlancoUnitCoordinator + +_LOGGER = logging.getLogger(__name__) + +# Service schemas +def _validate_amount_ml(value: int) -> int: + """Validate amount is minimum 50.""" + if value < 50: + raise vol.Invalid("Amount must be at least 50ml") + return value + + +SERVICE_DISPENSE_WATER_SCHEMA = vol.Schema( + { + vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, + vol.Required(HA_SERVICE_ATTR_AMOUNT_ML): vol.All( + vol.Coerce(int), + vol.Range(min=50), + _validate_amount_ml, + ), + vol.Required(HA_SERVICE_ATTR_CO2_INTENSITY): vol.All( + vol.Coerce(int), vol.In([1, 2, 3]) + ), + } +) + +SERVICE_CHANGE_PIN_SCHEMA = vol.Schema( + { + vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, + vol.Required(HA_SERVICE_ATTR_NEW_PIN): vol.All( + cv.string, + vol.Length(min=5, max=5), + vol.Match(r"^\d{5}$"), + ), + vol.Optional(HA_SERVICE_ATTR_UPDATE_CONFIG, default=False): cv.boolean, + } +) + +SERVICE_SCAN_PROTOCOL_SCHEMA = vol.Schema( + { + vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, + vol.Required(HA_SERVICE_ATTR_DATA): dict, + } +) + +SERVICE_DEVICE_ONLY_SCHEMA = vol.Schema( + { + vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, + } +) + +SERVICE_CONNECT_WIFI_SCHEMA = vol.Schema( + { + vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, + vol.Required(HA_SERVICE_ATTR_SSID): cv.string, + vol.Required(HA_SERVICE_ATTR_PASSWORD): cv.string, + } +) + +SERVICE_ALLOW_CLOUD_SCHEMA = vol.Schema( + { + vol.Required(HA_SERVICE_ATTR_DEVICE_ID): cv.string, + vol.Optional(HA_SERVICE_ATTR_RCA_ID, default=""): cv.string, + } +) + + +def async_setup_services(hass: HomeAssistant) -> None: + """Set up Blanco Unit integration services.""" + _LOGGER.debug("async_setup_services called") + + # Only register services once + if hass.services.has_service(DOMAIN, HA_SERVICE_DISPENSE_WATER): + _LOGGER.debug("Services already registered, skipping") + return + + async def handle_dispense_water(call: ServiceCall) -> None: + """Handle the dispense_water service call.""" + _LOGGER.debug("Dispense water service called with data: %s", call.data) + coordinator = _get_coordinator(hass, call) + amount_ml = call.data[HA_SERVICE_ATTR_AMOUNT_ML] + co2_intensity = call.data[HA_SERVICE_ATTR_CO2_INTENSITY] + + await coordinator.dispense_water(amount_ml, co2_intensity) + + async def handle_change_pin(call: ServiceCall) -> None: + """Handle the change_pin service call.""" + _LOGGER.debug("Change PIN service called with data: %s", call.data) + coordinator = _get_coordinator(hass, call) + new_pin = call.data[HA_SERVICE_ATTR_NEW_PIN] + update_config = call.data[HA_SERVICE_ATTR_UPDATE_CONFIG] + + # Change the PIN on the device + await coordinator.change_pin(new_pin) + + # If update_config is True, update the config entry with the new PIN + if update_config: + device_id = call.data[HA_SERVICE_ATTR_DEVICE_ID] + registry = dr.async_get(hass) + device = registry.async_get(device_id) + if device: + entry_id = next(iter(device.config_entries)) + entry = hass.config_entries.async_get_entry(entry_id) + if entry: + # Update the config entry with the new PIN + hass.config_entries.async_update_entry( + entry, + data={**entry.data, CONF_PIN: int(new_pin)}, + ) + _LOGGER.info("Updated config entry with new PIN") + # Reload the config entry to reconnect with new PIN + await hass.config_entries.async_reload(entry_id) + + async def handle_scan_protocol(call: ServiceCall) -> dict: + """Handle the scan_protocol_parameters service call.""" + + _LOGGER.debug("Scan protocol service called with data: %s", call.data) + coordinator = _get_coordinator(hass, call) + + # Extract parameters from the data object + data = call.data[HA_SERVICE_ATTR_DATA] + evt_type = data.get("evt_type", 7) + ctrl = data.get("ctrl") + pars = data.get("pars") + + _LOGGER.info( + "Testing protocol parameters: evt_type=%d, ctrl=%s, pars=%s", + evt_type, + ctrl, + pars, + ) + + # Test the specific parameter combination + response = await coordinator.test_protocol_parameters( + evt_type, ctrl, pars + ) + + result = { + "evt_type": evt_type, + "ctrl": ctrl, + "pars": pars, + "success": response is not None, + "response": response if response else None, + } + + _LOGGER.info("Response: %s", json.dumps(response, indent=2)) + return result + + async def handle_scan_wifi_networks(call: ServiceCall) -> dict: + """Handle the scan_wifi_networks service call.""" + _LOGGER.debug("Scan WiFi networks service called with data: %s", call.data) + coordinator = _get_coordinator(hass, call) + networks = await coordinator.scan_wifi_networks() + return { + "networks": [ + { + "ssid": n.ssid, + "signal": n.signal, + "auth_mode": n.auth_mode, + } + for n in networks + ] + } + + async def handle_connect_wifi(call: ServiceCall) -> None: + """Handle the connect_wifi service call.""" + _LOGGER.debug("Connect WiFi service called with data: %s", call.data) + coordinator = _get_coordinator(hass, call) + ssid = call.data[HA_SERVICE_ATTR_SSID] + password = call.data[HA_SERVICE_ATTR_PASSWORD] + await coordinator.connect_wifi(ssid, password) + + async def handle_disconnect_wifi(call: ServiceCall) -> None: + """Handle the disconnect_wifi service call.""" + _LOGGER.debug("Disconnect WiFi service called") + coordinator = _get_coordinator(hass, call) + await coordinator.disconnect_wifi() + + async def handle_allow_cloud_services(call: ServiceCall) -> None: + """Handle the allow_cloud_services service call.""" + _LOGGER.debug( + "Allow cloud services called with data: %s", call.data + ) + coordinator = _get_coordinator(hass, call) + rca_id = call.data[HA_SERVICE_ATTR_RCA_ID] + await coordinator.allow_cloud_services(rca_id) + + async def handle_factory_reset(call: ServiceCall) -> None: + """Handle the factory_reset service call.""" + _LOGGER.debug("Factory reset service called") + coordinator = _get_coordinator(hass, call) + await coordinator.factory_reset() + + hass.services.async_register( + DOMAIN, + HA_SERVICE_DISPENSE_WATER, + handle_dispense_water, + schema=SERVICE_DISPENSE_WATER_SCHEMA, + ) + + hass.services.async_register( + DOMAIN, + HA_SERVICE_CHANGE_PIN, + handle_change_pin, + schema=SERVICE_CHANGE_PIN_SCHEMA, + ) + + hass.services.async_register( + DOMAIN, + HA_SERVICE_SCAN_PROTOCOL, + handle_scan_protocol, + schema=SERVICE_SCAN_PROTOCOL_SCHEMA, + supports_response=SupportsResponse.ONLY, + ) + + hass.services.async_register( + DOMAIN, + HA_SERVICE_SCAN_WIFI, + handle_scan_wifi_networks, + schema=SERVICE_DEVICE_ONLY_SCHEMA, + supports_response=SupportsResponse.ONLY, + ) + + hass.services.async_register( + DOMAIN, + HA_SERVICE_CONNECT_WIFI, + handle_connect_wifi, + schema=SERVICE_CONNECT_WIFI_SCHEMA, + ) + + hass.services.async_register( + DOMAIN, + HA_SERVICE_DISCONNECT_WIFI, + handle_disconnect_wifi, + schema=SERVICE_DEVICE_ONLY_SCHEMA, + ) + + hass.services.async_register( + DOMAIN, + HA_SERVICE_ALLOW_CLOUD, + handle_allow_cloud_services, + schema=SERVICE_ALLOW_CLOUD_SCHEMA, + ) + + hass.services.async_register( + DOMAIN, + HA_SERVICE_FACTORY_RESET, + handle_factory_reset, + schema=SERVICE_DEVICE_ONLY_SCHEMA, + ) + + +def _get_coordinator(hass: HomeAssistant, call: ServiceCall) -> BlancoUnitCoordinator: + """Extract device_id from service call and return the coordinator.""" + device_id = call.data.get(HA_SERVICE_ATTR_DEVICE_ID) + if not device_id: + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="device_id_not_specified", + ) + + registry = dr.async_get(hass) + device = registry.async_get(device_id) + if not device: + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="device_not_found", + translation_placeholders={ + "device_id": str(device_id), + }, + ) + + entry_id = next(iter(device.config_entries)) + entry = hass.config_entries.async_get_entry(entry_id) + if entry is None: + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="config_entry_not_found", + translation_placeholders={ + "device_id": str(device_id), + }, + ) + + runtime_data = entry.runtime_data + if not isinstance(runtime_data, BlancoUnitCoordinator): + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="invalid_runtime_data", + translation_placeholders={ + "device_id": str(device_id), + }, + ) + + return runtime_data diff --git a/custom_components/blanco_unit/services.yaml b/custom_components/blanco_unit/services.yaml index 1c5d170..0145018 100644 --- a/custom_components/blanco_unit/services.yaml +++ b/custom_components/blanco_unit/services.yaml @@ -14,14 +14,13 @@ dispense_water: integration: blanco_unit amount_ml: name: Amount (ml) - description: Amount of water to dispense in milliliters (100-1500, must be multiple of 100). + description: Amount of water to dispense in milliliters. required: true example: 250 selector: number: - min: 100 - max: 1500 - step: 100 + min: 50 + step: 1 unit_of_measurement: "ml" co2_intensity: name: CO2 intensity diff --git a/test_cli.py b/test_cli.py index 60b43a9..712fc7f 100755 --- a/test_cli.py +++ b/test_cli.py @@ -389,7 +389,7 @@ async def test_set_water_hardness(self) -> None: async def test_dispense_water(self) -> None: """Test dispense_water().""" print("\n--- Dispense Water ---") - print("Amount: 100-1500ml (multiples of 100)") + print("Amount: ml (minimum 50ml)") print("CO2 Intensity: 1=still, 2=medium, 3=high") while True: @@ -397,8 +397,8 @@ async def test_dispense_water(self) -> None: amount = input("Enter amount (ml): ").strip() amount_int = int(amount) - if not (100 <= amount_int <= 1500 and amount_int % 100 == 0): - print("Amount must be 100-1500ml in multiples of 100") + if amount_int < 50: + print("Amount must be minimum 50ml") continue co2 = input("Enter CO2 intensity (1-3): ").strip() diff --git a/tests/test_client.py b/tests/test_client.py index 1f2d7e0..6d204c9 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1668,36 +1668,8 @@ async def test_bluetooth_client_dispense_water_invalid_amount_low(): pin="12345", device=device, connection_callback=callback ) - with pytest.raises(ValueError, match="Amount must be between 100ml and 1500ml"): - await client.dispense_water(amount_ml=50, co2_intensity=1) - - -@pytest.mark.asyncio -async def test_bluetooth_client_dispense_water_invalid_amount_high(): - """Test dispense_water with amount too high.""" - device = BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Test Device", details={}) - callback = MagicMock() - - client = BlancoUnitBluetoothClient( - pin="12345", device=device, connection_callback=callback - ) - - with pytest.raises(ValueError, match="Amount must be between 100ml and 1500ml"): - await client.dispense_water(amount_ml=2000, co2_intensity=1) - - -@pytest.mark.asyncio -async def test_bluetooth_client_dispense_water_invalid_amount_not_multiple(): - """Test dispense_water with amount not a multiple of 100.""" - device = BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Test Device", details={}) - callback = MagicMock() - - client = BlancoUnitBluetoothClient( - pin="12345", device=device, connection_callback=callback - ) - - with pytest.raises(ValueError, match="Amount must be a multiple of 100ml"): - await client.dispense_water(amount_ml=150, co2_intensity=1) + with pytest.raises(ValueError, match="Amount must be at least 50ml"): + await client.dispense_water(amount_ml=1, co2_intensity=1) @pytest.mark.asyncio diff --git a/tests/test_services.py b/tests/test_services.py index 3e0727d..4b4e7f2 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -47,16 +47,6 @@ def test_validate_amount_ml_valid(): assert _validate_amount_ml(1500) == 1500 -def test_validate_amount_ml_invalid(): - """Test _validate_amount_ml with invalid values.""" - with pytest.raises(vol.Invalid, match="Amount must be a multiple of 100ml"): - _validate_amount_ml(150) - with pytest.raises(vol.Invalid, match="Amount must be a multiple of 100ml"): - _validate_amount_ml(99) - with pytest.raises(vol.Invalid, match="Amount must be a multiple of 100ml"): - _validate_amount_ml(1550) - - def test_dispense_water_schema_valid(): """Test dispense water schema with valid data.""" valid_data = { @@ -72,32 +62,12 @@ def test_dispense_water_schema_valid(): def test_dispense_water_schema_invalid_amount(): """Test dispense water schema with invalid amount.""" - # Not a multiple of 100 - with pytest.raises(vol.Invalid): - SERVICE_DISPENSE_WATER_SCHEMA( - { - HA_SERVICE_ATTR_DEVICE_ID: "test_device_id", - HA_SERVICE_ATTR_AMOUNT_ML: 150, - HA_SERVICE_ATTR_CO2_INTENSITY: 2, - } - ) - # Below minimum with pytest.raises(vol.Invalid): SERVICE_DISPENSE_WATER_SCHEMA( { HA_SERVICE_ATTR_DEVICE_ID: "test_device_id", - HA_SERVICE_ATTR_AMOUNT_ML: 50, - HA_SERVICE_ATTR_CO2_INTENSITY: 2, - } - ) - - # Above maximum - with pytest.raises(vol.Invalid): - SERVICE_DISPENSE_WATER_SCHEMA( - { - HA_SERVICE_ATTR_DEVICE_ID: "test_device_id", - HA_SERVICE_ATTR_AMOUNT_ML: 1600, + HA_SERVICE_ATTR_AMOUNT_ML: 1, HA_SERVICE_ATTR_CO2_INTENSITY: 2, } )