gandalf/tests/test_db.py

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


class TestPowerFeeds:
    def test_none_goes_back_to_automatic(self):
        patcher, cur = _fake_conn([])
        with patcher:
            db.set_power_feed('RPDU', 7, None, 'jared')
        sql, args = cur.execute.call_args.args
        assert sql.startswith('DELETE FROM power_feeds') and args == ('RPDU', 7)

    def test_empty_target_is_saved_as_nothing(self):
        patcher, cur = _fake_conn([])
        with patcher:
            db.set_power_feed('RPDU', 7, '', 'jared')
        sql, args = cur.execute.call_args.args
        assert 'INSERT INTO power_feeds' in sql and args == ('RPDU', 7, '', 'jared')

    def test_rows_come_back_with_iso_times(self):
        patcher, _ = _fake_conn([{'source': 'RPDU', 'outlet': 7, 'target': 'large1', 'updated_by': 'jared',
                                  'updated_at': NOW}])
        with patcher:
            assert db.get_power_feeds()[0]['updated_at'] == '2026-11-14T12:00:00Z'