KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
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 Kiwi 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?
