evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix MatrixBot not resolving room aliases per-command (#106347)

Browse source
This commit is contained in:
Paarth Shah 2024-01-16 01:07:51 -08:00 • committed by GitHub
parent 28281523ec
commit 5afe155cd9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 246 additions and 79 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
homeassistant/components/matrix/__init__.py
View file

@ -223,9 +223,12 @@ class MatrixBot:
def _load_commands(self, commands: list[ConfigCommand]) -> None:
for command in commands:
# Set the command for all listening_rooms, unless otherwise specified.
command.setdefault(CONF_ROOMS, list(self._listening_rooms.values()))
if rooms := command.get(CONF_ROOMS):
command[CONF_ROOMS] = [self._listening_rooms[room] for room in rooms]
else:
command[CONF_ROOMS] = list(self._listening_rooms.values())
# COMMAND_SCHEMA guarantees that exactly one of CONF_WORD and CONF_expression are set.
# COMMAND_SCHEMA guarantees that exactly one of CONF_WORD and CONF_EXPRESSION are set.
if (word_command := command.get(CONF_WORD)) is not None:
for room_id in command[CONF_ROOMS]:
self._word_commands.setdefault(room_id, {})

72
tests/components/matrix/conftest.py
View file

@ -31,6 +31,8 @@ from homeassistant.components.matrix import (
CONF_WORD,
EVENT_MATRIX_COMMAND,
MatrixBot,
RoomAlias,
RoomAnyID,
RoomID,
)
from homeassistant.components.matrix.const import DOMAIN as MATRIX_DOMAIN
@ -51,13 +53,15 @@ from tests.common import async_capture_events
TEST_NOTIFIER_NAME = "matrix_notify"
TEST_HOMESERVER = "example.com"
TEST_DEFAULT_ROOM = "!DefaultNotificationRoom:example.com"
TEST_ROOM_A_ID = "!RoomA-ID:example.com"
TEST_ROOM_B_ID = "!RoomB-ID:example.com"
TEST_ROOM_B_ALIAS = "#RoomB-Alias:example.com"
TEST_JOINABLE_ROOMS = {
TEST_DEFAULT_ROOM = RoomID("!DefaultNotificationRoom:example.com")
TEST_ROOM_A_ID = RoomID("!RoomA-ID:example.com")
TEST_ROOM_B_ID = RoomID("!RoomB-ID:example.com")
TEST_ROOM_B_ALIAS = RoomAlias("#RoomB-Alias:example.com")
TEST_ROOM_C_ID = RoomID("!RoomC-ID:example.com")
TEST_JOINABLE_ROOMS: dict[RoomAnyID, RoomID] = {
TEST_ROOM_A_ID: TEST_ROOM_A_ID,
TEST_ROOM_B_ALIAS: TEST_ROOM_B_ID,
TEST_ROOM_C_ID: TEST_ROOM_C_ID,
}
TEST_BAD_ROOM = "!UninvitedRoom:example.com"
TEST_MXID = "@user:example.com"
@ -74,7 +78,7 @@ class _MockAsyncClient(AsyncClient):
async def close(self):
return None
async def room_resolve_alias(self, room_alias: str):
async def room_resolve_alias(self, room_alias: RoomAnyID):
if room_id := TEST_JOINABLE_ROOMS.get(room_alias):
return RoomResolveAliasResponse(
room_alias=room_alias, room_id=room_id, servers=[TEST_HOMESERVER]
@ -150,6 +154,16 @@ MOCK_CONFIG_DATA = {
CONF_EXPRESSION: "My name is (?P<name>.*)",
CONF_NAME: "ExpressionTriggerEventName",
},
{
CONF_WORD: "WordTriggerSubset",
CONF_NAME: "WordTriggerSubsetEventName",
CONF_ROOMS: [TEST_ROOM_B_ALIAS, TEST_ROOM_C_ID],
},
{
CONF_EXPRESSION: "Your name is (?P<name>.*)",
CONF_NAME: "ExpressionTriggerSubsetEventName",
CONF_ROOMS: [TEST_ROOM_B_ALIAS, TEST_ROOM_C_ID],
},
],
},
NOTIFY_DOMAIN: {
@ -164,15 +178,32 @@ MOCK_WORD_COMMANDS = {
"WordTrigger": {
"word": "WordTrigger",
"name": "WordTriggerEventName",
"rooms": [TEST_ROOM_A_ID, TEST_ROOM_B_ID],
"rooms": list(TEST_JOINABLE_ROOMS.values()),
}
},
TEST_ROOM_B_ID: {
"WordTrigger": {
"word": "WordTrigger",
"name": "WordTriggerEventName",
"rooms": [TEST_ROOM_A_ID, TEST_ROOM_B_ID],
}
"rooms": list(TEST_JOINABLE_ROOMS.values()),
},
"WordTriggerSubset": {
"word": "WordTriggerSubset",
"name": "WordTriggerSubsetEventName",
"rooms": [TEST_ROOM_B_ID, TEST_ROOM_C_ID],
},
},
TEST_ROOM_C_ID: {
"WordTrigger": {
"word": "WordTrigger",
"name": "WordTriggerEventName",
"rooms": list(TEST_JOINABLE_ROOMS.values()),
},
"WordTriggerSubset": {
"word": "WordTriggerSubset",
"name": "WordTriggerSubsetEventName",
"rooms": [TEST_ROOM_B_ID, TEST_ROOM_C_ID],
},
},
}
@ -181,15 +212,32 @@ MOCK_EXPRESSION_COMMANDS = {
{
"expression": re.compile("My name is (?P<name>.*)"),
"name": "ExpressionTriggerEventName",
"rooms": [TEST_ROOM_A_ID, TEST_ROOM_B_ID],
"rooms": list(TEST_JOINABLE_ROOMS.values()),
}
],
TEST_ROOM_B_ID: [
{
"expression": re.compile("My name is (?P<name>.*)"),
"name": "ExpressionTriggerEventName",
"rooms": [TEST_ROOM_A_ID, TEST_ROOM_B_ID],
}
"rooms": list(TEST_JOINABLE_ROOMS.values()),
},
{
"expression": re.compile("Your name is (?P<name>.*)"),
"name": "ExpressionTriggerSubsetEventName",
"rooms": [TEST_ROOM_B_ID, TEST_ROOM_C_ID],
},
],
TEST_ROOM_C_ID: [
{
"expression": re.compile("My name is (?P<name>.*)"),
"name": "ExpressionTriggerEventName",
"rooms": list(TEST_JOINABLE_ROOMS.values()),
},
{
"expression": re.compile("Your name is (?P<name>.*)"),
"name": "ExpressionTriggerSubsetEventName",
"rooms": [TEST_ROOM_B_ID, TEST_ROOM_C_ID],
},
],
}

