shawn
3ddf8170f8
- rooms: create, join, leave, forget, list, resolve alias, get info/members, invite/kick/ban/unban, set name/topic. RoomInfo/MemberInfo/PowerLevels dataclasses. - messaging: send text (with HTML, reply, thread), notice, emote, image, file, sticker, edit, redact, react, typing, read receipts, fully-read markers. - sync: SyncEngine with event-type/room/global handler dispatch, incremental sync with backoff retry, SyncFilter, paginated message history. - CLI: rooms (list/create/join/leave/info/members/invite), send (text/image/file), sync (listen with filters). - 182 tests pass (60 existing + 122 new).
303 lines
10 KiB
Python
303 lines
10 KiB
Python
"""Tests for matrix_hermes.cli module."""
|
|
|
|
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from matrix_hermes.cli import (
|
|
_mask_token,
|
|
build_parser,
|
|
cmd_check,
|
|
cmd_login,
|
|
cmd_logout,
|
|
cmd_show_config,
|
|
cmd_whoami,
|
|
main,
|
|
)
|
|
|
|
|
|
class TestMaskToken:
|
|
"""Tests for _mask_token helper."""
|
|
|
|
def test_short_token(self):
|
|
assert _mask_token("abc") == "***"
|
|
|
|
def test_long_token(self):
|
|
result = _mask_token("syt_abcdefghijklmnopqrstuvwxyz")
|
|
assert result.startswith("syt_ab")
|
|
assert result.endswith("uvwxyz")
|
|
assert "..." in result
|
|
|
|
def test_exact_length(self):
|
|
"""12-char token with visible=6: len<=12 means fully masked."""
|
|
result = _mask_token("123456789012")
|
|
assert result == "************"
|
|
|
|
def test_just_over_threshold(self):
|
|
"""13-char token with visible=6: len>12 means partial mask."""
|
|
result = _mask_token("1234567890123")
|
|
assert result == "123456...890123"
|
|
|
|
|
|
class TestParser:
|
|
"""Tests for CLI argument parser."""
|
|
|
|
def test_build_parser(self):
|
|
parser = build_parser()
|
|
assert "login" in parser._subparsers._group_actions[0].choices
|
|
assert "whoami" in parser._subparsers._group_actions[0].choices
|
|
assert "check" in parser._subparsers._group_actions[0].choices
|
|
assert "show-config" in parser._subparsers._group_actions[0].choices
|
|
|
|
def test_no_args_shows_help(self, capsys):
|
|
"""Running with no args shows help."""
|
|
result = main([])
|
|
# Should not crash
|
|
captured = capsys.readouterr()
|
|
assert "usage" in captured.out.lower() or result == 0
|
|
|
|
|
|
class TestCmdCheck:
|
|
"""Tests for 'check' command."""
|
|
|
|
def test_not_configured(self, clean_env, capsys):
|
|
"""When no Matrix vars are set, check should fail."""
|
|
args = build_parser().parse_args(["check"])
|
|
result = cmd_check(args)
|
|
assert result == 1
|
|
captured = capsys.readouterr()
|
|
assert "not configured" in captured.out.lower()
|
|
|
|
def test_configured(self, mock_env, capsys):
|
|
"""When Matrix vars are set, check should succeed."""
|
|
args = build_parser().parse_args(["check"])
|
|
result = cmd_check(args)
|
|
assert result == 0
|
|
captured = capsys.readouterr()
|
|
assert "valid" in captured.out.lower()
|
|
|
|
|
|
class TestCmdShowConfig:
|
|
"""Tests for 'show-config' command."""
|
|
|
|
def test_output_format(self, mock_env, capsys):
|
|
"""show-config outputs env var format."""
|
|
args = build_parser().parse_args(["show-config"])
|
|
result = cmd_show_config(args)
|
|
assert result == 0
|
|
captured = capsys.readouterr()
|
|
assert "MATRIX_HOMESERVER=" in captured.out
|
|
assert "MATRIX_ACCESS_TOKEN=" in captured.out
|
|
|
|
def test_redacted(self, mock_env, capsys):
|
|
"""show-config --redact masks sensitive values."""
|
|
args = build_parser().parse_args(["show-config", "--redact"])
|
|
result = cmd_show_config(args)
|
|
assert result == 0
|
|
captured = capsys.readouterr()
|
|
# Token should be masked
|
|
assert "syt_test_token" not in captured.out
|
|
assert "syt_test" in captured.out or "..." in captured.out
|
|
|
|
|
|
class TestCmdLogin:
|
|
"""Tests for 'login' command."""
|
|
|
|
def test_no_homeserver(self, clean_env, capsys):
|
|
"""Login fails without homeserver."""
|
|
args = build_parser().parse_args(["login"])
|
|
result = cmd_login(args)
|
|
assert result == 1
|
|
|
|
def test_with_valid_token(self, mock_env, capsys):
|
|
"""Login with valid token prints success."""
|
|
with patch("urllib.request.urlopen") as mock_urlopen:
|
|
mock_resp = MagicMock()
|
|
mock_resp.read.return_value = json.dumps(
|
|
{"user_id": "@bot:example.org", "device_id": "DEV1"}
|
|
).encode()
|
|
mock_resp.__enter__.return_value = mock_resp
|
|
mock_urlopen.return_value = mock_resp
|
|
|
|
args = build_parser().parse_args(["login"])
|
|
result = cmd_login(args)
|
|
assert result == 0
|
|
|
|
def test_with_bad_credentials(self, mock_env, capsys):
|
|
"""Login with bad credentials prints error."""
|
|
with patch("urllib.request.urlopen") as mock_urlopen:
|
|
mock_resp = MagicMock()
|
|
mock_resp.read.return_value = json.dumps(
|
|
{"errcode": "M_FORBIDDEN", "error": "Bad password"}
|
|
).encode()
|
|
mock_resp.__enter__.return_value = mock_resp
|
|
mock_urlopen.return_value = mock_resp
|
|
|
|
# Override with password-based login (no valid token)
|
|
import os
|
|
|
|
os.environ["MATRIX_ACCESS_TOKEN"] = ""
|
|
os.environ["MATRIX_PASSWORD"] = "wrong"
|
|
|
|
args = build_parser().parse_args(["login"])
|
|
result = cmd_login(args)
|
|
assert result == 1
|
|
|
|
|
|
class TestCmdWhoami:
|
|
"""Tests for 'whoami' command."""
|
|
|
|
def test_no_homeserver(self, clean_env, capsys):
|
|
"""whoami fails without homeserver."""
|
|
args = build_parser().parse_args(["whoami"])
|
|
result = cmd_whoami(args)
|
|
assert result == 1
|
|
|
|
def test_valid(self, mock_env, capsys):
|
|
"""whoami with valid token."""
|
|
with patch("urllib.request.urlopen") as mock_urlopen:
|
|
mock_resp = MagicMock()
|
|
mock_resp.read.return_value = json.dumps(
|
|
{"user_id": "@bot:example.org"}
|
|
).encode()
|
|
mock_resp.__enter__.return_value = mock_resp
|
|
mock_urlopen.return_value = mock_resp
|
|
|
|
args = build_parser().parse_args(["whoami"])
|
|
result = cmd_whoami(args)
|
|
assert result == 0
|
|
|
|
def test_invalid(self, mock_env, capsys):
|
|
"""whoami with invalid token."""
|
|
with patch("urllib.request.urlopen") as mock_urlopen:
|
|
mock_resp = MagicMock()
|
|
mock_resp.read.return_value = json.dumps(
|
|
{"errcode": "M_UNKNOWN_TOKEN", "error": "Unrecognised"}
|
|
).encode()
|
|
mock_resp.__enter__.return_value = mock_resp
|
|
mock_urlopen.return_value = mock_resp
|
|
|
|
args = build_parser().parse_args(["whoami"])
|
|
result = cmd_whoami(args)
|
|
assert result == 1
|
|
|
|
|
|
class TestCmdLogout:
|
|
"""Tests for 'logout' command."""
|
|
|
|
def test_no_homeserver(self, clean_env, capsys):
|
|
"""logout fails without homeserver."""
|
|
args = build_parser().parse_args(["logout"])
|
|
result = cmd_logout(args)
|
|
assert result == 1
|
|
|
|
def test_success(self, mock_env, capsys):
|
|
"""logout with valid token."""
|
|
with patch("urllib.request.urlopen") as mock_urlopen:
|
|
mock_resp = MagicMock()
|
|
mock_resp.read.return_value = json.dumps({}).encode()
|
|
mock_resp.__enter__.return_value = mock_resp
|
|
mock_urlopen.return_value = mock_resp
|
|
|
|
args = build_parser().parse_args(["logout"])
|
|
result = cmd_logout(args)
|
|
assert result == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# New async command parser tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNewCommandsParser:
|
|
"""Tests for parser entries of rooms, send, sync commands."""
|
|
|
|
def test_rooms_subparser_exists(self):
|
|
parser = build_parser()
|
|
choices = parser._subparsers._group_actions[0].choices
|
|
assert "rooms" in choices
|
|
assert "send" in choices
|
|
assert "sync" in choices
|
|
|
|
def test_rooms_list(self):
|
|
args = build_parser().parse_args(["rooms", "list"])
|
|
assert args.command == "rooms"
|
|
assert args.rooms_action == "list"
|
|
|
|
def test_rooms_create(self):
|
|
args = build_parser().parse_args(
|
|
["rooms", "create", "--name", "Test", "--topic", "Hello", "--public"]
|
|
)
|
|
assert args.rooms_action == "create"
|
|
assert args.name == "Test"
|
|
assert args.topic == "Hello"
|
|
assert args.public is True
|
|
|
|
def test_rooms_join(self):
|
|
args = build_parser().parse_args(["rooms", "join", "!room:example.org"])
|
|
assert args.rooms_action == "join"
|
|
assert args.room == "!room:example.org"
|
|
|
|
def test_rooms_leave(self):
|
|
args = build_parser().parse_args(["rooms", "leave", "!room:example.org"])
|
|
assert args.rooms_action == "leave"
|
|
assert args.room == "!room:example.org"
|
|
|
|
def test_rooms_info(self):
|
|
args = build_parser().parse_args(["rooms", "info", "!room:example.org"])
|
|
assert args.rooms_action == "info"
|
|
assert args.room == "!room:example.org"
|
|
|
|
def test_rooms_members(self):
|
|
args = build_parser().parse_args(["rooms", "members", "!room:example.org"])
|
|
assert args.rooms_action == "members"
|
|
|
|
def test_rooms_invite(self):
|
|
args = build_parser().parse_args(
|
|
["rooms", "invite", "!room:example.org", "@alice:example.org"]
|
|
)
|
|
assert args.rooms_action == "invite"
|
|
assert args.user == "@alice:example.org"
|
|
|
|
def test_send_text(self):
|
|
args = build_parser().parse_args(["send", "text", "!room:example.org", "Hello world"])
|
|
assert args.send_action == "text"
|
|
assert args.room == "!room:example.org"
|
|
assert args.body == "Hello world"
|
|
|
|
def test_send_text_with_reply(self):
|
|
args = build_parser().parse_args(
|
|
["send", "text", "!room:example.org", "Replying", "--reply-to", "$evt:example.org"]
|
|
)
|
|
assert args.reply_to == "$evt:example.org"
|
|
|
|
def test_send_image(self):
|
|
args = build_parser().parse_args(
|
|
["send", "image", "!room:example.org", "/tmp/photo.jpg", "--caption", "My photo"]
|
|
)
|
|
assert args.send_action == "image"
|
|
assert args.path == "/tmp/photo.jpg"
|
|
assert args.caption == "My photo"
|
|
|
|
def test_send_file(self):
|
|
args = build_parser().parse_args(
|
|
["send", "file", "!room:example.org", "/tmp/doc.pdf"]
|
|
)
|
|
assert args.send_action == "file"
|
|
|
|
def test_sync_listen(self):
|
|
args = build_parser().parse_args(["sync", "listen"])
|
|
assert args.sync_action == "listen"
|
|
assert args.filter_type == []
|
|
assert args.room == []
|
|
assert args.show_all is False
|
|
|
|
def test_sync_listen_with_filters(self):
|
|
args = build_parser().parse_args(
|
|
["sync", "listen", "--filter-type", "m.room.message", "--room", "!test:example.org", "--show-all"]
|
|
)
|
|
assert args.filter_type == ["m.room.message"]
|
|
assert args.room == ["!test:example.org"]
|
|
assert args.show_all is True
|