Merge origin/main (West Jeff) into wilm

Bring in the natural-key import fix, Metrics->Monitors rename, PC detail tabbing and panel groups, guest search grants, and the 2026-08-30 security fixes. Reconcile endpoint packaging/installer, migrations, contract version, and tests; keep the delivery uninstaller and 6-step installer alongside main's permission-based RBAC and pin-drift protection.

tests/test_plugins/test_endpoint_applications_model.py

@@ -0,0 +1,87 @@
+"""The installed-software snapshot table: it must actually accept inserts.
+
+This table is rewritten wholesale on every endpoint check-in - delete the
+asset's rows, insert the current inventory - so its surrogate key climbs at
+(endpoints x installed apps) per report and never reuses a value. On MySQL a
+32-bit key exhausts and inserts stop working permanently.
+
+Widening it to BigInteger broke every insert on SQLite, because SQLite only
+autoincrements a column typed exactly INTEGER PRIMARY KEY. Neither fault was
+caught by the suite, for the same reason: nothing inserted into this table at
+all. This does.
+"""
+from shopdb.api import db
+from plugins.endpoints.models.endpoint import EndpointApplication
+
+
+def test_rows_insert_and_the_key_autoincrements(app):
+    with app.app_context():
+        db.create_all()
+        first = EndpointApplication(assetid=1, displayname='Probe', displayversion='1.0')
+        second = EndpointApplication(assetid=1, displayname='Probe 2', displayversion='2.0')
+        db.session.add_all([first, second])
+        db.session.commit()
+
+        assert first.endpointapplicationid is not None
+        assert second.endpointapplicationid is not None
+        assert second.endpointapplicationid != first.endpointapplicationid
+
+
+def test_the_key_is_wide_enough_for_a_table_that_is_rewritten_every_cycle(app):
+    """BIGINT on MySQL, INTEGER on SQLite - both halves matter.
+
+    Dropping the variant breaks SQLite inserts; dropping the BigInteger breaks
+    MySQL years later, which is worse because it is silent until it is not.
+    """
+    from sqlalchemy.dialects import mysql, sqlite
+
+    with app.app_context():
+        column = EndpointApplication.__table__.c.endpointapplicationid
+        assert 'BIGINT' in str(column.type.compile(dialect=mysql.dialect()))
+        assert 'INTEGER' in str(column.type.compile(dialect=sqlite.dialect()))
+
+
+def test_over_long_registry_values_are_clamped_not_fatal(app):
+    """One malformed Uninstall entry must not cost the whole check-in.
+
+    On MySQL in strict mode an over-long value raises Data too long and fails the
+    transaction, taking telemetry and events down with it - so that endpoint goes
+    silent until somebody uninstalls the offending program. SQLite does not
+    enforce declared lengths, so the suite would never have shown it.
+    """
+    from plugins.endpoints.plugin import EndpointsPlugin
+
+    with app.app_context():
+        db.create_all()
+        warnings = []
+        stored = EndpointsPlugin()._replace_applications(
+            assetid=4242,
+            applications=[{
+                'displayname': 'A' * 400,
+                'displayversion': 'V' * 400,
+                'publisher': 'P' * 400,
+            }],
+            warnings=warnings,
+        )
+        db.session.commit()
+
+        assert stored == 1
+        row = EndpointApplication.query.filter_by(assetid=4242).one()
+        assert len(row.displayname) == 255
+        assert len(row.displayversion) == 128
+        assert len(row.publisher) == 255
+
+
+def test_a_blank_value_becomes_null_rather_than_an_empty_string(app):
+    from plugins.endpoints.plugin import EndpointsPlugin
+
+    with app.app_context():
+        db.create_all()
+        EndpointsPlugin()._replace_applications(
+            assetid=4243,
+            applications=[{'displayname': 'Thing', 'publisher': '   '}],
+            warnings=[],
+        )
+        db.session.commit()
+        row = EndpointApplication.query.filter_by(assetid=4243).one()
+        assert row.publisher is None

