NestQuest/tests/test_admin_allowlist.py

"""Tests for the Feature 03 admin allowlist business layer (core.admin_allowlist)."""
from __future__ import annotations

import asyncio
from unittest.mock import AsyncMock, MagicMock

import pytest

from custom_components.nestquest.core.admin_allowlist import (
    list_admin_ids,
    seed_setup_admin,
    set_admin_ids,
)
from custom_components.nestquest.core.db import NestQuestDatabase
from custom_components.nestquest.core.migrations import apply_migrations


def _run(coro):
    return asyncio.new_event_loop().run_until_complete(coro)


def _make_hass_mock():
    hass = MagicMock()
    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
    return hass


def _with_db(tmp_path, name):
    def _run_test(body):
        async def _main():
            database = NestQuestDatabase(_make_hass_mock().async_add_executor_job)
            await database.open(tmp_path / name)
            try:
                await apply_migrations(database)
                return await body(database)
            finally:
                await database.close()

        return _run(_main())

    return _run_test


# ---------------------------------------------------------------------------
# set_admin_ids: read/write with the fail-closed guarantee
# ---------------------------------------------------------------------------


def test_set_admin_ids_round_trip(tmp_path) -> None:
    async def _body(database):
        await set_admin_ids(database, ["user-owner", "user-parent"])
        # Membership is the contract; AdminUsersDao.list orders by id.
        assert sorted(await list_admin_ids(database)) == [
            "user-owner",
            "user-parent",
        ]
        # Replace: the removed id is gone, the new one present.
        await set_admin_ids(database, ["user-parent", "user-grandma"])
        assert sorted(await list_admin_ids(database)) == [
            "user-grandma",
            "user-parent",
        ]
        return await list_admin_ids(database)

    _with_db(tmp_path, "set-round-trip.db")(_body)


def test_set_admin_ids_rejects_empty_allowlist(tmp_path) -> None:
    """Removing the last admin (or saving an empty selection) is
    refused with a clear error: an empty allowlist fails closed, so
    the owner must never be able to reach one through this API."""
    async def _body(database):
        await set_admin_ids(database, ["user-owner"])
        with pytest.raises(ValueError, match="at least one admin"):
            await set_admin_ids(database, [])
        # The refused write left the allowlist intact.
        assert await list_admin_ids(database) == ["user-owner"]
        return None

    _with_db(tmp_path, "set-empty-refused.db")(_body)


@pytest.mark.parametrize(
    "bad", [None, "notalist", [42], [None], [""], ["  "], 5]
)
def test_set_admin_ids_rejects_malformed_input(tmp_path, bad) -> None:
    async def _body(database):
        with pytest.raises(ValueError):
            await set_admin_ids(database, bad)
        assert await list_admin_ids(database) == []
        return None

    _with_db(tmp_path, "set-malformed.db")(_body)


def test_set_admin_ids_rejects_duplicates(tmp_path) -> None:
    async def _body(database):
        with pytest.raises(ValueError, match="duplicates"):
            await set_admin_ids(database, ["user-a", "user-a"])
        assert await list_admin_ids(database) == []
        return None

    _with_db(tmp_path, "set-duplicates.db")(_body)


def test_set_admin_ids_trims_and_replaces_atomically(tmp_path) -> None:
    async def _body(database):
        await set_admin_ids(database, ["  user-a  "])
        assert await list_admin_ids(database) == ["user-a"]
        # Sync that both adds and removes in one write.
        await set_admin_ids(database, ["user-b", "user-c"])
        assert await list_admin_ids(database) == ["user-b", "user-c"]
        return None

    _with_db(tmp_path, "set-trim.db")(_body)


# ---------------------------------------------------------------------------
# seed_setup_admin: the owner is never locked out on first setup
# ---------------------------------------------------------------------------


def test_seed_adds_context_user_on_first_setup(tmp_path) -> None:
    async def _body(database):
        seeded = await seed_setup_admin(
            database, stored_admin_ids=None, context_user_id="owner-1"
        )
        assert seeded == ["owner-1"]
        assert await list_admin_ids(database) == ["owner-1"]
        return seeded

    _with_db(tmp_path, "seed-context.db")(_body)