180
tests/components/matrix/test_commands.py Normal file
View file

@ -0,0 +1,180 @@
"""Test MatrixBot's ability to parse and respond to commands in matrix rooms."""
from functools import partial
from itertools import chain
from typing import Any
from nio import MatrixRoom, RoomMessageText
from pydantic.dataclasses import dataclass
import pytest
from homeassistant.components.matrix import MatrixBot, RoomID
from homeassistant.core import Event, HomeAssistant
from tests.components.matrix.conftest import (
MOCK_EXPRESSION_COMMANDS,
MOCK_WORD_COMMANDS,
TEST_MXID,
TEST_ROOM_A_ID,
TEST_ROOM_B_ID,
TEST_ROOM_C_ID,
)
ALL_ROOMS = (TEST_ROOM_A_ID, TEST_ROOM_B_ID, TEST_ROOM_C_ID)
SUBSET_ROOMS = (TEST_ROOM_B_ID, TEST_ROOM_C_ID)
@dataclass
class CommandTestParameters:
"""Dataclass of parameters representing the command config parameters and expected result state.
Switches behavior based on `room_id` and `expected_event_room_data`.
"""
room_id: RoomID
room_message: RoomMessageText
expected_event_data_extra: dict[str, Any] | None
@property
def expected_event_data(self) -> dict[str, Any] | None:
"""Fully-constructed expected event data.
Commands that are named with 'Subset' are expected not to be read from Room A.
"""
if (
self.expected_event_data_extra is None
or "Subset" in self.expected_event_data_extra["command"]
and self.room_id not in SUBSET_ROOMS
):
return None
return {
"sender": "@SomeUser:example.com",
"room": self.room_id,
} | self.expected_event_data_extra
room_message_base = partial(
RoomMessageText,
formatted_body=None,
format=None,
source={
"event_id": "fake_event_id",
"sender": "@SomeUser:example.com",
"origin_server_ts": 123456789,
},
)
word_command_global = partial(
CommandTestParameters,
room_message=room_message_base(body="!WordTrigger arg1 arg2"),
expected_event_data_extra={
"command": "WordTriggerEventName",
"args": ["arg1", "arg2"],
},
)
expr_command_global = partial(
CommandTestParameters,
room_message=room_message_base(body="My name is FakeName"),
expected_event_data_extra={
"command": "ExpressionTriggerEventName",
"args": {"name": "FakeName"},
},
)
word_command_subset = partial(
CommandTestParameters,
room_message=room_message_base(body="!WordTriggerSubset arg1 arg2"),
expected_event_data_extra={
"command": "WordTriggerSubsetEventName",
"args": ["arg1", "arg2"],
},
)
expr_command_subset = partial(
CommandTestParameters,
room_message=room_message_base(body="Your name is FakeName"),
expected_event_data_extra={
"command": "ExpressionTriggerSubsetEventName",
"args": {"name": "FakeName"},
},
)
# Messages without commands should trigger nothing
fake_command_global = partial(
CommandTestParameters,
room_message=room_message_base(body="This is not a real command!"),
expected_event_data_extra=None,
)
# Valid commands sent by the bot user should trigger nothing
self_command_global = partial(
CommandTestParameters,
room_message=room_message_base(
body="!WordTrigger arg1 arg2",
source={
"event_id": "fake_event_id",
"sender": TEST_MXID,
"origin_server_ts": 123456789,
},
),
expected_event_data_extra=None,
)
@pytest.mark.parametrize(
"command_params",
chain(
(word_command_global(room_id) for room_id in ALL_ROOMS),
(expr_command_global(room_id) for room_id in ALL_ROOMS),
(word_command_subset(room_id) for room_id in SUBSET_ROOMS),
(expr_command_subset(room_id) for room_id in SUBSET_ROOMS),
),
)
async def test_commands(
hass: HomeAssistant,
matrix_bot: MatrixBot,
command_events: list[Event],
command_params: CommandTestParameters,
):
"""Test that the configured commands are used correctly."""
room = MatrixRoom(room_id=command_params.room_id, own_user_id=matrix_bot._mx_id)
await hass.async_start()
assert len(command_events) == 0
await matrix_bot._handle_room_message(room, command_params.room_message)
await hass.async_block_till_done()
# MatrixBot should emit exactly one Event with matching data from this Command
assert len(command_events) == 1
event = command_events[0]
assert event.data == command_params.expected_event_data
@pytest.mark.parametrize(
"command_params",
chain(
(word_command_subset(TEST_ROOM_A_ID),),
(expr_command_subset(TEST_ROOM_A_ID),),
(fake_command_global(room_id) for room_id in ALL_ROOMS),
(self_command_global(room_id) for room_id in ALL_ROOMS),
),
)
async def test_non_commands(
hass: HomeAssistant,
matrix_bot: MatrixBot,
command_events: list[Event],
command_params: CommandTestParameters,
):
"""Test that normal/non-qualifying messages don't wrongly trigger commands."""
room = MatrixRoom(room_id=command_params.room_id, own_user_id=matrix_bot._mx_id)
await hass.async_start()
assert len(command_events) == 0
await matrix_bot._handle_room_message(room, command_params.room_message)
await hass.async_block_till_done()
# MatrixBot should not treat this message as a Command
assert len(command_events) == 0
async def test_commands_parsing(hass: HomeAssistant, matrix_bot: MatrixBot):
"""Test that the configured commands were parsed correctly."""
await hass.async_start()
assert matrix_bot._word_commands == MOCK_WORD_COMMANDS
assert matrix_bot._expression_commands == MOCK_EXPRESSION_COMMANDS

