How to use _on_message method in tavern

Best Python code snippet using tavern

test_command_decorator.py

Source:test_command_decorator.py

...79        create_unique_id() for x in range(0, numusers)80    ]81    for user_id in user_ids:82        message.author.id = user_id83        await plugin._on_message(message)84    # Test for admin running the command85    message.author.id = admin_id = create_unique_id()86    message.author.guild_permissions.administrator = True87    await plugin._on_message(message)88    message.author.guild_permissions.administrator = False89    # Test for owner running the command.90    message.author.id = 6615385382480281691    await plugin._on_message(message)92    return user_ids, admin_id93@pytest.mark.asyncio94async def test_owner_only(dummy_plugin, message, caplog):95    await run_normal_admin_owner("!owneronly", dummy_plugin, message)96    # Assert all possibilities happened.97    assert sum(1 for x in caplog.records if "à² _à² " in x.msg) == 298    assert any("owner=true" in x.msg for x in caplog.records)99@pytest.mark.asyncio100async def test_nonglobal_command(dummy_plugin, message, caplog):101    message.content = "!nonglobal"102    await run_normal_admin_owner("!nonglobal", dummy_plugin, message, numusers=4)103    # Assert that tests has been successful.104    assert sum(1 for x in caplog.records if "You are not allowed" in x.msg) == 4105    assert sum(1 for x in caplog.records if "AUTHORIZED" in x.msg) == 2106@pytest.mark.asyncio107async def test_granted_permissions(monkeypatch, dummy_plugin, message, caplog):108    async def patched_perms(*args, **kwargs):109        return ["test.restricted"]110    monkeypatch.setattr("homura.lib.permissions.Permissions.get_perms", patched_perms)111    await run_normal_admin_owner("!nonglobal", dummy_plugin, message, numusers=4)112    # Assert that tests has been successful.113    assert sum(1 for x in caplog.records if "You are not allowed" in x.msg) == 0114    assert sum(1 for x in caplog.records if "AUTHORIZED" in x.msg) == 6115@pytest.mark.asyncio116async def test_denied_permissions(monkeypatch, dummy_plugin, message, caplog):117    async def patched_perms(*args, **kwargs):118        return ["-test"]119    monkeypatch.setattr("homura.lib.permissions.Permissions.get_perms", patched_perms)120    await run_normal_admin_owner("!kek", dummy_plugin, message, numusers=4)121    # Assert that tests has been successful.122    assert sum(1 for x in caplog.records if "You are not allowed" in x.msg) == 4123    assert sum(1 for x in caplog.records if "SINGLEPATTERN" in x.msg) == 2124"""125Command trigger tests.126* Tagging the bot with the normal prefix127* Tagging the bot with its mention.128* Multiple pattern commands with a match129* Single pattern commands130"""131@pytest.mark.asyncio132async def test_multi_triggers_normal_and_mention(bot, dummy_plugin, message, caplog):133    # Test for the normal ! prefix.134    message.content = "!test"135    await dummy_plugin._on_message(message)136    # Test for the bot's mention as a prefix.137    message.content = f"<@{dummy_plugin.bot.user.id}> catch ARGUMENT_REAL"138    await dummy_plugin._on_message(message)139    # Test for the bot's nickname mention as a prefix.140    message.content = f"<@!{dummy_plugin.bot.user.id}> catch ARGUMENT_NICK"141    await dummy_plugin._on_message(message)142    # Assert that tests has been successful.143    assert_has = ["TESTCOMMAND", "ARGUMENT_REAL", "ARGUMENT_NICK"]144    for has in assert_has:145        assert any(has in x.msg for x in caplog.records)146@pytest.mark.asyncio147async def test_single_triggers_normal_and_mention(bot, dummy_plugin, message, caplog):148    # Test for the normal ! prefix.149    message.content = "!kek"150    await dummy_plugin._on_message(message)151    # Test for the bot's mention as a prefix.152    message.content = f"<@{dummy_plugin.bot.user.id}> kek"153    await dummy_plugin._on_message(message)154    # Test for the bot's nickname mention as a prefix.155    message.content = f"<@!{dummy_plugin.bot.user.id}> kek"156    await dummy_plugin._on_message(message)157    # Assert that test_command has been executed three times....

