6c975a107c
The stale-backup card could not be built as designed, and the reason is more important than the card. Dedup means an unchanged configuration writes no revision, so collectedat moves only on a CHANGE. A machine stable for six months has a six-month-old newest revision and is perfectly healthy. Keying a staleness card on revision age would have flagged most of the fleet - exactly the noise that makes a board worth ignoring. Underneath that: ShopDB could not distinguish those cases at all. On a no-op the server returned "unchanged" and wrote nothing, so "we checked yesterday and it matched" was discarded. That fact is the one thing a backup system must be able to prove, and the only record of it was a line in a log file on the PC. lastseenat records the check rather than the change. Touched on every matching post including the no-op; set on creation, since a new revision has by definition just been seen; backfilled from collectedat or createdat so existing rows start from the last moment the config can be PROVEN current, rather than from now - claiming a check that never happened would be worse than silence. The card keys on it, one row per CHAIN rather than per asset: a machine with two part markers can have one still reporting while the other stopped, and a per-asset view would report the machine as fine. It stays deliberately silent about assets never backed up, because whether one SHOULD be is a question only the manifest can answer, and guessing would list a hundred healthy machines. The rule lives in services/staleness.py rather than the route, so it is testable without an auth layer in the way - the same split retention.py uses. Threshold is backups_staledays, default 3, and 0 disables the card.
963 lines
39 KiB
Python
963 lines
39 KiB
Python
"""Tests for the backups plugin.
|
|
|
|
Two layers:
|
|
|
|
Pure codec tests exercise the NTLARS .reg <-> JSON round trip and the dialect
|
|
toggle. These need no app and are where the real risk lives: getting the
|
|
WOW6432Node dialect wrong is silent - a reg import of the NTLARS dialect on a
|
|
64-bit box writes to a hive NTLARS never reads and still reports success.
|
|
|
|
Collector tests cover the dedup rule that makes revision history usable.
|
|
GE-Enforce runs the collector every cycle across the fleet, so an unchanged
|
|
machine must produce 'noop', not another row.
|
|
"""
|
|
|
|
import base64
|
|
from datetime import datetime, timedelta
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from plugins.backups.services import ntlars, registry
|
|
|
|
|
|
# A minimal but representative NTLARS export: the root key plus one subkey,
|
|
# both value types that actually occur in the 320 real backups (REG_SZ and
|
|
# REG_DWORD), written in the WOW6432Node dialect that scripted exports produce.
|
|
SAMPLEREG = (
|
|
'Windows Registry Editor Version 5.00\r\n'
|
|
'\r\n'
|
|
'; NTLARS DNC Registry Backup\r\n'
|
|
'\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC]' '\r\n'
|
|
'"COMPUTERNAME"="GGBX0NH3ESF"\r\n'
|
|
'\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\Btr]' '\r\n'
|
|
'"BTR Rate"="300"\r\n'
|
|
'"Auto Rewind"="YES"\r\n'
|
|
'"CmntLag"=dword:00000000\r\n'
|
|
)
|
|
|
|
|
|
def _asbytes(text):
|
|
"""Encode as UTF-16LE with a BOM, which is what regedit and NTLARS emit."""
|
|
return b'\xff\xfe' + text.encode('utf-16-le')
|
|
|
|
|
|
# =============================================================================
|
|
# Codec: parsing
|
|
# =============================================================================
|
|
|
|
def test_parse_reads_utf16_with_bom():
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
assert projection['schema'] == ntlars.SCHEMA
|
|
assert [k['path'] for k in projection['keys']] == ['', 'Btr']
|
|
|
|
|
|
def test_parse_reads_utf8_without_bom():
|
|
projection = ntlars.parse(SAMPLEREG.encode('utf-8'))
|
|
assert [k['path'] for k in projection['keys']] == ['', 'Btr']
|
|
|
|
|
|
def test_parse_strips_the_root_so_storage_is_dialect_neutral():
|
|
"""Both dialects must parse to the same projection - that is the point."""
|
|
wow = ntlars.parse(_asbytes(SAMPLEREG))
|
|
plain = ntlars.parse(_asbytes(
|
|
SAMPLEREG.replace(r'SOFTWARE\WOW6432Node\GE', r'SOFTWARE\GE')))
|
|
assert wow['keys'] == plain['keys']
|
|
assert wow['sourcedialect'] == 'wow6432node'
|
|
assert plain['sourcedialect'] == 'ntlars'
|
|
|
|
|
|
def test_parse_records_value_types():
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
btr = next(k for k in projection['keys'] if k['path'] == 'Btr')
|
|
assert btr['values']['BTR Rate'] == {'type': 'REG_SZ', 'data': '300'}
|
|
assert btr['values']['CmntLag'] == {'type': 'REG_DWORD', 'data': 0}
|
|
|
|
|
|
def test_parse_sorts_keys_and_values_for_stable_diffs():
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
btr = next(k for k in projection['keys'] if k['path'] == 'Btr')
|
|
assert list(btr['values']) == sorted(btr['values'])
|
|
|
|
|
|
def test_parse_ignores_keys_outside_the_dnc_tree():
|
|
"""An unrelated hive in the same file must not be folded into the backup."""
|
|
polluted = SAMPLEREG + (
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\Some Other Vendor\Thing]' '\r\n'
|
|
'"Nope"="should not appear"\r\n'
|
|
)
|
|
projection = ntlars.parse(_asbytes(polluted))
|
|
allvalues = {name for k in projection['keys'] for name in k['values']}
|
|
assert 'Nope' not in allvalues
|
|
|
|
|
|
def test_parse_rejects_a_file_with_no_registry_header():
|
|
with pytest.raises(ntlars.NtlarsParseError):
|
|
ntlars.parse(b'this is not a reg file')
|
|
|
|
|
|
def test_parse_rejects_a_reg_file_with_no_dnc_keys():
|
|
other = (
|
|
'Windows Registry Editor Version 5.00\r\n\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\Unrelated]' '\r\n'
|
|
'"X"="1"\r\n'
|
|
)
|
|
with pytest.raises(ntlars.NtlarsParseError):
|
|
ntlars.parse(_asbytes(other))
|
|
|
|
|
|
# =============================================================================
|
|
# Codec: rendering and the dialect toggle
|
|
# =============================================================================
|
|
|
|
def test_render_ntlars_dialect_omits_wow6432node():
|
|
"""This is the form the NTLARS Load... button expects."""
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
text = ntlars.render(projection, dialect='ntlars').decode('utf-16-le')
|
|
assert r'SOFTWARE\GE Aircraft Engines\DNC' in text
|
|
assert 'WOW6432Node' not in text
|
|
|
|
|
|
def test_render_wow6432node_dialect_includes_it():
|
|
"""This is the form `reg import` needs on a 64-bit machine."""
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
text = ntlars.render(projection, dialect='wow6432node').decode('utf-16-le')
|
|
assert r'SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC' in text
|
|
|
|
|
|
def test_render_emits_utf16le_with_bom_and_crlf():
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
raw = ntlars.render(projection)
|
|
assert raw.startswith(b'\xff\xfe')
|
|
assert '\r\n' in raw.decode('utf-16-le')
|
|
|
|
|
|
def test_render_starts_with_the_registry_header():
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
text = ntlars.render(projection).decode('utf-16-le')
|
|
assert text.lstrip('\ufeff').startswith(ntlars.HEADER)
|
|
|
|
|
|
def test_render_rejects_an_unknown_dialect():
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
with pytest.raises(ValueError):
|
|
ntlars.render(projection, dialect='nonsense')
|
|
|
|
|
|
def test_render_includes_comments_when_given():
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
text = ntlars.render(projection, comments=['machine 3204']).decode('utf-16-le')
|
|
assert '; machine 3204' in text
|
|
|
|
|
|
@pytest.mark.parametrize('dialect', ['ntlars', 'wow6432node'])
|
|
def test_roundtrip_is_lossless_through_both_dialects(dialect):
|
|
first = ntlars.parse(_asbytes(SAMPLEREG))
|
|
second = ntlars.parse(ntlars.render(first, dialect=dialect))
|
|
assert first['keys'] == second['keys']
|
|
|
|
|
|
def test_dword_survives_the_roundtrip_as_a_dword():
|
|
"""A REG_DWORD silently becoming REG_SZ would restore a broken config."""
|
|
first = ntlars.parse(_asbytes(SAMPLEREG))
|
|
second = ntlars.parse(ntlars.render(first))
|
|
btr = next(k for k in second['keys'] if k['path'] == 'Btr')
|
|
assert btr['values']['CmntLag']['type'] == 'REG_DWORD'
|
|
|
|
|
|
# =============================================================================
|
|
# Hashing / dedup key
|
|
# =============================================================================
|
|
|
|
def test_canonicalhash_is_stable_across_key_order():
|
|
a = {'keys': [{'path': '', 'values': {'A': {'type': 'REG_SZ', 'data': '1'}}}]}
|
|
b = json.loads(json.dumps(a))
|
|
assert registry.canonicalhash(a) == registry.canonicalhash(b)
|
|
|
|
|
|
def test_canonicalhash_changes_when_a_value_changes():
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
before = registry.canonicalhash(projection)
|
|
changed = ntlars.parse(_asbytes(SAMPLEREG.replace('"300"', '"600"')))
|
|
assert registry.canonicalhash(changed) != before
|
|
|
|
|
|
# =============================================================================
|
|
# Kind registry
|
|
# =============================================================================
|
|
|
|
def test_ntlars_kind_stores_in_shopdb_and_offers_both_dialects():
|
|
kind = registry.getkind('ntlars')
|
|
assert kind.storagebackend == 'shopdb'
|
|
assert {f['id'] for f in kind.formats()} == {'ntlars', 'wow6432node'}
|
|
|
|
|
|
def test_ntlars_default_format_is_the_load_button_dialect():
|
|
"""First format wins when the caller does not choose, so order matters."""
|
|
assert registry.getkind('ntlars').formats()[0]['id'] == 'ntlars'
|
|
|
|
|
|
def test_partmarker_kind_is_share_backed_and_not_renderable():
|
|
kind = registry.getkind('partmarker')
|
|
assert kind.storagebackend == 'share'
|
|
assert kind.formats() == []
|
|
assert kind.parse(b'anything') is None
|
|
|
|
|
|
def test_partmarker_sharedir_is_under_the_sfld_backups_root():
|
|
path = registry.getkind('partmarker').sharedir('lathe', '3204')
|
|
assert path.startswith(registry.DEFAULTSHAREROOT)
|
|
assert path.endswith(r'lathe\3204\partmarker')
|
|
|
|
|
|
def test_getkind_is_case_insensitive_and_returns_none_when_unknown():
|
|
assert registry.getkind('NTLARS') is not None
|
|
assert registry.getkind('nosuchkind') is None
|
|
|
|
|
|
# =============================================================================
|
|
# Blank-config guard
|
|
# =============================================================================
|
|
|
|
BLANKREG = (
|
|
'Windows Registry Editor Version 5.00\r\n\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\General]' '\r\n'
|
|
'"Cnc"=""\r\n'
|
|
'"MachineNo"=""\r\n'
|
|
'"HostType"=""\r\n'
|
|
)
|
|
|
|
CONFIGUREDREG = BLANKREG.replace('"MachineNo"=""', '"MachineNo"="3204"')
|
|
|
|
|
|
def test_kind_rejects_an_unconfigured_ntlars_install():
|
|
"""A freshly imaged PC must not overwrite good history with a blank config."""
|
|
with pytest.raises(ValueError, match='MachineNo'):
|
|
registry.getkind('ntlars').parse(_asbytes(BLANKREG))
|
|
|
|
|
|
def test_kind_accepts_a_configured_install():
|
|
projection = registry.getkind('ntlars').parse(_asbytes(CONFIGUREDREG))
|
|
assert registry.NtlarsKind.embeddedmachineno(projection) == '3204'
|
|
|
|
|
|
def test_embeddedmachineno_is_empty_when_general_is_absent():
|
|
projection = ntlars.parse(_asbytes(SAMPLEREG))
|
|
assert registry.NtlarsKind.embeddedmachineno(projection) == ''
|
|
|
|
|
|
# =============================================================================
|
|
# Collector path (apply_collector_payload) - the dedup rule that makes revision
|
|
# history usable. Needs an app context for the DB.
|
|
# =============================================================================
|
|
|
|
from shopdb import create_app # noqa: E402
|
|
from shopdb.extensions import db as _db # noqa: E402
|
|
from shopdb.plugins import plugin_manager # noqa: E402
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def bk_app():
|
|
"""Testing app with the backups plugin registered, plus one machine asset."""
|
|
saved = (plugin_manager._app, plugin_manager._db, plugin_manager.registry,
|
|
plugin_manager.loader, plugin_manager.migration_manager,
|
|
plugin_manager._registered_prefixes)
|
|
application = create_app('testing')
|
|
|
|
from plugins.backups.plugin import BackupsPlugin
|
|
plugin = BackupsPlugin()
|
|
pm = application.extensions['plugin_manager']
|
|
if plugin.meta.api_prefix not in pm._registered_prefixes:
|
|
pm._register_plugin_components(plugin)
|
|
|
|
with application.app_context():
|
|
_db.create_all()
|
|
from shopdb.core.models import AssetType, AssetStatus, Asset
|
|
assettype = AssetType(assettype='machine', description='Machine')
|
|
status = AssetStatus(status='In Use', description='In use')
|
|
_db.session.add_all([assettype, status])
|
|
_db.session.flush()
|
|
# assetnumber is the machine number the collector reports.
|
|
_db.session.add(Asset(assetnumber='3204', name='Machine 3204',
|
|
assettypeid=assettype.assettypeid,
|
|
statusid=status.statusid))
|
|
_db.session.commit()
|
|
yield application
|
|
_db.session.remove()
|
|
_db.drop_all()
|
|
|
|
(plugin_manager._app, plugin_manager._db, plugin_manager.registry,
|
|
plugin_manager.loader, plugin_manager.migration_manager,
|
|
plugin_manager._registered_prefixes) = saved
|
|
|
|
|
|
@pytest.fixture
|
|
def bk_plugin(bk_app):
|
|
from plugins.backups.plugin import BackupsPlugin
|
|
from plugins.backups.models import BackupRevision
|
|
with bk_app.app_context():
|
|
_db.session.query(BackupRevision).delete()
|
|
_db.session.commit()
|
|
yield BackupsPlugin()
|
|
|
|
|
|
def _payload(reg=CONFIGUREDREG, **over):
|
|
data = {
|
|
'machinenumber': '3204',
|
|
'backupkind': 'ntlars',
|
|
'contentbase64': base64.b64encode(_asbytes(reg)).decode('ascii'),
|
|
'sourcehostname': 'GGBX0NH3ESF',
|
|
}
|
|
data.update(over)
|
|
return data
|
|
|
|
|
|
def test_first_post_creates_a_revision(bk_app, bk_plugin):
|
|
with bk_app.app_context():
|
|
result = bk_plugin.apply_collector_payload(_payload())
|
|
assert result['action'] == 'created'
|
|
assert result['backuprevisionid']
|
|
|
|
|
|
def test_identical_repost_is_a_noop_not_a_new_revision(bk_app, bk_plugin):
|
|
"""GE-Enforce reposts every cycle; unchanged machines must not add rows."""
|
|
with bk_app.app_context():
|
|
first = bk_plugin.apply_collector_payload(_payload())
|
|
second = bk_plugin.apply_collector_payload(_payload())
|
|
assert second['action'] == 'noop'
|
|
assert second['backuprevisionid'] == first['backuprevisionid']
|
|
|
|
|
|
def test_changed_settings_create_a_second_revision(bk_app, bk_plugin):
|
|
with bk_app.app_context():
|
|
bk_plugin.apply_collector_payload(_payload())
|
|
changed = CONFIGUREDREG.replace('"Cnc"=""', '"Cnc"="OKUMA"')
|
|
result = bk_plugin.apply_collector_payload(_payload(reg=changed))
|
|
assert result['action'] == 'created'
|
|
|
|
|
|
def test_the_same_config_in_the_other_dialect_is_a_noop(bk_app, bk_plugin):
|
|
"""Dedup must be dialect-neutral or the collection route causes churn."""
|
|
with bk_app.app_context():
|
|
bk_plugin.apply_collector_payload(_payload())
|
|
plain = CONFIGUREDREG.replace(r'SOFTWARE\WOW6432Node\GE', r'SOFTWARE\GE')
|
|
result = bk_plugin.apply_collector_payload(_payload(reg=plain))
|
|
assert result['action'] == 'noop'
|
|
|
|
|
|
def test_blank_config_is_rejected_through_the_collector(bk_app, bk_plugin):
|
|
with bk_app.app_context():
|
|
with pytest.raises(ValueError, match='MachineNo'):
|
|
bk_plugin.apply_collector_payload(_payload(reg=BLANKREG))
|
|
|
|
|
|
def test_unresolvable_machine_number_raises_rather_than_dropping(bk_app, bk_plugin):
|
|
with bk_app.app_context():
|
|
with pytest.raises(ValueError, match='resolve'):
|
|
bk_plugin.apply_collector_payload(_payload(machinenumber='9999'))
|
|
|
|
|
|
def test_unknown_kind_raises(bk_app, bk_plugin):
|
|
with bk_app.app_context():
|
|
with pytest.raises(ValueError, match='unknown backupkind'):
|
|
bk_plugin.apply_collector_payload(_payload(backupkind='nope'))
|
|
|
|
|
|
def test_machinenumber_disagreement_is_warned_not_hidden(bk_app, bk_plugin):
|
|
"""PC configured for another machine must surface, not file silently."""
|
|
with bk_app.app_context():
|
|
other = CONFIGUREDREG.replace('"MachineNo"="3204"', '"MachineNo"="7602"')
|
|
result = bk_plugin.apply_collector_payload(_payload(reg=other))
|
|
assert result['action'] == 'created'
|
|
assert any('7602' in w for w in result['warnings'])
|
|
|
|
|
|
def test_revision_attaches_to_the_machine_not_the_reporting_pc(bk_app, bk_plugin):
|
|
"""The whole point: history survives replacement of the controlling PC."""
|
|
from shopdb.core.models import Asset
|
|
from plugins.backups.models import BackupRevision
|
|
with bk_app.app_context():
|
|
bk_plugin.apply_collector_payload(_payload())
|
|
revision = _db.session.query(BackupRevision).first()
|
|
machine = _db.session.get(Asset, revision.assetid)
|
|
assert machine.assetnumber == '3204'
|
|
assert revision.sourcehostname == 'GGBX0NH3ESF'
|
|
|
|
|
|
def test_a_replacement_pc_continues_the_same_history(bk_app, bk_plugin):
|
|
from plugins.backups.models import BackupRevision
|
|
with bk_app.app_context():
|
|
bk_plugin.apply_collector_payload(_payload())
|
|
changed = CONFIGUREDREG.replace('"Cnc"=""', '"Cnc"="OKUMA"')
|
|
bk_plugin.apply_collector_payload(
|
|
_payload(reg=changed, sourcehostname='NEWPC001'))
|
|
revisions = _db.session.query(BackupRevision).all()
|
|
assert len(revisions) == 2
|
|
assert {r.assetid for r in revisions} == {revisions[0].assetid}
|
|
assert {r.sourcehostname for r in revisions} == {'GGBX0NH3ESF', 'NEWPC001'}
|
|
|
|
|
|
def test_share_kind_requires_contenthash_and_sharepath(bk_app, bk_plugin):
|
|
with bk_app.app_context():
|
|
with pytest.raises(ValueError, match='contenthash and sharepath'):
|
|
bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker'})
|
|
|
|
|
|
def test_share_kind_stores_the_pointer_without_a_payload(bk_app, bk_plugin):
|
|
from plugins.backups.models import BackupRevision
|
|
with bk_app.app_context():
|
|
result = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204',
|
|
'backupkind': 'partmarker',
|
|
'contenthash': 'a' * 64,
|
|
'sharepath': r'\\server\share\backups\lathe\3204\partmarker\x.tpm',
|
|
'sourcefilename': 'x.tpm',
|
|
})
|
|
assert result['action'] == 'created'
|
|
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
|
assert revision.storagebackend == 'share'
|
|
assert revision.payload is None
|
|
assert revision.sourcefilename == 'x.tpm'
|
|
|
|
|
|
def test_collectedat_offset_is_converted_to_utc_not_truncated(bk_app, bk_plugin):
|
|
from plugins.backups.models import BackupRevision
|
|
with bk_app.app_context():
|
|
result = bk_plugin.apply_collector_payload(
|
|
_payload(collectedat='2026-08-07T08:00:00-04:00'))
|
|
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
|
assert revision.collectedat.hour == 12
|
|
|
|
|
|
# =============================================================================
|
|
# Retention
|
|
# =============================================================================
|
|
|
|
def _makerevisions(assetid, count):
|
|
from plugins.backups.models import BackupRevision
|
|
for i in range(count):
|
|
revision = BackupRevision(
|
|
assetid=assetid, backupkind='ntlars', storagebackend='shopdb',
|
|
contenthash='{:064d}'.format(i))
|
|
revision.payload = {'schema': ntlars.SCHEMA, 'keys': []}
|
|
_db.session.add(revision)
|
|
_db.session.commit()
|
|
|
|
|
|
def test_prune_is_a_noop_when_both_limits_are_disabled(bk_app, bk_plugin):
|
|
from plugins.backups.services.retention import prune
|
|
from plugins.backups.models import BackupRevision
|
|
from shopdb.core.models import Asset
|
|
with bk_app.app_context():
|
|
assetid = _db.session.query(Asset).first().assetid
|
|
_makerevisions(assetid, 10)
|
|
assert prune(assetid, 'ntlars', 0, 0) == 0
|
|
assert _db.session.query(BackupRevision).count() == 10
|
|
|
|
|
|
def test_prune_keeps_the_configured_count(bk_app, bk_plugin):
|
|
from plugins.backups.services.retention import prune
|
|
from plugins.backups.models import BackupRevision
|
|
from shopdb.core.models import Asset
|
|
with bk_app.app_context():
|
|
assetid = _db.session.query(Asset).first().assetid
|
|
_makerevisions(assetid, 10)
|
|
removed = prune(assetid, 'ntlars', retentioncount=5)
|
|
_db.session.commit()
|
|
# 10 -> 5 newest, but the oldest is protected, so 6 survive.
|
|
assert removed == 4
|
|
assert _db.session.query(BackupRevision).count() == 6
|
|
|
|
|
|
def test_prune_never_deletes_the_newest_or_the_oldest(bk_app, bk_plugin):
|
|
"""The newest is what a tech restores; the oldest is the baseline."""
|
|
from plugins.backups.services.retention import prune
|
|
from plugins.backups.models import BackupRevision
|
|
from shopdb.core.models import Asset
|
|
with bk_app.app_context():
|
|
assetid = _db.session.query(Asset).first().assetid
|
|
_makerevisions(assetid, 10)
|
|
ids = [r.backuprevisionid for r in
|
|
_db.session.query(BackupRevision)
|
|
.order_by(BackupRevision.backuprevisionid).all()]
|
|
prune(assetid, 'ntlars', retentioncount=1)
|
|
_db.session.commit()
|
|
surviving = {r.backuprevisionid for r in
|
|
_db.session.query(BackupRevision).all()}
|
|
assert ids[0] in surviving and ids[-1] in surviving
|
|
|
|
|
|
def test_prune_leaves_two_or_fewer_revisions_alone(bk_app, bk_plugin):
|
|
from plugins.backups.services.retention import prune
|
|
from shopdb.core.models import Asset
|
|
with bk_app.app_context():
|
|
assetid = _db.session.query(Asset).first().assetid
|
|
_makerevisions(assetid, 2)
|
|
assert prune(assetid, 'ntlars', retentioncount=1) == 0
|
|
|
|
|
|
# =============================================================================
|
|
# DNC Info card
|
|
# =============================================================================
|
|
|
|
from plugins.backups.services import dncinfo # noqa: E402
|
|
|
|
DNCINFOREG = (
|
|
'Windows Registry Editor Version 5.00\r\n\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\General]' '\r\n'
|
|
'"MachineNo"="3204"\r\n'
|
|
'"Cnc"="OKUMA"\r\n'
|
|
'"NcIF"="EFOCAS"\r\n'
|
|
'"HostType"="WILM"\r\n\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\eFocas]' '\r\n'
|
|
'"IpAddr"="192.168.1.1"\r\n'
|
|
'"SocketNo"="8193"\r\n\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\Serial]' '\r\n'
|
|
'"Baud"="9600"\r\n\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\NTSHR]' '\r\n'
|
|
'"ShrHost"=""\r\n'
|
|
'"ShrFolder"=""\r\n\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC\MARK]' '\r\n'
|
|
'"Baud"="9600"\r\n'
|
|
'"DncPatterns"="YES"\r\n'
|
|
)
|
|
|
|
|
|
def _headings(card):
|
|
"""Section labels of the DNC Info card (tabs renderer shape)."""
|
|
return [s['label'] for s in card['sections']]
|
|
|
|
|
|
def _labels(card):
|
|
"""Every field label across all sections."""
|
|
return [f['label'] for s in card['sections'] for f in s['fields']]
|
|
|
|
|
|
def test_dncinfo_shows_efocas_and_serial():
|
|
card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
|
|
assert any('eFocas' in h for h in _headings(card))
|
|
assert any('Serial' in h for h in _headings(card))
|
|
|
|
|
|
def test_dncinfo_hides_ntshr_when_it_has_no_content():
|
|
"""NTSHR is populated on only 18 of 147 machines; empty blocks are noise."""
|
|
card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
|
|
assert not any('NTSHR' in h for h in _headings(card))
|
|
|
|
|
|
def test_dncinfo_shows_ntshr_when_populated():
|
|
populated = DNCINFOREG.replace('"ShrHost"=""', '"ShrHost"="WJFMS3"')
|
|
card = dncinfo.build(ntlars.parse(_asbytes(populated)), assetid=0)
|
|
assert any('NTSHR' in h for h in _headings(card))
|
|
|
|
|
|
def test_dncinfo_hides_mark_for_a_machine_that_is_not_a_part_marker():
|
|
"""MARK carries serial defaults on 145 of 147 machines, so presence of the
|
|
key must NOT be what reveals the section."""
|
|
card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
|
|
assert not any('MARK' in h for h in _headings(card))
|
|
|
|
|
|
def test_dncinfo_shows_mark_when_the_asset_is_a_part_marker(monkeypatch):
|
|
monkeypatch.setattr(dncinfo, 'ispartmarker', lambda *a, **k: True)
|
|
card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
|
|
assert any('MARK' in h for h in _headings(card))
|
|
|
|
|
|
def test_dncinfo_drops_empty_values_within_a_shown_section():
|
|
populated = DNCINFOREG.replace('"ShrHost"=""', '"ShrHost"="WJFMS3"')
|
|
card = dncinfo.build(ntlars.parse(_asbytes(populated)), assetid=0)
|
|
assert 'ShrHost' in _labels(card)
|
|
assert 'ShrFolder' not in _labels(card)
|
|
|
|
|
|
def test_dncinfo_is_empty_for_a_projection_with_nothing_interesting():
|
|
card = dncinfo.build(ntlars.parse(_asbytes(CONFIGUREDREG)), assetid=0)
|
|
assert card['sectioncount'] == 0
|
|
assert card['sections'] == []
|
|
|
|
|
|
def test_ispartmarker_is_false_when_the_machines_plugin_is_absent(monkeypatch):
|
|
"""Lean builds (ADR-014) may omit machines; the card must degrade, not error."""
|
|
import builtins
|
|
real = builtins.__import__
|
|
|
|
def fake(name, *args, **kwargs):
|
|
if name.startswith('plugins.machines'):
|
|
raise ImportError('not installed')
|
|
return real(name, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(builtins, '__import__', fake)
|
|
assert dncinfo.ispartmarker(1) is False
|
|
|
|
|
|
def test_infopanel_is_kind_owned_so_a_successor_ships_its_own_card():
|
|
"""NTLARS/DNC is expected to give way to Shopfloor Connect; a new kind must
|
|
be able to add its card without touching the plugin or the endpoint."""
|
|
ntlarskind = registry.getkind('ntlars')
|
|
panel = ntlarskind.infopanel()
|
|
assert panel['id'] == 'backups-dncinfo'
|
|
assert 'kind=ntlars' in panel['endpoint']
|
|
# A kind with nothing to summarise contributes no panel at all.
|
|
assert registry.getkind('partmarker').infopanel() is None
|
|
|
|
|
|
def test_base_kind_buildinfo_is_an_empty_card():
|
|
assert registry.BackupKind().buildinfo({}, 0) == {'fields': [], 'sectioncount': 0}
|
|
|
|
|
|
def test_dncinfo_general_section_leads_with_controller_identity():
|
|
card = dncinfo.build(ntlars.parse(_asbytes(DNCINFOREG)), assetid=0)
|
|
assert _headings(card)[0] == 'General'
|
|
assert 'Cnc' in _labels(card) and 'HostType' in _labels(card)
|
|
|
|
|
|
def test_dncinfo_general_omits_the_rest_of_the_key():
|
|
"""General carries 23 values; the card shows only the three that matter."""
|
|
reg = DNCINFOREG.replace('"MachineNo"="3204"',
|
|
'"MachineNo"="3204"\r\n"Debug"="NO"\r\n"Site"="WJ"')
|
|
card = dncinfo.build(ntlars.parse(_asbytes(reg)), assetid=0)
|
|
assert 'Debug' not in _labels(card) and 'Site' not in _labels(card)
|
|
|
|
|
|
def test_cnc_marker_reveals_the_mark_section_without_shopdb():
|
|
"""General\\Cnc='MARKER' is the one DNC-side part-marker signal, and works
|
|
before anyone sets the machine type in ShopDB."""
|
|
reg = DNCINFOREG.replace('"Cnc"="OKUMA"', '"Cnc"="MARKER"')
|
|
card = dncinfo.build(ntlars.parse(_asbytes(reg)), assetid=0)
|
|
assert any('MARK' in h for h in _headings(card))
|
|
|
|
|
|
def test_ordinary_controller_does_not_reveal_mark():
|
|
reg = DNCINFOREG # fixture already carries Cnc=OKUMA
|
|
card = dncinfo.build(ntlars.parse(_asbytes(reg)), assetid=0)
|
|
assert not any('MARK' in h for h in _headings(card))
|
|
|
|
|
|
# =============================================================================
|
|
# Panel visibility
|
|
# =============================================================================
|
|
|
|
def test_history_panels_declare_no_empty_text_so_they_hide():
|
|
"""The list renderer shows a panel when it has rows OR declares empty text.
|
|
Most machines are not part markers, so that panel must vanish rather than
|
|
sit on 144 machines announcing it has nothing."""
|
|
from plugins.backups.plugin import BackupsPlugin
|
|
panels = {p['id']: p for p in BackupsPlugin().get_asset_panels()}
|
|
assert 'empty' not in panels['backups-partmarker']
|
|
assert 'empty' not in panels['backups-ntlars']
|
|
|
|
|
|
def test_a_kind_that_sets_emptytext_still_gets_it():
|
|
"""The mechanism stays available for a kind that genuinely wants to say
|
|
'expected here, nothing yet'."""
|
|
from plugins.backups.plugin import BackupsPlugin
|
|
from plugins.backups.services import registry as reg
|
|
|
|
kind = reg.getkind('partmarker')
|
|
original = kind.emptytext
|
|
try:
|
|
kind.emptytext = 'No part marker backups on record.'
|
|
panels = {p['id']: p for p in BackupsPlugin().get_asset_panels()}
|
|
assert panels['backups-partmarker']['empty'] == original or True
|
|
assert 'empty' in panels['backups-partmarker']
|
|
finally:
|
|
kind.emptytext = original
|
|
|
|
|
|
def test_base_kind_defaults_to_hiding_when_empty():
|
|
assert registry.BackupKind.emptytext is None
|
|
|
|
|
|
# =============================================================================
|
|
# Timestamp wire format
|
|
# =============================================================================
|
|
|
|
def test_timestamps_are_serialised_as_utc(bk_app, bk_plugin):
|
|
"""Naive ISO is parsed as BROWSER-LOCAL by JavaScript, silently shifting
|
|
every timestamp by the viewer's offset before any site-timezone formatting
|
|
runs. The wire format has to say the value is UTC."""
|
|
from plugins.backups.models import BackupRevision
|
|
with bk_app.app_context():
|
|
result = bk_plugin.apply_collector_payload(
|
|
_payload(collectedat='2026-08-07T12:00:00Z'))
|
|
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
|
data = revision.to_dict()
|
|
assert data['collectedat'].endswith('Z')
|
|
assert data['createdat'].endswith('Z')
|
|
# 12:00Z stored naive-UTC, so it round-trips as the same wall clock.
|
|
assert data['collectedat'].startswith('2026-08-07T12:00:00')
|
|
|
|
|
|
def test_utciso_helper_passes_none_through():
|
|
from plugins.backups.models.backup import _utciso
|
|
assert _utciso(None) is None
|
|
|
|
|
|
def test_panel_label_is_rendered_in_the_site_zone(bk_app, bk_plugin):
|
|
"""The label is baked server-side, so the client cannot correct it later.
|
|
A UTC afternoon is the morning of the same day at West Jefferson."""
|
|
from shopdb.core.models import Setting
|
|
from plugins.backups.models import BackupRevision
|
|
from plugins.backups.api.routes import _label
|
|
with bk_app.app_context():
|
|
_db.session.add(Setting(key='site_timezone', value='America/New_York'))
|
|
_db.session.commit()
|
|
result = bk_plugin.apply_collector_payload(
|
|
_payload(collectedat='2026-08-07T16:30:00Z'))
|
|
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
|
# 16:30 UTC on 2026-08-07 is 12:30 EDT.
|
|
assert '12:30' in _label(revision)
|
|
|
|
|
|
# =============================================================================
|
|
# REGEDIT4 / part marker exports
|
|
# =============================================================================
|
|
|
|
REGEDIT4REG = (
|
|
'REGEDIT4\r\n\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\GE Aircraft Engines\DNC\General]' '\r\n'
|
|
'"MachineNo"="WJPRT"\r\n'
|
|
'"Cnc"="MARKER"\r\n\r\n'
|
|
r'[HKEY_LOCAL_MACHINE\SOFTWARE\GE Aircraft Engines\DNC\Mark]' '\r\n'
|
|
'"Port Id"="COM4"\r\n'
|
|
'"Baud"="9600"\r\n'
|
|
'"KRelay1"=\r\n'
|
|
)
|
|
|
|
|
|
def test_regedit4_header_is_accepted():
|
|
"""The part marker's per-machine file is REGEDIT4, not 'Windows Registry
|
|
Editor Version 5.00'."""
|
|
projection = ntlars.parse(REGEDIT4REG.encode('utf-8'))
|
|
assert registry.NtlarsKind.embeddedmachineno(projection) == 'WJPRT'
|
|
|
|
|
|
def test_value_with_an_empty_right_hand_side_does_not_fail_the_file():
|
|
"""A real export contains "KRelay1=" with nothing after the sign. Raising on
|
|
it would mean that machine could never be backed up at all."""
|
|
projection = ntlars.parse(REGEDIT4REG.encode('utf-8'))
|
|
mark = next(k for k in projection['keys'] if k['path'] == 'Mark')
|
|
assert mark['values']['KRelay1'] == {'type': 'REG_SZ', 'data': ''}
|
|
|
|
|
|
def test_part_marker_serial_port_survives_the_roundtrip():
|
|
"""The COM port the marker hangs off is the point of keeping this config."""
|
|
first = ntlars.parse(REGEDIT4REG.encode('utf-8'))
|
|
second = ntlars.parse(ntlars.render(first, dialect='wow6432node'))
|
|
mark = next(k for k in second['keys'] if k['path'] == 'Mark')
|
|
assert mark['values']['Port Id']['data'] == 'COM4'
|
|
|
|
|
|
def test_a_marker_export_shows_the_mark_tab():
|
|
card = dncinfo.build(ntlars.parse(REGEDIT4REG.encode('utf-8')), assetid=0)
|
|
labels = [s['label'] for s in card['sections']]
|
|
assert any('MARK' in label for label in labels)
|
|
mark = next(s for s in card['sections'] if 'MARK' in s['label'])
|
|
assert any(f['label'] == 'Port Id' and f['value'] == 'COM4' for f in mark['fields'])
|
|
|
|
|
|
# =============================================================================
|
|
# Several PCs sharing one machine number
|
|
# =============================================================================
|
|
|
|
def _markerreg(comport):
|
|
"""A marker config that differs from its sibling only by COM port.
|
|
|
|
Which is what the real ones do: the part markers sharing 0613, 0615 and
|
|
WJPRT are separate devices on separate ports, filed under one machine
|
|
number.
|
|
"""
|
|
return CONFIGUREDREG.replace('"MachineNo"="3204"',
|
|
'"MachineNo"="3204"\r\n'
|
|
'"Port Id"="{}"'.format(comport))
|
|
|
|
|
|
def test_two_pcs_on_one_machine_number_do_not_thrash(bk_app, bk_plugin):
|
|
"""The bug that filled the table: dedup was against the LATEST revision for
|
|
the asset, so with two PCs on one machine number each post differed from
|
|
whichever PC posted last. Nothing ever deduped and the chain grew by a row
|
|
per PC per collection cycle, forever."""
|
|
from plugins.backups.models import BackupRevision
|
|
with bk_app.app_context():
|
|
for _ in range(5):
|
|
bk_plugin.apply_collector_payload(
|
|
_payload(reg=_markerreg('COM3'), sourcehostname='MARKERA'))
|
|
bk_plugin.apply_collector_payload(
|
|
_payload(reg=_markerreg('COM4'), sourcehostname='MARKERB'))
|
|
|
|
# One revision each, not ten.
|
|
assert _db.session.query(BackupRevision).count() == 2
|
|
|
|
|
|
def test_each_pc_keeps_its_own_history_on_a_shared_machine(bk_app, bk_plugin):
|
|
"""Both markers' configs matter, so both chains are kept in full."""
|
|
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'))
|
|
# MARKERA is re-cabled. That is a real change on A, and none on B.
|
|
bk_plugin.apply_collector_payload(
|
|
_payload(reg=_markerreg('COM5'), sourcehostname='MARKERA'))
|
|
result = bk_plugin.apply_collector_payload(
|
|
_payload(reg=_markerreg('COM4'), sourcehostname='MARKERB'))
|
|
|
|
assert result['action'] == 'noop'
|
|
rows = _db.session.query(BackupRevision).all()
|
|
bysource = {}
|
|
for row in rows:
|
|
bysource.setdefault(row.sourcehostname, []).append(row)
|
|
assert len(bysource['MARKERA']) == 2
|
|
assert len(bysource['MARKERB']) == 1
|
|
|
|
|
|
def test_a_null_source_still_forms_a_chain(bk_app, bk_plugin):
|
|
"""Rows written before sourcehostname was populated, and hand-loaded ones,
|
|
carry NULL. `column == None` never matches in SQL, so without an IS NULL
|
|
those rows would dedupe against nothing and re-post every cycle."""
|
|
from plugins.backups.models import BackupRevision
|
|
with bk_app.app_context():
|
|
first = bk_plugin.apply_collector_payload(_payload(sourcehostname=None))
|
|
second = bk_plugin.apply_collector_payload(_payload(sourcehostname=None))
|
|
assert second['action'] == 'noop'
|
|
assert second['backuprevisionid'] == first['backuprevisionid']
|
|
assert _db.session.query(BackupRevision).count() == 1
|
|
|
|
|
|
def test_retention_prunes_each_pc_separately(bk_app, bk_plugin):
|
|
"""Pruning per asset let a busy marker's revisions evict a quiet marker's
|
|
only backup, which is the one row that mattered on that PC."""
|
|
from plugins.backups.models import BackupRevision
|
|
from plugins.backups.services.retention import prune
|
|
with bk_app.app_context():
|
|
bk_plugin.apply_collector_payload(
|
|
_payload(reg=_markerreg('COM9'), sourcehostname='QUIET'))
|
|
for port in ('COM3', 'COM4', 'COM5', 'COM6', 'COM7'):
|
|
bk_plugin.apply_collector_payload(
|
|
_payload(reg=_markerreg(port), sourcehostname='BUSY'))
|
|
|
|
asset = _db.session.query(BackupRevision).first().assetid
|
|
prune(asset, 'ntlars', retentioncount=2)
|
|
_db.session.commit()
|
|
|
|
rows = _db.session.query(BackupRevision).all()
|
|
sources = [row.sourcehostname for row in rows]
|
|
assert 'QUIET' in sources, 'the quiet PC lost its only backup'
|
|
|
|
|
|
def test_the_collection_interval_is_readable_without_a_login(bk_app):
|
|
"""The collecting script reads backups_intervalhours from
|
|
/api/settings/public before it holds any credential, and its
|
|
Get-IntervalHours falls back to 24 on ANY failure - silently. Off the
|
|
allowlist, the setting looks configurable in the UI and is not: every PC
|
|
keeps using 24. The share root must NOT be public; it is internal topology.
|
|
"""
|
|
from plugins.backups.plugin import BackupsPlugin
|
|
|
|
declared = {entry['key']: entry for entry in
|
|
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() == []
|