tests/test_plugins/test_endpoint_delivery_descriptor.py

@@ -0,0 +1,151 @@
+"""An update_app delivery must pin the bytes it actually delivers.
+
+The agent verifies the downloaded file against the descriptor's sha256 and fails
+closed on a mismatch - which is worthless if the descriptor carries no digest,
+and misleading if it carries a digest for a different file.
+
+Both were true of the version: branch. It pinned AppVersion.sha256, but
+/api/endpoints/apppackage/<appversionid> resolves the version to its application
+and serves application.installpath, so the digest described an artifact that
+route never sends; where the version had no sha256 the descriptor carried none
+at all and the agent had nothing to check.
+
+It also ended the URL at the id, so an MSI arrived with no extension and the
+agent's Path.GetExtension could not route it through msiexec.
+"""
+import pytest
+
+from shopdb.api import db
+from shopdb.core.models import Application, AppVersion
+
+
+@pytest.fixture
+def hosted_app(app, request):
+    # Unique per test: appname is unique-constrained and the database outlives
+    # a single test here, so a fixed name collides on the second use.
+    with app.app_context():
+        db.create_all()
+        application = Application(
+            appname='Widget-%s' % request.node.name[-24:],
+            installpath='/api/applications/package/widget-setup.msi',
+            installersha256='a' * 64,
+        )
+        db.session.add(application)
+        db.session.commit()
+        version = AppVersion(appid=application.appid, version='2.1', sha256=None)
+        db.session.add(version)
+        db.session.commit()
+        yield application.appid, version.appversionid
+
+
+def test_a_version_delivery_carries_a_digest_even_when_the_version_has_none(app, hosted_app):
+    from plugins.endpoints.api.routes import _resolve_update_app
+
+    appid, appversionid = hosted_app
+    with app.app_context():
+        descriptor, error = _resolve_update_app(f'version:{appversionid}')
+
+    assert error is None, error
+    # Not None: this is the whole point. A blank pin makes the agent fail open.
+    assert descriptor['sha256'] == 'a' * 64
+
+
+def test_a_version_delivery_keeps_the_installer_extension(app, hosted_app):
+    """Without the filename an MSI is not routed through msiexec."""
+    from plugins.endpoints.api.routes import _resolve_update_app
+
+    appid, appversionid = hosted_app
+    with app.app_context():
+        descriptor, _ = _resolve_update_app(f'version:{appversionid}')
+
+    assert descriptor['installpath'].endswith('.msi')
+
+
+def test_a_package_with_no_recorded_digest_is_refused_not_delivered_unverified(app):
+    """Refusing to enqueue beats delivering something nothing can verify."""
+    from plugins.endpoints.api.routes import _resolve_update_app
+
+    with app.app_context():
+        db.create_all()
+        application = Application(appname='Unhashed-%s' % id(app),
+                                  installpath='/api/applications/package/x.exe')
+        db.session.add(application)
+        db.session.commit()
+        version = AppVersion(appid=application.appid, version='1.0', sha256=None)
+        db.session.add(version)
+        db.session.commit()
+
+        descriptor, error = _resolve_update_app(f'version:{version.appversionid}')
+
+    assert descriptor is None
+    assert 'SHA-256' in error
+
+
+# ---------------------------------------------------------------------------
+# The digest is pinned when an operator approves the push, not when the agent
+# claims it. Between those two moments the bytes behind a descriptor can be
+# replaced by anyone who can edit the catalog, and the re-resolved descriptor
+# would faithfully describe the NEW bytes.
+# ---------------------------------------------------------------------------
+
+def test_an_unpinned_command_still_claims(app):
+    """Actions that deliver no content, and rows queued before the pin existed."""
+    from plugins.endpoints.api.routes import _pin_holds
+    from plugins.endpoints.models import EndpointCommand
+
+    with app.app_context():
+        command = EndpointCommand(assetid=1, action='mtconnect_start',
+                                  pinnedsha256=None)
+        assert _pin_holds(command, None) is True
+        assert _pin_holds(command, {'sha256': 'b' * 64}) is True
+
+
+def test_a_matching_digest_claims(app):
+    from plugins.endpoints.api.routes import _pin_holds
+    from plugins.endpoints.models import EndpointCommand
+
+    with app.app_context():
+        command = EndpointCommand(assetid=1, action='run_script',
+                                  pinnedsha256='c' * 64)
+        assert _pin_holds(command, {'sha256': 'c' * 64}) is True
+
+
+def test_a_swapped_digest_does_not_claim(app):
+    """The finding this exists for: approve one script, ship another.
+
+    An operator queues run_script against a reviewed body. Before the agent's
+    long poll returns, somebody with applications.edit replaces the body and
+    the digest is restamped. Without the pin the claim re-resolves, hands the
+    agent the attacker's digest, and the agent verifies it correctly.
+    """
+    from plugins.endpoints.api.routes import _pin_holds
+    from plugins.endpoints.models import EndpointCommand
+
+    with app.app_context():
+        command = EndpointCommand(assetid=1, action='run_script',
+                                  pinnedsha256='c' * 64)
+        assert _pin_holds(command, {'sha256': 'd' * 64}) is False
+
+
+def test_a_vanished_descriptor_is_not_treated_as_a_swap(app):
+    """A deleted script already sends the command through with no descriptor so
+    the agent fails closed; that path must not be reported as tampering."""
+    from plugins.endpoints.api.routes import _pin_holds
+    from plugins.endpoints.models import EndpointCommand
+
+    with app.app_context():
+        command = EndpointCommand(assetid=1, action='run_script',
+                                  pinnedsha256='c' * 64)
+        assert _pin_holds(command, None) is True
+
+
+def test_enqueue_records_the_digest_it_approved(app, hosted_app):
+    """The resolved descriptor used to be discarded at enqueue (`_descriptor`)."""
+    from plugins.endpoints.api.routes import _resolve_update_app
+
+    appid, appversionid = hosted_app
+    with app.app_context():
+        descriptor, error = _resolve_update_app(str(appversionid))
+        assert error is None
+        # This is the value the enqueue path now writes to pinnedsha256.
+        assert descriptor['sha256'] == 'a' * 64