def test_seed_prefers_stored_entry_ids_over_context(tmp_path) -> None:
    async def _body(database):
        seeded = await seed_setup_admin(
            database,
            stored_admin_ids=["user-a", "user-b"],
            context_user_id="owner-1",
        )
        assert seeded == ["user-a", "user-b"]
        return seeded

    _with_db(tmp_path, "seed-stored.db")(_body)


def test_seed_leaves_nonempty_allowlist_untouched(tmp_path) -> None:
    """Re-running setup must never resurrect a deliberately narrowed
    list: only an EMPTY allowlist is seeded."""
    async def _body(database):
        await set_admin_ids(database, ["user-kept"])
        seeded = await seed_setup_admin(
            database,
            stored_admin_ids=["user-a", "user-b"],
            context_user_id="owner-1",
        )
        assert seeded == ["user-kept"]
        assert await list_admin_ids(database) == ["user-kept"]
        return seeded

    _with_db(tmp_path, "seed-narrowed.db")(_body)


def test_seed_with_nothing_available_leaves_empty(tmp_path) -> None:
    async def _body(database):
        seeded = await seed_setup_admin(
            database, stored_admin_ids=None, context_user_id=None
        )
        assert seeded == []
        return seeded

    _with_db(tmp_path, "seed-nothing.db")(_body)


def test_seed_rejects_malformed_candidates(tmp_path) -> None:
    async def _body(database):
        with pytest.raises(ValueError):
            await seed_setup_admin(
                database, stored_admin_ids=[42], context_user_id=None
            )
        with pytest.raises(ValueError):
            await seed_setup_admin(
                database, stored_admin_ids=None, context_user_id="  "
            )
        assert await list_admin_ids(database) == []
        return None

    _with_db(tmp_path, "seed-malformed.db")(_body)


# ---------------------------------------------------------------------------
def test_seed_owner_ids_argument(tmp_path) -> None:
    async def _body(database):
        # Malformed owner lists are rejected like any other candidate —
        # while the allowlist is still empty, so the seed proceeds far
        # enough to validate.
        with pytest.raises(ValueError):
            await seed_setup_admin(
                database,
                stored_admin_ids=None,
                context_user_id=None,
                owner_ids=[7],
            )
        assert await list_admin_ids(database) == []
        seeded = await seed_setup_admin(
            database,
            stored_admin_ids=None,
            context_user_id=None,
            owner_ids=["owner-1"],
        )
        assert seeded == ["owner-1"]
        return seeded

    _with_db(tmp_path, "seed-owner-arg.db")(_body)


# is_admin: the fail-closed resolver (Feature 09's verdict function)
# ---------------------------------------------------------------------------


def test_is_admin_verdict_matrix(tmp_path) -> None:
    async def _body(database):
        from custom_components.nestquest.core.admin_allowlist import is_admin

        await set_admin_ids(database, ["user-owner"])
        assert await is_admin(database, "user-owner") is True
        assert await is_admin(database, "user-other") is False
        # Unknown, malformed, and absent identities all fail closed.
        assert await is_admin(database, "") is False
        assert await is_admin(database, "  user-owner  ") is False, (
            "ids are stored trimmed; an untrimmed lookup is a different "
            "id and must not match"
        )
        assert await is_admin(database, None) is False
        assert await is_admin(database, 42) is False
        assert await is_admin(database, ["user-owner"]) is False
        assert await is_admin(database, True) is False
        return None

    _with_db(tmp_path, "resolver-matrix.db")(_body)


def test_is_admin_empty_allowlist_fails_closed(tmp_path) -> None:
    """An empty allowlist returns False for EVERYONE, not True."""
    async def _body(database):
        from custom_components.nestquest.core.admin_allowlist import is_admin

        for candidate in ("user-owner", "", None, 1):
            assert await is_admin(database, candidate) is False
        return None

    _with_db(tmp_path, "resolver-empty.db")(_body)


def test_is_admin_false_after_last_admin_removal_attempt(tmp_path) -> None:
    """The refusal keeps the verdict sound: a refused removal leaves
    the admin in place, so is_admin still resolves True."""
    async def _body(database):
        from custom_components.nestquest.core.admin_allowlist import is_admin

        await set_admin_ids(database, ["user-owner"])
        with pytest.raises(ValueError, match="at least one admin"):
            await set_admin_ids(database, [])
        assert await is_admin(database, "user-owner") is True
        return None

    _with_db(tmp_path, "resolver-refusal.db")(_body)