minecraft-bot/tests/test_modern_chat.py

import contextlib
import io
import struct
import unittest
import uuid

from munch import Munch
from minecraft.networking.modern import ClientboundPlayerChatPacket

from mosfet.commands import Commands
from mosfet.connect import component_text
from mosfet.protocol.modern import (
    ModernChatManager, parse_incoming_private_message,
)


class FakeConnection:
    def register_packet_listener(self, *_args, **_kwargs):
        pass


class FakeChat:
    def __init__(self):
        self.messages = []

    def set_handler(self, _handler):
        pass

    def send(self, message):
        self.messages.append(message)


class TestModernChat(unittest.TestCase):
    def test_incoming_dm_translation_is_rendered_and_parsed(self):
        rendered = component_text({
            'translate': 'commands.message.display.incoming',
            'with': ['tanner', '!ping'],
        })
        self.assertEqual(rendered, 'tanner whispers to you: !ping')
        self.assertEqual(parse_incoming_private_message(rendered),
                         ('tanner', '!ping'))
        self.assertEqual(parse_incoming_private_message(
            'tanner -> me: 1ping'), ('tanner', '1ping'))

    def test_player_chat_console_uses_name_but_keeps_uuid_for_commands(self):
        player_uuid = 'f6126022-2c68-37f7-be93-aaa551374663'
        state = Munch(
            connection=FakeConnection(),
            player_names={player_uuid: 'tanner'},
            reload_lock=None,
        )
        manager = ModernChatManager(state)
        received = []
        manager.set_handler(received.append)
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            manager.handle_player_chat(Munch(
                sender=player_uuid, message='!ping', chat_type_id=1,
                network_target_name_present=False))

        self.assertEqual(output.getvalue(), '[chat] tanner: !ping\n')
        self.assertEqual(received, [
            ('CHAT', player_uuid, 'tanner: !ping'),
        ])

    def test_player_chat_target_name_marks_dm_and_routes_reply(self):
        player_uuid = 'f6126022-2c68-37f7-be93-aaa551374663'
        state = Munch(
            connection=FakeConnection(),
            player_names={player_uuid: 'tanner'},
            reload_lock=None,
        )
        manager = ModernChatManager(state)
        received = []
        manager.set_handler(received.append)
        output = io.StringIO()
        packet = Munch(
            sender=player_uuid, message='!ping', chat_type_id=2,
            network_target_name_present=True)
        with contextlib.redirect_stdout(output):
            manager.handle_player_chat(packet)

        self.assertEqual(output.getvalue(), '[chat] [DM] tanner: !ping\n')
        self.assertEqual(received, [
            ('DM', player_uuid, 'tanner -> me: !ping', 'tanner'),
        ])

    def test_player_chat_parser_reads_optional_target_name(self):
        def varint(value):
            encoded = bytearray()
            while value > 0x7f:
                encoded.append((value & 0x7f) | 0x80)
                value >>= 7
            encoded.append(value)
            return bytes(encoded)

        message = b'!ping'
        payload = (
            varint(0) + uuid.UUID(int=0).bytes + varint(0) + b'\x00' +
            varint(len(message)) + message +
            struct.pack('>qq', 0, 0) + varint(0) + b'\x00' +
            varint(0) + varint(2) + b'\x00\x01\x00'
        )
        packet = ClientboundPlayerChatPacket()
        packet.read(io.BytesIO(payload))

        self.assertEqual(packet.message, '!ping')
        self.assertEqual(packet.chat_type_id, 2)
        self.assertTrue(packet.network_target_name_present)

    def test_chat_type_registry_identifies_dm_without_target_name(self):
        player_uuid = 'f6126022-2c68-37f7-be93-aaa551374663'
        state = Munch(
            connection=FakeConnection(),
            player_names={player_uuid: 'tanner'},
            reload_lock=None,
        )
        state.connection.registry_ids = {
            'minecraft:chat_type': {
                'minecraft:msg_command_incoming': 2,
            },
        }
        manager = ModernChatManager(state)
        received = []
        manager.set_handler(received.append)
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            manager.handle_player_chat(Munch(
                sender=player_uuid, message='!ping', chat_type_id=2,
                network_target_name_present=False))

        self.assertEqual(output.getvalue(), '[chat] [DM] tanner: !ping\n')
        self.assertEqual(received, [
            ('DM', player_uuid, 'tanner -> me: !ping', 'tanner'),
        ])

    def test_system_dm_console_uses_dm_prefix(self):
        state = Munch(connection=FakeConnection(), reload_lock=None)
        manager = ModernChatManager(state)
        received = []
        manager.set_handler(received.append)
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            handled = manager.handle_system_chat(
                None, 'tanner whispers to you: !ping')

        self.assertTrue(handled)
        self.assertEqual(output.getvalue(), '[chat] [DM] tanner: !ping\n')
        self.assertEqual(received, [
            ('DM', 'tanner', 'tanner -> me: !ping', 'tanner'),
        ])

    def test_dm_command_replies_to_named_sender(self):
        chat = FakeChat()
        command = Commands.__new__(Commands)
        command.g = Munch(name='mosfet1', chat=chat)

        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            command.handle_chat((
                'DM', 'f6126022-2c68-37f7-be93-aaa551374663',
                'tanner -> me: !ping', 'tanner'))

        self.assertIn('Reply: pong', output.getvalue())
        self.assertEqual(chat.messages, ['/msg tanner pong'])

    def test_here_dm_uses_sender_uuid_and_private_reply_name(self):
        sender_uuid = 'f6126022-2c68-37f7-be93-aaa551374663'
        chat = FakeChat()
        command = Commands.__new__(Commands)
        command.g = Munch(
            name='mosfet1', chat=chat, pos=(0, 64, 0), path=[], job=None,
            players={1: Munch(player_uuid=sender_uuid, x=10, y=64, z=20)},
            world=Munch(path_to_place=lambda start, goal: [start, goal]),
        )

        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            command.handle_chat((
                'DM', sender_uuid, 'tanner -> me: 1here', 'tanner'))

        self.assertEqual(command.g.path, [(0, 64, 0), (10, 64, 20)])
        self.assertIn('Reply: ok', output.getvalue())
        self.assertEqual(chat.messages, ['/msg tanner ok'])


if __name__ == '__main__':
    unittest.main()