Merge origin/main into wilm: dashboard/fleet console, vuln upgrades, colours

tests/test_core/test_relationship_directional.py

@@ -104,3 +104,50 @@ def test_relationships_row_carries_isdirectional(client, db, auth_headers):
     ctrl_rows = [r for r in data['outgoing']
                  if r['relationshiptypename'] == 'controls']
     assert ctrl_rows and all(r['isdirectional'] is True for r in ctrl_rows)
+
+def test_the_controlled_asset_sees_the_link_from_its_own_page(client, db, auth_headers):
+    """The target of a directional edge must receive it in `incoming`.
+
+    A PC controls a machine, and the row is stored once, source=PC. The PC's
+    page reads it from `outgoing`; the machine's page can only ever see it in
+    `incoming`. Nothing asserted that second half, and a UI was shipped that
+    rendered the link on the PC page and showed the machine page nothing - the
+    reciprocal-links defect. The endpoint was correct all along, which is
+    exactly why this assertion is worth having: it pins the contract the
+    frontend depends on, so the next such bug cannot be blamed on the API.
+    """
+    atype_machine = AssetType(assettype='machine')
+    atype_computer = AssetType(assettype='computer')
+    db.session.add_all([atype_machine, atype_computer])
+    db.session.flush()
+
+    controls = RelationshipType(relationshiptype='controls', isdirectional=True)
+    db.session.add(controls)
+    db.session.flush()
+
+    pc = _make_asset('PC-1', atype_computer.assettypeid)
+    machine = _make_asset('M-1', atype_machine.assettypeid)
+
+    # Stored ONCE, in the canonical direction only.
+    db.session.add(AssetRelationship(sourceassetid=pc.assetid,
+                                     targetassetid=machine.assetid,
+                                     relationshiptypeid=controls.relationshiptypeid))
+    db.session.commit()
+
+    resp = client.get(f'/api/assets/{machine.assetid}/relationships',
+                      headers=auth_headers)
+    assert resp.status_code == 200
+    data = resp.get_json()['data']
+
+    assert data['outgoing'] == [], 'the machine controls nothing'
+    assert len(data['incoming']) == 1, 'the machine must see the PC that controls it'
+
+    row = data['incoming'][0]
+    assert row['relationshiptypename'] == 'controls'
+    # The peer has to arrive with the fields the UI routes on. assetid alone
+    # is not enough: the detail routes key on pluginid, and the type name
+    # chooses the route base.
+    peer = row['sourceasset']
+    assert peer is not None
+    assert peer['assetid'] == pc.assetid
+    assert peer['assettypename'] == 'computer'

tests/test_plugin_migrations.py

@@ -70,8 +70,10 @@ EXPECTED_HEAD_REVISION['printers'] = 'printers0005observedqueues'
 EXPECTED_HEAD_REVISION['machines'] = 'machines0002rename'
 # employees adds the photofilename column on top of its cutover anchor.
 EXPECTED_HEAD_REVISION['employees'] = 'employees0002photo'
-# usb drops the dead usbcheckouts.machineid column on top of its anchor.
-EXPECTED_HEAD_REVISION['usb'] = 'usb0002dropmachineid'
+# usb drops the dead usbcheckouts.machineid column on top of its anchor, then
+# adds scannedviruses so an imported external history keeps the scan result -
+# the field a removable-media audit actually asks about.
+EXPECTED_HEAD_REVISION['usb'] = 'usb0003scanfields'
 # network links a device to a catalog model, which is where its photo comes
 # from - machines, PCs and printers already had that link.
 EXPECTED_HEAD_REVISION['network'] = 'network0003prefix'

tests/test_plugins/test_supplies_card_blindness.py

