How to use notify_admins method in Kiwi

Best Python code snippet using Kiwi_python

mailman.py

Source:mailman.py

1import json2import logging3import os4import subprocess5import tempfile6log = logging.getLogger(__name__)7default_path = '/usr/lib/mailman/bin'8class MailmanError(Exception):9    '''Base class for all exceptions generated by this module'''10class CalledProcessError(MailmanError):11    def __init__(self, command, returncode, stdout, stderr):12        self.command = command13        self.returncode = returncode14        self.stdout = stdout15        self.stderr = stderr16class InvalidCommandError(MailmanError):17    pass18class Mailman(object):19    def __init__(self, path=None, allowed_commands=None):20        self.path = path or default_path21        self.allowed_commands = allowed_commands or []22    # we have to use **kwargs here because of python2 (we can't23    # have keyword arguments following *args)24    def run(self, cmd, *args, **kwargs):25        '''Run a mailman command, returning the decoded output.26        Raises InvalidCommandError if the given command is not27        in self.allowed_commands, starts with a '/', or contains '..'.28        Raises CalledProcessError if a command returns a nonzero29        exit status.30        '''31        if cmd.startswith('/'):32            raise InvalidCommandError(cmd)33        if '..' in cmd:34            raise InvalidCommandError(cmd)35        if self.allowed_commands and cmd not in self.allowed_commands:36            raise InvalidCommandError(cmd)37        stdin = kwargs['stdin'].encode() if 'stdin' in kwargs else None38        raw = kwargs.get('raw')39        cmdvec = [os.path.join(self.path, cmd)]40        cmdvec.extend(args)41        log.debug('running command: %s', cmdvec)42        p = subprocess.Popen(43            cmdvec,44            stdin=subprocess.PIPE,45            stdout=subprocess.PIPE,46            stderr=subprocess.PIPE,47        )48        out, err = p.communicate(input=stdin)49        ret = p.wait()50        if ret != 0:51            raise CalledProcessError(cmdvec, ret, out, err)52        return out if raw else out.decode()53    def list_lists(self,54                   advertised_only=False,55                   domain=None):56        '''Returns a list of mailing list names'''57        args = ['-b']58        if advertised_only:59            args.append('-a')60        if domain:61            args.extend(('-V', domain))62        return self.run('list_lists', *args).splitlines()63    def list_all_members(self, listname):64        '''Returns a list of all members of the mailing list'''65        return self.run('list_members', listname).splitlines()66    def list_regular_members(self, listname):67        '''Returns a list of all regular (non-digest) members of the mailing list'''68        return self.run('list_members', '-r', listname).splitlines()69    def list_digest_members(self, listname):70        '''Returns a list of all digest members of the mailing list'''71        return self.run('list_members', '-d', listname).splitlines()72    def list_exists(self, listname):73        '''Returns True if the named list exists'''74        return listname in self.list_lists()75    def is_subscribed(self, email, listname):76        '''Returns True if the given email address is subscribed to the list'''77        return email in self.list_all_members(listname)78    def create_list(self, name, owner, password,79                    urlhost=None,80                    emailhost=None,81                    notify_owner=False):82        '''Create a mailing list.83        This will send a notification to the list owner unless84        notify_owner is False.85        '''86        args = []87        if urlhost:88            args.extend(('--urlhost', urlhost))89        if emailhost:90            args.extend(('--emailhost', emailhost))91        if not notify_owner:92            args.append('-q')93        args.extend((name, owner, password))94        return self.run('newlist', *args)95    def remove_list(self, name, remove_archives=False):96        args = []97        if remove_archives:98            args.append('-a')99        args.append(name)100        return self.run('rmlist', *args)101    def add_members(self, listname, members,102                    digest=False,103                    notify_members=False,104                    notify_admins=False):105        '''Add members to a list'''106        return self.run(107            'add_members',108            '-w{}'.format('y' if notify_members else 'n'),109            '-a{}'.format('y' if notify_admins else 'n'),110            '-d' if digest else '-r', '-',111            listname,112            stdin='\n'.join(members))113    def add_regular_members(self, listname, members,114                            notify_members=False,115                            notify_admins=False):116        '''Add regular (non-digest) members to a list'''117        return self.add_members(listname, members, digest=False,118                                notify_members=notify_members,119                                notify_admins=notify_admins)120    def add_digest_members(self, listname, members,121                           notify_members=False,122                           notify_admins=False):123        '''Add digest members to a list'''124        return self.add_members(listname, members, digest=True,125                                notify_members=notify_members,126                                notify_admins=notify_admins)127    def remove_members(self, listname, members,128                       notify_members=False,129                       notify_admins=False):130        '''Remove members from a list'''131        args = ['-f', '-']132        if not notify_members:133            args.append('-n')134        if not notify_admins:135            args.append('-N')136        args.append(listname)137        return self.run(138            'remove_members',139            *args,140            stdin='\n'.join(members))141    def get_list_config(self, listname):142        '''Return the list configuration as a dictionary'''143        config = self.run('config_list', '-o', '-', listname, raw=True)144        ns = {}145        exec(config, ns)146        return {k: v147                for k, v in ns.items()148                if not k.startswith('_')}149    def set_list_config(self, listname, config):150        '''Configure listname using the contents of config.151        This is a bit of hack that takes advantage of the fact that152        the native JSON representation of types is compatible153        with Python. There's probably a better way to do it.154        '''155        with tempfile.NamedTemporaryFile() as fd:156            fd.write(b'true = True\n')157            fd.write(b'false = False\n')158            for k, v in config.items():159                fd.write('{} = {}\n'.format(k, json.dumps(v)).encode())160            fd.flush()161            return self.run('config_list', '-i', fd.name, listname)162if __name__ == '__main__':...

