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

How to use check_service_health method in localstack

Best Python code snippet using localstack_python

plugins.py

Source:plugins.py

...141        print_error = retries <= 0142        # loop through plugins and check service status143        for name, plugin in SERVICE_PLUGINS.items():144            if name in apis:145                check_service_health(146                    api=name, print_error=print_error, expect_shutdown=expect_shutdown147                )148        for additional in additional_checks:149            additional(expect_shutdown=expect_shutdown)150    except Exception as e:151        if retries <= 0:152            LOG.exception("Error checking state of local environment (after some retries)")153            raise e154        time.sleep(3)155        check_infra(156            retries - 1,157            expect_shutdown=expect_shutdown,158            apis=apis,159            additional_checks=additional_checks,160        )161def wait_for_infra_shutdown(apis=None):162    apis = apis or canonicalize_api_names()163    names = [name for name, plugin in SERVICE_PLUGINS.items() if name in apis]164    def check(name):165        check_service_health(api=name, expect_shutdown=True)166        LOG.debug("[shutdown] api %s has shut down", name)167    # no special significance to 10 workers, seems like a reasonable number given the number of services we have168    with ThreadPoolExecutor(max_workers=10) as executor:169        executor.map(check, names)170def check_service_health(api, print_error=False, expect_shutdown=False):171    try:172        plugin = SERVICE_PLUGINS.get(api)173        plugin.check(expect_shutdown=expect_shutdown, print_error=print_error)174        record_service_health(api, "running")175    except Exception as e:176        if not expect_shutdown:177            LOG.warning('Service "%s" not yet available, retrying...' % api)178        else:179            LOG.warning('Service "%s" still shutting down, retrying...' % api)180        raise e181def reload_services_health():182    check_infra(retries=0)183def record_service_health(api, status):184    # TODO: consider making in-memory calls here, to optimize performance...

consul_zabbix.py

Source:consul_zabbix.py

...108        :param data_center:109        :return:110        """111        return self.consul.catalog.services(dc=data_center)112    def check_service_health(self, service):113        """114        :param service:115        :return:116        """117        c = self.consul.health.checks(service=service)118        return c119    def list(self, subject):120        """121        :param subject:122        :return:123        """124        assert not hasattr(self.consul, subject)125        obj_getattr = getattr(self.consul, subject)126        return obj_getattr.list()127def get_service_list(host="127.0.0.1", port=8500, token=None):128    """129    :return:130    """131    data = list()132    if token is None:133        params = {134            "host": host,135            "port": port136        }137    else:138        params = {139            "host": host,140            "port": port,141            "token": token142        }143    cq = ConsulQuery(**params)144    for key, value in cq.get_services(data_center="dc1")[1].items():145        for ser in cq.check_service_health(service=key)[1]:146            data.append({"{#SERVICE}": key, "{#NODE}": ser["Node"], "{#NAME}": ser['Name']})147    return json.dumps({"data": data}, indent=4)148def check_service(service, node, host="127.0.0.1", port=8500, token=None):149    """150    :param service:151    :param node:152    :param host:153    :param port:154    :param token:155    :return:156    """157    data = list()158    if token is None:159        params = {160            "host": host,161            "port": port162        }163    else:164        params = {165            "host": host,166            "port": port,167            "token": token168        }169    cq = ConsulQuery(**params)170    for ser in cq.check_service_health(service=service)[1]:171        if ser['Node'] == node and ser['Status'] == 'passing':172            return 1173        elif ser['Node'] == node and ser['Status'] != 'passing':174            return 0175    return 0176def useage():177    print("%s -h\t#å¸®å©ææ¡£" % sys.argv[0])178    print("%s -h IP -pç«¯å£ -t tokenï¼å¯éï¼ -d\t#è¿åæå¡åç§°åè¡¨" % sys.argv[0])179    print("%s -s\tæå¡åç§°\t#è¿åæå¡ç¶æ" % sys.argv[0])180def main():181    if len(sys.argv) == 1:182        useage()183        sys.exit()184    try:...

healthservicehandler.py

Source:healthservicehandler.py

...15        self.result = []16        microservice = 017        microservices = len(service)18        while microservice < microservices:19            self.check_service_health(service[microservice])20            microservice += 121        return self.result22    # Check the service status and availabilty23    def check_service_health(self, service):24        try:25            now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')26            harmless_status = range(200, 308)27            client_error_status = range(400, 499)28            server_error_status = range(500, 599)29            self.output['Service Url'] = service30            self.result.append(self.output['Service Url'])31            response = requests.head(service) # request url32            response_status = response.status_code # response status code33            for x in harmless_status:34                if response_status == x : # The http response is harmless to the functionality of the service35                    self.output['Health-Check Status'] = "Online"36                    self.result.append(self.output['Health-Check Status'])37            for y in client_error_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.