@@ -0,0 +1,90 @@
+"""The supplies dashboard card must not report "nothing low" when it cannot see.
+
+_get_low_supplies_data returned the same empty result for three different
+situations - Zabbix not configured, Zabbix unreachable, and every cartridge
+full - so the dashboard displayed "Low Supplies 0 / Critical 0" while the
+monitoring was dead. That is not a missing feature; it is a wrong statement, and
+it reads as reassurance.
+
+These pin the three cases apart.
+"""
+from unittest.mock import patch
+
+import pytest
+
+from shopdb.api import db
+from shopdb.core.models import Setting
+
+
+def _call(client, auth_headers):
+    response = client.get('/api/printers/dashboard/supplies', headers=auth_headers)
+    assert response.status_code == 200
+    return response.get_json()['data']
+
+
+class FakeService:
+    def __init__(self, configured, reachable):
+        self.isconfigured = configured
+        self.isreachable = reachable
+
+
+@pytest.fixture(autouse=True)
+def _clear_cache(app):
+    from shopdb.extensions import cache
+    cache.delete('printers_low_supplies')
+    yield
+    cache.delete('printers_low_supplies')
+
+
+def test_unreachable_says_so_rather_than_reporting_zero(client, auth_headers):
+    """Configured and unanswerable: the cartridges are NOT known to be fine."""
+    with patch('plugins.printers.api.asset_routes.ZabbixService',
+               return_value=FakeService(configured=True, reachable=False)):
+        data = _call(client, auth_headers)
+
+    assert data['rows'] == []
+    assert data['unavailable'] == 'Supply monitoring is not reachable'
+
+
+def test_unconfigured_stays_quiet(client, auth_headers):
+    """The site did not ask for supply monitoring.
+
+    A permanent "not configured" line would be a card reporting nothing wrong
+    every day, which is how a board teaches people to stop reading it. It has
+    nothing to say, so it says nothing and the card hides.
+    """
+    with patch('plugins.printers.api.asset_routes.ZabbixService',
+               return_value=FakeService(configured=False, reachable=False)):
+        data = _call(client, auth_headers)
+
+    assert data['rows'] == []
+    assert data.get('unavailable') is None
+
+
+def test_reachable_with_nothing_low_is_not_flagged_as_blind(client, auth_headers):
+    """Zabbix answered and every cartridge is fine. That is a real zero."""
+    with patch('plugins.printers.api.asset_routes._get_low_supplies_data',
+               return_value={'printers': [],
+                             'summary': {'total_checked': 3, 'low': 0, 'critical': 0}}):
+        data = _call(client, auth_headers)
+
+    assert data['rows'] == []
+    assert data.get('unavailable') is None
+
+
+def test_the_envelope_carries_rows_so_the_card_can_still_render(client, auth_headers):
+    """The response became an envelope to carry the reason; rows must survive."""
+    printer = {
+        'printerid': 1, 'assetnumber': 'PRN-1', 'printername': 'Front office',
+        'location': 'Bay 3', 'mapx': None, 'mapy': None, 'levelid': None,
+        'supplies': [{'name': 'Black', 'remaining': 2, 'status': 'critical'}],
+    }
+    with patch('plugins.printers.api.asset_routes._get_low_supplies_data',
+               return_value={'printers': [printer],
+                             'summary': {'total_checked': 1, 'low': 1, 'critical': 1}}):
+        data = _call(client, auth_headers)
+
+    assert data.get('unavailable') is None
+    assert len(data['rows']) == 1
+    assert data['rows'][0]['printername'] == 'Front office'
+    assert data['rows'][0]['iscritical'] is True

tests/test_plugins/test_usb_import_pairing.py

