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

Source:test_health.py

...4from hamcrest import assert_that, equal_to, is_5from microcosm.api import create_object_graph6from microcosm.loaders import load_from_dict7from parameterized import parameterized8def test_health_check():9    """10    Default health check returns OK.11    """12    graph = create_object_graph(name="example", testing=True)13    graph.use("health_convention")14    client = graph.flask.test_client()15    graph.health_convention.optional_checks["foo"] = _health_check()16    response = client.get("/api/health")17    assert_that(response.status_code, is_(equal_to(200)))18    assert_that(response.json, is_(equal_to({19        "name": "example",20        "ok": True,21        "checks": {22            "build_num": {23                "message": "undefined",24                "ok": True25            },26            "sha1": {27                "message": "undefined",28                "ok": True29            },30        },31    })))32def test_health_check_required_check_failed():33    """34    Should return 503 on health check failure.35    """36    loader = load_from_dict(37        health_convention=dict(38            include_build_info="false",39        ),40    )41    graph = create_object_graph(name="example", testing=True, loader=loader)42    graph.use("health_convention")43    client = graph.flask.test_client()44    graph.health_convention.checks["foo"] = _health_check(False)45    response = client.get("/api/health")46    assert_that(response.status_code, is_(equal_to(503)))47    assert_that(response.json, is_(equal_to({48        "name": "example",49        "ok": False,50        "checks": {51            "foo": {52                "message": "failure!",53                "ok": False,54            },55        },56    })))57def test_health_check_optional_check_failed():58    """59    Optional checks should not be evaluated by default60    """61    loader = load_from_dict(62        health_convention=dict(63            include_build_info="false",64        ),65    )66    graph = create_object_graph(name="example", testing=True, loader=loader)67    graph.use("health_convention")68    client = graph.flask.test_client()69    graph.health_convention.optional_checks["foo"] = _health_check(False)70    response = client.get("/api/health")71    assert_that(response.status_code, is_(equal_to(200)))72    assert_that(response.json, is_(equal_to({73        "name": "example",74        "ok": True,75    })))76def test_health_check_with_build_info():77    graph = create_object_graph(name="example", testing=True)78    graph.use("health_convention")79    client = graph.flask.test_client()80    response = client.get("/api/health", query_string=dict(full=True))81    assert_that(response.status_code, is_(equal_to(200)))82    assert_that(response.json, is_(equal_to(dict(83        name="example",84        ok=True,85        checks=dict(86            build_num=dict(87                message="undefined",88                ok=True,89            ),90            sha1=dict(91                message="undefined",92                ok=True,93            ),94        ),95    ))))96@parameterized([97    # When non-optional, always end up in the response98    (False, False, True),99    (False, True, True),100    # Optional checks conditionally show up101    (True, False, False),102    (True, True, True),103])104def test_health_check_custom_checks(optional_check, full_check, expect_check_response):105    loader = load_from_dict(106        health_convention=dict(107            include_build_info="false",108        ),109    )110    graph = create_object_graph(name="example", testing=True, loader=loader)111    graph.use("health_convention")112    client = graph.flask.test_client()113    if optional_check:114        graph.health_convention.optional_checks["foo"] = _health_check()115    else:116        graph.health_convention.checks["foo"] = _health_check()117    response = client.get("/api/health", query_string=dict(full=full_check))118    assert_that(response.status_code, is_(equal_to(200)))119    expected_response = {120        "name": "example",121        "ok": True,122    }123    if expect_check_response:124        expected_response.update({125            "checks": {126                "foo": {127                    "message": "hi",128                    "ok": True,129                },130            },131        })132    assert_that(response.json, is_(equal_to(expected_response)))133@parameterized([134    # When non-optional, always end up in the response and fail135    (False, False, True),136    (False, True, True),137    # Optional checks conditionally show up, and only fail is specified138    (True, False, False),139    (True, True, True),140])141def test_health_check_custom_check_failed(optional_check, full_check, expect_failure):142    loader = load_from_dict(143        health_convention=dict(144            include_build_info="false",145        ),146    )147    graph = create_object_graph(name="example", testing=True, loader=loader)148    graph.use("health_convention")149    client = graph.flask.test_client()150    if optional_check:151        graph.health_convention.optional_checks["foo"] = _health_check(False)152    else:153        graph.health_convention.checks["foo"] = _health_check(False)154    response = client.get("/api/health", query_string=dict(full=full_check))155    if expect_failure:156        assert_that(response.status_code, is_(equal_to(503)))157        assert_that(response.json, is_(equal_to({158            "name": "example",159            "ok": False,160            "checks": {161                "foo": {162                    "message": "failure!",163                    "ok": False,164                },165            },166        })))167    else:168        assert_that(response.status_code, is_(equal_to(200)))169        assert_that(response.json, is_(equal_to({170            "name": "example",171            "ok": True,172        })))173def _health_check(success=True):174    if success:175        return lambda graph: "hi"176    def fail(graph):177        raise Exception("failure!")...

health_monitor.py

Source:health_monitor.py

...15        self._failure_count = 016    @property17    def check_name(self):18        return self._check_name19    def do_health_check(self):20        """21        Executes the check defined by do_health_check_impl, and keeps track of the failure counts. This method22        returns True only if the number of failures exceeds the threshold set, otherwise, False.23        """24        is_passing = False25        try:26            is_passing = self.do_health_check_impl()27        except:28            logging.exception("An error occurred during health check!")29        self._failure_count = 0 if is_passing else self._failure_count + 130        if self._failure_count > 0:31            logging.info("Failure count/threshold: %d/%d", self._failure_count, self._failure_threshold)32        return self._failure_count < self._failure_threshold33    @abstractmethod34    def do_health_check_impl(self):35        """Defines the check logic (to be implemented by subclasses)."""36        pass37    @abstractmethod38    def handle_status(self, is_passing):39        """40        Defines the logic to handle the check status (to be implemented by subclasses).41        """42        pass43    @abstractmethod44    def continue_checking(self):45        """46        Return True to signal the HealthMonitor thread to continue executing this check (to be implemented by47        subclasses).48        """49        pass50class HealthMonitor(looping_thread.LoopingThread):51    """52    Defines a Consul TTL check, keeps executing the supplied HealthCheck, and updates the Consul check status53    accordingly.54    """55    CONSUL_BASE_URL = "http://localhost:8500/v1"56    CONSUL_REGISTER_CHECK_URL = CONSUL_BASE_URL + "/agent/check/register"57    CONSUL_UPDATE_CHECK_URL = CONSUL_BASE_URL + "/agent/check/update/{}"58    def __init__(self, health_check, check_interval_seconds):59        """60        :param health_check: A HealthCheck instance that implements the check logic.61        :param check_interval_seconds: The time interval (in seconds) between two consecutive checks.62        """63        super().__init__(check_interval_seconds)64        self._health_check = health_check65        self._create_consul_check()66    def _create_consul_check(self):67        ttl = self._interval_seconds + 568        logging.info("Creating Consul TTL check: %s, with TTL: %ds", self._health_check.check_name, ttl)69        body = {70            "Name": self._health_check.check_name,71            "TTL": "%ds" % ttl,72        }73        response = requests.put(self.CONSUL_REGISTER_CHECK_URL, json=body)74        logging.info("Response (%d) %s", response.status_code, response.text)75        response.raise_for_status()76    def _update_consul_check(self, is_passing):77        status = "passing" if is_passing else "critical"78        logging.info("Updating Consul TTL check: %s, with status: %s", self._health_check.check_name, status)79        response = requests.put(self.CONSUL_UPDATE_CHECK_URL.format(self._health_check.check_name),80                                json={"Status": status})81        logging.info("Response (%d) %s", response.status_code, response.text)82        response.raise_for_status()83    def do_one_run(self):84        """85        Executes the supplied HealthCheck's do_health_check method, then passes the result to the HealthCheck's86        handle_status method, It also updates the Consul check status with the result, and finally, evaluates the87        HealthCheck's continue_checking method to decide whether to stop or not.88        """89        is_passing = self._health_check.do_health_check()90        try:91            self._update_consul_check(is_passing)92        except:93            logging.exception("An error occurred during updating Consul's check!")94        self._health_check.handle_status(is_passing)95        if self._health_check.continue_checking() is False:96            logging.info("HealthCheck decided to stop the monitoring loop!")...

health_checker.py

Source:health_checker.py

1import os2from rectifier import settings3from healthcheck import HealthCheck4from redis.exceptions import ConnectionError, TimeoutError5class HealthChecker:6    """7    Configures a health checker8    """9    def __init__(self, redis_storage):10        self._redis_store = redis_storage11        self._health_check = HealthCheck(12            failed_status=settings.HEALTH_CHECKER_FAILED_STATUS13        )14        self._health_check.add_section(15            'commit', os.environ.get('HEROKU_SLUG_COMMIT', None)16        )17        self._health_check.add_section(18            'release', {"version": os.environ.get('HEROKU_RELEASE_VERSION', None)}19        )20        self._health_check.add_check(self._redis_available)21    def _redis_available(self):22        try:23            info = self._redis_store.redis.info()24        except ConnectionError:25            return False, "Could not connect to Redis instance"26        except TimeoutError:27            return False, "Redis connection timed out"28        return True, info29    def run(self):...

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.