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'
|