shopdb-flask/plugins/endpoints/tests/test_plugin.py

"""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) == 5
    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_recent_events_card_is_last_and_starts_collapsed():
    """Events is the lowest-priority card: forced to the bottom of the column
    and collapsed by default."""
    plugin = EndpointsPlugin()
    panels = plugin.get_asset_panels()
    events = next(p for p in panels if p['id'] == 'endpoints-events')
    assert events['collapsedDefault'] is True
    assert events['position'] == max(p['position'] for p in panels)


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'
        assert widget['permission'] == 'endpoints.view'


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',
                'applications'):
        assert key not 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__)
    application.register_blueprint(EndpointsPlugin().get_blueprint())
    rules = {rule.rule for rule in application.url_map.iter_rules()}
    assert any(rule.endswith('/asset/<int:asset_id>/telemetrymeta') for rule in rules)