6516e76bf4
Backup-NtlarsSettings reads backups_intervalhours from /api/settings/public, because it runs before it holds any credential, and its Get-IntervalHours falls back to 24 on any failure. The plugin never declared the key public, so the endpoint did not return it, the fallback fired on every PC, and the setting looked configurable in the UI while changing nothing. The fleet log shows the symptom plainly: "Throttled: last attempt under 24h ago", every cycle, regardless of what the setting said. Declared public. A collection cadence is not a secret. backups_shareroot stays private - it is internal topology - and the test asserts both directions so a later edit cannot quietly widen it. Same defect as the 3D parts kiosk label prefix already in this changelog: a logged-out reader against an allowlist its key was not on. Worth noticing that the pattern has now bitten twice.
865 lines
35 KiB
Python
865 lines
35 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
|
|
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')
|