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

How to use health_check method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

test_health_check.py

Source:test_health_check.py

...17from openstack import resource218from openstack.tests.functional import base19from openstack.tests.functional.load_balancer.v1 import test_listener20from openstack.tests.functional.load_balancer.v1 import test_load_balancer21def auto_create_health_check(conn, listener_id):22    """auto create a load balancer listener for functional test23    :param conn:24    :param listener_id: listener id this health check belongs to25    :return:26    """27    health_check = {28        "listener_id": listener_id,29        "healthcheck_protocol": "HTTP",30        "healthcheck_connect_port": 80,31        "healthcheck_interval": 5,32        "healthcheck_timeout": 10,33        "healthcheck_uri": "/health",34        "healthy_threshold": 3,35        "unhealthy_threshold": 336    }37    return conn.load_balancer.create_health_check(**health_check)38class TestHealthCheck(base.BaseFunctionalTest):39    NAME = "SDK-" + uuid.uuid4().hex40    lb = None41    listener = None42    health_check = None43    @classmethod44    def setUpClass(cls):45        super(TestHealthCheck, cls).setUpClass()46        # create an external load balancer47        router = cls.get_first_router()48        cls.lb = test_load_balancer.auto_create_external_lb(cls.conn,49                                                            cls.NAME,50                                                            router.id)51        # create a listener52        cls.listener = test_listener.auto_create_listener(cls.conn,53                                                          cls.NAME,54                                                          cls.lb.id)55        # create health check56        cls.health_check = auto_create_health_check(cls.conn, cls.listener.id)57    @classmethod58    def tearDownClass(cls):59        # delete health check created in setup60        cls.conn.load_balancer.delete_health_check(cls.health_check)61        # delete listener created in setup62        cls.conn.load_balancer.delete_listener(cls.listener)63        # delete load balancer created in setup64        job = cls.conn.load_balancer.delete_load_balancer(cls.lb)65        job = _job.Job(id=job.job_id)66        resource2.wait_for_status(cls.conn.load_balancer._session,67                                  job,68                                  "SUCCESS",69                                  interval=5,70                                  failures=["FAIL"])71    def test_update_health_check(self):72        updated = {73            "healthcheck_connect_port": 88,74            "healthcheck_interval": 5,75            "healthcheck_protocol": "HTTP",76            "healthcheck_timeout": 10,77            "healthcheck_uri": "/",78            "healthy_threshold": 3,79            "unhealthy_threshold": 280        }81        health_check = self.conn.load_balancer.update_health_check(82            self.health_check, **updated)83        self.assertEqual(88, health_check.healthcheck_connect_port)84        self.assertEqual(5, health_check.healthcheck_interval)85        self.assertEqual("HTTP", health_check.healthcheck_protocol)86        self.assertEqual(10, health_check.healthcheck_timeout)87        self.assertEqual("/", health_check.healthcheck_uri)88        self.assertEqual(3, health_check.healthy_threshold)89        self.assertEqual(2, health_check.unhealthy_threshold)90        self.health_check = health_check91    def test_get_health_check(self):92        health_check = self.conn.load_balancer.get_health_check(93            self.health_check)...

health_check_test.py

Source:health_check_test.py

...5from ..di import StandardDependencies6class HealthCheckTest(unittest.TestCase):7    def test_simple_success(self):8        health_check = test({'handler_class': HealthCheck, 'handler_config': {}})9        response = health_check()10        response_data = response[0]['data']11        self.assertEquals('success', response[0]['status'])12        self.assertEquals(200, response[1])13    def test_callable_success(self):14        test_callable = MagicMock(return_value=True)15        health_check = test({16            'handler_class': HealthCheck,17            'handler_config': {18                'callable': test_callable,19            }20        })21        response = health_check()22        response_data = response[0]['data']23        self.assertEquals('success', response[0]['status'])24        self.assertEquals(200, response[1])25        test_callable.assert_called_once()26    def test_callable_failure(self):27        test_callable = MagicMock(return_value=False)28        health_check = test({29            'handler_class': HealthCheck,30            'handler_config': {31                'callable': test_callable,32            }33        })34        response = health_check()35        response_data = response[0]['data']36        self.assertEquals('failure', response[0]['status'])37        self.assertEquals(500, response[1])38        test_callable.assert_called_once()39    def test_check_dependencies_success(self):40        health_check = test(41            {42                'handler_class': HealthCheck,43                'handler_config': {44                    'services': ['sup'],45                },46            },47            bindings={'sup': 'hey'},48        )49        response = health_check()50        response_data = response[0]['data']51        self.assertEquals('success', response[0]['status'])52        self.assertEquals(200, response[1])53    def test_check_dependencies_failure(self):54        health_check = test({55            'handler_class': HealthCheck,56            'handler_config': {57                'services': ['sup'],58            },59        })60        response = health_check()61        response_data = response[0]['data']62        self.assertEquals('failure', response[0]['status'])63        self.assertEquals(500, response[1])64    def test_documentation(self):65        health_check = HealthCheck(StandardDependencies())66        health_check.configure({})67        documentation = health_check.documentation()[0]68        self.assertEquals(0, len(documentation.parameters))69        self.assertEquals(1, len(documentation.responses))70        self.assertEquals([200], [response.status for response in documentation.responses])71        success_response = documentation.responses[0]72        self.assertEquals(['status', 'data', 'pagination', 'error', 'input_errors'],...

provider.py

Source:provider.py

...19class Route53Provider(Route53Api, ServiceLifecycleHook):20    def get_change(self, context: RequestContext, id: ResourceId) -> GetChangeResponse:21        change_info = ChangeInfo(Id=id, Status=ChangeStatus.INSYNC, SubmittedAt=datetime.now())22        return GetChangeResponse(ChangeInfo=change_info)23    def get_health_check(24        self, context: RequestContext, health_check_id: HealthCheckId25    ) -> GetHealthCheckResponse:26        health_check: Optional[route53_models.HealthCheck] = route53_backends[context.account_id][27            "global"28        ].health_checks.get(health_check_id, None)29        if not health_check:30            raise NoSuchHealthCheck(31                f"No health check exists with the specified ID {health_check_id}"32            )33        health_check_config = {34            "Disabled": health_check.disabled,35            "EnableSNI": health_check.enable_sni,36            "FailureThreshold": health_check.failure_threshold,37            "FullyQualifiedDomainName": health_check.fqdn,38            "HealthThreshold": health_check.health_threshold,39            "Inverted": health_check.inverted,40            "IPAddress": health_check.ip_address,41            "MeasureLatency": health_check.measure_latency,42            "Port": health_check.port,43            "RequestInterval": health_check.request_interval,44            "ResourcePath": health_check.resource_path,45            "Type": health_check.type_,46        }47        return GetHealthCheckResponse(48            HealthCheck=HealthCheck(49                Id=health_check.id,50                CallerReference=health_check.caller_reference,51                HealthCheckConfig=health_check_config,52            )53        )54    def delete_health_check(55        self, context: RequestContext, health_check_id: HealthCheckId56    ) -> DeleteHealthCheckResponse:57        health_check = route53_backends[context.account_id]["global"].delete_health_check(58            health_check_id59        )60        if not health_check:61            raise NoSuchHealthCheck(62                f"No health check exists with the specified ID {health_check_id}"63            )...

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.