nanobot/tests/channels/test_base_channel.py

from types import SimpleNamespace

import pytest

from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel


class _DummyChannel(BaseChannel):
    name = "dummy"
    _sent: list[OutboundMessage]

    def __init__(self, config, bus):
        super().__init__(config, bus)
        self._sent = []

    async def start(self) -> None:
        return None

    async def stop(self) -> None:
        return None

    async def send(self, msg: OutboundMessage) -> None:
        self._sent.append(msg)


def test_is_allowed_requires_exact_match() -> None:
    channel = _DummyChannel(SimpleNamespace(allow_from=["allow@email.com"]), MessageBus())

    assert channel.is_allowed("allow@email.com") is True
    assert channel.is_allowed("attacker|allow@email.com") is False


def test_is_allowed_supports_dict_allow_from_alias() -> None:
    channel = _DummyChannel({"allowFrom": ["alice"]}, MessageBus())

    assert channel.is_allowed("alice") is True


def test_is_allowed_denies_empty_dict_allow_from() -> None:
    channel = _DummyChannel({"allow_from": []}, MessageBus())

    assert channel.is_allowed("alice") is False


def test_is_allowed_handles_none_allow_from() -> None:
    channel = _DummyChannel({"allow_from": None}, MessageBus())
    assert channel.is_allowed("alice") is False

    channel2 = _DummyChannel({"allowFrom": None}, MessageBus())
    assert channel2.is_allowed("alice") is False


def test_is_allowed_star_allows_all() -> None:
    channel = _DummyChannel({"allowFrom": ["*"]}, MessageBus())
    assert channel.is_allowed("anyone") is True


def test_is_allowed_pairing_fallback(monkeypatch) -> None:
    channel = _DummyChannel({"allowFrom": []}, MessageBus())
    monkeypatch.setattr(
        "nanobot.channels.base.is_approved", lambda _ch, sid: sid == "paired"
    )
    assert channel.is_allowed("paired") is True
    assert channel.is_allowed("unknown") is False


@pytest.mark.asyncio
async def test_handle_message_dm_sends_pairing_code(monkeypatch) -> None:
    channel = _DummyChannel({"allowFrom": []}, MessageBus())
    monkeypatch.setattr(
        "nanobot.channels.base.generate_code", lambda _ch, sid: "ABCD-EFGH"
    )

    await channel._handle_message(
        sender_id="stranger", chat_id="chat1", content="hello", is_dm=True
    )

    assert len(channel._sent) == 1
    msg = channel._sent[0]
    assert "ABCD-EFGH" in msg.content
    assert msg.metadata.get("_pairing_code") == "ABCD-EFGH"


@pytest.mark.asyncio
async def test_dm_during_transient_store_failure_keeps_approvals(
    tmp_path, monkeypatch
) -> None:
    """An unapproved DM while pairing.json is unreadable must not wipe approvals.

    The pairing store treated a transient OSError like corruption and returned
    an empty store; the DM pairing path then persisted that empty view,
    erasing every approved sender.
    """
    import builtins
    from pathlib import Path

    from nanobot.pairing import store

    path = tmp_path / "pairing.json"
    monkeypatch.setattr(store, "_store_path", lambda: path)
    code = store.generate_code("dummy", "friend")
    store.approve_code(code)

    channel = _DummyChannel({"allowFrom": []}, MessageBus())

    real_open = builtins.open

    def flaky_open(file, mode="r", *args, **kwargs):
        try:
            same = Path(file) == path
        except TypeError:
            same = False
        if same and "r" in mode and "+" not in mode:
            raise PermissionError(13, "temporarily locked", str(path))
        return real_open(file, mode, *args, **kwargs)

    with monkeypatch.context() as m:
        m.setattr(builtins, "open", flaky_open)
        await channel._handle_message(
            sender_id="stranger", chat_id="chat1", content="hello", is_dm=True
        )

    assert channel._sent == []
    assert store.is_approved("dummy", "friend") is True


@pytest.mark.asyncio
async def test_handle_message_group_ignores_unknown() -> None:
    channel = _DummyChannel({"allowFrom": []}, MessageBus())

    await channel._handle_message(
        sender_id="stranger", chat_id="chat1", content="hello", is_dm=False
    )

    assert channel._sent == []


@pytest.mark.asyncio
async def test_handle_message_uses_authorization_id_without_changing_sender() -> None:
    bus = MessageBus()
    channel = _DummyChannel({"allowFrom": ["group@g.us"]}, bus)

    await channel._handle_message(
        sender_id="member-lid",
        authorization_id="group@g.us",
        chat_id="group@g.us",
        content="hello",
    )

    msg = await bus.consume_inbound()
    assert msg.sender_id == "member-lid"
    assert msg.chat_id == "group@g.us"


@pytest.mark.asyncio
async def test_handle_message_rejects_when_authorization_id_is_not_allowed() -> None:
    bus = MessageBus()
    channel = _DummyChannel({"allowFrom": ["member-lid"]}, bus)

    await channel._handle_message(
        sender_id="member-lid",
        authorization_id="other-group@g.us",
        chat_id="other-group@g.us",
        content="hello",
    )

    assert bus.inbound_size == 0