adb8542018
Backup timestamps read wrong because of two faults stacked, which is why it
looked like a single offset.
The API serialised naive ISO ("2026-08-07T12:00:00"), with nothing saying the
value was UTC. JavaScript's new Date() parses that as BROWSER-LOCAL, so every
timestamp shifted by the viewer's offset before any timezone formatting ran.
Every datetime this plugin stores is naive UTC, so the wire format now carries
a trailing Z.
The history view then formatted with toLocaleString(), i.e. the viewer's zone,
ignoring the site_timezone setting entirely. It now loads that setting and
formats through the shared formatInZone helper, matching NotificationsList.
The panel list label is built server-side with strftime, so a client cannot
correct it afterwards. It now converts to the site zone using the same Setting
lookup the notifications plugin uses - without that it showed UTC, four hours
out at West Jefferson.
Tests cover the wire format and that 16:30Z renders as 12:30 in
America/New_York.
713 lines
28 KiB
Python
713 lines
28 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)
|