Unlock 30% off on Manual Testing Annual Plans this Holiday Season. 
Use Code: LCMYEARENDEXCL30
Use Code: LCMYEARENDEXCL30Check icon
Copied to Clipboard!
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run tavern automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
