Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use safe_send method in Airtest

Best Python code snippet using Airtest

events.py

Source:events.py

...20            ctx, error, "The response had an invalid body, likely due to length."21        )22    if error.code == 30003:23        # pin limit24        await safe_send(ctx, "Uh oh! The pin limit has been reached.")25        raise error26async def _generic_error_handler(ctx: Context, error, text):27    await safe_send(ctx, f"Uh oh! {text} {_report_bug_message}")28    if safe_bug_report(ctx):29        await safe_send(ctx, f"```{error}```")30    raise error31def _update_player_members(bot, after):32    """Update player members when they change."""33    try:34        player = get_player(bot.game, after)35        player.member = after36    except PlayerNotFoundError:37        pass38def _update_storyteller_list(bot, after, before):39    """Add new storytellers to the Storyteller list."""40    if bot.storyteller_role not in before.roles and bot.storyteller_role in after.roles:41        bot.game.storytellers.append(Player(after, Storyteller, None))42    bot.backup()43class Events(commands.Cog):44    """Listeners for event handling."""45    def __init__(self, bot: BOTCBot):46        self.bot = bot47    @commands.Cog.listener()48    async def on_ready(self):49        """Run immediately after startup."""50        print("Logged in as", self.bot.user.name)51        print("ID:", self.bot.user.id)52        print("Server:", self.bot.server)53        print("Gameplay Channel: #", self.bot.channel.name)54        # restore backups55        await self.bot.restore_backup()56        # update status57        await self.bot.update_status()58        print("------")59    @commands.Cog.listener()60    async def on_command_error(self, ctx: Context, error: Exception):61        """Handle command errors."""62        # Ignore commands with local handling63        if hasattr(ctx.command, "on_error"):64            return65        # Check original exceptions for commands.CommandInvokeError66        error = getattr(error, "original", error)67        if isinstance(error, commands.CheckFailure):68            # commands which fail contextual checks69            if isinstance(error, commands.NotOwner):70                # commands.is_owner()71                return await safe_send(72                    ctx, "Stop trying to play around with debug tools, please! ;)"73                )74            if str(error):75                # most checks76                return await safe_send(ctx, str(error))77            # checks.is_dm() and checks.is_in_channel()78            return79        if isinstance(error, ValueError):80            # value errors81            if str(error) == "cancelled":82                # raised by lib.utils.get_input if input is cancel83                await safe_send(ctx, "Cancelled!")84                return85            if str(error) == "command called":86                # raised by lib.utils.get_input if another command is called87                return88        elif isinstance(error, HTTPException):89            # errors in HTTP request operations90            await _http_error_handler(ctx, error)91        elif isinstance(error, SyntaxError):92            # syntax errors93            await _generic_error_handler(94                ctx, error, "There's a syntax error somewhere."95            )96        elif isinstance(error, asyncio.TimeoutError):97            # timeout error98            return await safe_send(ctx, "Timed out.")99        elif isinstance(error, commands.CommandNotFound):100            # unknown commands101            if ctx.guild is None:102                return await safe_send(103                    ctx,104                    (105                        f"Command not found: {ctx.invoked_with}. "106                        f"For a list of commands, try `{ctx.prefix}help`."107                    ),108                )109            return110        elif isinstance(error, commands.errors.MissingRequiredArgument):111            # not enough arguments112            await safe_send(113                ctx, "Uh oh! You're missing arguments. Hopefully this helps:"114            )115            return await ctx.send_help(ctx.command)116        elif isinstance(error, commands.BadArgument):117            # converter error118            return await safe_send(ctx, str(error))119        elif isinstance(error, commands.DisabledCommand):120            # disabled command121            return await safe_send(ctx, f"{ctx.command.name} has been disabled.")122        await _generic_error_handler(ctx, error, "An unknown error occurred.")123    @commands.Cog.listener()124    async def on_member_update(self, before, after):125        """Handle member updates."""126        if self.bot.game:127            # update player objects with changes128            _update_player_members(self.bot, after)129            # add new storytellers to the seating order130            _update_storyteller_list(self.bot, after, before)131    @commands.Cog.listener()132    async def on_message(self, message):133        """Handle messages."""134        if message.author.bot:135            return...

debug.py

Source:debug.py

