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

How to use sleeping method in Robotframework

Best Python code snippet using robotframework

load_balancing.py

Source:load_balancing.py

1# --------------------------------------------------------------------------2# Source file provided under Apache License, Version 2.0, January 2004,3# http://www.apache.org/licenses/4# (c) Copyright IBM Corp. 2015, 20185# --------------------------------------------------------------------------6# Source: http://blog.yhathq.com/posts/how-yhat-does-cloud-balancing.html7from collections import namedtuple8from docplex.mp.model import Model9# ----------------------------------------------------------------------------10# Initialize the problem data11# ----------------------------------------------------------------------------12class TUser(namedtuple("TUser", ["id", "running", "sleeping", "current_server"])):13    def __str__(self):14        return self.id15SERVERS = ["server002", "server003", "server001", "server006", "server007", "server004", "server005"]16USERS = [("user013", 2, 1, "server002"),17         ("user014", 0, 2, "server002"),18         ("user015", 0, 4, "server002"),19         ("user016", 1, 4, "server002"),20         ("user017", 0, 3, "server002"),21         ("user018", 0, 2, "server002"),22         ("user019", 0, 2, "server002"),23         ("user020", 0, 1, "server002"),24         ("user021", 4, 4, "server002"),25         ("user022", 0, 1, "server002"),26         ("user023", 0, 3, "server002"),27         ("user024", 1, 2, "server002"),28         ("user025", 0, 1, "server003"),29         ("user026", 0, 1, "server003"),30         ("user027", 1, 1, "server003"),31         ("user028", 0, 1, "server003"),32         ("user029", 2, 1, "server003"),33         ("user030", 0, 5, "server003"),34         ("user031", 0, 2, "server003"),35         ("user032", 0, 3, "server003"),36         ("user033", 1, 1, "server003"),37         ("user034", 0, 1, "server003"),38         ("user035", 0, 1, "server003"),39         ("user036", 4, 1, "server003"),40         ("user037", 7, 1, "server003"),41         ("user038", 2, 1, "server003"),42         ("user039", 0, 3, "server003"),43         ("user040", 1, 2, "server003"),44         ("user001", 0, 2, "server001"),45         ("user002", 0, 3, "server001"),46         ("user003", 5, 4, "server001"),47         ("user004", 0, 1, "server001"),48         ("user005", 0, 1, "server001"),49         ("user006", 0, 2, "server001"),50         ("user007", 0, 4, "server001"),51         ("user008", 0, 1, "server001"),52         ("user009", 5, 1, "server001"),53         ("user010", 7, 1, "server001"),54         ("user011", 4, 5, "server001"),55         ("user012", 0, 4, "server001"),56         ("user062", 0, 1, "server006"),57         ("user063", 3, 5, "server006"),58         ("user064", 0, 1, "server006"),59         ("user065", 0, 3, "server006"),60         ("user066", 3, 1, "server006"),61         ("user067", 0, 1, "server006"),62         ("user068", 0, 1, "server006"),63         ("user069", 0, 2, "server006"),64         ("user070", 3, 2, "server006"),65         ("user071", 0, 1, "server006"),66         ("user072", 5, 3, "server006"),67         ("user073", 0, 1, "server006"),68         ("user074", 0, 1, "server006"),69         ("user075", 0, 2, "server007"),70         ("user076", 1, 1, "server007"),71         ("user077", 1, 1, "server007"),72         ("user078", 0, 1, "server007"),73         ("user079", 0, 3, "server007"),74         ("user080", 0, 1, "server007"),75         ("user081", 4, 1, "server007"),76         ("user082", 1, 1, "server007"),77         ("user041", 0, 1, "server004"),78         ("user042", 2, 1, "server004"),79         ("user043", 5, 2, "server004"),80         ("user044", 5, 2, "server004"),81         ("user045", 0, 2, "server004"),82         ("user046", 1, 5, "server004"),83         ("user047", 0, 1, "server004"),84         ("user048", 0, 3, "server004"),85         ("user049", 5, 1, "server004"),86         ("user050", 0, 2, "server004"),87         ("user051", 0, 3, "server004"),88         ("user052", 0, 3, "server004"),89         ("user053", 0, 1, "server004"),90         ("user054", 0, 2, "server004"),91         ("user055", 0, 3, "server005"),92         ("user056", 3, 1, "server005"),93         ("user057", 0, 3, "server005"),94         ("user058", 0, 2, "server005"),95         ("user059", 0, 1, "server005"),96         ("user060", 0, 5, "server005"),97         ("user061", 0, 2, "server005")98         ]99# ----------------------------------------------------------------------------100# Prepare the data for modeling101# ----------------------------------------------------------------------------102DEFAULT_MAX_PROCESSES_PER_SERVER = 50103def _is_migration(user, server):104    """ Returns True if server is not the user's current105        Used in setup of constraints.106    """107    return server != user.current_server108# ----------------------------------------------------------------------------109# Build the model110# ----------------------------------------------------------------------------111def build_load_balancing_model(servers, users_, max_process_per_server=DEFAULT_MAX_PROCESSES_PER_SERVER, **kwargs):112    m = Model(name='load_balancing', **kwargs)113    # decision objects114    users = [TUser(*user_row) for user_row in users_]115    active_var_by_server = m.binary_var_dict(servers, name='isActive')116    def user_server_pair_namer(u_s):117        u, s = u_s118        return '%s_to_%s' % (u.id, s)119    assign_user_to_server_vars = m.binary_var_matrix(users, servers, user_server_pair_namer)120    m.add_constraints(121        m.sum(assign_user_to_server_vars[u, s] * u.running for u in users) <= max_process_per_server for s in servers)122    # each assignment var <u, s>  is <= active_server(s)123    for s in servers:124        for u in users:125            ct_name = 'ct_assign_to_active_{0!s}_{1!s}'.format(u, s)126            m.add_constraint(assign_user_to_server_vars[u, s] <= active_var_by_server[s], ct_name)127        # sum of assignment vars for (u, all s in servers) == 1128        for u in users:129            ct_name = 'ct_unique_server_%s' % (u[0])130            m.add_constraint(m.sum((assign_user_to_server_vars[u, s] for s in servers)) == 1, ct_name)131    number_of_active_servers = m.sum((active_var_by_server[svr] for svr in servers))132    m.add_kpi(number_of_active_servers, "Number of active servers")133    number_of_migrations = m.sum(134        assign_user_to_server_vars[u, s] for u in users for s in servers if135        _is_migration(u, s))136    m.add_kpi(number_of_migrations, "Total number of migrations")137    max_sleeping_workload = m.integer_var(name="max_sleeping_processes")138    for s in servers:139        ct_name = 'ct_define_max_sleeping_%s' % s140        m.add_constraint(141            m.sum(142                assign_user_to_server_vars[u, s] * u.sleeping for u in users) <= max_sleeping_workload,143            ct_name)144    m.add_kpi(max_sleeping_workload, "Max sleeping workload")145    # Set objective function146    # m.minimize(number_of_active_servers)147    m.minimize_static_lex([number_of_active_servers, number_of_migrations, max_sleeping_workload])148    # attach artefacts to model for reporting149    m.users = users150    m.servers = servers151    m.active_var_by_server = active_var_by_server152    m.assign_user_to_server_vars = assign_user_to_server_vars153    m.max_sleeping_workload = max_sleeping_workload154    return m155def lb_report(mdl):156    active_servers = sorted([s for s in mdl.servers if mdl.active_var_by_server[s].solution_value == 1])157    print("Active Servers: {0} = {1}".format(len(active_servers), active_servers))158    print("*** User/server assignments , #migrations={0} ***".format(159        mdl.kpi_by_name("number of migrations").solution_value))160    # for (u, s) in sorted(mdl.assign_user_to_server_vars):161    #     if mdl.assign_user_to_server_vars[(u, s)].solution_value == 1:162    #         print("{} uses {}, migration: {}".format(u, s, "yes" if _is_migration(u, s) else "no"))163    print("*** Servers sleeping processes ***")164    for s in active_servers:165        sleeping = sum(mdl.assign_user_to_server_vars[u, s].solution_value * u.sleeping for u in mdl.users)166        print("Server: {} #sleeping={}".format(s, sleeping))167def make_default_load_balancing_model(**kwargs):168    return build_load_balancing_model(SERVERS, USERS, **kwargs)169def lb_save_solution_as_json(mdl, json_file):170    """Saves the solution for this model as JSON.171    Note that this is not a CPLEX Solution file, as this is the result of post-processing a CPLEX solution172    """173    import json174    solution_dict = {}175    # active server176    active_servers = sorted([s for s in mdl.servers if mdl.active_var_by_server[s].solution_value == 1])177    solution_dict["active servers"] = active_servers178    # sleeping processes by server179    sleeping_processes = {}180    for s in active_servers:181        sleeping = sum(mdl.assign_user_to_server_vars[u, s].solution_value * u.sleeping for u in mdl.users)182        sleeping_processes[s] = sleeping183    solution_dict["sleeping processes by server"] = sleeping_processes184# user assignment185    user_assignment = []186    for (u, s) in sorted(mdl.assign_user_to_server_vars):187        if mdl.assign_user_to_server_vars[(u, s)].solution_value == 1:188            n = {189                'user': u.id,190                'server': s,191                'migration': "yes" if _is_migration(u, s) else "no"192            }193            user_assignment.append(n)194    solution_dict['user assignment'] = user_assignment195    json_file.write(json.dumps(solution_dict, indent=3).encode('utf-8'))196# ----------------------------------------------------------------------------197# Solve the model and display the result198# ----------------------------------------------------------------------------199if __name__ == '__main__':200    lbm = make_default_load_balancing_model()201    # Run the model.202    lbs = lbm.solve(log_output=True)203    lb_report(lbm)204    # save json, used in worker tests205    from docplex.util.environment import get_environment206    with get_environment().get_output_stream("solution.json") as fp:...