tests/test_plugins/test_endpoints_plugin.py

@@ -0,0 +1,273 @@
+"""Endpoints plugin smoke + collector-contract tests.
+
+Asserts the plugin loads cleanly, satisfies the framework contract, and that
+its ADR-006 collector schema is well formed. Extend with DB-backed ingest tests
+as the plugin grows.
+"""
+
+from flask import Blueprint
+
+from plugins.endpoints.plugin import EndpointsPlugin, CURATED_EVENT_IDS
+from plugins.endpoints.eventcatalog import EVENT_CATALOG, CRITICAL_EVENT_IDS
+from plugins.endpoints.commands import (
+    COMMAND_ALLOWLIST, is_allowed_action, elevated_permission_for,
+    requires_target,
+)
+
+
+def test_endpoints_plugin_meta_is_valid():
+    """meta returns a PluginMeta with the expected name and prefix."""
+    plugin = EndpointsPlugin()
+    assert plugin.meta.name == 'endpoints'
+    assert plugin.meta.api_prefix == '/api/endpoints'
+    assert 'computers' in plugin.meta.dependencies
+
+
+def test_endpoints_plugin_get_blueprint_returns_blueprint():
+    """get_blueprint returns a Flask Blueprint, not None."""
+    plugin = EndpointsPlugin()
+    assert isinstance(plugin.get_blueprint(), Blueprint)
+
+
+def test_endpoints_plugin_get_models_returns_models():
+    """get_models returns the two telemetry tables."""
+    plugin = EndpointsPlugin()
+    models = plugin.get_models()
+    assert len(models) >= 2
+    for model in models:
+        assert hasattr(model, '__tablename__')
+
+
+def test_endpoints_collector_schema_identity_is_hostname():
+    """ADR-006 schema declares hostname as the identity field."""
+    plugin = EndpointsPlugin()
+    schema = plugin.get_collector_schema()
+    assert schema is not None
+    assert schema['identityfield'] == 'hostname'
+    assert 'hostname' in schema['fields']['required']
+
+
+def test_endpoints_curated_event_ids_include_core_security_ids():
+    """The curated catalog covers the agreed logon/stability IDs."""
+    for eventid in (4624, 4625, 4634, 4672, 4697, 4698, 41, 4618, 55, 7011,
+                    10016, 1001):
+        assert eventid in CURATED_EVENT_IDS
+
+
+def test_curated_ids_match_the_event_catalog():
+    """Ingest filter and display catalog cannot drift apart."""
+    assert CURATED_EVENT_IDS == frozenset(EVENT_CATALOG)
+
+
+def test_critical_event_ids_are_a_subset_of_the_catalog():
+    """Dashboard stability card only queries known critical IDs."""
+    assert set(CRITICAL_EVENT_IDS)
+    assert set(CRITICAL_EVENT_IDS).issubset(set(EVENT_CATALOG))
+
+
+def test_endpoints_asset_panels_are_computer_scoped():
+    """Every asset panel targets computer assets only, with an endpoint."""
+    plugin = EndpointsPlugin()
+    panels = plugin.get_asset_panels()
+    assert len(panels) == 6
+    for panel in panels:
+        assert panel['assettypes'] == ['computer']
+        assert panel['endpoint'].startswith('/api/endpoints/asset/{assetid}')
+        assert panel['render'] in ('keyvalue', 'list', 'table')
+
+
+def test_services_and_processes_are_split_into_two_filtered_cards():
+    """The single services card is now two cards filtered by row kind, both
+    reading the same services snapshot so a future top-20 collector is just
+    another filtered panel over the same feed."""
+    plugin = EndpointsPlugin()
+    panels = {p['id']: p for p in plugin.get_asset_panels()}
+    services = panels['endpoints-tracked-services']
+    processes = panels['endpoints-tracked-processes']
+    assert services['title'] == 'Tracked Services'
+    assert processes['title'] == 'Tracked Processes'
+    assert services['filter'] == {'key': 'kind', 'equals': 'service'}
+    assert processes['filter'] == {'key': 'kind', 'equals': 'process'}
+    # Both slice one endpoint feed.
+    assert services['endpoint'] == processes['endpoint']
+    assert services['endpoint'].endswith('/services')
+    # Command wiring carries over to both cards.
+    assert services['permission'] == 'endpoints.command'
+    assert processes['permission'] == 'endpoints.command'
+
+
+def test_activity_card_starts_collapsed_and_sits_below_live_state():
+    """The activity feed is history, so it must not push live state down.
+
+    Renamed from the events card when events and commands merged into one feed.
+    It is no longer the LAST panel - installed software sits below it - so the
+    assertion is that it ranks below every live-state panel rather than being
+    the maximum, which would break again the next time a reference panel is
+    added underneath.
+    """
+    plugin = EndpointsPlugin()
+    panels = plugin.get_asset_panels()
+    activity = next(p for p in panels if p['id'] == 'endpoints-activity')
+    # NOT collapsed by default, unlike the events card it replaced. The feed
+    # surfaces critical events, and a card that hides the one thing worth seeing
+    # is the reason the health dot and the visible rows disagreed. Collapsible,
+    # so a reader can put it away.
+    assert activity.get('collapsible') is True
+    assert activity.get('collapsedDefault') is not True
+    live = [p['position'] for p in panels
+            if p['id'] in ('endpoints-status', 'endpoints-tracked-services',
+                           'endpoints-tracked-processes', 'endpoints-interfaces')]
+    assert activity['position'] > max(live)
+
+
+def test_endpoints_dashboard_widgets_declared():
+    """Quiet + stability exception cards are declared and hide when empty."""
+    plugin = EndpointsPlugin()
+    widgets = plugin.get_dashboard_widgets()
+    ids = {w['id'] for w in widgets}
+    assert {'endpoints-quiet', 'endpoints-stability'} <= ids
+    for widget in widgets:
+        assert widget['render'] == 'exceptions'
+        assert widget['empty'] == 'hide'
+        # The fleet-wide cards read the whole roster, so they are gated on
+        # endpoints.fleet, not the per-endpoint endpoints.view a guest may hold.
+        assert widget['permission'] == 'endpoints.fleet'
+
+
+def test_endpoints_navigation_item_points_at_fleet_view():
+    """The fleet list view is exposed as a single /endpoints nav item."""
+    plugin = EndpointsPlugin()
+    items = plugin.get_navigation_items()
+    routes = {item['route'] for item in items}
+    assert '/endpoints' in routes
+
+
+def test_system_restart_is_allowlisted_and_policy_selectable():
+    """Restart is a real allowlisted action, so the policy builder (which
+    reads COMMAND_ALLOWLIST) surfaces it as a selectable action."""
+    assert is_allowed_action('system_restart')
+    assert 'system_restart' in COMMAND_ALLOWLIST
+    assert COMMAND_ALLOWLIST['system_restart']['label'] == 'Restart computer'
+
+
+def test_system_restart_requires_the_elevated_restart_permission():
+    """Restart maps to endpoints.restart so the enqueue route can demand a
+    heavier grant than the routine endpoints.command toggle. Routine actions
+    carry no elevated requirement."""
+    assert elevated_permission_for('system_restart') == 'endpoints.restart'
+    assert elevated_permission_for('mtconnect_start') is None
+
+
+def test_endpoints_declares_the_restart_permission():
+    """The dedicated restart permission is declared for role-grid seeding."""
+    plugin = EndpointsPlugin()
+    permissions = {name for name, *_ in plugin.get_permissions()}
+    assert 'endpoints.restart' in permissions
+    assert 'endpoints.command' in permissions
+
+
+def test_collect_now_is_allowlisted_targetless_and_routine():
+    """collect_now forces an off-cadence telemetry report. It is allowlisted so
+    the enqueue route accepts it, carries no wire target (fixed action), and
+    needs no elevated grant beyond endpoints.command."""
+    assert is_allowed_action('collect_now')
+    assert 'collect_now' in COMMAND_ALLOWLIST
+    assert not requires_target('collect_now')
+    assert elevated_permission_for('collect_now') is None
+
+
+def test_run_script_and_update_app_are_allowlisted_delivery_actions():
+    """The two software-delivery actions are allowlisted with a targetkind so
+    the enqueue route accepts them and carries their subject on the wire:
+    run_script carries a script name, update_app carries an appversionid."""
+    for action in ('run_script', 'update_app'):
+        assert is_allowed_action(action)
+        assert action in COMMAND_ALLOWLIST
+        assert requires_target(action)
+    assert COMMAND_ALLOWLIST['run_script']['targetkind'] == 'script'
+    assert COMMAND_ALLOWLIST['update_app']['targetkind'] == 'app'
+
+
+def test_delivery_actions_require_the_elevated_deploy_permission():
+    """run_script and update_app both demand endpoints.deploy ON TOP OF
+    endpoints.command, so pushing software is a strictly heavier grant than a
+    routine service/process toggle."""
+    assert elevated_permission_for('run_script') == 'endpoints.deploy'
+    assert elevated_permission_for('update_app') == 'endpoints.deploy'
+
+
+def test_endpoints_declares_the_deploy_permission():
+    """The dedicated software-delivery permission is declared for role seeding."""
+    plugin = EndpointsPlugin()
+    permissions = {name for name, *_ in plugin.get_permissions()}
+    assert 'endpoints.deploy' in permissions
+
+
+def test_known_payload_keys_cover_persisted_and_relayed_fields():
+    """The ingest allow-set must list every key the handler consumes today, so
+    those keys are never mislogged as 'unpersisted'. Anything the agent sends
+    outside this set is logged for observability before columns exist."""
+    from plugins.endpoints.plugin import KNOWN_PAYLOAD_KEYS
+
+    persisted_or_relayed = {
+        'hostname', 'adopt', 'loggedinuser', 'serialnumber', 'pctype',
+        'machinenumber', 'vendorname', 'modelnumber', 'osname', 'ipaddress',
+        'agentversion', 'lastboottime', 'uptimeseconds', 'idleseconds',
+        'cpupercent', 'memorypercent', 'services', 'ports', 'interfaces',
+        'scheduledtasks', 'events',
+    }
+    assert persisted_or_relayed <= KNOWN_PAYLOAD_KEYS
+
+
+def test_new_hardware_telemetry_keys_are_logged_not_silently_known():
+    """Disk/bios/memory inventory is collected at the agent before the server
+    persists it. Those keys must be OUTSIDE the known set so ingest logs them
+    rather than dropping them silently."""
+    from plugins.endpoints.plugin import KNOWN_PAYLOAD_KEYS
+
+    for key in ('disks', 'bios', 'memory', 'processor', 'osinfo', 'domain'):
+        assert key not in KNOWN_PAYLOAD_KEYS
+
+    # 'applications' USED to be in this list. It is persisted now - that is what
+    # the installed-software work added - so it belongs in the known set, and
+    # asserting its absence would fail the moment the feature it tests works.
+    assert 'applications' in KNOWN_PAYLOAD_KEYS
+
+
+def test_telemetrymeta_route_is_registered_for_freshness_gating():
+    """The WebUI gates panel refreshes on lastreportedat advancing, so the
+    lightweight freshness probe must be exposed on the plugin blueprint."""
+    from flask import Flask
+
+    application = Flask(__name__)
+    # A url_prefix is REQUIRED here: the fleet list registers as route(''), which
+    # Flask rejects as a bare rule and accepts under a prefix. That is deliberate
+    # - CONTRIBUTING says API routes carry no trailing slash - so the test has to
+    # mount the blueprint the way the app does.
+    application.register_blueprint(EndpointsPlugin().get_blueprint(),
+                                   url_prefix='/api/endpoints')
+    rules = {rule.rule for rule in application.url_map.iter_rules()}
+    assert any(rule.endswith('/asset/<int:asset_id>/telemetrymeta') for rule in rules)
+
+
+def test_fleet_list_route_resolves_without_trailing_slash():
+    """The Fleet list route must match the slash-less URL the SPA calls.
+
+    The list is the only route on the bare blueprint prefix. If it demands a
+    trailing slash, Flask 308-redirects /api/endpoints to /api/endpoints/, and
+    behind the /shopdb reverse proxy that redirect drops the path prefix and
+    the list call fails while the detail routes (which carry a path suffix)
+    keep working. strict_slashes=False makes both forms resolve directly.
+    """
+    from flask import Flask
+
+    application = Flask(__name__)
+    application.register_blueprint(
+        EndpointsPlugin().get_blueprint(), url_prefix='/api/endpoints')
+    adapter = application.url_map.bind('localhost')
+
+    endpoint, _ = adapter.match('/api/endpoints', method='GET')
+    assert endpoint.endswith('list_endpoints')
+    endpoint_slash, _ = adapter.match('/api/endpoints/', method='GET')
+    assert endpoint_slash.endswith('list_endpoints')
+