...18        """Unpin all messages in channel."""19        with ctx.typing():20            for msg in await ctx.bot.channel.pins():21                await msg.unpin()22        await safe_send(ctx, "Unpinned all messages.")23    @commands.command(name="eval")24    @commands.is_owner()25    @checks.is_dm()26    async def _eval(self, ctx: Context, *, code: str):27        """Execute code, sending its output to ctx."""28        await safe_send(ctx, await aexec("return " + code, ctx))29    @commands.command(name="exec")30    @commands.is_owner()31    @checks.is_dm()32    async def _exec(self, ctx: Context, *, code: str):33        """Execute code."""34        await aexec(code, ctx)35    @commands.command(name="load")36    @commands.is_owner()37    @checks.is_dm()38    async def _load(self, ctx: Context, *, cog: str):39        """Load cog."""40        try:41            self.bot.load_extension(cog)42            await safe_send(ctx, "Load successful.")43        except commands.errors.ExtensionNotFound:44            try:45                self.bot.load_extension("lib.cogs." + cog)46                await safe_send(ctx, "Load successful.")47            except commands.errors.ExtensionNotFound:48                await safe_send(ctx, f"Extension not found: {cog}.")49    @commands.command(name="reload")50    @commands.is_owner()51    @checks.is_dm()52    async def _reload(self, ctx: Context, *, cog: str):53        """Reload cog."""54        try:55            self.bot.reload_extension(cog)56            await safe_send(ctx, "Reload successful.")57        except commands.errors.ExtensionNotLoaded:58            try:59                self.bot.reload_extension("lib.cogs." + cog)60                await safe_send(ctx, "Reload successful.")61            except commands.errors.ExtensionNotLoaded:62                await safe_send(ctx, f"Extension not loaded: {cog}.")63    @commands.command(name="unload")64    @commands.is_owner()65    @checks.is_dm()66    async def _unload(self, ctx: Context, *, cog: str):67        """Unload cog."""68        try:69            self.bot.unload_extension(cog)70            await safe_send(ctx, "Unload successful.")71        except commands.errors.ExtensionNotLoaded:72            try:73                self.bot.unload_extension("lib.cogs." + cog)74                await safe_send(ctx, "Unload successful.")75            except commands.errors.ExtensionNotLoaded:76                await safe_send(ctx, f"Extension not loaded: {cog}.")77    @commands.command(name="detailedgrimoire")78    @commands.is_owner()79    @checks.is_game()80    @checks.is_dm()81    async def _detailedgrimoire(self, ctx: GameContext):82        """Display a detailed text grimoire."""83        message_text = "**Grimoire:**"84        for player in ctx.bot.game.seating_order:85            message_text += f"\n\n__{player.epithet}:__"86            for effect in player.effects:87                message_text += f"\n{effect.name}:"88                message_text += f"\n> Source: {effect.source_player.epithet}"89                message_text += f"\n> Disabled: {effect.disabled}"90                statuses = []  # type: List[str]91                registers_statuses = []  # type: List[str]92                for status in status_list:93                    if effect.status(ctx.bot.game, status):94                        statuses.append(status)95                    if effect.registers_status(ctx.bot.game, status):96                        registers_statuses.append(status)97                message_text += "\n> Causes: "98                message_text += list_to_plural_string(statuses, "none")[0]99                message_text += "\n> Causes registering: "100                message_text += list_to_plural_string(registers_statuses, "none")[0]101        await safe_send(ctx, message_text)102def setup(bot: BOTCBot):103    """Set the cog up."""...

listener.py

Source:listener.py

...19                        try:20                            function, args, kwargs = request_params21                        except ValueError as e:22                            live_logger.critical(f"Invalid number of request paramters {request_params=}")23                            self.safe_send(exception_traceback(e) + str(request_params))24                            break25                        try:26                            rpc_result = getattr(self, function)(*args, **kwargs)27                        except Exception as e:28                            traceback = exception_traceback(e)29                            live_logger.critical(traceback)30                            self.safe_send(traceback)31                        else:32                            self.safe_send(rpc_result)33    def safe_send(self, data: object) -> None:34        try:35            self.conn.send(data)36        except OSError as e:37            live_logger.critical(str(e))38    def close_connection(self) -> None:39        self.conn.close()40    @staticmethod41    def preview_node(*args, **kwargs) -> Li:...

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.