sleeping_barber.py

Source:sleeping_barber.py

1import time2import random3from threading import Thread, Event, Lock4event = Event() #The internal flag is initially false.5lock = Lock()67class SleepingBarber:89    def __init__(self, barber, num_seats_in_waiting_room):10        self.barber = barber11        self.num_seats_in_waiting_room = num_seats_in_waiting_room1213    def start_day(self):14        print("The barber shop is opening for the day")15        barber_working1 = Thread(target=self.work) #A thread on the work function that will continuously look for new customers in the waiting room, and if none left, will allow the barber to go to sleep.16        barber_working1.start()#Start the thread.17        barber_working2 = Thread(target=self.work)18        barber_working2.start()19        barber_working3 = Thread(target=self.work)20        barber_working3.start()21        #Each thread acts as a different barber. The more threads there are, the more barbers in the shop.22    23    waiting_room = [] #A list used to store the customers waiting for a hair-cut.242526    def customer_enters_barbers(self, customer):27        lock.acquire() #Get the lock inorder to protect the waiting room as this is a shared resource between the work() function and the customer_enters_barbers() function.28        print("{} entered the barber shop".format(customer))29        if len(self.waiting_room) != self.num_seats_in_waiting_room: # If there's still seats available in the waiting room:30            print("{} took a seat in the waiting room".format(customer))31            self.waiting_room.append(customer) # append the customer to the waiting room list32            lock.release()#Release the lock so the work() function can then acquire it.33            event.set()#The barber wakes up as a customer walks in. Set the internal flag to true. All threads waiting for it to become true are awakened.34            35        else:36            print("The waiting room is full at the moment. {} leaves the barber shop".format(customer))37            lock.release()#Release the lock so the work() function can then acquire it.383940    def work(self):41        while True: #while True means loop forever. True always evaluates to boolean "true" and thus executes the loop body indefinitely.42            lock.acquire() #Get the lock inorder to protect the waiting room as this is a shared resource between the work() function and the customer_enters_barbers() function.43            if len(self.waiting_room) != 0: # If there's someone in the waiting room44                customer = self.waiting_room.pop(0) #Let this person be the next customer and delete this person from the waiting room45                lock.release() #Release the lock so the customer_enters_barbers() function can then acquire it.46                self.barber.hair_cut(customer) #Pass this customer into the haircut function47            48            else:49                lock.release()#Release the lock so the customer_enters_barbers() function can then acquire it.50                print("There is no one left to get a hair cut, the barber goes to sleep")51                event.wait()#The barber goes to sleep. Block until the internal flag is true. If the internal flag is true on entry, return immediately. Otherwise, block until another thread calls set() to set the flag to true, or until the optional timeout occurs5253                print("A new customer arrived. The barber wakes up")5455class Barber:5657    def hair_cut(self, customer):58        event.clear() #Reset the internal flag to false.59        print("{} is getting their hair cut".format(customer))60        time.sleep(random.randint(3, 12)) # The amount of time it takes to get a haircut.61        print("{} is finished getting their hair cut".format(customer))626364if __name__ == '__main__':65    arriving_customers = []6667    arriving_customers.append('customer26')68    arriving_customers.append('customer25')69    arriving_customers.append('customer24')70    arriving_customers.append('customer23')71    arriving_customers.append('customer22')72    arriving_customers.append('customer21')73    arriving_customers.append('customer20')74    arriving_customers.append('customer19')75    arriving_customers.append('customer18')76    arriving_customers.append('customer17')77    arriving_customers.append('customer16')78798081    SleepingBarber.waiting_room.append('customer1')82    SleepingBarber.waiting_room.append('customer2')83    SleepingBarber.waiting_room.append('customer3')84    SleepingBarber.waiting_room.append('customer4')85    SleepingBarber.waiting_room.append('customer5')86    SleepingBarber.waiting_room.append('customer6')87    SleepingBarber.waiting_room.append('customer7')88    SleepingBarber.waiting_room.append('customer8')89    SleepingBarber.waiting_room.append('customer9')90    SleepingBarber.waiting_room.append('customer10')91    SleepingBarber.waiting_room.append('customer11')92    SleepingBarber.waiting_room.append('customer12')93    SleepingBarber.waiting_room.append('customer13')94    SleepingBarber.waiting_room.append('customer14')95    SleepingBarber.waiting_room.append('customer15')9697    barber = Barber()98    sleeping_barber = SleepingBarber(barber, num_seats_in_waiting_room=15)99    sleeping_barber.start_day() # Call the start of the program100    101    while len(arriving_customers) > 0: # While there are customers in the arriving_customers list:102        customer = arriving_customers.pop()#New customer enters the barbers103        sleeping_barber.customer_enters_barbers(customer)#Pass this customer to the customer_enters_barbers() function.
...

