Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use run_with_timeout method in lisa

Best Python code snippet using lisa_python

appcontroller_client.py

Source:appcontroller_client.py

...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,...

grade_best4.py

Source:grade_best4.py

...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)...

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.