@@ -0,0 +1,99 @@
+"""Folding an external check-in/out LOG into self-hosted checkout PAIRS.
+
+The external cmmc_usb database records one row per action; the self-hosted
+usbcheckouts table records one row per checkout, carrying both times. The
+import has to convert between them, and the interesting cases are all the ways
+a real log is untidy after years of use: a drive still out, a return nobody
+has a matching issue for, two issues in a row.
+
+The rule these tests pin down is that NOTHING is invented and NOTHING vanishes
+silently. An unclosed checkout stays unclosed; anything that cannot be paired
+is reported. A quieter import that guessed the missing halves would produce a
+tidier database and a false audit history, which for removable-media records is
+the worst of the available outcomes.
+"""
+
+from plugins.usb.cli import _pair, _tri
+
+
+def _row(action, when, badge='B1'):
+    return {'action': action, 'timestamp': when, 'badge_number': badge}
+
+
+def test_a_clean_out_and_in_becomes_one_closed_checkout():
+    problems = []
+    pairs = _pair([_row('check-out', 1), _row('check-in', 2)], problems, 'DEV1')
+
+    assert len(pairs) == 1
+    out, back = pairs[0]
+    assert out['timestamp'] == 1
+    assert back['timestamp'] == 2
+    assert problems == []
+
+
+def test_a_drive_still_out_stays_open_and_is_not_reported_as_a_problem():
+    # This is the normal state of any drive currently issued to somebody, not
+    # a data fault, so it must not appear in the problem list.
+    problems = []
+    pairs = _pair([_row('check-out', 1)], problems, 'DEV1')
+
+    assert len(pairs) == 1
+    out, back = pairs[0]
+    assert back is None
+    assert problems == []
+
+
+def test_a_return_with_no_matching_issue_is_skipped_and_reported():
+    problems = []
+    pairs = _pair([_row('check-in', 5)], problems, 'DEV1')
+
+    assert pairs == []
+    assert len(problems) == 1
+    assert 'no matching check-out' in problems[0]
+    assert 'DEV1' in problems[0]
+
+
+def test_two_issues_in_a_row_leave_the_first_open_rather_than_inventing_a_return():
+    problems = []
+    pairs = _pair([_row('check-out', 1, 'ALICE'),
+                   _row('check-out', 2, 'BOB'),
+                   _row('check-in', 3, 'BOB')], problems, 'DEV1')
+
+    assert len(pairs) == 2
+    first, second = pairs
+    assert first[0]['badge_number'] == 'ALICE'
+    assert first[1] is None, 'the first checkout must not be given a return time'
+    assert second[0]['badge_number'] == 'BOB'
+    assert second[1]['timestamp'] == 3
+    assert len(problems) == 1
+    assert 'still out' in problems[0]
+
+
+def test_an_unrecognised_action_is_skipped_and_reported():
+    problems = []
+    pairs = _pair([_row('destroyed', 1)], problems, 'DEV1')
+
+    assert pairs == []
+    assert len(problems) == 1
+    assert 'unrecognised action' in problems[0]
+
+
+def test_action_spelling_variants_are_accepted():
+    # Years of a hand-maintained log produce more than one spelling.
+    for out_word, in_word in (('check-out', 'check-in'),
+                              ('checkout', 'checkin'),
+                              ('out', 'in'),
+                              ('CHECK_OUT', 'RETURN')):
+        problems = []
+        pairs = _pair([_row(out_word, 1), _row(in_word, 2)], problems, 'DEV1')
+        assert len(pairs) == 1, out_word
+        assert pairs[0][1] is not None, out_word
+        assert problems == [], out_word
+
+
+def test_unrecorded_flags_stay_unknown_rather_than_becoming_false():
+    # NULL in the source means nobody wrote it down. False would assert that
+    # the drive was checked and failed, which is a different claim entirely.
+    assert _tri(None) is None
+    assert _tri(0) is False
+    assert _tri(1) is True

tests/test_scan_sbom.py

@@ -0,0 +1,225 @@
+"""The vulnerability scanner's judgement, without touching the network.
+
+scan_sbom.py decides three things that matter: whether a finding can stop a
+release, which version clears it, and whether somebody already accepted it.
+Each is pure given an OSV entry, so each is tested against a hand-written entry
+rather than a live query. The network call itself is not tested - if OSV is
+down the script says so and exits 2, which is the only sensible behaviour and
+not worth a mock.
+"""
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+
+REPO = Path(__file__).resolve().parents[1]
+SCANNER = REPO / 'scripts' / 'scan_sbom.py'
+
+
+def load():
+    """Import the script by path - scripts/ is not a package."""
+    spec = importlib.util.spec_from_file_location('scan_sbom', SCANNER)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.fixture(scope='module')
+def scan():
+    return load()
+
+
+class TestSeverity:
+    def test_word_is_taken_from_the_database(self, scan):
+        entry = {'database_specific': {'severity': 'HIGH'}}
+        assert scan.severity_of(entry) == 'HIGH'
+
+    def test_lowercase_is_accepted(self, scan):
+        assert scan.severity_of(
+            {'database_specific': {'severity': 'critical'}}) == 'CRITICAL'
+
+    def test_a_cvss_vector_alone_is_unknown_not_guessed(self, scan):
+        # Deliberate: parsing a vector into a score we might get wrong is worse
+        # than an honest gap, because somebody would act on the wrong number.
+        entry = {'severity': [{'type': 'CVSS_V3',
+                               'score': 'CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/'
+                                        'C:H/I:H/A:H'}]}
+        assert scan.severity_of(entry) == 'UNKNOWN'
+
+    def test_missing_severity_is_unknown(self, scan):
+        assert scan.severity_of({}) == 'UNKNOWN'
+
+    def test_unknown_never_reaches_the_gate(self, scan):
+        assert not scan.at_or_above('UNKNOWN', scan.GATE_AT)
+
+    def test_gate_floor_and_above(self, scan):
+        assert scan.at_or_above('HIGH', 'HIGH')
+        assert scan.at_or_above('CRITICAL', 'HIGH')
+        assert not scan.at_or_above('MODERATE', 'HIGH')
+        assert not scan.at_or_above('LOW', 'HIGH')
+
+
+class TestPurl:
+    def test_npm(self, scan):
+        assert scan.parse_purl('pkg:npm/axios@1.13.2') == ('npm', 'axios')
+
+    def test_pypi_maps_to_the_osv_name(self, scan):
+        assert scan.parse_purl('pkg:pypi/pyjwt@2.12.1') == ('PyPI', 'pyjwt')
+
+    def test_scoped_npm_name_survives(self, scan):
+        assert scan.parse_purl('pkg:npm/%40vue/compiler-sfc@3.5.0') == (
+            'npm', '@vue/compiler-sfc')
+
+    def test_no_version_is_fine(self, scan):
+        assert scan.parse_purl('pkg:npm/axios') == ('npm', 'axios')
+
+
+class TestFixedVersions:
+    def test_reads_the_fixed_event(self, scan):
+        entry = {'affected': [{
+            'package': {'ecosystem': 'PyPI', 'name': 'pyjwt'},
+            'ranges': [{'events': [{'introduced': '0'}, {'fixed': '2.13.0'}]}],
+        }]}
+        assert scan.fixed_versions(entry, 'pkg:pypi/pyjwt@2.12.1') == ['2.13.0']
+
+    def test_another_packages_fix_is_not_attributed_here(self, scan):
+        # An advisory covering several packages must not lend one package's
+        # fix version to another - that would send somebody to a wrong version.
+        entry = {'affected': [
+            {'package': {'ecosystem': 'npm', 'name': 'other'},
+             'ranges': [{'events': [{'fixed': '9.9.9'}]}]},
+            {'package': {'ecosystem': 'npm', 'name': 'axios'},
+             'ranges': [{'events': [{'fixed': '1.16.0'}]}]},
+        ]}
+        assert scan.fixed_versions(entry, 'pkg:npm/axios@1.13.2') == ['1.16.0']
+
+    def test_same_name_in_another_ecosystem_is_ignored(self, scan):
+        entry = {'affected': [{
+            'package': {'ecosystem': 'npm', 'name': 'pyjwt'},
+            'ranges': [{'events': [{'fixed': '9.9.9'}]}],
+        }]}
+        assert scan.fixed_versions(entry, 'pkg:pypi/pyjwt@2.12.1') == []
+
+    def test_no_fix_published_is_empty(self, scan):
+        entry = {'affected': [{
+            'package': {'ecosystem': 'npm', 'name': 'axios'},
+            'ranges': [{'events': [{'introduced': '0'}]}],
+        }]}
+        assert scan.fixed_versions(entry, 'pkg:npm/axios@1.13.2') == []
+
+
+class TestVersionOrder:
+    def test_ten_is_above_nine(self, scan):
+        # A string compare puts 1.9.0 above 1.10.0, which would recommend
+        # downgrading.
+        assert scan.version_key('1.10.0') > scan.version_key('1.9.0')
+
+    def test_prerelease_does_not_displace_a_release(self, scan):
+        assert scan.version_key('2.0.0') > scan.version_key('2.0.0-rc1')
+
+    def test_versions_with_different_segment_counts_compare(self, scan):
+        # OSV fix versions are not normalised, so a two-segment and a
+        # three-segment version land in the same sort.
+        assert scan.version_key('2.0.1') > scan.version_key('2.0')
+        assert scan.version_key('1.16.0') > scan.version_key('1.9')
+
+
+class TestUpgradePlan:
+    def _finding(self, component, version, fixedin, allowlisted=False):
+        return {'component': component, 'version': version, 'fixedin': fixedin,
+                'allowlisted': allowlisted}
+
+    def test_highest_fix_wins_and_counts_what_it_clears(self, scan):
+        findings = [
+            self._finding('axios 1.13.2', '1.13.2', ['1.13.5']),
+            self._finding('axios 1.13.2', '1.13.2', ['1.16.0']),
+            self._finding('axios 1.13.2', '1.13.2', ['1.15.1']),
+        ]
+        assert scan.upgrade_plan(findings) == [('axios 1.13.2', '1.16.0', 3, False)]
+
+    def test_a_major_jump_is_flagged(self, scan):
+        findings = [self._finding('nanoid 3.3.11', '3.3.11', ['5.1.16'])]
+        assert scan.upgrade_plan(findings)[0][3] is True
+
+    def test_a_branch_patch_beats_a_major_on_the_same_advisory(self, scan):
+        # The real case this was found on: nanoid is fixed in BOTH 3.3.18 and
+        # 5.1.16, and it is transitive under a postcss range of ^3.3.11 - so
+        # 5.x is not merely a bigger job, it is unreachable. Recommending the
+        # highest fix sent somebody to a major upgrade they could not make.
+        findings = [self._finding('nanoid 3.3.11', '3.3.11',
+                                  ['3.3.18', '5.1.16'])]
+        component, version, _count, major = scan.upgrade_plan(findings)[0]
+        assert version == '3.3.18'
+        assert major is False
+
+    def test_component_target_satisfies_every_advisory_against_it(self, scan):
+        # Two advisories, minimal fixes 3.3.16 and 3.3.18: the row must be the
+        # higher, or the upgrade clears one and leaves the other.
+        findings = [
+            self._finding('nanoid 3.3.11', '3.3.11', ['3.3.16', '5.1.16']),
+            self._finding('nanoid 3.3.11', '3.3.11', ['3.3.18', '5.1.6']),
+        ]
+        assert scan.upgrade_plan(findings)[0][1] == '3.3.18'
+
+    def test_a_fix_on_an_older_branch_is_never_a_downgrade(self, scan):
+        # axios advisories list fixes on the 0.30.x line too; on 1.13.2 those
+        # are behind us and must not be recommended.
+        findings = [self._finding('axios 1.13.2', '1.13.2',
+                                  ['0.30.2', '1.15.2'])]
+        assert scan.upgrade_plan(findings)[0][1] == '1.15.2'
+
+    def test_only_older_branch_fixes_means_no_row(self, scan):
+        findings = [self._finding('axios 1.13.2', '1.13.2', ['0.30.2'])]
+        assert scan.upgrade_plan(findings) == []
+
+    def test_a_patch_bump_is_not_flagged(self, scan):
+        findings = [self._finding('pyjwt 2.12.1', '2.12.1', ['2.13.0'])]
+        assert scan.upgrade_plan(findings)[0][3] is False
+
+    def test_allowlisted_findings_do_not_ask_for_an_upgrade(self, scan):
+        findings = [self._finding('pyjwt 2.12.1', '2.12.1', ['2.13.0'],
+                                  allowlisted=True)]
+        assert scan.upgrade_plan(findings) == []
+
+    def test_a_finding_with_no_fix_is_omitted(self, scan):
+        findings = [self._finding('thing 1.0', '1.0', [])]
+        assert scan.upgrade_plan(findings) == []
+
+
+class TestAllowlist:
+    def test_the_shipped_allowlist_parses_and_is_complete(self, scan):
+        """Every entry must carry the reasoning, or it is just a mute."""
+        data = json.loads(scan.ALLOWLIST.read_text(encoding='utf8'))
+        for entry in data.get('accepted', []):
+            for field in ('id', 'reason', 'acceptedby', 'accepted'):
+                assert entry.get(field), '%s missing %s' % (entry, field)
+            assert len(entry['reason']) > 40, (
+                'reason for %s is too short to be a reason' % entry['id'])
+
+    def test_allowlist_is_keyed_by_advisory_id(self, scan):
+        allowed = scan.load_allowlist()
+        for key, entry in allowed.items():
+            assert key == entry['id']
+
+
+class TestScopeSeparation:
+    def test_build_only_components_are_read_as_not_shipped(self, scan, tmp_path):
+        sbom = tmp_path / 'sbom.json'
+        sbom.write_text(json.dumps({'components': [
+            {'name': 'axios', 'version': '1.0', 'purl': 'pkg:npm/axios@1.0'},
+            {'name': 'vite', 'version': '5.0', 'purl': 'pkg:npm/vite@5.0',
+             'scope': 'excluded'},
+        ]}), encoding='utf8')
+        components = scan.load_components(sbom)
+        shipped = {name: ship for _p, name, _v, ship in components}
+        assert shipped == {'axios': True, 'vite': False}
+
+    def test_a_component_without_a_purl_is_skipped(self, scan, tmp_path):
+        # OSV is queried by purl; a component without one cannot be looked up,
+        # and inventing an identifier would produce confident wrong answers.
+        sbom = tmp_path / 'sbom.json'
+        sbom.write_text(json.dumps({'components': [
+            {'name': 'local-thing', 'version': '1.0'},
+        ]}), encoding='utf8')
+        assert scan.load_components(sbom) == []