add_sleeping_furniture.py

Source:add_sleeping_furniture.py

1import sys2import os.path3sys.path.append(os.path.join(os.path.dirname(__file__), '..'))45from PyQt5.QtWidgets import (QLabel, QLineEdit, QComboBox, QMessageBox,6                             QWidget, QApplication, QGridLayout, QLayout)78from data_correctness import Validations9from classes.sleeping_furniture import SleepingFurniture10from management.room_management import (get_all_rooms_number,11                                        add_sleeping_furniture)121314class SleepingFurniture_Form(QWidget):15    def __init__(self):16        super(SleepingFurniture_Form, self).__init__()17        self.setupUi(self)1819    def setupUi(self, SleepingFurniture_Form):20        layout = QGridLayout(self)21        self.sleeping_furnitures = []2223        if get_all_rooms_number() != []:24            min_room_number = min(get_all_rooms_number())25            max_room_number = max(get_all_rooms_number())26            room_number_label = "Enter room number({0}-{1}):".format(27                                str(min_room_number), str(max_room_number))28        else:29            room_number_label = ''30        self.room_number_label = QLabel(room_number_label)31        self.room_number_line_edit = QLineEdit()32        self.sleeping_furniture_name_label = QLabel("Name:")33        self.sleeping_furniture_name_line_edit = QLineEdit()34        self.sleeping_seats_label = QLabel("Sleeping Seats")35        self.sleeping_seats_line_edit = QLineEdit()36        self.quality_label = QLabel("Quality")37        self.quality_combo_box = QComboBox()3839        self.quality_combo_box.addItem("Excellent")40        self.quality_combo_box.addItem("Good")41        self.quality_combo_box.addItem("Bad")4243        layout.addWidget(self.room_number_label, 0, 0)44        layout.addWidget(self.room_number_line_edit, 0, 1)45        layout.addWidget(self.sleeping_furniture_name_label, 1, 0)46        layout.addWidget(self.sleeping_furniture_name_line_edit, 1, 1)47        layout.addWidget(self.sleeping_seats_label, 2, 0)48        layout.addWidget(self.sleeping_seats_line_edit, 2, 1)49        layout.addWidget(self.quality_label, 3, 0)50        layout.addWidget(self.quality_combo_box, 3, 1)5152        self.setLayout(layout)53        self.layout().setSizeConstraint(QLayout.SetFixedSize)5455    def is_information_invalid(self, name, sleeping_seats, room_number):56        return (not Validations.is_name(name) or57                not Validations.is_positive_integer(sleeping_seats) or58                not Validations.is_positive_integer(room_number) or59                not int(room_number) in get_all_rooms_number())6061    def error_message(self, name, sleeping_seats, room_number):62        properties = [Validations.is_name, Validations.is_positive_integer]63        data = [name, sleeping_seats]64        messages = ['name', 'sleeping seats']65        result = []66        for i in range(len(data)):67            if not properties[i](data[i]):68                result.append(messages[i] + ',')69        if (not Validations.is_positive_integer(room_number) or70           not int(room_number) in get_all_rooms_number()):71            result.append('room number' + ',')72        return ' '.join(result)7374    def add_button_click(self):75        room_number = self.room_number_line_edit.text()76        name = self.sleeping_furniture_name_line_edit.text()77        sleeping_seats = self.sleeping_seats_line_edit.text()78        quality = self.quality_combo_box.currentText()7980        if self.is_information_invalid(name, sleeping_seats, room_number):81            error_message = self.error_message(name, sleeping_seats,82                                               room_number)83            QMessageBox(QMessageBox.Critical, "Error",84                        "Invalid " + error_message[:len(error_message) - 1] +85                        ". Correct it!!!").exec_()86            return8788        new_sleeping_furniture = SleepingFurniture(name, quality,89                                                   int(sleeping_seats),90                                                   int(room_number))9192        if new_sleeping_furniture in self.sleeping_furnitures:93            notifier = QMessageBox(QMessageBox.Question,94                                   "Add New Sleeping Furniture",95                                   "You have already added this Sleeping" +96                                   " Furniture. Do you want to add it again?",97                                   QMessageBox.Yes | QMessageBox.No)98            notifier.setDefaultButton(QMessageBox.No)99            choosed_option = notifier.exec_()100101            if choosed_option == QMessageBox.No:102                return103            elif choosed_option == QMessageBox.Yes:104                self.sleeping_furnitures.append(new_sleeping_furniture)105                add_sleeping_furniture(new_sleeping_furniture)106                return107108        add_sleeping_furniture(new_sleeping_furniture)109        self.sleeping_furnitures.append(new_sleeping_furniture)110        QMessageBox(QMessageBox.Information, "Add New sleeping furniture",111                    "Congratulations. You successfully" +
...

12108.py

Source:12108.py

1#!/usr/bin/python32# uva: 121083def gcd(a, b):4    r = b % a5    while r != 0:6        b = a7        a = r8        r = b % a9    return a10def gcd_array(numbers):11    '''Provide an array of numbers. Array size must be > 1. Return value is gcd of all numbers'''12    if len(numbers) == 1:13        print('numbers array size = 1')14        return15    else:16        g = numbers[0]17        for i in range(1, len(numbers)):18            g = gcd(g, numbers[i])19    return g20def lcm(numbers):21    '''Provide an array of numbers. Array size must be > 1. Return value is lcm of all numbers'''22    if len(numbers) == 1:23        print('numbers array size = 1')24        return25    else:26        g = gcd_array(numbers)27        numbers = [i/g for i in numbers]28        m = 129        for i in numbers:30            m = m * i31        return int(g * m)32class Student:33    def __init__(self, student_id, awake_time, sleep_time, position):34        self.student_id = student_id35        self.sleep_time = sleep_time36        self.awake_time = awake_time37        self.state = 038        self.state_pos = 039        self.starting_state = position40        if self.starting_state <= self.awake_time:41            #self.state = 0 # I am awake42            self.state = 'awake'43            self.state_pos = self.awake_time - self.starting_state44        else:45            #self.state = 1 # I am sleeping46            self.state = 'sleeping'47            self.state_pos = self.sleep_time - (self.starting_state - self.awake_time)48        return49    def getState(self):50        return self.state, self.state_pos51    def nextState(self, awake_no, sleeping_no):52        if self.state == 'sleeping':53            if self.state_pos > 0:54                self.state_pos -= 155            elif self.state_pos == 0:56                self.state = 'awake'57                self.state_pos = self.awake_time - 158            else:59                print('I should not be printed!!')60        elif self.state == 'awake':61            if self.state_pos > 0:62                self.state_pos -= 163            elif self.state_pos == 0:64                if sleeping_no > awake_no:65                    self.state = 'sleeping'66                    self.state_pos = self.sleep_time - 167                else:68                    self.state = 'awake'69                    self.state_pos = self.awake_time - 170            else:71                print('I should not be printed!!')72        else:73            print('I should not be printed!!')74        return75class SleepingClass:76    def __init__(self, n):77        '''students is an array Student class'''78        self.sleeping = 079        self.awake = 080        self.n = n81        self.students = []82        return83    def getStudentState(self, student_id, awake_time, sleep_time, position):84        self.students.append(Student(student_id, awake_time, sleep_time, position))85    def calculateClassState(self):86        self.awake = 087        self.sleeping = 088        for s in self.students:89            # print(s.state)90            if s.state == 'awake':91                self.awake += 192            elif s.state == 'sleeping':93                self.sleeping += 194            else:95                print("1. I should not be printed!!")96        if self.n != self.awake + self.sleeping:97            print("Logic Error: Number of students don't match")98        return99    def nextTick(self):100        self.calculateClassState()101        for s in self.students:102            s.nextState(self.awake, self.sleeping)103        return104'''105s1 = Student(2, 4, 1)106s2 = Student(1, 5, 2)107s3 = Student(1, 4, 3)108print(s1.getstate())109print(s2.getstate())110print(s3.getstate())111'''112n = int(input())113caseno = 1114found = False115while n!= 0:116    c = SleepingClass(n)117    for i in range(n):118        line = [int(j) for j in input().strip().split()]119        c.getStudentState(i, line[0], line[1], line[2])120    t = [(s.awake_time + s.sleep_time) for s in c.students]121    l = lcm(t)122    #for s in c.students:123    #    print(s.getState())124    c.calculateClassState()125    for i in range(2*l):126        #print('pass no: {}'.format(i))127        if c.sleeping == 0:128            found = True129            break130        else:131            c.nextTick()132    if found == True:133        print('Case {}: {}'.format(caseno, i))134    elif found == False:135        print('Case {}: {}'.format(caseno, -1))136    # next input137    n = int(input())138    caseno += 1...

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.