jaredandClaude Opus 5.5
415d2222e7
Lint / JS (eslint) (push) Successful in 21s
Lint / Python (flake8) (push) Successful in 2m36s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Test / Python Tests (pytest) (push) Successful in 2m46s
Lint / JS (eslint) (pull_request) Successful in 15s
Lint / Python (flake8) (pull_request) Successful in 2m14s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (push) Successful in 1m33s
Security / Python Security (bandit) (pull_request) Successful in 54s
Test / Python Tests (pytest) (pull_request) Successful in 1m3s
- Each PDU, UPS and power strip card lists its outlets that feed something, draw power or are off, with live watts. Each device fed gets a power badge with what it draws. Hovering a power card or a device draws its feeds as dotted lines and keeps both ends lit. - What an outlet feeds: a saved assignment, else the outlet's UniFi name when it names a device (or the modem), else a MAC in its name that belongs to a host NIC (the monitor's host_macs) or UniFi device. - Admins click an outlet to set what it feeds, nothing, or automatic. Saved in a new power_feeds table (created with the cutsheet tables); PUT /api/topology/power/<device>/<outlet>, admin and JSON only, with the device, outlet and target checked. UniFi is never changed. - An outlet that is off but feeds something is listed under "to look at". - Power gear is shown by default again, now that it has something to say. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
87 lines
3.3 KiB
Python
87 lines
3.3 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'
|
|
|
|
|
|
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'
|