tests/test_plugins/test_geenforce_payload_hardening.py

@@ -107,21 +107,47 @@ def test_payload_download_is_rate_limited(client, app, auth_headers):
 
 
 def test_rate_limit_is_per_ip(client, app, auth_headers):
-    """A second caller (different X-Forwarded-For) has its own budget: one IP
-    being throttled must not throttle everyone."""
+    """A second caller has its own budget: one IP being throttled must not
+    throttle everyone.
+
+    Callers are distinguished by REMOTE_ADDR, the peer that actually connected,
+    which is what ProxyFix fills from X-Forwarded-For when a proxy is really in
+    front of this instance.
+    """
     sha = _store(app, b'per-ip blob')
     key = _fetch_key(client, auth_headers)
-    hot = {**key, 'X-Forwarded-For': '203.0.113.202'}
-    cool = {**key, 'X-Forwarded-For': '203.0.113.203'}
+    hot = {'REMOTE_ADDR': '203.0.113.202'}
+    cool = {'REMOTE_ADDR': '203.0.113.203'}
     with _config(app, GEENFORCE_PAYLOAD_RATELIMIT_MAX=1,
                  GEENFORCE_PAYLOAD_RATELIMIT_WINDOW_SECONDS=300):
         assert client.get(f'/api/geenforce/payload/{sha}',
-                          headers=hot).status_code == 200
+                          headers=key, environ_base=hot).status_code == 200
         assert client.get(f'/api/geenforce/payload/{sha}',
-                          headers=hot).status_code == 429
+                          headers=key, environ_base=hot).status_code == 429
         # Fresh IP still gets its first request through.
         assert client.get(f'/api/geenforce/payload/{sha}',
-                          headers=cool).status_code == 200
+                          headers=key, environ_base=cool).status_code == 200
+
+
+def test_rate_limit_cannot_be_reset_by_forging_a_forwarded_header(
+        client, app, auth_headers):
+    """The budget follows the connection, not a header the caller writes.
+
+    The limiter used to key on the first X-Forwarded-For hop, so rotating that
+    header per request bought an unbounded budget and the limit stopped being
+    a limit.
+    """
+    sha = _store(app, b'forged header blob')
+    key = _fetch_key(client, auth_headers)
+    caller = {'REMOTE_ADDR': '203.0.113.210'}
+    with _config(app, GEENFORCE_PAYLOAD_RATELIMIT_MAX=1,
+                 GEENFORCE_PAYLOAD_RATELIMIT_WINDOW_SECONDS=300):
+        assert client.get(f'/api/geenforce/payload/{sha}', headers=key,
+                          environ_base=caller).status_code == 200
+        for hop in ('198.51.100.1', '198.51.100.2', '198.51.100.3'):
+            spoofed = {**key, 'X-Forwarded-For': hop}
+            assert client.get(f'/api/geenforce/payload/{sha}', headers=spoofed,
+                              environ_base=caller).status_code == 429
 
 
 def test_rate_limit_can_be_disabled(client, app, auth_headers):