66
tests/components/matrix/test_matrix_bot.py
View file

@ -1,5 +1,4 @@
"""Configure and test MatrixBot."""
from nio import MatrixRoom, RoomMessageText
from homeassistant.components.matrix import (
DOMAIN as MATRIX_DOMAIN,
@ -9,12 +8,7 @@ from homeassistant.components.matrix import (
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
from homeassistant.core import HomeAssistant
from .conftest import (
MOCK_EXPRESSION_COMMANDS,
MOCK_WORD_COMMANDS,
TEST_NOTIFIER_NAME,
TEST_ROOM_A_ID,
)
from .conftest import TEST_NOTIFIER_NAME
async def test_services(hass: HomeAssistant, matrix_bot: MatrixBot):
@ -29,61 +23,3 @@ async def test_services(hass: HomeAssistant, matrix_bot: MatrixBot):
# Verify that the matrix notifier is registered
assert (notify_service := services.get(NOTIFY_DOMAIN))
assert TEST_NOTIFIER_NAME in notify_service
async def test_commands(hass, matrix_bot: MatrixBot, command_events):
"""Test that the configured commands were parsed correctly."""
await hass.async_start()
assert len(command_events) == 0
assert matrix_bot._word_commands == MOCK_WORD_COMMANDS
assert matrix_bot._expression_commands == MOCK_EXPRESSION_COMMANDS
room_id = TEST_ROOM_A_ID
room = MatrixRoom(room_id=room_id, own_user_id=matrix_bot._mx_id)
# Test single-word command.
word_command_message = RoomMessageText(
body="!WordTrigger arg1 arg2",
formatted_body=None,
format=None,
source={
"event_id": "fake_event_id",
"sender": "@SomeUser:example.com",
"origin_server_ts": 123456789,
},
)
await matrix_bot._handle_room_message(room, word_command_message)
await hass.async_block_till_done()
assert len(command_events) == 1
event = command_events.pop()
assert event.data == {
"command": "WordTriggerEventName",
"sender": "@SomeUser:example.com",
"room": room_id,
"args": ["arg1", "arg2"],
}
# Test expression command.
room = MatrixRoom(room_id=room_id, own_user_id=matrix_bot._mx_id)
expression_command_message = RoomMessageText(
body="My name is FakeName",
formatted_body=None,
format=None,
source={
"event_id": "fake_event_id",
"sender": "@SomeUser:example.com",
"origin_server_ts": 123456789,
},
)
await matrix_bot._handle_room_message(room, expression_command_message)
await hass.async_block_till_done()
assert len(command_events) == 1
event = command_events.pop()
assert event.data == {
"command": "ExpressionTriggerEventName",
"sender": "@SomeUser:example.com",
"room": room_id,
"args": {"name": "FakeName"},
}