palette.py

Source:palette.py

1import os2from Messages.administration  import notify_admins, authentificate, check_serial_aviability3from Messages.messages import MessageParser4from config import serial_initialized5from data import users6from serial_talker import serial_talker7from speech import speech8from webcam import shooter9telegramBot = [None, ]10def set_bot(new_bot):11    telegramBot[0] = new_bot12    print("Bot inited")13@notify_admins14def time(message):15    import time16    chat_id = message.chat_info.id17    text = str(time.time())18    telegramBot[0].sendMessage(chat_id, text)19@notify_admins20def launch_game(message):21    game_link = "http://192.168.0.136"22    telegramBot[0].sendMessage(message.chat_instance, game_link)23@notify_admins    24def my_name(message):25    chat_id = message.chat_info.id    26    text = os.name27    telegramBot[0].sendMessage(chat_id, text)    28    29@check_serial_aviability30@notify_admins    31def temperature(message):32    chat_id = message.chat_info.id33    degrees = serial_talker.execute_command("/t")34    text = "Temperature is {}".format(degrees)35    telegramBot[0].sendMessage(chat_id, text)36def test_audio(message):37    chat_id = message.chat_info.id38    text = "Testing speech module."39    telegramBot[0].sendMessage(chat_id, text)40    speech.ding()    41    42    43@authentificate44@notify_admins45def room_photo(message):46    pic_path = shooter.take_picture()47    # text notification48    chat_id = message.chat_info.id49    text = "Picture taken. Here you go:"50    telegramBot[0].sendMessage(chat_id, text)51    # send file52    with open(pic_path, 'rb') as photo:53        telegramBot[0].sendPhoto(chat_id, photo)54    55@authentificate56@notify_admins57def subscribe_to_notifications(message):58    chat_id = message.chat_info.id59    users.add_subscriber(message.from_info.username, chat_id)60    text = "Subscribed"61    telegramBot[0].sendMessage(chat_id, text)62    63def default_answer(message):64    chat_id = message.chat_info.id65    text = "I read you {}. No callback found for {}".format(message.from_info.username,  str(message.text))66    telegramBot[0].sendMessage(chat_id, text,67                    parse_mode=None, disable_web_page_preview=None,68                    disable_notification=None, reply_to_message_id=None, reply_markup=None)    69def custom_answer(chat_id, text):70    print (text)71    telegramBot[0].sendMessage(chat_id, text)                   72callbacks = {73            "/myname" : my_name,74            "/temperature" : temperature,75            "/test_audio" : test_audio,76            "/time" : time,77            "/game" : launch_game,78            "/subscribe" : subscribe_to_notifications,79            "/webcam" : room_photo,80            }81    82def message_callback(new_messages):83    print("______________new message!____________")84    parser = MessageParser()85    parsed_messages = parser.parse(new_messages)86    for message in parsed_messages:87        if message is None:88            continue89        print ("from: {}{}Message: {}".format( message.from_info.username, os.linesep, message.text))90        callback = callbacks.get(message.text, default_answer)...

__init__.py

Source:__init__.py

1from .default_commands import setup_default_commands2from .logger import setup_logger3from .notify_admins import notify_admins4__all__ = [5    "setup_logger",6    "setup_default_commands",7    "notify_admins",...

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.