Best Python code snippet using autotest_python
jobs.py
Source:jobs.py
...25 table.align['Description'] = 'l'26 table.align['Cronexp'] = 'l'27 # prepare request to listener28 request = Request()29 request.set_classname('JobListEvent')30 request.set_classpath(self.CLASSPATH)31 # send request to listener32 sc = SocketClient(config)33 response = sc.send_request(request)34 if response.status == StatusCode.STATUS_OK:35 job_list = response.data['Result']36 for s in job_list:37 uid = s.get('id')38 name = s.get('name')39 description = s.get('description')40 cronexp = s.get('cronexp')41 enabled = s.get('enabled')42 table.add_row([uid, name, description, cronexp, enabled])43 print(table)44 print()45 else:46 print('Impossible to execute the command.')47class JobAdd(RemoteCommand):48 def __init__(self, config):49 super().__init__(config)50 def execute(self, cmd_args):51 config = self.config52 # test socket connection53 self.test_socket_connection()54 registered_tasks = config['tasks']55 job = self.ask_basics()56 # get job data from the user57 job['parameters'] = self.set_job_parameters(registered_tasks)58 # prepare request to listener59 request = Request()60 request.set_classname('JobAddEvent')61 request.set_classpath(self.CLASSPATH)62 request.set_data(job)63 # send request to listener64 sc = SocketClient(config)65 response = sc.send_request(request)66 # init result table67 table = PrettyTable()68 table.field_names = ['Status', 'Message']69 table.align['Status'] = 'c'70 table.align['Message'] = 'l'71 table.add_row([response.status, response.data['msg']])72 print(table)73 print()74 return75 def ask_basics(self):76 questions = [77 inquirer.Text(78 'name',79 message = 'Name of job'80 ),81 inquirer.Text(82 'description',83 message = 'Description of job',84 ),85 inquirer.Text(86 'cronexp',87 message = 'Cron expression',88 validate = check_cron,89 ),90 inquirer.Confirm(91 'enabled',92 message = 'Enable the job?',93 default = True,94 )95 ]96 job = inquirer.prompt(questions)97 return job98 def set_task(self, registered_tasks, job_parameters):99 classnames = [tsk['classname'] for tsk in registered_tasks]100 # select a task101 choices = [f'{tsk["classname"]} - {tsk["description"]}' for tsk in registered_tasks]102 questions = [103 inquirer.List('task',104 message = 'Select task to execute',105 choices = choices,106 carousel = True,107 ),108 inquirer.Text(109 'timeout',110 message = 'Timeout',111 default = '600',112 validate = check_int113 ),114 inquirer.Text(115 'arg_list',116 message = 'Arguments (comma-separated list of key=value)',117 default = None118 ),119 ]120 ans = inquirer.prompt(questions)121 task_idx = choices.index(ans['task'])122 task_to_run = registered_tasks[task_idx]123 classname = task_to_run['classname']124 # set task data125 task = {}126 task['filepath'] = task_to_run['filepath']127 task['timeout'] = ans['timeout']128 task['args'] = get_dict_from_args(ans['arg_list'])129 task['onPipe'] = None130 task['onSuccess'] = None131 task['onFail'] = None132 job_parameters['tasklist'][classname] = task133 return job_parameters, classname134 def set_connection(self, registered_tasks, parameters, base_task):135 question_pipe = [136 inquirer.Confirm('onPipe',137 message = f'[{base_task}] Do you want to pipe another task?',138 default = False,139 )140 ]141 question_success = [142 inquirer.Confirm('onSuccess',143 message = f'[{base_task}] Do you want to execute another task on success?',144 default = False,145 )146 ]147 question_fail = [148 inquirer.Confirm('onFail',149 message = f'[{base_task}] Do you want to execute another task on fail?',150 default = False,151 )152 ]153 ans = inquirer.prompt(question_pipe)154 if ans['onPipe']:155 parameters, next_task = self.set_task(registered_tasks, parameters)156 parameters['tasklist'][base_task]['onPipe'] = next_task157 parameters = self.set_connection(registered_tasks, parameters, next_task)158 ans = inquirer.prompt(question_success)159 if ans['onSuccess']:160 parameters, next_task = self.set_task(registered_tasks, parameters)161 parameters['tasklist'][base_task]['onSuccess'] = next_task162 parameters = self.set_connection(registered_tasks, parameters, next_task)163 ans = inquirer.prompt(question_fail)164 if ans['onFail']:165 parameters, next_task = self.set_task(registered_tasks, parameters)166 parameters['tasklist'][base_task]['onFail'] = next_task167 parameters = self.set_connection(registered_tasks, parameters, next_task)168 return parameters169 def set_job_parameters(self, registered_tasks):170 job_parameters = {'tasklist': {}}171 # get a task and set connections172 job_parameters, classname = self.set_task(registered_tasks, job_parameters)173 job_parameters['root'] = classname174 job_parameters = self.set_connection(registered_tasks, job_parameters, classname)175 # serialize to json176 job_parameters = json.dumps(job_parameters)177 return job_parameters178class JobRemove(RemoteCommand):179 def __init__(self, config):180 super().__init__(config)181 def execute(self, cmd_args):182 config = self.config183 # test socket connection184 self.test_socket_connection()185 # prepare request to listener186 request = Request()187 request.set_classname('JobRemoveEvent')188 request.set_classpath(self.CLASSPATH)189 request.set_data(cmd_args)190 # send request to listener191 sc = SocketClient(config)192 response = sc.send_request(request)193 # init result table194 table = PrettyTable()195 table.field_names = ['Status', 'Message']196 table.align['Status'] = 'c'197 table.align['Message'] = 'l'198 table.add_row([response.status, response.data['msg']])199 print(table)200 print()201 return202class JobEnable(RemoteCommand):203 def __init__(self, config):204 super().__init__(config)205 def execute(self, cmd_args):206 config = self.config207 # test socket connection208 self.test_socket_connection()209 # prepare request to listener210 request = Request()211 request.set_classname('JobEnableEvent')212 request.set_classpath(self.CLASSPATH)213 request.set_data(cmd_args)214 # send request to listener215 sc = SocketClient(config)216 response = sc.send_request(request)217 # init result table218 table = PrettyTable()219 table.field_names = ['Status', 'Message']220 table.align['Status'] = 'c'221 table.align['Message'] = 'l'222 table.add_row([response.status, response.data['msg']])223 print(table)224 print()225 return226class JobDisable(RemoteCommand):227 def __init__(self, config):228 super().__init__(config)229 def execute(self, cmd_args):230 config = self.config231 # test socket connection232 self.test_socket_connection()233 # prepare request to listener234 request = Request()235 request.set_classname('JobDisableEvent')236 request.set_classpath(self.CLASSPATH)237 request.set_data(cmd_args)238 # send request to listener239 sc = SocketClient(config)240 response = sc.send_request(request)241 # init result table242 table = PrettyTable()243 table.field_names = ['Status', 'Message']244 table.align['Status'] = 'c'245 table.align['Message'] = 'l'246 table.add_row([response.status, response.data['msg']])247 print(table)248 print()249 return250class JobExec(RemoteCommand):251 def __init__(self, config):252 super().__init__(config)253 def execute(self, cmd_args):254 config = self.config255 # test socket connection256 self.test_socket_connection()257 # prepare request to listener258 request = Request()259 request.set_classname('JobExecEvent')260 request.set_classpath(self.CLASSPATH)261 request.set_data(cmd_args)262 # send request to listener263 sc = SocketClient(config)264 response = sc.send_request(request)265 # init result table266 table = PrettyTable()267 table.field_names = ['Status', 'Message']268 table.align['Status'] = 'c'269 table.align['Message'] = 'l'270 table.add_row([response.status, response.data['msg']])271 print(table)272 print()273 return274class JobGet(RemoteCommand):275 def __init__(self, config):276 super().__init__(config)277 def execute(self, cmd_args):278 config = self.config279 # test socket connection280 self.test_socket_connection()281 # init overview table282 table_overview = PrettyTable()283 table_overview.field_names = ['ID', 'Name', 'Description', 'Cronexp', 'Enabled']284 table_overview.align['Name'] = 'l'285 table_overview.align['Description'] = 'l'286 table_overview.align['Cronexp'] = 'l'287 # init task table288 table_task = PrettyTable()289 table_task.field_names = ['No.', 'Task', 'Filepath', 'Timeout', 'Args', 'onPipe', 'onSuccess', 'onFail']290 table_task.align = 'l'291 # prepare request to listener292 request = Request()293 request.set_classname('JobGetEvent')294 request.set_classpath(self.CLASSPATH)295 request.set_data(cmd_args)296 # send request to listener297 sc = SocketClient(config)298 response = sc.send_request(request)299 if response.status == StatusCode.STATUS_OK:300 schedule_info = response.data['Result']301 uid = schedule_info.get('id')302 name = schedule_info.get('name')303 description = schedule_info.get('description')304 cronexp = schedule_info.get('cronexp')305 enabled = schedule_info.get('enabled')306 table_overview.add_row([uid, name, description, cronexp, enabled])307 print()...
Event.py
Source:Event.py
1import datetime23class Event:4 def __init__(self):5 self._id = 06 self._content = ""7 self._className = ""8 self._start = None9 10 # Setters/Getters11 def set_id(self,id):12 self._id = id 13 def get_id(self):14 return self._id 15 def set_content(self,content):16 self._content = content 17 def get_content(self):18 return self._content 19 def set_className(self, className):20 self._className = className 21 def get_className(self):22 return self._className 23 # Start is datetime object24 def set_start(self, start):25 if '/' in start:26 try:27 self._start = datetime.datetime.strptime(start, "%Y/%m/%dT%H:%M:%S")28 except Exception:29 try:30 self._start = datetime.datetime.strptime(start, "%d/%m/%YT%H:%M:%S")31 except:32 self._start = datetime.datetime.strptime(start, "%m/%d/%YT%H:%M:%S")33 elif '-' in start:34 try:35 self._start = datetime.datetime.strptime(start, "%Y-%m-%dT%H:%M:%S")36 except Exception:37 self._start = datetime.datetime.strptime(start, "%d-%m-%YT%H:%M:%S")38 def get_start(self):39 return self._start40 # Returns string of start datetime object41 def get_start_tostring(self):42 return self._start.strftime("%m/%d/%YT%H:%M:%S")4344# Each of the following event classes inherit the Event class and contain additional fields unique to their event45# Each class contains a tojson that returns the object information in json data format46class Auditd(Event):47 def __init__(self, id, content, className, start):48 self.set_id(id)49 self.set_content(content)50 self.set_className(className)51 self.set_start(start)52 def tojson(self):53 return {'auditd_id' : self.get_id(), 'content' : self.get_content(), 'className' : self.get_className(), 'start' : self.get_start_tostring()}5455class Clicks(Event):56 def __init__(self, id, content, type, className, start):57 self.set_id(id)58 self.set_content(content)59 self.set_className(className)60 self.set_start(start)61 self._type = type62 def get_type(self):63 return self._type64 def tojson(self):65 return {'clicks_id' : self.get_id(), 'content' : self.get_content(), 'type' : self.get_type(), 'classname' : self.get_className(), 'start' : self.get_start_tostring()}6667class Keypresses(Event):68 def __init__(self, id, content, className, start):69 self.set_id(id)70 self.set_content(content)71 self.set_className(className)72 self.set_start(start)73 def tojson(self):74 return {'keypresses_id' : self.get_id(), 'content' : self.get_content(), 'className' : self.get_className(), 'start' : self.get_start_tostring()}7576class Traffic(Event):77 def __init__(self, id, content, className, title, start):78 self.set_id(id)79 self.set_content(content)80 self.set_className(className)81 self.set_start(start)82 self._title = title83 def get_title(self):84 return self._title85 def tojson(self):86 return {'traffic_all_id' : self.get_id(), 'content' : self.get_content(), 'className' : self.get_className(), 'title' : self.get_title(), 'start' : self.get_start_tostring()}8788class TrafficThroughput(Event):89 def __init__(self, id, className, start, y):90 self.set_id(id)91 self.set_className(className)92 self.set_start(start)93 self._y = y94 def get_y(self):95 return self._y96 def tojson(self):97 return {'traffic_xy_id' : self.get_id(), 'className' : self.get_className(), 'start' : self.get_start_tostring(), 'y' : self.get_y()}9899class Timed(Event):100 def __init__(self, id, type, className, content, start):101 self.set_id(id)102 self.set_content(content)103 self.set_className(className)104 self.set_start(start)105 self._type = type 106 def get_type(self):107 return self._type108 def tojson(self):109 return {'timed_id' : self.get_id(), 'type' : self.get_type(), 'classname' : self.get_className(), 'content' : self.get_content(), 'start' : self.get_start_tostring()}110111class Suricata(Event):112 def __init__(self, id, rule, content, className, start):113 self.set_id(id)114 self._rule = rule115 self.set_content(content)116 self.set_className(className)117 self.set_start(start)118 def get_rule(self):119 return self._rule120 def tojson(self):121 return {"suricata_id" : self.get_id(), "suricata_rule_id" : self.get_rule(), "content" : self.get_content(), "className" : self.get_className(), "start" : self.get_start_tostring()}
...
test.py
Source:test.py
...6 colors_rgb = [(204,244,202),(10,30,12)]7 super().__init__(max_num,colors_rgb)8pset = palette(size=5, color="purple")9cm = confusion_matrix([[200,0,0,0],[0,156,8,14],[0,18,131,30],[0,60,28,75]])10cm.set_classname(["Acute", "Non-resolving","Normal","Inactive"])11cm.set_title("Retinal Specialist2")12print(cm)13pset.draw(cm, "./temp.png")14import numpy as np15y_true = np.array([1,1,1,1,0,0,0,0,1,1,1])16y_pred = np.array([1,1,0,1,0,0,0,1,1,0,1])17cm =confusion_matrix_by_cal(y_true, y_pred)18cm.set_classname(["Acute", "Inactive"])19cm.set_title("Retinal Specialist1")20print(cm)...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!