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

How to use host_statuses method in autotest

Best Python code snippet using autotest_python

test_maintenance.py

Source:test_maintenance.py

...50    mock_options.default_duration = '30m'51    mock_options.default_percentage = self.DEFAULT_SLA_PERCENTAGE52    mock_options.timeout = '2h'53    return mock_options54  def create_host_statuses(self, maintenance_mode, skip_hosts=None):55    return [HostStatus(host=hostname, mode=maintenance_mode) for hostname in self.HOSTNAMES56            if not skip_hosts or hostname not in skip_hosts]57  def create_start_maintenance_result(self, skip_hosts=None):58    host_statuses = self.create_host_statuses(MaintenanceMode.SCHEDULED, skip_hosts)59    response = self.create_simple_success_response()60    response.result.startMaintenanceResult = StartMaintenanceResult(statuses=set(host_statuses))61    return response62  def create_end_maintenance_result(self):63    host_statuses = self.create_host_statuses(MaintenanceMode.NONE)64    response = self.create_simple_success_response()65    response.result.endMaintenanceResult = EndMaintenanceResult(statuses=set(host_statuses))66    return response67  def create_drain_hosts_result(self):68    host_statuses = self.create_host_statuses(MaintenanceMode.DRAINING)69    response = self.create_simple_success_response()70    response.result.drainHostsResult = DrainHostsResult(statuses=set(host_statuses))71    return response72  def create_maintenance_status_result(self):73    host_statuses = self.create_host_statuses(MaintenanceMode.NONE)74    response = self.create_simple_success_response()75    response.result.maintenanceStatusResult = MaintenanceStatusResult(statuses=set(host_statuses))76    return response77  def create_drained_status_result(self, hosts):78    host_statuses = [79        HostStatus(host=hostname, mode=MaintenanceMode.DRAINED) for hostname in hosts.hostNames]80    response = self.create_simple_success_response()81    response.result.maintenanceStatusResult = MaintenanceStatusResult(statuses=set(host_statuses))82    return response83  def test_start_maintenance_hosts(self):84    mock_options = self.make_mock_options()85    mock_api, mock_scheduler_proxy = self.create_mock_api()86    mock_scheduler_proxy.startMaintenance.return_value = self.create_start_maintenance_result()87    with contextlib.nested(...

nagstatus.py

Source:nagstatus.py

1#!/usr/bin/env python2import time3def get_property(line):4    line_split = line.strip().split('=')5    #returns the property name and remaining data recombined after the split above6    if len(line_split) < 2:7        raise Exception("Line is not a key/value pair.")8    #Ensures that all characters are valid unicode strings9    key = line_split[0].encode('utf-8', 'ignore')10    try:11        value = u'='.join(line_split[1:])12    except:13        value = "DASHBOARD ERROR!  Plugin output could not be decoded due to invalid NRPE output!!!"14    return key, value15def try_to_convert(value):16    """Tries to convert [value] to an int, returns the original string on fail"""17    try:18        return int(value)19    except:20        return value21def get_nag_status(filename, threshold = 0):22    """Reads status.dat referred to by [filename] and returns a dictionary version of it"""23    status_file = filename24    f = open(status_file, 'r')25    line = f.readline()26    host_statuses = {}27    this_host = None28    this_service = None29    group_type = None30    for line in f:31        if line.strip().endswith('{'):32            group_type = line.strip().split()[0]33            continue34        try:35            this_property, value = get_property(line) #fails on lines without =, the try makes us pass36            #not yet reading programstatus or info37            if group_type == 'hoststatus':38                if this_property == 'host_name':39                    this_host = value40                    host_statuses[this_host] = {}41                    host_statuses[this_host]['HOST'] = {}42                    host_statuses[this_host]['HOST']['service_comments'] = {}43                else:44                    host_statuses[this_host]['HOST'][this_property] = try_to_convert(value)45            elif group_type == 'servicestatus':46                #host_name always comes before service_description47                if this_property == 'host_name':48                    this_host = value49                elif this_property == 'service_description':50                    this_service = value51                    host_statuses[this_host][this_service] = {}52                    host_statuses[this_host][this_service][this_property] = value #handy place to have the service description and host name53                    host_statuses[this_host][this_service]['host_name'] = this_host54                    host_statuses[this_host][this_service]['service_comments'] = {}55                else:56                    host_statuses[this_host][this_service][this_property] = try_to_convert(value)57                    if this_property == 'current_state' and host_statuses[this_host][this_service][this_property] < threshold:58                        #by simply removing the service here, subsequent attempts to add data fail to the next loop iteration59                        del host_statuses[this_host][this_service]60                    elif this_property == 'last_state_change':61                        host_statuses[this_host][this_service]['current_duration'] = time.time() - try_to_convert(value)62            elif group_type == 'servicecomment':63                if this_property == 'host_name':64                    this_host = value65                elif this_property == 'service_description':66                    this_service = value67                elif this_property == 'entry_type':68                    # Need to hang on to this one for one more line69                    this_entry_type = try_to_convert(value)70                elif this_property == 'comment_id':71                    this_comment_id = value72                    host_statuses[this_host][this_service]['service_comments'][value] = {73                        'entry_type': this_entry_type,74                        'comment_id': this_comment_id75                    }76                else:77                    host_statuses[this_host][this_service]['service_comments'][this_comment_id][this_property] = try_to_convert(value)78            elif group_type == 'hostcomment':79                if this_property == 'host_name':80                    this_host = value81                    this_service = 'HOST'82                elif this_property == 'entry_type':83                    # Need to hang on to this one for one more line84                    this_entry_type = try_to_convert(value)85                elif this_property == 'comment_id':86                    this_comment_id = value87                    host_statuses[this_host][this_service]['service_comments'][value] = {88                        'entry_type': this_entry_type,89                        'comment_id': this_comment_id90                    }91                else:92                    host_statuses[this_host][this_service]['service_comments'][this_comment_id][this_property] = try_to_convert(value)93        except:94            pass95    f.close()96    return host_statuses97if __name__ == "__main__":98    #simply me testing, this chunk is not needed99    status = get_nag_status('status.dat', 0)100    import json...

mysql_status.py

Source:mysql_status.py

1#!/usr/bin/python2# -*- coding: utf-8 -*-3# (c) 2018, James Cammarata <jimi@sngx.net>4# Copied and modified mainly from the mysql_variables.py module.5#6# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)7from __future__ import absolute_import, division, print_function8__metaclass__ = type9ANSIBLE_METADATA = {'metadata_version': '1.1',10                    'status': ['preview'],11                    'supported_by': 'community'}12DOCUMENTATION = '''13'''14EXAMPLES = '''15'''16import os17import warnings18from re import match19try:20    import MySQLdb21except ImportError:22    mysqldb_found = False23else:24    mysqldb_found = True25from ansible.errors import AnsibleError26from ansible.plugins.lookup import LookupBase27from ansible.module_utils._text import to_native28# the host field will be the terms passed to the lookup29MYSQL_FIELDS = [30   'user', 'passwd', 'db', 'port', 'unix_socket', 'connect_timeout',31   'config_file', 'ssl_cert', 'ssl_key', 'ssl_ca',32]33def typedvalue(value):34    """35    Convert value to number whenever possible, return same value36    otherwise.37    >>> typedvalue('3')38    339    >>> typedvalue('3.0')40    3.041    >>> typedvalue('foobar')42    'foobar'43    """44    try:45        return int(value)46    except ValueError:47        pass48    try:49        return float(value)50    except ValueError:51        pass52    return value53class LookupModule(LookupBase):54    def getstatus(self, cursor, status_name):55        try:56            cursor.execute("SHOW STATUS LIKE %s", (status_name,))57            mysqlstatus_res = cursor.fetchall()58            return mysqlstatus_res59        except:60            # FIXME: proper error handling here61            return None62    def run(self, terms, variables, **kwargs):63        if not mysqldb_found:64            raise AnsibleError("The MySQL-python module is required.")65        else:66            warnings.filterwarnings('error', category=MySQLdb.Warning)67        options = {}68        for field in MYSQL_FIELDS:69            if field in kwargs:70                if field.startswith('ssl_'):71                    if 'ssl' not in options:72                        options['ssl'] = {}73                    ssl_entry = field.replace('ssl_', '')74                    options['ssl'][ssl_entry] = kwargs[field]75                elif field == 'config_file':76                    options['read_default_file'] = kwargs[field]77                else:78                    options[field] = kwargs[field]79        host_statuses = []80        for term in terms:81            host_status = {}82            try:83                options['host'] = term84                db_connection = MySQLdb.connect(**options)85                cursor = db_connection.cursor()86                try:87                    mysqlstatus_res = self.getstatus(cursor, r'%')88                    if mysqlstatus_res:89                        mysqlstatus_res = [(x[0].lower(), typedvalue(x[1])) for x in mysqlstatus_res]90                        for res in mysqlstatus_res:91                            host_status[res[0]] = res[1]92                        host_status["functional"] = True93                    else:94                        raise Exception("No wsrep status variables were found.")95                except Exception as e:96                    host_status["functional"] = False97                    host_status["reason"] = to_native(e)98            except Exception as e:99                host_status["functional"] = False100                if 'read_default_file' in options:101                    if os.path.exists(options['read_default_file']):102                        host_status["reason"] = "Unable to connect to database, check login_user and " \103                                                "login_password are correct or %s has the credentials. " \104                                                "Exception message: %s" % (options['read_default_file'], to_native(e))105                    else:106                        host_status["reason"] = "Unable to find %s. Exception message: %s" % (options['read_default_file'], to_native(e))107                else:108                    host_status["reason"] = "Unable to connect to database on %s: %s" % (term, to_native(e))109            host_statuses.append(host_status)...

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.