How to use _health_check method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

test_health.py

Source:test_health.py Github

copy

Full Screen

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

Full Screen

Full Screen

health_monitor.py

Source:health_monitor.py Github

copy

Full Screen

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

Full Screen

Full Screen

health_checker.py

Source:health_checker.py Github

copy

Full Screen

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

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run dbt-osmosis automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful