{{ assetname }}
+| Captured | +Kind | +From PC | +Hash | +Changes | +Download | +
|---|---|---|---|---|---|
| + {{ formatWhen(rev.collectedat || rev.createdat) }} + current + | +{{ rev.backupkind }} | +{{ rev.sourcehostname || '-' }} | +{{ rev.shorthash }} | ++ + - + | +
+
+
+ {{ rev.sharepath }}
+
+
+
+
+ |
+
+ {{ diffs[rev.backuprevisionid].message }} +
+| Key | Value | Before | After |
|---|---|---|---|
| {{ c.keypath || '(root)' }} | +{{ c.valuename }} | +{{ display(c.before) }} | +{{ display(c.after) }} | +
No differences.
+[0-9a-fA-F]+)\))?:(?P.*)$', body, re.S)
+ if not m:
+ raise NtlarsParseError('unparseable hex value: {!r}'.format(body))
+ code = int(m.group('code'), 16) if m.group('code') else 3
+ tokens = [t.strip() for t in m.group('bytes').split(',') if t.strip()]
+ try:
+ data = bytes(int(t, 16) for t in tokens)
+ except ValueError as exc:
+ raise NtlarsParseError('bad hex byte in value: {}'.format(exc))
+
+ typename = HEXTYPES.get(code, 'REG_BINARY')
+
+ # Wide-string hex types decode back to text so diffs stay readable.
+ if typename in ('REG_SZ', 'REG_EXPAND_SZ'):
+ return typename, data.decode('utf-16-le', errors='replace').rstrip('\x00')
+ if typename == 'REG_MULTI_SZ':
+ text = data.decode('utf-16-le', errors='replace')
+ return typename, [p for p in text.split('\x00') if p]
+ # REG_QWORD must come back as an int: _rendervalue turns it back into
+ # little-endian bytes via int(), so storing the "aa,bb" byte form here
+ # would raise at download time - i.e. precisely when someone is trying to
+ # restore a machine.
+ if typename == 'REG_QWORD':
+ return typename, int.from_bytes(data, 'little')
+ return typename, ','.join('{:02x}'.format(b) for b in data)
+
+
+def _parsevalue(body):
+ """Parse the right-hand side of a .reg value assignment."""
+ body = body.strip()
+ if body.startswith('"'):
+ if not body.endswith('"') or len(body) < 2:
+ raise NtlarsParseError('unterminated string value: {!r}'.format(body))
+ return 'REG_SZ', _unescape(body[1:-1])
+ if body.lower().startswith('dword:'):
+ try:
+ return 'REG_DWORD', int(body.split(':', 1)[1].strip(), 16)
+ except ValueError:
+ raise NtlarsParseError('bad dword value: {!r}'.format(body))
+ if body.lower().startswith('hex'):
+ return _parsehexvalue(body)
+ if body == '-':
+ return 'DELETE', None
+ raise NtlarsParseError('unrecognised value form: {!r}'.format(body))
+
+
+def _striproot(keypath):
+ """Strip a known DNC root, returning the relative subkey path.
+
+ Returns None for keys outside the DNC tree so callers can ignore them
+ rather than silently folding unrelated hives into the backup.
+ """
+ for root in KNOWNROOTS:
+ if keypath.upper() == root.upper():
+ return ''
+ prefix = root.upper() + '\\'
+ if keypath.upper().startswith(prefix):
+ return keypath[len(prefix):]
+ return None
+
+
+def parse(raw):
+ """Parse .reg bytes/text into the dialect-neutral JSON projection.
+
+ Returns {'schema', 'sourcedialect', 'keys': [{'path', 'values': {...}}]}
+ with keys and value names sorted for deterministic diffing.
+ """
+ text = decodereg(raw)
+ lines = _joincontinuations(text)
+
+ if not any(line.strip().lower().startswith('windows registry editor')
+ or line.strip().lower().startswith('regedit4')
+ for line in lines[:5]):
+ raise NtlarsParseError('missing "Windows Registry Editor" header')
+
+ keys = {}
+ current = None
+ sawwow = False
+ sawplain = False
+
+ for line in lines:
+ stripped = line.strip()
+ if not stripped or stripped.startswith(';'):
+ continue
+
+ if stripped.startswith('[') and stripped.endswith(']'):
+ keypath = stripped[1:-1].strip()
+ if keypath.startswith('-'):
+ current = None # key deletion, not a backup concern
+ continue
+ if keypath.upper().startswith(ROOTWOW.upper()):
+ sawwow = True
+ elif keypath.upper().startswith(ROOTNTLARS.upper()):
+ sawplain = True
+ rel = _striproot(keypath)
+ if rel is None:
+ current = None # outside the DNC tree - ignore
+ continue
+ current = rel
+ keys.setdefault(current, {})
+ continue
+
+ if current is None or '=' not in stripped:
+ continue
+
+ # Match the QUOTED name and split at the '=' that follows its closing
+ # quote. A plain split('=', 1) breaks on any value name containing '='
+ # or an escaped quote - both legal in the registry - and silently drops
+ # the value.
+ match = re.match(r'^(?:@|"((?:[^"\\]|\\.)*)")\s*=\s*(.*)$', stripped, re.S)
+ if not match:
+ continue
+ name = '' if match.group(1) is None else _unescape(match.group(1))
+ body = match.group(2)
+
+ # Deliberately NOT caught. A backup that silently dropped an
+ # unparseable value would present as complete and restore a machine
+ # with a setting missing - the exact silent-failure class this project
+ # has repeatedly been bitten by. Fail the whole parse instead; the
+ # collector reports it and the previous good revision stays newest.
+ typename, data = _parsevalue(body)
+ if typename == 'DELETE':
+ continue
+ keys[current][name] = {'type': typename, 'data': data}
+
+ if not keys:
+ raise NtlarsParseError(
+ 'no keys under {} or {} - not an NTLARS DNC backup'.format(
+ ROOTNTLARS, ROOTWOW))
+
+ dialect = 'wow6432node' if sawwow else ('ntlars' if sawplain else 'unknown')
+
+ return {
+ 'schema': SCHEMA,
+ 'sourcedialect': dialect,
+ 'keys': [
+ {'path': path, 'values': dict(sorted(keys[path].items()))}
+ for path in sorted(keys)
+ ],
+ }
+
+
+def _rendervalue(name, entry):
+ typename = entry.get('type', 'REG_SZ')
+ data = entry.get('data')
+ lhs = '@' if name == '' else '"{}"'.format(_escape(name))
+
+ if typename == 'REG_SZ':
+ return '{}="{}"'.format(lhs, _escape('' if data is None else str(data)))
+ if typename == 'REG_DWORD':
+ return '{}=dword:{:08x}'.format(lhs, int(data) & 0xFFFFFFFF)
+ if typename == 'REG_QWORD':
+ return '{}=hex(b):{}'.format(lhs, _hexbytes(
+ int(data).to_bytes(8, 'little')))
+ if typename == 'REG_EXPAND_SZ':
+ payload = ('' if data is None else str(data)).encode('utf-16-le') + b'\x00\x00'
+ return '{}=hex(2):{}'.format(lhs, _hexbytes(payload))
+ if typename == 'REG_MULTI_SZ':
+ parts = data if isinstance(data, list) else [str(data)]
+ payload = ''.join(p + '\x00' for p in parts).encode('utf-16-le') + b'\x00\x00'
+ return '{}=hex(7):{}'.format(lhs, _hexbytes(payload))
+
+ # REG_BINARY / REG_NONE: data is the "aa,bb,cc" form parse() produced.
+ raw = bytes(int(t, 16) for t in str(data).split(',') if t.strip()) if data else b''
+ code = HEXTYPECODES.get(typename, 3)
+ prefix = 'hex:' if code == 3 else 'hex({:x}):'.format(code)
+ return '{}={}{}'.format(lhs, prefix, _hexbytes(raw))
+
+
+def _hexbytes(raw):
+ return ','.join('{:02x}'.format(b) for b in raw)
+
+
+def render(projection, dialect='ntlars', comments=None):
+ """Render the JSON projection back to .reg bytes (UTF-16LE, BOM, CRLF).
+
+ dialect='ntlars' omits WOW6432Node - use with the NTLARS Load... button
+ dialect='wow6432node' includes it - use with `reg import` on 64-bit
+ """
+ if dialect not in ('ntlars', 'wow6432node'):
+ raise ValueError('unknown dialect: {!r}'.format(dialect))
+ root = ROOTWOW if dialect == 'wow6432node' else ROOTNTLARS
+
+ out = [HEADER, '']
+ for line in (comments or []):
+ out.append('; {}'.format(line))
+ if comments:
+ out.append('')
+
+ for key in projection.get('keys', []):
+ path = key.get('path', '')
+ out.append('[{}]'.format(root + ('\\' + path if path else '')))
+ for name, entry in (key.get('values') or {}).items():
+ out.append(_rendervalue(name, entry))
+ out.append('')
+
+ text = '\r\n'.join(out)
+ if not text.endswith('\r\n'):
+ text += '\r\n'
+ return b'\xff\xfe' + text.encode('utf-16-le')
diff --git a/plugins/backups/services/registry.py b/plugins/backups/services/registry.py
new file mode 100644
index 00000000..449a9071
--- /dev/null
+++ b/plugins/backups/services/registry.py
@@ -0,0 +1,237 @@
+"""Backup kind registry.
+
+A "kind" is one category of configuration backup (NTLARS/DNC registry, part
+marker, UDC, CMM ...). Each kind declares three things the rest of the plugin
+needs and nothing else:
+
+ where its bytes live 'shopdb' (parsed into payloadjson) or 'share'
+ (opaque vendor file left on the SFLD share)
+ how to resolve an asset the collector reports an identifier; the kind turns
+ it into the assetid the revision belongs to
+ how to render downloads one or more output formats
+
+Adding a kind is a class plus one REGISTRY entry. Nothing else in the plugin
+knows kind names.
+"""
+
+import hashlib
+import json
+
+from . import ntlars as ntlarscodec
+
+# Default root of the opaque-backup tree on the SFLD share. Connected PCs write
+# here directly (they need SFLD creds - a SYSTEM process hitting a UNC path
+# without them gets an access-denied that Test-Path reports as "not found").
+#
+# This is the WEST JEFFERSON path and is only a DEFAULT: the live value is the
+# backups_shareroot setting, because a bundled plugin in a multi-site product
+# must not hardcode one site's file server.
+DEFAULTSHAREROOT = r'\\tsgwp00525.wjs.geaerospace.net\shared\dt\shopfloor\backups'
+
+
+def canonicalhash(projection):
+ """sha256 over the SEMANTIC content only. Dedup key for 'shopdb' kinds.
+
+ Deliberately hashes just schema + keys, excluding sourcedialect. The same
+ settings exported through NTLARS's Save... button and through a scripted
+ reg export parse to identical keys but differing sourcedialect; hashing the
+ whole projection would record that as a change and produce a spurious
+ revision every time the collection route changed. Storage is
+ dialect-neutral, so the dedup key has to be too.
+ """
+ subset = {
+ 'schema': projection.get('schema'),
+ 'keys': projection.get('keys'),
+ }
+ blob = json.dumps(subset, sort_keys=True, separators=(',', ':'))
+ return hashlib.sha256(blob.encode('utf-8')).hexdigest()
+
+
+def byteshash(raw):
+ """sha256 over raw bytes. Dedup key for 'share' kinds."""
+ return hashlib.sha256(raw).hexdigest()
+
+
+class BackupKind:
+ """Base class. Subclasses override what applies to them."""
+
+ key = None
+ displayname = None
+ storagebackend = 'shopdb'
+ assettypes = ['*']
+ # Human note shown on the panel when there is nothing yet.
+ emptytext = 'No backups on record.'
+
+ def parse(self, raw):
+ """Opaque kinds return None; parseable kinds return the projection."""
+ return None
+
+ def formats(self):
+ """Downloadable formats: [{'id','label','ext','mimetype'}]."""
+ return []
+
+ def render(self, projection, formatid, comments=None):
+ """Return (bytes, extension, mimetype) for a 'shopdb' kind."""
+ raise NotImplementedError
+
+ def resolveassetid(self, payload):
+ """Map a collector payload to the assetid the backup belongs to."""
+ raise NotImplementedError
+
+ def infopanel(self):
+ """Optional 'at a glance' card built from this kind's LATEST revision.
+
+ Returns an ADR-010 panel dict, or None when the kind has no summary
+ worth surfacing. Declared by the kind rather than hardcoded in the
+ plugin so a successor technology (NTLARS/DNC is expected to give way to
+ Shopfloor Connect) ships its own card by adding a class, without
+ touching the plugin or the panel wiring.
+ """
+ return None
+
+ def buildinfo(self, projection, assetid, **options):
+ """Build this kind's info-card payload. Only called when infopanel()."""
+ return {'fields': [], 'sectioncount': 0}
+
+ def sharedir(self, machinetype, identifier, shareroot=None):
+ """UNC directory a 'share' kind's files are expected under.
+
+ Advisory only - the authoritative path is the sharepath the collector
+ reports, since the PC is what actually wrote the file. This builds the
+ conventional location for display and for validating a reported path.
+ """
+ return '{}\\{}\\{}\\{}'.format(
+ shareroot or DEFAULTSHAREROOT,
+ machinetype or 'unknown', identifier or 'unknown', self.key)
+
+
+class NtlarsKind(BackupKind):
+ """NTLARS / DNC registry settings.
+
+ Collected from the CONTROLLING PC's registry but belongs to the MACHINE:
+ the settings describe how to talk to that machine's controller, so they
+ follow the machine when a PC is swapped. resolveassetid therefore keys on
+ the reported machine number, not the hostname.
+ """
+
+ key = 'ntlars'
+ displayname = 'NTLARS / DNC Settings'
+ storagebackend = 'shopdb'
+ assettypes = ['machine']
+ emptytext = 'No NTLARS settings captured yet.'
+
+ @staticmethod
+ def embeddedmachineno(projection):
+ """The MachineNo NTLARS itself is configured with (General tab)."""
+ for key in (projection or {}).get('keys', []):
+ if (key.get('path') or '').lower() == 'general':
+ entry = (key.get('values') or {}).get('MachineNo') or {}
+ return str(entry.get('data') or '').strip()
+ return ''
+
+ def parse(self, raw):
+ """Parse, then refuse to record an unconfigured NTLARS install.
+
+ A freshly imaged PC opens NTLARS with a blank General tab (no MachineNo,
+ CNC, Host or Interface type) until a tech restores the config. Storing
+ that would make an empty config the newest revision at exactly the
+ moment someone needs the last good one - so a blank MachineNo is
+ rejected as "nothing worth backing up" rather than accepted as a change.
+ Two of the 320 known-good backups on the share already have this shape.
+ """
+ projection = ntlarscodec.parse(raw)
+ if not self.embeddedmachineno(projection):
+ raise ValueError(
+ 'NTLARS General\\MachineNo is empty - unconfigured install, '
+ 'refusing to record it as a revision')
+ return projection
+
+ def formats(self):
+ return [
+ {
+ 'id': 'ntlars',
+ 'label': 'NTLARS Load... (.reg)',
+ 'ext': '.reg',
+ 'mimetype': 'application/octet-stream',
+ 'hint': 'Restore with the Load... button in the NTLARS settings dialog.',
+ },
+ {
+ 'id': 'wow6432node',
+ 'label': 'Direct reg import (.reg, WOW6432Node)',
+ 'ext': '.reg',
+ 'mimetype': 'application/octet-stream',
+ 'hint': 'Use when importing outside NTLARS on a 64-bit machine.',
+ },
+ ]
+
+ def render(self, projection, formatid, comments=None):
+ if formatid not in ('ntlars', 'wow6432node'):
+ raise ValueError('unknown format for ntlars kind: {}'.format(formatid))
+ raw = ntlarscodec.render(projection, dialect=formatid, comments=comments)
+ return raw, '.reg', 'application/octet-stream'
+
+ def infopanel(self):
+ return {
+ 'id': 'backups-dncinfo',
+ 'title': 'DNC Info',
+ 'assettypes': ['machine'],
+ 'endpoint': '/api/backups/asset/{assetid}/info?kind=ntlars',
+ 'render': 'keyvalue',
+ 'empty': 'No NTLARS settings captured for this machine yet.',
+ # Above the history panels: this answers the question a tech
+ # arrives with, while history is for the rarer restore case.
+ 'position': 38,
+ }
+
+ def buildinfo(self, projection, assetid, **options):
+ from . import dncinfo
+ return dncinfo.build(projection, assetid,
+ partmarkertypes=options.get('partmarkertypes'))
+
+ def resolveassetid(self, payload):
+ from shopdb.api import db, Asset
+
+ machinenumber = (payload.get('machinenumber') or '').strip()
+ if not machinenumber:
+ return None, 'no machinenumber in payload'
+ asset = db.session.query(Asset).filter(
+ Asset.assetnumber == machinenumber).first()
+ if asset is None:
+ return None, 'no asset with assetnumber {!r}'.format(machinenumber)
+ return asset.assetid, None
+
+
+class PartMarkerKind(BackupKind):
+ """Telesis part marker configuration.
+
+ Opaque vendor filetype with no useful JSON representation, so ShopDB stores
+ metadata and a UNC pointer while the file itself stays on the share. The
+ original filename and extension are preserved because the vendor tool
+ rejects a renamed file.
+ """
+
+ key = 'partmarker'
+ displayname = 'Part Marker Configuration'
+ storagebackend = 'share'
+ assettypes = ['machine', 'measuring_tool']
+ emptytext = 'No part marker backups on record.'
+
+ def resolveassetid(self, payload):
+ from shopdb.api import db, Asset
+
+ identifier = (payload.get('machinenumber')
+ or payload.get('assetnumber') or '').strip()
+ if not identifier:
+ return None, 'no machinenumber/assetnumber in payload'
+ asset = db.session.query(Asset).filter(
+ Asset.assetnumber == identifier).first()
+ if asset is None:
+ return None, 'no asset with assetnumber {!r}'.format(identifier)
+ return asset.assetid, None
+
+
+REGISTRY = {k.key: k for k in (NtlarsKind(), PartMarkerKind())}
+
+
+def getkind(key):
+ return REGISTRY.get((key or '').strip().lower())
diff --git a/plugins/backups/services/retention.py b/plugins/backups/services/retention.py
new file mode 100644
index 00000000..35abf25d
--- /dev/null
+++ b/plugins/backups/services/retention.py
@@ -0,0 +1,68 @@
+"""Revision retention.
+
+Dedup already keeps growth low - a revision appears only when a setting really
+changed - so retention exists for the pathological case, not the normal one: a
+value that flaps (or two PCs alternately claiming one machine number) would
+otherwise append a revision every collection cycle forever.
+
+Two independent limits, both off by default at 0:
+
+ retentioncount keep at most N revisions per asset per kind
+ retentiondays additionally drop anything older than N days
+
+The NEWEST and the OLDEST revision are never pruned. The newest is the one a
+tech restores from; the oldest is the earliest known-good baseline, which is
+usually the most valuable row in the table and the one a naive "keep last N"
+would delete first.
+"""
+
+from datetime import datetime, timedelta
+
+from shopdb.api import db
+
+from ..models import BackupRevision
+
+
+def prune(assetid, backupkind, retentioncount=0, retentiondays=0):
+ """Delete surplus revisions for one asset+kind. Returns the number removed.
+
+ Caller commits. Returns 0 when both limits are disabled.
+ """
+ retentioncount = int(retentioncount or 0)
+ retentiondays = int(retentiondays or 0)
+ if retentioncount <= 0 and retentiondays <= 0:
+ return 0
+
+ revisions = (db.session.query(BackupRevision)
+ .filter(BackupRevision.assetid == assetid,
+ BackupRevision.backupkind == backupkind)
+ .order_by(BackupRevision.backuprevisionid.desc())
+ .all())
+ if len(revisions) <= 2:
+ return 0 # newest + oldest are both protected
+
+ newest = revisions[0]
+ oldest = revisions[-1]
+ protected = {newest.backuprevisionid, oldest.backuprevisionid}
+
+ doomed = []
+
+ if retentioncount > 0 and len(revisions) > retentioncount:
+ # Walk from the oldest END of the middle, so the surplus dropped is
+ # always the least recent, never the newest.
+ for revision in revisions[retentioncount:]:
+ if revision.backuprevisionid not in protected:
+ doomed.append(revision)
+
+ if retentiondays > 0:
+ cutoff = datetime.utcnow() - timedelta(days=retentiondays)
+ for revision in revisions:
+ if revision.backuprevisionid in protected:
+ continue
+ stamp = revision.collectedat or revision.createdat
+ if stamp and stamp < cutoff and revision not in doomed:
+ doomed.append(revision)
+
+ for revision in doomed:
+ db.session.delete(revision)
+ return len(doomed)
diff --git a/scripts/import_ntlars_backups.py b/scripts/import_ntlars_backups.py
new file mode 100644
index 00000000..95b69aad
--- /dev/null
+++ b/scripts/import_ntlars_backups.py
@@ -0,0 +1,137 @@
+#!/usr/bin/env python
+"""Seed backup history from a directory of existing NTLARS .reg files.
+
+A site's file share typically already holds per-machine backups collected by
+hand over years, named by machine number. Importing them gives every machine a
+baseline revision on day one, so the feature is useful before the fleet
+collector has run even once.
+
+Files are keyed by machine number in the FILENAME, which is how the share names
+them. The MachineNo embedded in the file is compared against it and any
+disagreement is reported: in the curated set the two always agree, so a
+mismatch means the file is misfiled and should not be trusted as that machine's
+baseline.
+
+Dry run by default. Nothing is written without --commit.
+
+ venv/bin/python scripts/import_ntlars_backups.py /path/to/regs
+ venv/bin/python scripts/import_ntlars_backups.py /path/to/regs --commit
+"""
+
+import argparse
+import os
+import sys
+from datetime import datetime
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from shopdb import create_app # noqa: E402
+from shopdb.api import db # noqa: E402
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument('directory', help='directory holding .reg files')
+ parser.add_argument('--commit', action='store_true',
+ help='actually write (default is a dry run)')
+ parser.add_argument('--hostname', default='imported',
+ help="sourcehostname to record (default 'imported')")
+ args = parser.parse_args()
+
+ if not os.path.isdir(args.directory):
+ parser.error('not a directory: {}'.format(args.directory))
+
+ regs = sorted(f for f in os.listdir(args.directory) if f.lower().endswith('.reg'))
+ if not regs:
+ parser.error('no .reg files in {}'.format(args.directory))
+
+ app = create_app()
+ with app.app_context():
+ from shopdb.core.models import Asset
+ from plugins.backups.models import BackupRevision
+ from plugins.backups.services.registry import getkind, canonicalhash
+
+ kind = getkind('ntlars')
+
+ created = skipped = unresolved = rejected = duplicate = 0
+ mismatches = []
+
+ for filename in regs:
+ path = os.path.join(args.directory, filename)
+ machinenumber = os.path.splitext(filename)[0]
+
+ with open(path, 'rb') as handle:
+ raw = handle.read()
+
+ try:
+ projection = kind.parse(raw)
+ except ValueError as exc:
+ # Unconfigured NTLARS installs are in the corpus; they are not
+ # a usable baseline for anyone.
+ print(' REJECT {:<12} {}'.format(machinenumber, exc))
+ rejected += 1
+ continue
+
+ embedded = kind.embeddedmachineno(projection)
+ if embedded and embedded != machinenumber:
+ mismatches.append((machinenumber, embedded))
+
+ asset = db.session.query(Asset).filter(
+ Asset.assetnumber == machinenumber).first()
+ if asset is None:
+ print(' NOASSET {:<12} no asset with that assetnumber'.format(
+ machinenumber))
+ unresolved += 1
+ continue
+
+ contenthash = canonicalhash(projection)
+ latest = (db.session.query(BackupRevision)
+ .filter(BackupRevision.assetid == asset.assetid,
+ BackupRevision.backupkind == 'ntlars')
+ .order_by(BackupRevision.backuprevisionid.desc())
+ .first())
+ if latest is not None and latest.contenthash == contenthash:
+ duplicate += 1
+ continue
+
+ if args.commit:
+ revision = BackupRevision(
+ assetid=asset.assetid,
+ backupkind='ntlars',
+ storagebackend='shopdb',
+ contenthash=contenthash,
+ sourcefilename=filename,
+ bytesize=len(raw),
+ sourcehostname=args.hostname,
+ # The file's mtime is the closest thing to when the config
+ # was actually captured; createdat records the import.
+ collectedat=datetime.utcfromtimestamp(os.path.getmtime(path)),
+ )
+ revision.payload = projection
+ db.session.add(revision)
+ created += 1
+
+ if args.commit:
+ db.session.commit()
+
+ print()
+ print('files : {}'.format(len(regs)))
+ print('would create : {}'.format(created) if not args.commit
+ else 'created : {}'.format(created))
+ print('already current : {}'.format(duplicate))
+ print('no matching asset: {}'.format(unresolved))
+ print('rejected (blank) : {}'.format(rejected))
+ if mismatches:
+ print()
+ print('filename/embedded MachineNo disagreements ({}):'.format(
+ len(mismatches)))
+ for filename_no, embedded_no in mismatches[:20]:
+ print(' file={:<10} embedded={}'.format(filename_no, embedded_no))
+ print(' These are likely misfiled; review before trusting them.')
+ if not args.commit:
+ print()
+ print('DRY RUN - nothing written. Re-run with --commit to apply.')
+
+
+if __name__ == '__main__':
+ main()
diff --git a/shopdb/plugins/alembic_template.py b/shopdb/plugins/alembic_template.py
index 213a15c3..9f9cacaf 100644
--- a/shopdb/plugins/alembic_template.py
+++ b/shopdb/plugins/alembic_template.py
@@ -47,6 +47,7 @@ logger = logging.getLogger('alembic.env.plugin')
# Explicit table-ownership map. Adding tables to a plugin requires updating
# this dict so the per-plugin migration knows which tables to include.
PLUGIN_TABLE_OWNERS: dict[str, Iterable[str]] = {
+ 'backups': ('backuprevisions',),
'computers': ('computertypes', 'computers', 'computerinstalledapps',
'accessprotocols', 'computeraccess'),
'employees': ('directoryemployees',),
diff --git a/tests/test_plugin_migrations.py b/tests/test_plugin_migrations.py
index 5acbd2e3..a00b0643 100644
--- a/tests/test_plugin_migrations.py
+++ b/tests/test_plugin_migrations.py
@@ -46,6 +46,8 @@ CUTOVER_PLUGINS = (
# stamp '0001anchor'; measuringtools stamps its real baseline id.
EXPECTED_HEAD_REVISION = {plugin: f'{plugin}0001anchor' for plugin in CUTOVER_PLUGINS}
EXPECTED_HEAD_REVISION['measuringtools'] = 'measuringtools0001baseline'
+# backups is also post-cutover: its 0001 really creates backuprevisions.
+EXPECTED_HEAD_REVISION['backups'] = 'backups0001baseline'
# geenforce adds the content-addressed blob store (manifestblobs) on top of its
# baseline.
EXPECTED_HEAD_REVISION['geenforce'] = 'geenforce0002blobs'
diff --git a/tests/test_plugins/test_backups.py b/tests/test_plugins/test_backups.py
new file mode 100644
index 00000000..55d11599
--- /dev/null
+++ 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))