500f7d6ad6
The plugin shipped five settings and no page for any of them. The core settings catalog is a hardcoded list a plugin cannot extend, so backups_shareroot and the collection interval were reachable only through the API or a flask shell on the server - which is not a configurable setting in any useful sense, and it is how today went: the setting simply was not there to set. The page is contributed the ADR-010 way rather than by editing core - a card from get_settings_cards plus a route in the plugin's own routes.js, both of which disappear when the plugin is disabled. Filed under Machines, next to Machine Types, because that is where the assets these back up live. It also states the two things the settings do not say for themselves: that health is judged on when a backup was last CONFIRMED rather than when it last changed, and that retention prunes database revisions only - files on the share stay, so enabling it on a share-backed kind leaves orphaned snapshot directories. Snapshot paths are now shown as full UNC text in both places: a Snapshot row on the Marker Contents card, and the existing path on the history page, which gains a copy button. Deliberately NOT a link - browsers refuse to navigate from an https page to file://, so a clickable path looks actionable and silently does nothing, which is worse than no link. Copy, then paste into Explorer. The path comes off the revision, never composed from the configured share root: a snapshot written before the layout changed still lives where it was put.
1419 lines
60 KiB
Python
1419 lines
60 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_configured_backups_root():
|
|
"""The marker kind keys its tree on the MARKER, not the machine type.
|
|
|
|
It overrides the base layout (<root>\\<machinetype>\\<identifier>\\<kind>)
|
|
because the site files marker snapshots as <root>\\partmarker\\<asset>\\,
|
|
with one timestamped snapshot directory per capture below that. The
|
|
override exists so what ShopDB displays matches what the collector writes.
|
|
"""
|
|
root = r'\\fileserver\shopfloor\backups'
|
|
path = registry.getkind('partmarker').sharedir('lathe', '3204', shareroot=root)
|
|
assert path.startswith(root)
|
|
assert path.endswith(r'partmarker\3204')
|
|
# The machine type is deliberately NOT in the path: a marker is identified
|
|
# by its own asset, and the type of the machine it serves is irrelevant to
|
|
# where its patterns live.
|
|
assert 'lathe' not in path
|
|
|
|
|
|
def test_sharedir_is_empty_when_no_root_is_configured():
|
|
"""No default root ships (ADR-015), so an unconfigured site gets nothing
|
|
rather than a path composed from another site's file server."""
|
|
assert registry.getkind('partmarker').sharedir('lathe', '3204') == ''
|
|
|
|
|
|
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']
|
|
|
|
# The part marker kind now ships a card of its own, built from the manifest
|
|
# the collector sends beside the hash. It could not before: a share kind
|
|
# stored no description of its contents, and reading the share from the app
|
|
# server is refused on purpose.
|
|
markerpanel = registry.getkind('partmarker').infopanel()
|
|
assert markerpanel['id'] == 'backups-partmarker-contents'
|
|
assert 'kind=partmarker' in markerpanel['endpoint']
|
|
# Marker assets are core type 'machine' like the other 144, and the panel
|
|
# is kept off those by having no rows and no empty text rather than by any
|
|
# special-casing.
|
|
assert markerpanel['assettypes'] == ['machine']
|
|
assert 'empty' not in markerpanel
|
|
|
|
|
|
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() == []
|
|
|
|
|
|
def test_the_history_reports_when_a_config_was_last_verified(bk_app, bk_plugin):
|
|
"""A machine checked daily for a year still shows one revision, because an
|
|
unchanged config writes none. The API has to say the check happened or that
|
|
machine reads as abandoned."""
|
|
from plugins.backups.models import BackupRevision
|
|
with bk_app.app_context():
|
|
result = bk_plugin.apply_collector_payload(_payload())
|
|
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
|
data = revision.to_dict()
|
|
assert data['lastseenat'] is not None
|
|
assert data['lastseenat'].endswith('Z')
|
|
|
|
|
|
def test_the_panel_shows_verified_beside_captured(bk_app, bk_plugin):
|
|
with bk_app.app_context():
|
|
panels = bk_plugin.get_asset_panels()
|
|
listpanel = next(p for p in panels if p.get('render') == 'list')
|
|
labels = [m['label'] for m in listpanel['map']['meta']]
|
|
assert 'Captured' in labels and 'Verified' in labels
|
|
|
|
|
|
def test_retention_reads_the_key_the_settings_page_writes(app, db):
|
|
"""The plugin helper namespaces to `plugin.backups.<key>`; the declared key
|
|
is bare. Reading through the helper meant retention was configurable and
|
|
inert - it always fell back to 0, which means keep everything."""
|
|
from shopdb.core.models import Setting
|
|
from plugins.backups.plugin import BackupsPlugin
|
|
|
|
declared = {entry['key'] for entry in BackupsPlugin().get_settings_defaults()}
|
|
assert 'backups_retentioncount' in declared
|
|
|
|
Setting.set('backups_retentioncount', 5)
|
|
db.session.commit()
|
|
assert int(Setting.get('backups_retentioncount', 0)) == 5
|
|
# The helper cannot see it, which is the whole defect: it looks under
|
|
# plugin.backups.backups_retentioncount, which nothing writes.
|
|
assert int(BackupsPlugin().get_setting('backups_retentioncount', 0) or 0) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Share kinds that describe themselves
|
|
#
|
|
# A 'share' kind keeps its bytes on SFLD, so ShopDB knew a snapshot existed and
|
|
# where, but nothing about what was in it. The collector now accepts an optional
|
|
# manifest - the index, not the content - which is what lets a marker's asset
|
|
# page list its patterns and name the one that disappeared. Reading the share
|
|
# from the app server is refused on purpose, so this is the only route.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _markermanifest(names, sourcehost='FBYTNCX3', family='TMC470'):
|
|
return {
|
|
'capturedat': '2026-08-25T12:00:00Z',
|
|
'machine': '3204',
|
|
'sourcehost': sourcehost,
|
|
'family': family,
|
|
'systems': ['TMP1700/470'],
|
|
'files': [{'name': n, 'bytes': 266, 'sha256': 'ab' * 32,
|
|
'system': 'TMP1700/470'} for n in names],
|
|
}
|
|
|
|
|
|
def test_a_share_kind_may_send_a_manifest_and_it_is_stored(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'\\host\share\partmarker\3204\20260825-120000',
|
|
'manifest': _markermanifest(['TESTPAT.PAT', 'NET.DAT']),
|
|
})
|
|
assert result['action'] == 'created'
|
|
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
|
# Bytes still live on the share; only the index is in the database.
|
|
assert revision.storagebackend == 'share'
|
|
assert revision.sharepath.endswith('20260825-120000')
|
|
assert [f['name'] for f in revision.payload['files']] == \
|
|
['TESTPAT.PAT', 'NET.DAT']
|
|
|
|
|
|
def test_a_share_kind_without_a_manifest_still_works(bk_app, bk_plugin):
|
|
"""The manifest is advisory. A client that sends none behaves as before."""
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
result = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204',
|
|
'backupkind': 'partmarker',
|
|
'contenthash': 'b' * 64,
|
|
'sharepath': r'\\host\share\partmarker\3204\20260825-130000',
|
|
})
|
|
assert result['action'] == 'created'
|
|
assert _db.session.get(
|
|
BackupRevision, result['backuprevisionid']).payload is None
|
|
|
|
|
|
def test_a_manifest_that_is_not_an_object_is_rejected(bk_app, bk_plugin):
|
|
with bk_app.app_context():
|
|
with pytest.raises(ValueError):
|
|
bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204',
|
|
'backupkind': 'partmarker',
|
|
'contenthash': 'c' * 64,
|
|
'sharepath': r'\\host\share\x',
|
|
'manifest': 'not an object',
|
|
})
|
|
|
|
|
|
def test_marker_card_counts_by_kind_and_names_what_was_removed(bk_app, bk_plugin):
|
|
"""The removed line is the reason the snapshot layout exists.
|
|
|
|
A pattern deleted from a controller is invisible everywhere else until
|
|
somebody needs it. It is NAMED rather than counted - "1 removed" would send
|
|
a tech to the share to find out which one.
|
|
"""
|
|
from plugins.backups.services import partmarkerinfo
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
first = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': 'd' * 64, 'sharepath': r'\\host\s\1',
|
|
'sourcehostname': 'FBYTNCX3',
|
|
'manifest': _markermanifest(['A.PAT', 'B.PAT', 'NET.DAT']),
|
|
})
|
|
_db.session.commit()
|
|
second = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': 'e' * 64, 'sharepath': r'\\host\s\2',
|
|
'sourcehostname': 'FBYTNCX3',
|
|
'manifest': _markermanifest(['A.PAT', 'C.PAT', 'NET.DAT']),
|
|
})
|
|
_db.session.commit()
|
|
assert first['action'] == 'created' and second['action'] == 'created'
|
|
|
|
revision = _db.session.get(BackupRevision, second['backuprevisionid'])
|
|
card = partmarkerinfo.build(revision.payload, revision.assetid,
|
|
revisionid=revision.backuprevisionid)
|
|
fields = {f['label']: f['value'] for f in card['fields']}
|
|
|
|
assert fields['Patterns'] == '2'
|
|
assert fields['Settings files'] == '1'
|
|
assert fields['Controller'] == 'TMC470'
|
|
assert fields['Added'] == 'C.PAT'
|
|
assert fields['Removed'] == 'B.PAT'
|
|
|
|
|
|
def test_the_first_snapshot_has_nothing_to_diff_against(bk_app, bk_plugin):
|
|
"""Every file would otherwise read as 'added', which says nothing."""
|
|
from plugins.backups.services import partmarkerinfo
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
result = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': 'f' * 64, 'sharepath': r'\\host\s\1',
|
|
'manifest': _markermanifest(['A.PAT']),
|
|
})
|
|
_db.session.commit()
|
|
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
|
card = partmarkerinfo.build(revision.payload, revision.assetid,
|
|
revisionid=revision.backuprevisionid)
|
|
labels = [f['label'] for f in card['fields']]
|
|
assert 'Added' not in labels and 'Removed' not in labels
|
|
|
|
|
|
def test_a_revision_with_no_manifest_renders_no_card():
|
|
"""Older revisions predate manifests; they must not error the panel."""
|
|
from plugins.backups.services import partmarkerinfo
|
|
assert partmarkerinfo.build(None, 1)['sectioncount'] == 0
|
|
assert partmarkerinfo.build({}, 1)['fields'] == []
|
|
|
|
|
|
def test_a_noop_backfills_a_manifest_the_row_never_had(bk_app, bk_plugin):
|
|
"""The hash attests the BYTES, not the manifest.
|
|
|
|
A client that starts describing itself posts the same contenthash as
|
|
before, so without this the manifest would never be stored on exactly the
|
|
markers that are healthy and unchanging - which is most of them. The panel
|
|
would then be visible only on markers that had recently changed.
|
|
"""
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
bare = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '1' * 64, 'sharepath': r'\\host\s\old',
|
|
'sourcehostname': 'FBYTNCX3',
|
|
})
|
|
_db.session.commit()
|
|
assert _db.session.get(BackupRevision, bare['backuprevisionid']).payload is None
|
|
|
|
described = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '1' * 64, 'sharepath': r'\\host\s\new',
|
|
'sourcehostname': 'FBYTNCX3',
|
|
'manifest': _markermanifest(['A.PAT']),
|
|
})
|
|
_db.session.commit()
|
|
|
|
# Still a no-op: nothing changed on the controller, so no new history.
|
|
assert described['action'] == 'noop'
|
|
assert described['backuprevisionid'] == bare['backuprevisionid']
|
|
|
|
revision = _db.session.get(BackupRevision, bare['backuprevisionid'])
|
|
assert [f['name'] for f in revision.payload['files']] == ['A.PAT']
|
|
# And the pointer follows a snapshot that moved on the share.
|
|
assert revision.sharepath == r'\\host\s\new'
|
|
|
|
|
|
def test_a_noop_does_not_overwrite_a_manifest_that_is_already_there(bk_app, bk_plugin):
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
first = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '2' * 64, 'sharepath': r'\\host\s\1',
|
|
'sourcehostname': 'FBYTNCX3',
|
|
'manifest': _markermanifest(['ORIGINAL.PAT']),
|
|
})
|
|
_db.session.commit()
|
|
bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '2' * 64, 'sharepath': r'\\host\s\1',
|
|
'sourcehostname': 'FBYTNCX3',
|
|
'manifest': _markermanifest(['REPLACEMENT.PAT']),
|
|
})
|
|
_db.session.commit()
|
|
revision = _db.session.get(BackupRevision, first['backuprevisionid'])
|
|
assert [f['name'] for f in revision.payload['files']] == ['ORIGINAL.PAT']
|
|
|
|
|
|
def test_a_file_edited_in_place_is_named_as_changed(bk_app, bk_plugin):
|
|
"""Same file set, different bytes.
|
|
|
|
A revision exists only because the hash moved, so reporting "no files added
|
|
or removed" here would contradict the reason the row is there at all.
|
|
"""
|
|
from plugins.backups.services import partmarkerinfo
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
before = _markermanifest(['A.PAT', 'B.PAT'])
|
|
after = _markermanifest(['A.PAT', 'B.PAT'])
|
|
after['files'][1]['sha256'] = 'cd' * 32
|
|
|
|
bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '3' * 64, 'sharepath': r'\\host\s\1',
|
|
'sourcehostname': 'FBYTNCX3', 'manifest': before})
|
|
_db.session.commit()
|
|
second = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '4' * 64, 'sharepath': r'\\host\s\2',
|
|
'sourcehostname': 'FBYTNCX3', 'manifest': after})
|
|
_db.session.commit()
|
|
|
|
revision = _db.session.get(BackupRevision, second['backuprevisionid'])
|
|
card = partmarkerinfo.build(revision.payload, revision.assetid,
|
|
revisionid=revision.backuprevisionid)
|
|
fields = {f['label']: f['value'] for f in card['fields']}
|
|
assert fields['Changed'] == 'B.PAT'
|
|
assert 'Added' not in fields and 'Removed' not in fields
|
|
|
|
|
|
def test_two_pcs_reporting_one_marker_do_not_cross_diff(bk_app, bk_plugin):
|
|
"""Dedup chains per source host, and the diff must follow the same chain.
|
|
|
|
Diffing across chains would report every file as added and removed at once
|
|
the moment a second PC reported the same marker.
|
|
"""
|
|
from plugins.backups.services import partmarkerinfo
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '5' * 64, 'sharepath': r'\\host\s\a',
|
|
'sourcehostname': 'PC-A',
|
|
'manifest': _markermanifest(['ONLY-ON-A.PAT'])})
|
|
_db.session.commit()
|
|
onb = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '6' * 64, 'sharepath': r'\\host\s\b',
|
|
'sourcehostname': 'PC-B',
|
|
'manifest': _markermanifest(['ONLY-ON-B.PAT'])})
|
|
_db.session.commit()
|
|
|
|
revision = _db.session.get(BackupRevision, onb['backuprevisionid'])
|
|
card = partmarkerinfo.build(revision.payload, revision.assetid,
|
|
revisionid=revision.backuprevisionid)
|
|
fields = {f['label']: f['value'] for f in card['fields']}
|
|
# PC-B's first snapshot has no predecessor OF ITS OWN, so no diff at all
|
|
# rather than a diff against PC-A's unrelated history.
|
|
assert 'Added' not in fields and 'Removed' not in fields
|
|
|
|
|
|
def test_a_predecessor_without_a_manifest_yields_no_diff(bk_app, bk_plugin):
|
|
"""Mixed rollout: the previous snapshot predates manifests."""
|
|
from plugins.backups.services import partmarkerinfo
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '7' * 64, 'sharepath': r'\\host\s\1',
|
|
'sourcehostname': 'FBYTNCX3'})
|
|
_db.session.commit()
|
|
second = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '8' * 64, 'sharepath': r'\\host\s\2',
|
|
'sourcehostname': 'FBYTNCX3',
|
|
'manifest': _markermanifest(['A.PAT'])})
|
|
_db.session.commit()
|
|
|
|
revision = _db.session.get(BackupRevision, second['backuprevisionid'])
|
|
card = partmarkerinfo.build(revision.payload, revision.assetid,
|
|
revisionid=revision.backuprevisionid)
|
|
labels = [f['label'] for f in card['fields']]
|
|
assert 'Added' not in labels and 'Removed' not in labels
|
|
assert 'Patterns' in labels # the card itself still renders
|
|
|
|
|
|
def test_a_wiped_controller_still_renders_and_reports_the_loss(bk_app, bk_plugin):
|
|
"""files: [] is a fact, not an absence.
|
|
|
|
A controller wiped or swapped is the disaster the Removed line exists for,
|
|
so it must not render as no panel at all.
|
|
"""
|
|
from plugins.backups.services import partmarkerinfo
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': '9' * 64, 'sharepath': r'\\host\s\1',
|
|
'sourcehostname': 'FBYTNCX3',
|
|
'manifest': _markermanifest(['GONE.PAT'])})
|
|
_db.session.commit()
|
|
empty = _markermanifest([])
|
|
wiped = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': 'a1' * 32, 'sharepath': r'\\host\s\2',
|
|
'sourcehostname': 'FBYTNCX3', 'manifest': empty})
|
|
_db.session.commit()
|
|
|
|
revision = _db.session.get(BackupRevision, wiped['backuprevisionid'])
|
|
card = partmarkerinfo.build(revision.payload, revision.assetid,
|
|
revisionid=revision.backuprevisionid)
|
|
fields = {f['label']: f['value'] for f in card['fields']}
|
|
assert card['sectioncount'] == 1
|
|
assert fields['Removed'] == 'GONE.PAT'
|
|
|
|
|
|
def test_a_malformed_manifest_is_refused_before_it_can_break_a_panel(bk_app, bk_plugin):
|
|
"""A client bug must not become a page that fails until content changes."""
|
|
bad = [
|
|
{'files': ['A.PAT']}, # not objects
|
|
{'files': [{'bytes': 10}]}, # nameless
|
|
{'files': [{'name': 'A.PAT', 'bytes': '1,024'}]}, # size not an int
|
|
{'files': 'A.PAT'}, # not a list
|
|
]
|
|
with bk_app.app_context():
|
|
for manifest in bad:
|
|
with pytest.raises(ValueError):
|
|
bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': 'b2' * 32, 'sharepath': r'\\host\s\x',
|
|
'manifest': manifest})
|
|
|
|
|
|
def test_buildinfo_kwargs_do_not_disturb_the_live_ntlars_card(bk_app, bk_plugin):
|
|
"""The endpoint passes revisionid and sourcehostname to EVERY kind.
|
|
|
|
NTLARS is live in production; a new keyword reaching its buildinfo must be
|
|
absorbed by **options rather than raising.
|
|
"""
|
|
with bk_app.app_context():
|
|
result = bk_plugin.apply_collector_payload(_payload())
|
|
_db.session.commit()
|
|
from plugins.backups.models import BackupRevision
|
|
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
|
# The assertion is that this does not raise TypeError on the new
|
|
# keywords, not that this particular fixture produces sections.
|
|
card = registry.getkind('ntlars').buildinfo(
|
|
revision.payload, revision.assetid,
|
|
partmarkertypes=None, revisionid=revision.backuprevisionid,
|
|
sourcehostname=revision.sourcehostname)
|
|
assert 'sectioncount' in card
|
|
|
|
|
|
def test_the_card_shows_the_snapshot_path_it_was_actually_written_to(bk_app, bk_plugin):
|
|
"""Plain text, and from the REVISION - not composed from the setting.
|
|
|
|
A snapshot written before the share layout changed still lives where it was
|
|
put, so the stored path is the only honest answer. It is not a link: a
|
|
browser refuses to navigate from https to file://, so a clickable path
|
|
would look actionable and do nothing.
|
|
"""
|
|
from plugins.backups.services import partmarkerinfo
|
|
from plugins.backups.models import BackupRevision
|
|
|
|
with bk_app.app_context():
|
|
result = bk_plugin.apply_collector_payload({
|
|
'machinenumber': '3204', 'backupkind': 'partmarker',
|
|
'contenthash': 'c3' * 32,
|
|
'sharepath': r'\\host\share\partmarker\3204\20260825-144500',
|
|
'sourcehostname': 'FBYTNCX3',
|
|
'manifest': _markermanifest(['A.PAT'])})
|
|
_db.session.commit()
|
|
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
|
|
|
card = partmarkerinfo.build(revision.payload, revision.assetid,
|
|
revisionid=revision.backuprevisionid,
|
|
sharepath=revision.sharepath)
|
|
snapshot = [f for f in card['fields'] if f['label'] == 'Snapshot']
|
|
assert snapshot and snapshot[0]['value'] == revision.sharepath
|
|
assert snapshot[0]['mono'] is True
|
|
|
|
|
|
def test_the_card_omits_the_snapshot_row_when_no_path_is_passed(bk_app, bk_plugin):
|
|
from plugins.backups.services import partmarkerinfo
|
|
card = partmarkerinfo.build(_markermanifest(['A.PAT']), 1)
|
|
assert 'Snapshot' not in [f['label'] for f in card['fields']]
|