jaredandClaude Opus 5.5
0921e82c54
Lint / JS (eslint) (push) Successful in 19s
Lint / Python (flake8) (push) Successful in 1m7s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 52s
Test / Python Tests (pytest) (push) Successful in 44s
Lint / JS (eslint) (pull_request) Successful in 7s
Lint / Python (flake8) (pull_request) Successful in 50s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 52s
Test / Python Tests (pytest) (pull_request) Successful in 1m35s
Gandalf only comments on tickets, so a ticket can stay open long after its alert cleared (pbs's ticket sat open for a month), and an alert can stay active for days once nobody is looking. - The monitor reads Tinker's open ticket list (tickets_api.php, read scope) every 15 minutes and saves it. A failed read keeps the previous list. - Active alerts get a stale badge after monitor.stale_alert_days (3), and a "ticket closed" badge when their ticket was closed while still active. - A Needs Follow-up panel lists open tickets whose every alert cleared at least monitor.stale_ticket_hours (24) ago. - A "TO FOLLOW UP" status chip counts both. If the open ticket list is over 2 hours old, the ticket flags are left off rather than guessed. - Replaced an — in the alert count notice. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
"""Tests for db.py query helpers that do work beyond the SQL: the connection is faked."""
|
|
import os
|
|
import sys
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
|
|
|
import db # noqa: E402
|
|
|
|
NOW = datetime(2026, 11, 14, 12, 0, 0)
|
|
|
|
|
|
def _fake_conn(rows):
|
|
cur = MagicMock()
|
|
cur.fetchall.return_value = rows
|
|
cur.fetchone.return_value = {'now': NOW}
|
|
conn = MagicMock()
|
|
conn.cursor.return_value.__enter__.return_value = cur
|
|
|
|
@contextmanager
|
|
def get_conn():
|
|
yield conn
|
|
return patch.object(db, 'get_conn', get_conn), cur
|
|
|
|
|
|
def _event(tid, hours_ago_first, hours_ago_resolved, etype='host_unreachable', target='pbs'):
|
|
return {
|
|
'ticket_id': tid, 'event_type': etype, 'target_name': target, 'target_detail': '',
|
|
'severity': 'critical', 'first_seen': NOW - timedelta(hours=hours_ago_first),
|
|
'resolved_at': None if hours_ago_resolved is None else NOW - timedelta(hours=hours_ago_resolved),
|
|
}
|
|
|
|
|
|
class TestGetClearedTickets:
|
|
def test_no_ids_skips_the_query(self):
|
|
patcher, cur = _fake_conn([])
|
|
with patcher:
|
|
assert db.get_cleared_tickets([]) == []
|
|
cur.execute.assert_not_called()
|
|
|
|
def test_only_tickets_whose_every_alert_cleared_long_enough_ago(self):
|
|
rows = [
|
|
_event('1', 300, 250), _event('1', 100, 48), # two outages, both long cleared
|
|
_event('2', 50, 40), _event('2', 5, None), # still active
|
|
_event('3', 10, 2), # cleared too recently
|
|
]
|
|
patcher, cur = _fake_conn(rows)
|
|
with patcher:
|
|
out = db.get_cleared_tickets(['3', '1', '2', '1'], older_than_hours=24)
|
|
assert [r['ticket_id'] for r in out] == ['1']
|
|
assert out[0]['events'] == 2
|
|
assert out[0]['cleared_at'] == (NOW - timedelta(hours=48)).isoformat() + 'Z'
|
|
assert out[0]['first_seen'] == (NOW - timedelta(hours=300)).isoformat() + 'Z'
|
|
assert cur.execute.call_args_list[0].args[1] == ['1', '2', '3']
|
|
|
|
def test_newest_clearance_first(self):
|
|
rows = [_event('1', 400, 300), _event('2', 100, 30, target='micro1')]
|
|
patcher, _ = _fake_conn(rows)
|
|
with patcher:
|
|
out = db.get_cleared_tickets(['1', '2'])
|
|
assert [r['ticket_id'] for r in out] == ['2', '1']
|
|
assert out[0]['target_name'] == 'micro1'
|