tests/test_site_import_loader.py

@@ -0,0 +1,164 @@
+"""The WJ reference loader's resolve, crosswalk binding and source aliasing.
+
+The loader had no tests at all, and three of its defects were the silent kind:
+they did not raise, they imported fewer rows and finished cheerfully. These
+cover the logic that can be exercised without a MySQL source, which is exactly
+the part that was wrong.
+
+Not covered here: the stage functions themselves. They need the classic source
+schema, so they belong in a fixture-backed integration test against a scratch
+database, not in the unit suite.
+"""
+import json
+import os
+
+import pytest
+
+from scripts.site_imports.wjf import run as loader
+from scripts.site_imports.wjf.harness import IdMap, Source, StaleIdMapError
+
+
+class FakeHarness:
+    """Records the paths requested and replays canned list responses."""
+
+    def __init__(self, pages):
+        self.pages = pages
+        self.requested = []
+
+    def get(self, path):
+        self.requested.append(path)
+        return 200, self.pages.get(path, {'items': []})
+
+
+class TestResolveExisting:
+    def test_finds_a_row_beyond_the_first_page(self):
+        """The bug: per_page is clamped to 100, so row 101 was invisible.
+
+        The old code asked for per_page=10000, read one response, and returned
+        None for anything the server did not fit in its first 100 rows. The
+        target already holds more than 100 models and more than 100
+        applications, so this fired on every resume.
+        """
+        page1 = {'items': [{'modelnumber': 'M%03d' % i, 'modelnumberid': i}
+                           for i in range(100)]}
+        page2 = {'items': [{'modelnumber': 'LATE', 'modelnumberid': 999}]}
+        harness = FakeHarness({
+            '/api/models?modelnumber=LATE&per_page=100': {'items': []},
+            '/api/models?per_page=100&page=1': page1,
+            '/api/models?per_page=100&page=2': page2,
+        })
+        found = loader._resolve_existing(
+            harness, '/api/models', 'modelnumber', 'LATE', 'modelnumberid')
+        assert found == 999
+
+    def test_tries_the_exact_filter_first(self):
+        harness = FakeHarness({
+            '/api/models?modelnumber=WIDGET&per_page=100':
+                {'items': [{'modelnumber': 'WIDGET', 'modelnumberid': 7}]},
+        })
+        assert loader._resolve_existing(
+            harness, '/api/models', 'modelnumber', 'WIDGET', 'modelnumberid') == 7
+        # One request: it must not walk every page when the filter answered.
+        assert len(harness.requested) == 1
+
+    def test_values_needing_encoding_do_not_break_the_filter(self):
+        harness = FakeHarness({})
+        loader._resolve_existing(harness, '/api/vendors', 'vendor',
+                                 'Acme & Co/Ltd', 'vendorid')
+        assert 'Acme%20%26%20Co/Ltd' in harness.requested[0] or \
+               'Acme%20%26%20Co%2FLtd' in harness.requested[0]
+
+    def test_matching_ignores_case_and_padding(self):
+        harness = FakeHarness({
+            '/api/vendors?vendor=acme&per_page=100':
+                {'items': [{'vendor': '  ACME ', 'vendorid': 3}]},
+        })
+        assert loader._resolve_existing(
+            harness, '/api/vendors', 'vendor', 'acme', 'vendorid') == 3
+
+    def test_missing_row_returns_none_without_spinning(self):
+        harness = FakeHarness({})
+        assert loader._resolve_existing(
+            harness, '/api/models', 'modelnumber', 'NOPE', 'modelnumberid') is None
+        # Empty first page ends it: the filter attempt plus one page.
+        assert len(harness.requested) == 2
+
+    def test_a_short_page_ends_the_walk(self):
+        harness = FakeHarness({
+            '/api/models?per_page=100&page=1': {'items': [{'modelnumber': 'A', 'modelnumberid': 1}]},
+        })
+        assert loader._resolve_existing(
+            harness, '/api/models', 'modelnumber', 'ZZZ', 'modelnumberid') is None
+        assert len(harness.requested) == 2
+
+    def test_a_bare_list_response_is_accepted(self):
+        harness = FakeHarness({
+            '/api/x?name=a&per_page=100': [{'name': 'a', 'id': 5}],
+        })
+        assert loader._resolve_existing(harness, '/api/x', 'name', 'a', 'id') == 5
+
+    def test_an_existing_query_string_keeps_its_separator(self):
+        harness = FakeHarness({})
+        loader._resolve_existing(harness, '/api/models?category=pc',
+                                 'modelnumber', 'M1', 'modelnumberid')
+        assert harness.requested[0].startswith('/api/models?category=pc&modelnumber=')
+
+
+class TestIdMapBinding:
+    def test_a_crosswalk_from_another_database_is_refused(self, tmp_path):
+        """The silent corruption this guard exists for.
+
+        Keeping idmap.json across a drop-and-rebuild leaves legacy ids pointing
+        at assetids from the OLD database, so warranties and printer
+        assignments attach to whatever asset holds that id now. No error, wrong
+        data.
+        """
+        path = tmp_path / 'idmap.json'
+        path.write_text(json.dumps({'_target': 'aaaa', 'asset': {'1': 10}}))
+        with pytest.raises(StaleIdMapError) as caught:
+            IdMap(str(path), target='bbbb')
+        assert 'different database' in str(caught.value)
+
+    def test_the_matching_target_loads(self, tmp_path):
+        path = tmp_path / 'idmap.json'
+        path.write_text(json.dumps({'_target': 'aaaa', 'asset': {'1': 10}}))
+        ids = IdMap(str(path), target='aaaa')
+        assert ids.get('asset', 1) == 10
+
+    def test_an_unstamped_crosswalk_is_adopted_not_refused(self, tmp_path):
+        # Written before the guard existed. Refusing would strand a crosswalk
+        # that is probably fine; adopting binds it for every run after.
+        path = tmp_path / 'idmap.json'
+        path.write_text(json.dumps({'asset': {'1': 10}}))
+        ids = IdMap(str(path), target='ccc')
+        assert ids.get('asset', 1) == 10
+        assert json.loads(path.read_text())['_target'] == 'ccc'
+
+    def test_the_bookkeeping_key_is_not_counted_as_a_namespace(self, tmp_path):
+        path = tmp_path / 'idmap.json'
+        ids = IdMap(str(path), target='ccc')
+        ids.put('asset', 1, 10)
+        assert ids.count('asset') == 1
+        assert ids.count('_target') == 0
+
+    def test_a_fresh_crosswalk_needs_no_stamp_to_start(self, tmp_path):
+        ids = IdMap(str(tmp_path / 'new.json'), target='ddd')
+        ids.put('asset', 5, 50)
+        assert ids.get('asset', 5) == 50
+
+
+class TestSourceAliasing:
+    def test_logical_names_default_to_the_scratch_databases(self, monkeypatch):
+        for var in ('SOURCE_DB_MAIN', 'SOURCE_DB_USB', 'SOURCE_DB_EMPLOYEES'):
+            monkeypatch.delenv(var, raising=False)
+        assert Source.resolve('shopdb_src') == 'shopdb_src'
+        assert Source.resolve('cmmc_usb_src') == 'cmmc_usb_src'
+        assert Source.resolve('wjf_employees_src') == 'wjf_employees_src'
+
+    def test_the_environment_repoints_the_loader(self, monkeypatch):
+        """What unblocks the prodscratch run without editing run.py."""
+        monkeypatch.setenv('SOURCE_DB_MAIN', 'prodscratch')
+        assert Source.resolve('shopdb_src') == 'prodscratch'
+
+    def test_an_unknown_name_passes_through(self, monkeypatch):
+        assert Source.resolve('something_else') == 'something_else'