feat(ui): plugin get_nav() hook → sidebar Infrastructure links (slice 2)
CI / lint (push) Successful in 2s
CI / unit (push) Successful in 46s
CI / integration (push) Successful in 2m20s
CI / publish (push) Successful in 1m16s

Plugin UIs had no nav home (only dashboard widgets / typed URLs). Add an
optional get_nav() plugin export and surface it in the sidebar.

- plugin_manager: _PLUGIN_NAV registry + get_plugin_nav() getter + a tolerant
  _collect_plugin_nav() (missing hook = fine; raising/malformed = logged &
  skipped; idempotent per plugin so hot-reload re-runs cleanly). Collected in
  both the startup load path and the hot-reload path.
- app.py: inject plugin_nav into the template context, filtered to enabled
  plugins so a hot-disabled plugin's link can't linger before restart.
- base.html: render plugin links under the Infrastructure group, with the same
  request.path active-state treatment as core links.
- docker/snmp/unifi/traefik: each exports get_nav() → its /plugins/<name>/ view.
- Tests: collector behavior (collect, missing hook, reload-replace, malformed
  item, raising hook, sorted output).

With docker enabled, "Docker" now appears under Infrastructure → its fleet view.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC
This commit is contained in:
2026-06-19 18:00:12 -04:00
parent 95ebdf7045
commit c8b6719b37
8 changed files with 127 additions and 2 deletions
+5
View File
@@ -47,3 +47,8 @@ def get_scheduled_tasks() -> list:
def get_blueprint():
from .routes import docker_bp
return docker_bp
def get_nav() -> list[dict]:
# Surfaces in the sidebar's Infrastructure group (fleet-wide container view).
return [{"label": "Docker", "href": "/plugins/docker/"}]
+5
View File
@@ -21,3 +21,8 @@ def get_scheduled_tasks() -> list:
def get_blueprint():
from .routes import snmp_bp
return snmp_bp
def get_nav() -> list[dict]:
# Surfaces in the sidebar's Infrastructure group (SNMP device readings).
return [{"label": "SNMP", "href": "/plugins/snmp/"}]
+5
View File
@@ -28,3 +28,8 @@ def get_scheduled_tasks() -> list:
def get_blueprint():
from .routes import traefik_bp
return traefik_bp
def get_nav() -> list[dict]:
# Surfaces in the sidebar's Infrastructure group (Traefik services view).
return [{"label": "Traefik", "href": "/plugins/traefik/"}]
+5
View File
@@ -26,3 +26,8 @@ def get_scheduled_tasks() -> list:
def get_blueprint():
from .routes import unifi_bp
return unifi_bp
def get_nav() -> list[dict]:
# Surfaces in the sidebar's Infrastructure group (UniFi network overview).
return [{"label": "UniFi", "href": "/plugins/unifi/"}]
+5 -1
View File
@@ -177,7 +177,7 @@ def create_app(
# ── 10. Template context: inject plugin_failures into every response ───────
@app.context_processor
def _inject_plugin_failures():
from .core.plugin_manager import get_plugin_failures
from .core.plugin_manager import get_plugin_failures, get_plugin_nav
from .core.settings import get_undecryptable_secrets
# enabled_plugins lets templates gate cross-plugin embeds (e.g. the
# Hosts hub only fetches the docker fragment when docker is enabled),
@@ -186,9 +186,13 @@ def create_app(
name for name, cfg in (app.config.get("PLUGINS") or {}).items()
if isinstance(cfg, dict) and cfg.get("enabled")
}
# Sidebar entries for enabled plugins that expose a UI (get_nav). Filter
# by enabled so a hot-disabled plugin's link can't linger before restart.
plugin_nav = [e for e in get_plugin_nav() if e["plugin"] in enabled_plugins]
return {
"plugin_failures": get_plugin_failures(),
"enabled_plugins": enabled_plugins,
"plugin_nav": plugin_nav,
# Read from an in-memory cache (no DB hit per render); kept current by
# the startup scan + set_setting's discard-on-write.
"undecryptable_secrets": get_undecryptable_secrets(),
+42
View File
@@ -24,12 +24,52 @@ _LOADED_PLUGINS: set[str] = set()
# Track plugins that failed to load: name → human-readable reason.
_FAILED_PLUGINS: dict[str, str] = {}
# Nav entries contributed by plugins via the optional get_nav() export, so a
# plugin's UI gets a home in the sidebar. Each entry:
# {"plugin": <name>, "label": str, "href": str, "section": str}
_PLUGIN_NAV: list[dict] = []
def get_plugin_failures() -> dict[str, str]:
"""Return a copy of the failed-plugin registry (name → error message)."""
return dict(_FAILED_PLUGINS)
def get_plugin_nav() -> list[dict]:
"""Plugin-contributed sidebar entries, sorted by section then label."""
return sorted(_PLUGIN_NAV, key=lambda e: (e.get("section", ""), e.get("label", "")))
def _collect_plugin_nav(name: str, module) -> None:
"""Pull a plugin's optional get_nav() entries into the nav registry.
Tolerant by design: a missing hook is fine, and a raising or malformed hook
is logged and skipped — a bad nav contribution must never break loading.
Idempotent per plugin (drops prior entries first) so hot-reload re-runs cleanly.
"""
global _PLUGIN_NAV
_PLUGIN_NAV = [e for e in _PLUGIN_NAV if e.get("plugin") != name]
if not hasattr(module, "get_nav"):
return
try:
items = module.get_nav() or []
except Exception:
logger.exception("Plugin %r: get_nav() raised, skipping its nav entries", name)
return
for item in items:
try:
label, href = str(item["label"]), str(item["href"])
except (TypeError, KeyError):
logger.warning("Plugin %r: malformed nav item %r, skipping", name, item)
continue
_PLUGIN_NAV.append({
"plugin": name,
"label": label,
"href": href,
"section": str(item.get("section", "Infrastructure")),
})
def resolve_plugin_path(plugin_dirs: list[Path], name: str) -> Path | None:
"""Return the first plugin root that contains `name`, else None.
@@ -222,6 +262,7 @@ def load_plugins(app: "Quart") -> None:
_FAILED_PLUGINS.pop(name, None) # clear any stale failure from a previous reload attempt
_LOADED_PLUGINS.add(name)
_collect_plugin_nav(name, module)
logger.info("Plugin %r loaded (v%s)", name, meta.get("version", "?"))
@@ -440,6 +481,7 @@ def hot_reload_plugin(app: "Quart", name: str) -> tuple[bool, str]:
return False, f"get_scheduled_tasks() raised: {exc}"
_LOADED_PLUGINS.add(name)
_collect_plugin_nav(name, module)
logger.info("Plugin %r hot-reloaded (v%s)", name, meta.get("version", "?"))
return True, f"Plugin activated (v{meta.get('version', '?')})"
+3 -1
View File
@@ -287,7 +287,9 @@ body.dash-editing .widget-drawer { transform:translateY(0); }
<div class="nav-group">
<div class="nav-group-label">Infrastructure</div>
<a href="/hosts/" {% if _p.startswith('/hosts') %}class="active" aria-current="page"{% endif %}>Hosts</a>
{# plugin links land here in slice 2 (get_nav hook) #}
{% for item in plugin_nav %}
<a href="{{ item.href }}" {% if _p.startswith(item.href.rstrip('/')) %}class="active" aria-current="page"{% endif %}>{{ item.label }}</a>
{% endfor %}
</div>
<div class="nav-group">
<div class="nav-group-label">Monitoring</div>
+57
View File
@@ -0,0 +1,57 @@
"""Unit tests for plugin-contributed sidebar nav collection."""
import types
from steward.core import plugin_manager as pm
def _mod(nav):
"""A stand-in plugin module; omit get_nav by passing nav=None."""
m = types.SimpleNamespace()
if nav is not None:
m.get_nav = lambda: nav
return m
def test_collect_and_get(monkeypatch):
monkeypatch.setattr(pm, "_PLUGIN_NAV", [])
pm._collect_plugin_nav("docker", _mod([{"label": "Docker", "href": "/plugins/docker/"}]))
assert pm.get_plugin_nav() == [
{"plugin": "docker", "label": "Docker", "href": "/plugins/docker/", "section": "Infrastructure"}
]
def test_missing_hook_is_noop(monkeypatch):
monkeypatch.setattr(pm, "_PLUGIN_NAV", [])
pm._collect_plugin_nav("x", _mod(None))
assert pm.get_plugin_nav() == []
def test_reload_replaces_same_plugin_entries(monkeypatch):
monkeypatch.setattr(pm, "_PLUGIN_NAV", [])
pm._collect_plugin_nav("docker", _mod([{"label": "Docker", "href": "/old"}]))
pm._collect_plugin_nav("docker", _mod([{"label": "Docker", "href": "/new"}]))
assert [e["href"] for e in pm.get_plugin_nav() if e["plugin"] == "docker"] == ["/new"]
def test_malformed_item_skipped(monkeypatch):
monkeypatch.setattr(pm, "_PLUGIN_NAV", [])
pm._collect_plugin_nav("x", _mod([{"label": "NoHref"}, {"label": "Ok", "href": "/ok"}]))
assert [e["href"] for e in pm.get_plugin_nav()] == ["/ok"]
def test_raising_hook_skipped(monkeypatch):
monkeypatch.setattr(pm, "_PLUGIN_NAV", [])
def boom():
raise RuntimeError("nope")
pm._collect_plugin_nav("x", types.SimpleNamespace(get_nav=boom))
assert pm.get_plugin_nav() == []
def test_get_plugin_nav_sorted_by_section_then_label(monkeypatch):
monkeypatch.setattr(pm, "_PLUGIN_NAV", [])
pm._collect_plugin_nav("traefik", _mod([{"label": "Traefik", "href": "/t"}]))
pm._collect_plugin_nav("docker", _mod([{"label": "Docker", "href": "/d"}]))
# Same default section → sorted by label.
assert [e["label"] for e in pm.get_plugin_nav()] == ["Docker", "Traefik"]