backups plugin: per-asset config backups with revision history

Adds a kind-pluggable backups plugin. Configuration captured from a PC is filed against the MACHINE it controls, with a revision history and download back to the native format. NTLARS/DNC is the first kind. Settings live in the controlling PC's registry but describe the machine, so revisions attach to the machine's asset and carry no foreign key to the PC: history survives a PC being replaced or deleted, and sourcehostname records the handover. Storage splits by kind. Parseable kinds store a dialect-neutral JSON projection in ShopDB and re-render on download; opaque vendor formats (part marker and similar) keep their bytes on the SFLD share with ShopDB holding metadata and the UNC pointer. Two .reg dialects exist in the wild: NTLARS's own Save... export omits the WOW6432Node path segment, scripted exports include it. Parsing strips whichever root matched, so a stored revision commits to neither and download offers both (NTLARS Load... by default, WOW6432Node for direct reg import). Getting this backwards is silent, so the dedup hash deliberately excludes sourcedialect and both dialects of one config dedup to a single revision. Dedup is load-bearing: the collector runs every GE-Enforce cycle across the fleet, so a revision is inserted only when the content hash differs from that asset's latest for that kind. A freshly imaged PC opens NTLARS with a blank General tab. Recording that would make an empty config the newest revision exactly when someone needs the last good one, so a blank MachineNo is rejected rather than accepted as a change. Two of the 320 known-good backups on the share already have that shape. DNC Info card summarises the latest revision on the machine page: General (Cnc, NcIF, HostType), eFocas, Serial, NTSHR when populated (only 18 of 147 machines), and MARK when the machine is a marker. MARK is gated on Cnc=MARKER or the ShopDB machine type, not on the MARK key having content: MARK carries serial defaults on 145 of 147 machines and DncPatterns reads YES on 103 including ordinary lathes, so neither identifies a marker. The info card is owned by the kind (BackupKind.infopanel/buildinfo) and served by a generic endpoint, so the expected successor to DNC ships its own card by adding a class rather than changing the plugin or the panel wiring. Also: schedule and retention settings with a prune that never drops the newest or the oldest revision, and scripts/import_ntlars_backups.py to seed history from the existing per-machine .reg files (144 of 147 resolve to assets). Codec verified against all 320 real backups: round-trips clean through both dialects. Bay-side generation verified on Windows against reg.exe export.
2026-08-07 14:24:42 -04:00
parent d180a693ee
commit fca775c737
21 changed files with 2675 additions and 0 deletions
--- a/tests/test_plugins/test_backups.py
+++ b/tests/test_plugins/test_backups.py
@@ -0,0 +1,633 @@
+"""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):
+    return [f['label'] for f in card['fields'] if f.get('heading')]
+
+
+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)
+    labels = [f['label'] for f in card['fields'] if not f.get('heading')]
+    assert 'ShrHost' in labels
+    assert 'ShrFolder' not in labels
+
+
+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['fields'] == []
+
+
+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'
+    labels = [f['label'] for f in card['fields'] if not f.get('heading')]
+    assert 'Cnc' in labels and 'HostType' in labels
+
+
+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)
+    labels = [f['label'] for f in card['fields'] if not f.get('heading')]
+    assert 'Debug' not in labels and 'Site' not in labels
+
+
+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))