tests/test_plugins/test_printedparts_plugin.py

@@ -0,0 +1,30 @@
+"""Printedparts plugin smoke tests.
+
+Asserts the plugin loads cleanly and satisfies the framework contract.
+Replace and extend with domain tests as you build the plugin out.
+"""
+
+from plugins.printedparts.plugin import PrintedpartsPlugin
+
+
+def test_printedparts_plugin_meta_is_valid():
+    """PrintedpartsPlugin.meta returns a PluginMeta with the expected name."""
+    plugin = PrintedpartsPlugin()
+    assert plugin.meta.name == 'printedparts'
+    assert plugin.meta.api_prefix == '/api/printedparts'
+
+
+def test_printedparts_plugin_get_blueprint_returns_blueprint():
+    """get_blueprint returns a Flask Blueprint, not None."""
+    from flask import Blueprint
+    plugin = PrintedpartsPlugin()
+    assert isinstance(plugin.get_blueprint(), Blueprint)
+
+
+def test_printedparts_plugin_get_models_returns_a_model():
+    """get_models returns a list with at least one SQLAlchemy model."""
+    plugin = PrintedpartsPlugin()
+    models = plugin.get_models()
+    assert len(models) >= 1
+    for model in models:
+        assert hasattr(model, '__tablename__')

