test(snmp): the poll test reaches the plugin as the app loaded it (milestone 491)
CI / lint (push) Successful in 3s
CI / integration (push) Failing after 1m6s
CI / unit (push) Successful in 1m41s
CI / publish (push) Skipped

SNMP is on by default, so the loader has already mapped snmp_devices under
_steward_plugin_snmp; importing plugins.snmp.models defined the table twice
(integration lane, run 8788).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-08 10:57:16 -04:00
co-authored by Claude Opus 5.5
parent 9ca2e38fa1
commit 189b492397
+14 -3
View File
@@ -4,6 +4,7 @@ Requires STEWARD_DATABASE_URL."""
from __future__ import annotations
import asyncio
import importlib
import os
import uuid
@@ -20,14 +21,24 @@ def app():
return create_app(testing=False)
def _snmp(sub):
"""A submodule of the snmp plugin as the app loaded it: it is on by
default, so its models are already mapped under the loader's name, and
importing plugins.snmp.models would define the table a second time."""
from steward.core.plugin_manager import get_loaded_plugin_modules
loaded = get_loaded_plugin_modules().get("snmp")
return importlib.import_module(f"{loaded.__name__ if loaded else 'plugins.snmp'}.{sub}")
def test_a_poll_records_why_a_device_is_silent_and_what_another_answered(app):
from sqlalchemy import select
from plugins.snmp import scheduler
from plugins.snmp.models import SnmpDevice
from plugins.snmp.poller import PollResult
from steward.models.metrics import PluginMetricLatest
scheduler = _snmp("scheduler")
SnmpDevice = _snmp("models").SnmpDevice
PollResult = _snmp("poller").PollResult
tag = uuid.uuid4().hex[:8]
up, down = f"sw-{tag}", f"ups-{tag}"
counts = iter([1_000.0, 61_000.0])