Create health check endpoint
Some checks failed
Run tests and linters / test (pull_request) Successful in 2m47s
Run tests and linters / lint (pull_request) Failing after 2s

This is to prevent livenessProbe spamming of index page
This commit is contained in:
Jaroslav Groman 2026-02-26 15:49:10 +01:00 committed by Kamil Páral
commit c2c9d51a59
2 changed files with 45 additions and 0 deletions

View file

@ -279,6 +279,23 @@ def index():
return display_current()
@main.route('/_health')
def get_health_check():
"""Endpoint for OpenShift pod health checks (liveness and readiness probes).
Verifies database connectivity and returns HTTP 200 on success, 500 on failure.
"""
try:
# Simple DB query to verify database connectivity
db.session.execute(db.text('SELECT 1')).scalar()
response = make_response('OK')
response.mimetype = 'text/plain'
return response
except Exception as e: # pylint: disable=broad-except
app.logger.error('Health check failed: %s', e)
abort(500)
@main.route('/<int:num>/<release_name>')
def old_display_buglist(num, release_name):
return redirect(url_for('.display_buglist', num=num, release_name=release_name))

View file

@ -1,7 +1,9 @@
import datetime
import re
from unittest.mock import patch
import pytest
import sqlalchemy.exc
from blockerbugs.models.milestone import Milestone
from blockerbugs.models.release import Release
@ -490,3 +492,29 @@ class TestGetFunctions:
# make sure there are no additional updates listed
matches = re.findall(re.escape('<tr class="update">'), html)
assert len(matches) == len(bug.updates.all())
@pytest.mark.usefixtures('_postgres_db')
class TestHealthCheck:
"""Tests for the /_health endpoint"""
@pytest.fixture(autouse=True)
def setup_client(self):
"""Set up a Flask test client for each test."""
self.client = app.test_client()
def test_health_check_returns_200_when_db_is_accessible(self):
"""GET /_health returns HTTP 200 with body 'OK' when the DB is reachable."""
rv = self.client.get('/_health')
assert rv.status_code == 200
assert rv.data == b'OK'
assert rv.content_type == 'text/plain; charset=utf-8'
def test_health_check_returns_500_when_db_is_unreachable(self):
"""GET /_health returns HTTP 500 when the DB query raises an exception."""
db_error = sqlalchemy.exc.OperationalError(
'SELECT 1', {}, Exception('connection refused')
)
with patch.object(db.session, 'execute', side_effect=db_error):
rv = self.client.get('/_health')
assert rv.status_code == 500