KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use cmd_logs method in localstack

Best Python code snippet using localstack_python

interactive.py

Source:interactive.py

1# Copyright (C) 2003-2007  Robey Pointer <robeypointer@gmail.com>2#3# This file is part of paramiko.4#5# Paramiko is free software; you can redistribute it and/or modify it under the6# terms of the GNU Lesser General Public License as published by the Free7# Software Foundation; either version 2.1 of the License, or (at your option)8# any later version.9#10# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY11# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR12# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more13# details.14#15# You should have received a copy of the GNU Lesser General Public License16# along with Paramiko; if not, write to the Free Software Foundation, Inc.,17# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.18import socket19import sys20import select21from paramiko.py3compat import u22from  modules import models23import datetime24# windows does not have termios...25try:26    import termios27    import tty28    has_termios = True29except ImportError:30    has_termios = False31def interactive_shell(chan,user_obj,bind_host_obj,cmd_logs,log_recording):32    if has_termios:33        posix_shell(chan,user_obj,bind_host_obj,cmd_logs,log_recording)34    else:35        windows_shell(chan)36def posix_shell(chan,user_obj,bind_host_obj,cmd_logs,log_recording):37    # è·ååttyå±æ§38    oldtty = termios.tcgetattr(sys.stdin)39    try:40        # ä¸ºttyè®¾ç½®æ°å±æ§41        # é»è®¤å½åttyè®¾å¤å±æ§ï¼42        # è¾å¥ä¸è¡åè½¦ï¼æ§è¡43        # CTRL+C è¿ç¨éåºï¼éå°ç¹æ®åç¬¦ï¼ç¹æ®å¤çã44        # è¿æ¯ä¸ºåå§æ¨¡å¼ï¼ä¸è®¤è¯ææç¹æ®ç¬¦å·45        # æ¾ç½®ç¹æ®åç¬¦åºç¨å¨å½åç»ç«¯ï¼å¦æ¤è®¾ç½®ï¼å°ææçç¨æ·è¾å¥ååéå°è¿ç¨æå¡å¨46        tty.setraw(sys.stdin.fileno())47        tty.setcbreak(sys.stdin.fileno())48        chan.settimeout(0.0)49        cmd = ''50        tab_key = False51        # æ³¨æï¼ä¸é¢å¾ªç¯å¶å®æ¯åå¾ªç¯ if sys.stdin in r:(ç¨æ·åæè¾å¥)ï¼åé¢å¾ªç¯if chan in r:(æ¾åç»æï¼tabé®è¡¥å¨çåå®¹ä¹æ¯è¿åç»æ)52        while True:53            r, w, e = select.select([chan, sys.stdin], [], [])54            if chan in r:55                try:56                    x = u(chan.recv(1024))57                    # å¦æç¨æ·ä¸ä¸æ¬¡ç¹å»çæ¯tabé®ï¼åè·åè¿åçåå®¹åå¥å¨è®°å½ä¸58                    if tab_key:59                        #if x not in ('\x07' , '\r\n'):60                        cmd += x # è¿éä¼è®°å½ä¸ä¸ªtabé®å ä¸tabåé¢çstdout61                        tab_key = False62                    if len(x) == 0:63                        sys.stdout.write('\r\n*** EOF\r\n')64                        break65                    sys.stdout.write(x)66                    sys.stdout.flush()67                except socket.timeout:68                    pass69            if sys.stdin in r:70                # å ä¸ºè¦ä½¿ç¨tabé®ï¼æä»¥ä¸ä¸ªåèä¸ä¸ªåèçå»è¯»åæ°æ®71                x = sys.stdin.read(1)72                # ç¨æ·è¾å¥å½ä»¤çæ¶åæ²¡æè¾å¥åè½¦ï¼æ¯å¦ifconfigï¼ä¼ä¾æ¬¡å¾ªç¯æ¶åi,f,c,o,nï¼æ¤æ¶åè¾å¥tabé®ï¼ä¼å¾ªç¯å°ä¸é¢çif x == "\t":å¤æï¼tabé®è¡¥å¨çåå®¹å°±æ¯æ åè¾åºäºï¼stdoutä¼èµ°å°ä¸é¢çå¾ªç¯ä¸å»ï¼if tab_key73                if x != "\r":74                    cmd += x75                else: # è¿éç¨æ·è¾å¥äºåè½¦ï¼ä¸æ¡å½ä»¤æ§è¡å®æ¯76                    log_item = models.AuditLog(user_id=user_obj.id,77                                          bind_host_id=bind_host_obj.id,78                                          action_type='cmd',79                                          cmd=cmd.replace("\t",""), #æ¿æ¢æå½ä»¤ä¸çtabç©ºæ ¼80                                          date=datetime.datetime.now()81                                          )82                    cmd_logs.append(log_item)83                    cmd = ""84                    if len(cmd_logs)>= 10:85                        #log_recording(user_obj,bind_host_obj,cmd_caches) #è²ä¼¼user_objï¼bind_host_objåcmd_cachesä¸çæäºåæ®µéå¤äº86                        log_recording(cmd_logs) # è®°å½ç¨æ·æ¥å¿ï¼æ¯åæ¡åä¸æ¬¡æ°æ®åº87                        cmd_logs = []88                if x == "\t": # ç¨æ·è¾å¥tab89                    tab_key = True90                if len(x) == 0:91                    break92                chan.send(x)93    finally:94        # éæ°è®¾ç½®ç»ç«¯å±æ§95        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty)96    97# thanks to Mike Looijmans for this code98def windows_shell(chan):99    import threading100    sys.stdout.write("Line-buffered terminal emulation. Press F6 or ^Z to send EOF.\r\n\r\n")101        102    def writeall(sock):103        while True:104            data = sock.recv(256)105            if not data:106                sys.stdout.write('\r\n*** EOF ***\r\n\r\n')107                sys.stdout.flush()108                break109            sys.stdout.write(data.decode())110            sys.stdout.flush()111        112    writer = threading.Thread(target=writeall, args=(chan,))113    writer.start()114        115    try:116        while True:117            d = sys.stdin.read(1)118            if not d:119                break120            chan.send(d)121    except EOFError:122        # user hit ^Z or F6...

__init__.py

Source:__init__.py

1"""2This file is a part of the source code for the KithscordBot.3This project has been licensed under the MIT license.4Copyright (c) 2021-present Kithare Organization5This file is the main file of kithscord subdir6"""7import io8import signal9import sys10import discord11from kithscord import commands, common, routine12from kithscord.utils import utils13is_init = False14async def _init():15    """16    Startup call helper for kithscord bot17    """18    if not common.LOCAL_TEST:19        sys.stdout = sys.stderr = common.stdout = io.StringIO()20    print("The KithscordBot is now online!")21    print("Server(s):")22    for server in common.bot.guilds:23        prim = ""24        if common.guild is None and server.id == common.SERVER_ID:25            prim = "| Primary Guild"26            common.guild = server27        print(" -", server.name, "| Number of channels:", len(server.channels), prim)28        for channel in server.channels:29            if channel.id == common.LOG_CHANNEL_ID:30                common.log_channel = channel31            elif channel.id == common.CONSOLE_CHANNEL_ID:32                common.console_channel = channel33async def init():34    """35    Startup call helper for kithscord bot36    """37    global is_init38    if is_init:39        return40    try:41        await _init()42    except Exception:43        # error happened in the first init sequence. report error to stdout/stderr44        # note that the chances of this happening are pretty slim, but you never know45        sys.stdout = sys.__stdout__46        sys.stderr = sys.__stderr__47        raise48    if not common.LOCAL_TEST:49        routine.handle_console.start()50    await utils.setup_kcr()51    if common.guild is None:52        raise RuntimeWarning(53            "Primary guild was not set. Some features of bot would not run as usual."54            " People running commands via DMs might face some problems"55        )56    is_init = True57async def message_delete(msg: discord.Message):58    """59    This function is called for every message deleted by user.60    """61    if msg.id in common.cmd_logs:62        del common.cmd_logs[msg.id]63    elif msg.author.id == common.bot.user.id:64        for log, logmsg in common.cmd_logs.items():65            if logmsg.id == msg.id:66                del common.cmd_logs[log]67                return68async def message_edit(_: discord.Message, new: discord.Message):69    """70    This function is called for every message edited by user.71    """72    if not new.content.startswith(common.PREFIX):73        return74    try:75        if new.id in common.cmd_logs:76            await commands.handle(new, common.cmd_logs[new.id])77    except discord.HTTPException:78        pass79async def handle_message(msg: discord.Message):80    """81    Handle a message posted by user82    """83    if msg.channel.id in common.NO_TALK_CHANNELS and msg.author.id != common.BOT_ID:84        await msg.delete()85        return86    if not msg.content.startswith(common.PREFIX):87        return88    ret = await commands.handle(msg)89    if ret is not None:90        common.cmd_logs[msg.id] = ret91    if len(common.cmd_logs) > 100:92        del common.cmd_logs[list(common.cmd_logs)[0]]93def cleanup(*_):94    """95    Call cleanup functions96    """97    common.bot.loop.run_until_complete(common.bot.close())98    common.bot.loop.close()99def run():100    """101    Does what discord.Client.run does, except, handles custom cleanup functions102    """103    # use signal.signal to setup SIGTERM signal handler, runs after event loop104    # closes105    signal.signal(signal.SIGTERM, cleanup)106    try:107        common.bot.loop.run_until_complete(common.bot.start(common.TOKEN))108    except KeyboardInterrupt:109        # Silence keyboard interrupt traceback (it contains no useful info)110        pass111    finally:...

logging_commands.py

Source:logging_commands.py

...8async def setup(bot: commands.Bot):9    bot.add_command(cmd_logs)10@commands.command(name="logs")11@commands.has_guild_permissions(manage_guild=True)12async def cmd_logs(context: commands.Context):13    last_logs = datamanagement.getLastLogs(guild=context.guild, limit=10)14    if not last_logs:15        embed = discord.Embed(16            title="No logs found",17            color=discord.Color.orange()18        )19    else:20        embed = discord.Embed(21            color=discord.Color.green()22        )23        for log in last_logs:24            embed.add_field(25                name=str(log.timestamp),26                value=f"```txt\n{log.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.