zlock_slots/tests/test_config_flow.py

"""Tests for the config flow and options flow (staged commit semantics)."""
import pytest
from unittest.mock import patch

from homeassistant.config_entries import SOURCE_USER
from homeassistant.data_entry_flow import FlowResultType
from pytest_homeassistant_custom_component.common import MockConfigEntry

from custom_components.zlock_slots.const import (
  DOMAIN, CONF_LOCK_ENTITY_IDS, CONF_GROUPS, CONF_GROUP_NAME,
)
from tests.conftest import LOCK_ENTITY_ID, FakeNode

_ZWAVE_HELPER = "homeassistant.components.zwave_js.helpers.async_get_node_from_entity_id"
LOCK_2 = "lock.back_door"


@pytest.fixture(autouse=True)
def mock_zwave_node():
  with patch(_ZWAVE_HELPER, return_value=FakeNode()):
    yield


@pytest.fixture(autouse=True)
def bypass_manifest_dependencies():
  """zwave_js/frontend/lovelace can't be set up in the test harness."""
  with patch(
    "homeassistant.setup._async_process_dependencies", return_value=[]
  ):
    yield


# ---------------------------------------------------------------------------
# Config flow
# ---------------------------------------------------------------------------

async def test_user_flow_creates_entry(hass):
  result = await hass.config_entries.flow.async_init(
    DOMAIN, context={"source": SOURCE_USER}
  )
  assert result["type"] is FlowResultType.FORM
  assert result["step_id"] == "user"

  result = await hass.config_entries.flow.async_configure(
    result["flow_id"], {CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]}
  )
  await hass.async_block_till_done()

  assert result["type"] is FlowResultType.CREATE_ENTRY
  assert result["data"] == {CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]}


async def test_user_flow_no_locks_shows_error(hass):
  result = await hass.config_entries.flow.async_init(
    DOMAIN, context={"source": SOURCE_USER}
  )
  result = await hass.config_entries.flow.async_configure(
    result["flow_id"], {CONF_LOCK_ENTITY_IDS: []}
  )
  assert result["type"] is FlowResultType.FORM
  assert result["errors"] == {CONF_LOCK_ENTITY_IDS: "no_locks_selected"}


async def test_second_entry_blocked(hass, config_entry):
  """manifest single_config_entry: a second entry cannot be created."""
  config_entry.add_to_hass(hass)
  result = await hass.config_entries.flow.async_init(
    DOMAIN, context={"source": SOURCE_USER}
  )
  assert result["type"] is FlowResultType.ABORT
  assert result["reason"] == "single_instance_allowed"


# ---------------------------------------------------------------------------
# Options flow
# ---------------------------------------------------------------------------

def _entry_with_group() -> MockConfigEntry:
  return MockConfigEntry(
    domain=DOMAIN,
    data={CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2]},
    options={
      CONF_GROUPS: [{
        CONF_GROUP_NAME: "All Locks",
        CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2],
      }]
    },
    entry_id="test_entry_options",
  )


async def _setup_entry(hass, entry):
  entry.add_to_hass(hass)
  assert await hass.config_entries.async_setup(entry.entry_id)
  await hass.async_block_till_done()


async def _begin_options(hass, entry, step: str):
  result = await hass.config_entries.options.async_init(entry.entry_id)
  assert result["type"] is FlowResultType.MENU
  return await hass.config_entries.options.async_configure(
    result["flow_id"], {"next_step_id": step}
  )


async def _finish(hass, flow_id):
  result = await hass.config_entries.options.async_configure(
    flow_id, {"next_step_id": "finish"}
  )
  await hass.async_block_till_done()
  return result


async def test_options_add_group(hass, config_entry):
  await _setup_entry(hass, config_entry)

  result = await _begin_options(hass, config_entry, "add_group")
  result = await hass.config_entries.options.async_configure(
    result["flow_id"],
    {CONF_GROUP_NAME: "Front", CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]},
  )
  assert result["type"] is FlowResultType.MENU
  result = await _finish(hass, result["flow_id"])

  assert result["type"] is FlowResultType.CREATE_ENTRY
  assert config_entry.options[CONF_GROUPS] == [
    {CONF_GROUP_NAME: "Front", CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]}
  ]


async def test_options_duplicate_group_name_rejected(hass):
  entry = _entry_with_group()
  await _setup_entry(hass, entry)

  result = await _begin_options(hass, entry, "add_group")
  result = await hass.config_entries.options.async_configure(
    result["flow_id"],
    {CONF_GROUP_NAME: "All Locks", CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]},
  )
  assert result["type"] is FlowResultType.FORM
  assert result["errors"] == {CONF_GROUP_NAME: "group_name_exists"}


async def test_options_remove_lock_prunes_it_from_groups(hass):
  entry = _entry_with_group()
  await _setup_entry(hass, entry)

  result = await _begin_options(hass, entry, "manage_locks")
  result = await hass.config_entries.options.async_configure(
    result["flow_id"], {CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]}
  )
  assert result["type"] is FlowResultType.MENU
  await _finish(hass, result["flow_id"])

  assert entry.data[CONF_LOCK_ENTITY_IDS] == [LOCK_ENTITY_ID]
  # Group survives but no longer references the removed lock.
  assert entry.options[CONF_GROUPS] == [
    {CONF_GROUP_NAME: "All Locks", CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]}
  ]


async def test_options_group_dropped_when_all_its_locks_removed(hass):
  entry = MockConfigEntry(
    domain=DOMAIN,
    data={CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2]},
    options={
      CONF_GROUPS: [{
        CONF_GROUP_NAME: "Back only",
        CONF_LOCK_ENTITY_IDS: [LOCK_2],
      }]
    },
    entry_id="test_entry_prune",
  )
  await _setup_entry(hass, entry)

  result = await _begin_options(hass, entry, "manage_locks")
  result = await hass.config_entries.options.async_configure(
    result["flow_id"], {CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]}
  )
  await _finish(hass, result["flow_id"])

  assert entry.data[CONF_LOCK_ENTITY_IDS] == [LOCK_ENTITY_ID]
  assert entry.options[CONF_GROUPS] == []


async def test_options_abandoned_flow_changes_nothing(hass):
  """Lock changes are staged on the flow; without 'finish' the entry is untouched."""
  entry = _entry_with_group()
  await _setup_entry(hass, entry)

  result = await _begin_options(hass, entry, "manage_locks")
  result = await hass.config_entries.options.async_configure(
    result["flow_id"], {CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID]}
  )
  assert result["type"] is FlowResultType.MENU
  # No finish step — abandon the flow here.

  assert entry.data[CONF_LOCK_ENTITY_IDS] == [LOCK_ENTITY_ID, LOCK_2]
  assert entry.options[CONF_GROUPS] == [
    {CONF_GROUP_NAME: "All Locks", CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2]}
  ]