tests/test_plugins/test_supplies_card_ordering.py

@@ -0,0 +1,101 @@
+"""The supplies card must put the emptiest cartridge first.
+
+It sorted on iscritical, then on PRINTER NAME. Critical is everything at or
+below 5%, so a cartridge at 5% and one at 0% tied on the first key and the
+alphabet decided the order.
+
+That is bad ordering on its own, and worse than it looks: the dashboard queue
+takes only the first few rows per source, so a printer that is genuinely empty
+was truncated out of the card while a fuller one held its place. The card
+existed to answer "what do I walk out and change today" and was answering it
+with the alphabet.
+"""
+from unittest.mock import patch
+
+import pytest
+
+
+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']
+
+
+@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 _printer(printerid, name, remaining, status='critical'):
+    return {
+        'printerid': printerid,
+        'printername': name,
+        'assetnumber': 'A{}'.format(printerid),
+        'location': 'Shop floor',
+        'mapx': None, 'mapy': None, 'levelid': None,
+        'supplies': [{
+            'name': 'Black Toner', 'level': remaining, 'remaining': remaining,
+            'status': status, 'iswaste': False, 'isdrum': False,
+            'partnumbers': [], 'lastseen': None, 'color': 'black',
+            'supplytype': 'toner',
+        }],
+    }
+
+
+def _patched(rows):
+    return patch('plugins.printers.api.asset_routes._get_low_supplies_data',
+                 return_value={'printers': rows,
+                               'summary': {'total_checked': len(rows),
+                                           'low': 0, 'critical': len(rows)}})
+
+
+def test_the_empty_cartridge_comes_before_the_nearly_empty_one(client,
+                                                               auth_headers):
+    """'Alpha' at 5% used to precede 'Zulu' at 0% purely on the name."""
+    with _patched([_printer(1, 'Alpha', 5), _printer(2, 'Zulu', 0)]):
+        data = _call(client, auth_headers)
+
+    order = [row['printername'] for row in data['rows']]
+    assert order == ['Zulu', 'Alpha']
+
+
+def test_rows_are_ordered_by_how_empty_they_are(client, auth_headers):
+    with _patched([_printer(1, 'Alpha', 4),
+                   _printer(2, 'Bravo', 0),
+                   _printer(3, 'Charlie', 2),
+                   _printer(4, 'Delta', 5)]):
+        data = _call(client, auth_headers)
+
+    remaining = [row['worstremaining'] for row in data['rows']]
+    assert remaining == sorted(remaining)
+    assert remaining == [0, 2, 4, 5]
+
+
+def test_a_zero_percent_printer_survives_the_queue_truncation(client,
+                                                              auth_headers):
+    """The dashboard queue keeps only the first few rows per source.
+
+    With five fuller printers whose names sort earlier, the empty one used to
+    fall off the end of the card entirely.
+    """
+    rows = [_printer(index, name, 5) for index, name in enumerate(
+        ['Alpha', 'Bravo', 'Charlie', 'Delta', 'Echo'], start=1)]
+    rows.append(_printer(6, 'Zulu', 0))
+
+    with _patched(rows):
+        data = _call(client, auth_headers)
+
+    # QUEUEPERSOURCE is 5 on the frontend; first place is what matters here.
+    assert data['rows'][0]['printername'] == 'Zulu'
+    assert [r['printername'] for r in data['rows'][:5]].count('Zulu') == 1
+
+
+def test_the_name_still_breaks_a_tie(client, auth_headers):
+    with _patched([_printer(1, 'Bravo', 3), _printer(2, 'Alpha', 3)]):
+        data = _call(client, auth_headers)
+
+    assert [row['printername'] for row in data['rows']] == ['Alpha', 'Bravo']