How to use run_with_timeout method in lisa

Best Python code snippet using lisa_python

appcontroller_client.py

Source:appcontroller_client.py Github

copy

Full Screen

...59 """60 self.host = host61 self.server = SOAPpy.SOAPProxy('https://{0}:{1}'.format(host, self.PORT))62 self.secret = secret63 def run_with_timeout(self, timeout_time, default, num_retries,64 http_error_is_success, function, *args):65 """Runs the given function, aborting it if it runs too long.66 Args:67 timeout_time: The number of seconds that we should allow function to68 execute for.69 default: The value that should be returned if the timeout is exceeded.70 num_retries: The number of times we should retry the SOAP call if we see71 an unexpected exception.72 http_error_is_success: A bool that indicates if receiving a HTTPError is73 the expected result.74 function: The function that should be executed.75 *args: The arguments that will be passed to function.76 Returns:77 Whatever function(*args) returns if it runs within the timeout window, and78 default otherwise.79 Raises:80 AppControllerException: If the AppController we're trying to connect to is81 not running at the given IP address, or if it rejects the SOAP request.82 """83 def timeout_handler(_, __):84 """Raises a TimeoutException if the function we want to execute takes85 too long to run.86 Raises:87 TimeoutException: If a SIGALRM is raised.88 """89 raise TimeoutException()90 try:91 retval = function(*args)92 except TimeoutException:93 return default94 except socket.error as exception:95 if num_retries > 0:96 sys.stderr.write("Saw socket exception {0} when communicating with the " \97 "AppController, retrying momentarily. Message is {1}".format(exception, exception.msg))98 return self.run_with_timeout(timeout_time, default, num_retries - 1,99 http_error_is_success, function, *args)100 else:101 raise exception102 except SOAPpy.Errors.HTTPError as exception:103 if http_error_is_success:104 return "true"105 else:106 sys.stderr.write("Saw HTTPError {0} when communicating with the " \107 "AppController. Message is {1}".format(exception, exception.msg))108 return default109 except Exception as exception:110 # This 'except' should be catching ssl.SSLError, but that error isn't111 # available when running in the App Engine sandbox.112 # TODO(cgb): App Engine 1.7.7 adds ssl support, so we should be able to113 # fix this when we update our SDK to that version.114 sys.stderr.write("Saw exception {0} when communicating with the " \115 "AppController.".format(str(exception)))116 return default117 if retval == self.BAD_SECRET_MESSAGE:118 raise AppControllerException("Could not authenticate successfully" + \119 " to the AppController. You may need to change the keyname in use.")120 return retval121 def set_parameters(self, locations, credentials, app=None):122 """Passes the given parameters to an AppController, allowing it to start123 configuring API services in this AppScale deployment.124 Args:125 locations: A list that contains the first node's IP address.126 credentials: A list that contains API service-level configuration info,127 as well as a mapping of IPs to the API services they should host128 (excluding the first node).129 app: A list of the App Engine apps that should be started.130 Raises:131 AppControllerException: If the remote AppController indicates that there132 was a problem with the parameters passed to it.133 """134 if app is None:135 app = 'none'136 result = self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "Error",137 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.set_parameters,138 locations, credentials, [app], self.secret)139 if result.startswith('Error'):140 raise AppControllerException(result)141 def get_all_public_ips(self):142 """Queries the AppController for a list of all the machines running in this143 AppScale deployment, and returns their public IP addresses.144 Returns:145 A list of the public IP addresses of each machine in this AppScale146 deployment.147 """148 return json.loads(self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "[]",149 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR,150 self.server.get_all_public_ips, self.secret))151 def get_role_info(self):152 """Queries the AppController to determine what each node in the deployment153 is doing and how it can be externally or internally reached.154 Returns:155 A dict that contains the public IP address, private IP address, and a list156 of the API services that each node runs in this AppScale deployment.157 """158 return json.loads(self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "{}",159 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.get_role_info,160 self.secret))161 def get_status(self):162 """Queries the AppController to learn information about the machine it runs163 on.164 This includes information about the CPU, memory, and disk of that machine,165 as well as what machine that AppController connects to for database access166 (via the UserAppServer).167 Returns:168 A str containing information about the CPU, memory, and disk usage of that169 machine, as well as where the UserAppServer is located.170 """171 return self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "", 172 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.status,173 self.secret)174 def get_api_status(self):175 """Queries the AppController to see what the status of Google App Engine176 APIs are in this AppScale deployment, reported to it by the API Checker.177 APIs can be either 'running', 'failed', or 'unknown' (which typically178 occurs when AppScale is first starting up).179 Returns:180 A dict that maps each API name (a str) to its status (also a str).181 """182 return json.loads(self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "{}",183 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.get_api_status,184 self.secret))185 def get_database_information(self):186 """Queries the AppController to see what database is being used to implement187 support for the Google App Engine Datastore API, and how many replicas are188 present for each piece of data.189 Returns:190 A dict that indicates both the name of the database in use (with the key191 'table', for historical reasons) and the replication factor (with the192 key 'replication').193 """194 return json.loads(self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "{}",195 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR,196 self.server.get_database_information, self.secret))197 def upload_tgz(self, tgz_filename, email):198 """Tells the AppController to use the AppScale Tools to upload the Google199 App Engine application at the specified location.200 Args:201 tgz_filename: A str that points to a .tar.gz file on the local filesystem202 containing the user's Google App Engine application.203 email: A str containing an e-mail address that should be registered as the204 administrator of this application.205 Return:206 A str that indicates either that the app was successfully uploaded, or the207 reason why the application upload failed.208 """209 return self.run_with_timeout(self.UPLOAD_TAR_GZ_TIME,210 "Timeout uploading app.", self.UPLOAD_TAR_GZ_RETRIES,211 self.ALLOW_HTTP_ERROR, self.server.upload_tgz_file, tgz_filename, email,212 self.secret)213 def get_stats(self):214 """Queries the AppController to get server-level statistics and a list of215 App Engine apps running in this cloud deployment across all machines.216 Returns:217 A list of dicts, where each dict contains server-level statistics (e.g.,218 CPU, memory, disk usage) about one machine.219 """220 return json.loads(self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "[]",221 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.get_stats_json,222 self.secret))223 def is_initialized(self):224 """Queries the AppController to see if it has started up all of the API225 services it is responsible for on its machine.226 Returns:227 A bool that indicates if all API services have finished starting up on228 this machine.229 """230 return self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, False,231 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR,232 self.server.is_done_initializing, self.secret)233 def start_roles_on_nodes(self, roles_to_nodes):234 """Dynamically adds the given machines to an AppScale deployment, with the235 specified roles.236 Args:237 A JSON-dumped dict that maps roles to IP addresses.238 Returns:239 The result of executing the SOAP call on the remote AppController.240 """241 return self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "Error",242 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR,243 self.server.start_roles_on_nodes, roles_to_nodes, self.secret)244 def stop_app(self, app_id):245 """Tells the AppController to no longer host the named application.246 Args:247 app_id: A str that indicates which application should be stopped.248 Returns:249 The result of telling the AppController to no longer host the app.250 """251 return self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "Error",252 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.stop_app,253 app_id, self.secret)254 def is_app_running(self, app_id):255 """Queries the AppController to see if the named application is running.256 Args:257 app_id: A str that indicates which application we should be checking258 for.259 Returns:260 True if the application is running, False otherwise.261 """262 return self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "Error",263 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.is_app_running,264 app_id, self.secret)265 def done_uploading(self, app_id, remote_app_location):266 """Tells the AppController that an application has been uploaded to its267 machine, and where to find it.268 Args:269 app_id: A str that indicates which application we have copied over.270 remote_app_location: A str that indicates the location on the remote271 machine where the App Engine application can be found.272 """273 return self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "Error",274 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.done_uploading,275 app_id, remote_app_location, self.secret)276 def update(self, apps_to_run):277 """Tells the AppController which applications to run, which we assume have278 already been uploaded to that machine.279 Args:280 apps_to_run: A list of apps to start running on nodes running the App281 Engine service.282 """283 return self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "Error",284 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.update,285 apps_to_run, self.secret)286 def gather_logs(self):287 """ Tells the AppController to copy logs from all machines to a tar.gz file288 stored in the AppDashboard's static file directory, so that users can289 download it.290 """291 return self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, (False, ""),292 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.gather_logs,293 self.secret)294 def run_groomer(self):295 """ Tells the AppController to clean up entities in the Datastore that have296 been soft deleted, and to generate statistics about the entities still in297 the Datastore (which can be viewed in the AppDashboard).298 """299 return self.run_with_timeout(self.DEFAULT_TIMEOUT_TIME, "Error",300 self.DEFAULT_NUM_RETRIES, self.NO_HTTP_ERROR, self.server.run_groomer,...

