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_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)


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')