Dashboard cards, backup last-seen

This commit is contained in:
2026-08-11 14:01:57 -04:00
parent c30195a92b
commit 0ed8be75d9
18 changed files with 1167 additions and 9 deletions

View File

@@ -13,6 +13,7 @@ Two layers:
"""
import base64
from datetime import datetime, timedelta
import json
import pytest
@@ -862,3 +863,100 @@ def test_the_collection_interval_is_readable_without_a_login(bk_app):
BackupsPlugin().get_settings_defaults()}
assert declared['backups_intervalhours'].get('public') is True
assert not declared['backups_shareroot'].get('public')
# =============================================================================
# Dashboard: backups that have STOPPED, which is not the same as unchanged
# =============================================================================
def test_an_unchanged_config_records_the_check(bk_app, bk_plugin):
"""The fix that makes the card possible at all. Dedup writes no revision on
a no-op, so without recording the check there is no evidence a backup still
runs - a config stable for six months looks identical to a backup that died
six months ago."""
from plugins.backups.models import BackupRevision
with bk_app.app_context():
first = bk_plugin.apply_collector_payload(_payload())
revision = _db.session.get(BackupRevision, first['backuprevisionid'])
_db.session.query(BackupRevision).update(
{'lastseenat': datetime(2026, 1, 1)})
_db.session.commit()
second = bk_plugin.apply_collector_payload(_payload())
assert second['action'] == 'noop'
_db.session.refresh(revision)
assert revision.lastseenat > datetime(2026, 1, 2)
def test_a_stable_machine_is_not_called_stale(bk_app, bk_plugin):
"""A machine whose settings have not changed in months is HEALTHY. Keying
the card on revision age would have flagged most of the fleet."""
from plugins.backups.models import BackupRevision
with bk_app.app_context():
bk_plugin.apply_collector_payload(_payload())
# Changed long ago, checked just now: the normal steady state.
_db.session.query(BackupRevision).update({
'collectedat': datetime.utcnow() - timedelta(days=180),
'lastseenat': datetime.utcnow(),
})
_db.session.commit()
from plugins.backups.services.staleness import stalechains
assert stalechains() == []
def test_a_machine_that_stopped_checking_in_is_listed(bk_app, bk_plugin):
from plugins.backups.models import BackupRevision
with bk_app.app_context():
bk_plugin.apply_collector_payload(_payload())
_db.session.query(BackupRevision).update(
{'lastseenat': datetime.utcnow() - timedelta(days=10)})
_db.session.commit()
from plugins.backups.services.staleness import stalechains
[row] = stalechains()
assert row['quietdays'] == 10
assert row['backupkind'] == 'ntlars'
def test_one_marker_stopping_is_not_hidden_by_its_healthy_sibling(bk_app,
bk_plugin):
"""Per CHAIN, not per asset. Two part markers share one machine number, and
a per-asset view would report the machine as fine while one device has been
silent for a week."""
from plugins.backups.models import BackupRevision
with bk_app.app_context():
bk_plugin.apply_collector_payload(
_payload(reg=_markerreg('COM3'), sourcehostname='MARKERA'))
bk_plugin.apply_collector_payload(
_payload(reg=_markerreg('COM4'), sourcehostname='MARKERB'))
(_db.session.query(BackupRevision)
.filter_by(sourcehostname='MARKERA')
.update({'lastseenat': datetime.utcnow() - timedelta(days=9)}))
(_db.session.query(BackupRevision)
.filter_by(sourcehostname='MARKERB')
.update({'lastseenat': datetime.utcnow()}))
_db.session.commit()
from plugins.backups.services.staleness import stalechains
rows = stalechains()
assert [r['sourcehostname'] for r in rows] == ['MARKERA']
def test_the_threshold_is_a_setting_and_zero_disables_the_card(bk_app, bk_plugin):
from shopdb.core.models import Setting
from plugins.backups.models import BackupRevision
with bk_app.app_context():
bk_plugin.apply_collector_payload(_payload())
_db.session.query(BackupRevision).update(
{'lastseenat': datetime.utcnow() - timedelta(days=2)})
_db.session.add(Setting(key='backups_staledays', value='1'))
_db.session.commit()
from plugins.backups.services.staleness import stalechains
assert len(stalechains()) == 1
_db.session.query(Setting).filter_by(key='backups_staledays').update(
{'value': '0'})
_db.session.commit()
assert stalechains() == []