test_websocket.py

Source:test_websocket.py

...70    # connect external to internal ID71    handler.external_registry[72        "e0b0f42177341bff16987e1b12904d7dc8a6e4417dee0291fa134b5044763482"73    ] = list(handler.internal_registry.keys())[0]74    handler._on_message(None, DATA_RESPONSE)75    callback_mock.assert_called_with(None, json.loads(DATA_RESPONSE))76def test_on_message_subscription():77    handler = get_handler()78    assert_handler_initialized(handler)79    # add request ID to expected responses80    handler.expected_ids = {1}81    handler._on_message(None, SUBSCRIPTION_RESPONSE)82    # acknowledges, we don't expect another internal ID to come in83    assert handler.expected_ids == set()84    # successfully linked external to internal ID85    assert (86        handler.external_registry[87            "242d29d5c0ec9268f51a39aba4ed6a36c757c03c183633568edb0531658a9799"88        ]89        == 190    )91def test_on_message_unsubscription():92    handler = get_handler()93    assert_handler_initialized(handler)94    # add request ID to expected responses95    handler.expected_ids = {1}96    handler._on_message(None, UNSUBSCRIPTION_RESPONSE)97    # acknowledges, we don't expect another internal ID to come in98    assert handler.expected_ids == set()99def test_on_message_unknown():100    handler = get_handler()101    assert_handler_initialized(handler)102    with pytest.raises(APIError):103        handler._on_message(None, UNKNOWN_RESPONSE)104def test_on_message_invalid():105    handler = get_handler()106    assert_handler_initialized(handler)107    with pytest.raises(JSONDecodeError):108        handler._on_message(None, "invalid")109@pytest.mark.parametrize(110    "hook,args",111    (("on_open", (None,)), ("on_close", (None,)), ("on_error", (None, None))),112)113def test_custom_callbacks(hook, args):114    handler = get_handler()115    setattr(handler, hook, Mock())116    # simulate callbacks from websocket client117    getattr(handler, f"_{hook}")(*args)118    getattr(handler, hook).assert_called_with(*args)119def test_subscriptions_on_open():120    handler = get_handler()121    mock_callback = Mock()122    assert_handler_initialized(handler)...

signaling.py

Source:signaling.py

...40        self._websock = websock41        async for message in websock:42            data = json.loads(message)43            if self._on_message:44                await self._on_message(data)45# ÐÐ»Ð°ÑÑ Ð´Ð»Ñ ÑÐ¾Ð·Ð´Ð°Ð½Ð¸Ñ ÑÐ¸Ð³Ð½Ð°Ð»ÑÐ½Ð¾Ð³Ð¾ ÐºÐ°Ð½Ð°Ð»Ð°46class WebSocketClient(WebSocketBasic):47    def __init__(self, server, port):48        super().__init__()49        uri = f"wss://{server}:{port}"50        asyncio.get_event_loop().create_task(self.__connect(uri))51    async def __connect(self, uri):52        async with websockets.connect(uri, ssl=True) as self._websock:53            logger.info(f"Connected {self._websock.remote_address} websockets")54            # Ð°Ð²ÑÐ¾ÑÐ¸Ð·Ð°ÑÐ¸Ñ55            with open("rsa_key", "rb") as key_file:56                private_key = serialization.load_pem_private_key(key_file.read(), password=None)57            encrypted_key = await self._websock.recv()58            try:59                decrypted_key = private_key.decrypt(encrypted_key, padding.OAEP(60                        mgf=padding.MGF1(algorithm=hashes.SHA256()),61                        algorithm=hashes.SHA256(),62                        label=None63                    )64                )65            except ValueError:66                logger.error(f"Authentication failed")67                await self._websock.close()68                return69            await self._websock.send(decrypted_key)70            # Ð¿ÑÐ¾Ð´Ð¾Ð»Ð¶ÐµÐ½Ð¸Ðµ ÑÐ°Ð±Ð¾ÑÑ71            logger.info(f"Authentication succeed")72            if self._on_connected:73                await self._on_connected(self._websock)74            async for message in self._websock:75                data = json.loads(message)76                if self._on_message:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.