Add an alert history page and tie /health staleness to the poll interval
Lint / JS (eslint) (push) Successful in 6s
Lint / Python (flake8) (push) Successful in 1m56s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 2m3s
Test / Python Tests (pytest) (push) Successful in 1m0s
Lint / Python (flake8) (pull_request) Successful in 48s
Lint / JS (eslint) (pull_request) Successful in 6s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 2m7s
Test / Python Tests (pytest) (pull_request) Successful in 2m6s
Lint / JS (eslint) (push) Successful in 6s
Lint / Python (flake8) (push) Successful in 1m56s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 2m3s
Test / Python Tests (pytest) (push) Successful in 1m0s
Lint / Python (flake8) (pull_request) Successful in 48s
Lint / JS (eslint) (pull_request) Successful in 6s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 2m7s
Test / Python Tests (pytest) (pull_request) Successful in 2m6s
/events lists every alert from the last 24 hours to 90 days, open or cleared, with how long each lasted, failed checks and its ticket. Filters (device, type, severity, state, range) live in the URL, so device pages and tickets link straight to a device's history. A "Recurring" table lists alerts that fired more than once on the same target. On phones the rows become cards. /api/events/history backs it (100 per page). The existing /api/events is unchanged. /health and the dashboard's monitor indicator now call the monitor stalled after five missed polls (at least 10 minutes) instead of a fixed 20, and /health reports the monitor's age. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -316,3 +316,63 @@ class TestCutsheetLivePorts:
|
||||
def test_admin_only(self, cs_client):
|
||||
client, _ = cs_client
|
||||
assert client.get('/api/cutsheet/live-ports', headers=USER).status_code == 403
|
||||
|
||||
|
||||
class TestAlertHistory:
|
||||
HISTORY = {'events': [], 'total': 0, 'by_type': {}}
|
||||
|
||||
def _db(self):
|
||||
db = gandalf.db
|
||||
db.get_events_history.return_value = dict(self.HISTORY)
|
||||
db.get_flappers.return_value = []
|
||||
db.get_event_types.return_value = ['host_unreachable']
|
||||
return db
|
||||
|
||||
def test_page_and_nav(self, client):
|
||||
html = client.get('/events', headers=AUTH).get_data(as_text=True)
|
||||
assert 'Alert History' in html and 'events.js' in html
|
||||
assert 'href="/events"' in client.get('/', headers=AUTH).get_data(as_text=True)
|
||||
|
||||
def test_defaults_to_seven_days(self, client):
|
||||
db = self._db()
|
||||
data = client.get('/api/events/history', headers=AUTH).get_json()
|
||||
assert db.get_events_history.call_args.args == (168, '', '', '', 'all')
|
||||
assert data['types'] == ['host_unreachable'] and data['page_size'] == 100
|
||||
|
||||
def test_filters_are_passed_through(self, client):
|
||||
db = self._db()
|
||||
client.get('/api/events/history?range=90d&type=interface_down&target=+large1+&severity=critical'
|
||||
'&status=resolved&offset=200', headers=AUTH)
|
||||
assert db.get_events_history.call_args.args == (2160, 'interface_down', 'large1', 'critical', 'resolved')
|
||||
assert db.get_events_history.call_args.kwargs == {'limit': 100, 'offset': 200}
|
||||
assert db.get_flappers.call_args.args == (2160, 'interface_down', 'large1', 'critical')
|
||||
|
||||
@pytest.mark.parametrize('query', ['range=1y', 'severity=bad', 'status=open', 'offset=x'])
|
||||
def test_bad_filters_are_rejected(self, client, query):
|
||||
self._db()
|
||||
assert client.get(f'/api/events/history?{query}', headers=AUTH).status_code == 400
|
||||
|
||||
def test_needs_login(self, client):
|
||||
assert client.get('/api/events/history').status_code == 401
|
||||
|
||||
|
||||
class TestHealth:
|
||||
def _health(self, client, age_s, poll=120):
|
||||
from datetime import datetime, timedelta, timezone
|
||||
stamp = (datetime.now(timezone.utc) - timedelta(seconds=age_s)).strftime('%Y-%m-%d %H:%M:%S UTC')
|
||||
gandalf.db.get_state.side_effect = lambda key, default=None: stamp if key == 'last_check' else default
|
||||
with patch.object(gandalf, '_cfg', dict(CONFIG, monitor={'poll_interval': poll})):
|
||||
return client.get('/health')
|
||||
|
||||
def test_recent_check_is_healthy(self, client):
|
||||
resp = self._health(client, 150)
|
||||
assert resp.status_code == 200 and resp.get_json()['checks']['monitor'] == 'ok'
|
||||
|
||||
def test_stalled_monitor_fails_after_ten_minutes(self, client):
|
||||
resp = self._health(client, 660)
|
||||
assert resp.status_code == 503
|
||||
assert resp.get_json()['checks'] == {'db': 'ok', 'monitor': 'stale', 'monitor_age_s': 660}
|
||||
|
||||
def test_limit_follows_a_long_poll_interval(self, client):
|
||||
assert self._health(client, 1000, poll=300).status_code == 200
|
||||
assert self._health(client, 1600, poll=300).status_code == 503
|
||||
|
||||
Reference in New Issue
Block a user