Full Screen

Full Screen

grade_best4.py

Source:grade_best4.py Github

copy

Full Screen

...61 betterLearner, worseLearner = None, None62 if group=='author':63 try:64 from gen_data import author65 auth_string = run_with_timeout(author,seconds_per_test_case,(),{})66 if auth_string == 'tb34':67 incorrect = True68 msgs.append(" Incorrect author name (tb34)")69 points_earned = -1070 elif auth_string == '':71 incorrect = True72 msgs.append(" Empty author name")73 points_earned = -1074 else:75 incorrect = False76 except Exception as e:77 incorrect = True78 msgs.append(" Exception occured when calling author() method: {}".format(e))79 points_earned = -1080 else:81 if group=="best4dt":82 from gen_data import best4DT83 dataX, dataY = run_with_timeout(best4DT,seconds_per_test_case,(),{'seed':seed})84 same_dataX,same_dataY = run_with_timeout(best4DT,seconds_per_test_case,(),{'seed':seed})85 diff_dataX,diff_dataY = run_with_timeout(best4DT,seconds_per_test_case,(),{'seed':seed+1})86 betterLearner = DTLearner87 worseLearner = LinRegLearner88 elif group=='best4lr':89 from gen_data import best4LinReg90 dataX, dataY = run_with_timeout(best4LinReg,seconds_per_test_case,(),{'seed':seed})91 same_dataX, same_dataY = run_with_timeout(best4LinReg,seconds_per_test_case,(),{'seed':seed})92 diff_dataX, diff_dataY = run_with_timeout(best4LinReg,seconds_per_test_case,(),{'seed':seed+1})93 betterLearner = LinRegLearner94 worseLearner = DTLearner95 num_samples = dataX.shape[0]96 cutoff = int(num_samples*0.6)97 worse_better_err = []98 for run in range(max_tests):99 permutation = np.random.permutation(num_samples)100 train_X,train_Y = dataX[permutation[:cutoff]], dataY[permutation[:cutoff]]101 test_X,test_Y = dataX[permutation[cutoff:]], dataY[permutation[cutoff:]]102 better = betterLearner()103 worse = worseLearner()104 better.addEvidence(train_X,train_Y)105 worse.addEvidence(train_X,train_Y)106 better_pred = better.query(test_X)...

Full Screen

Full Screen

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run lisa 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful