feedBack/tests/test_plugins.py
Bret Mogilefsky af2949677a
rename: slopsmith → feedBack, byron → got-feedBack (#537)
* Update GitHub repo references from feedback* to feedBack*

* rename: slopsmith -> feedBack, byron -> got-feedBack

Renames across the entire codebase:
- slopsmith/Slopsmith/SLOPSMITH/SlopSmith -> feedBack/FeedBack/FEEDBACK/FeedBack
- byron/Byron/Byrongamatos -> got-feedBack/got-feedBack/got-feedBack
- /home/byron/ -> /opt/got-feedBack/
- byron@ougsoft.com -> hi@got-feedBack.org
- github.com/byrongamatos/ -> github.com/got-feedback/
- com.byron. -> com.got-feedback.
- SLOPSMITH_ env vars -> FEEDBACK_ with backward-compat fallback
- Protocol/storage strings migrated with read-old/write-new pattern
- window.slopsmith JS API -> window.feedBack (canonical) + backward-compat alias

Refs: #rename-slopsmith

* rename: complete regen against current main + fix backward-compat alias

Regenerated the slopsmith->feedBack / byron->got-feedBack rename on top of
current main (3 commits had landed since the branch: #572/#554/#574),
resolving the four content conflicts in favour of main's newer content
(autoplay/auto-exit, accuracy-badge, Virtuoso re-home, feedpak badge).

Completion fixes on top of the mechanical rename:
- Re-apply rename to post-branch content the original rename never saw:
  window.slopsmith(.Tour) consumers in lessons.js / notifications.js /
  onboarding-tour.js, and the matching JS + python tests (autoplay_exit,
  progression_*, test_feedpak_extension FEEDBACK_* env vars). The test env
  vars now match server.py (which reads FEEDBACK_SYNC_STARTUP /
  FEEDBACK_SKIP_STARTUP_TASKS), so the sync-startup test exercises the real
  path again.
- Restore the window.slopsmith backward-compat alias dropped during conflict
  resolution, and move the bus aliases to AFTER the _feedBackExisting merge
  block so they reference the fully-assembled object (also fixes the
  loop_api.test.js API-surface regex, which the original PR latently broke).
- Drop the stray empty data/web_library.db (runtime DB lives in CONFIG_DIR)
  and gitignore it.
- Fix stale tone-source test: feed[dB]ack -> fee[dB]ack to match shipped
  source labels.

Verified locally (org CI billing-blocked): JS 819/819 pass; pytest 1669
passed / 1683 collected with 0 import errors; zero residual slopsmith/byron
except the two intentional window.slopsmith aliases.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* rename: implement advertised backward-compat + prune dead community plugins

Address gaps where PR #537's "Backward compatibility" section was advertised
but not implemented, and clean up the community plugin list.

Env vars (FEEDBACK_* canonical, legacy SLOPSMITH_* honoured):
- New lib/env_compat.py (getenv_compat / env_flag_compat) + tests. server.py
  (_env_flag + all FEEDBACK_* reads), diagnostics_hardware, gp2midi and
  tailwind_rebuild now resolve the legacy alias, so existing SLOPSMITH_UI /
  SLOPSMITH_PLUGINS_DIR / etc. deployments keep working.
- Fix the rename collapsing plugins/__init__.py and minigames/routes.py from
  `FEEDBACK_PLUGINS_DIR or SLOPSMITH_PLUGINS_DIR` into a redundant
  `FEEDBACK_ or FEEDBACK_` (the fallback was silently lost).

Storage (app.js update-channel):
- Read feedBack-update-channel, fall back to legacy slopsmith-update-channel,
  and clear the legacy key on write — so a user's update-channel preference
  survives the rename instead of resetting to "stable".

Community plugin list (README): the rename rewrote third-party repo URLs we
don't own. Probed every one; their owners never renamed, so:
- Restore the 13 live community plugins to their real slopsmith-* names.
- Prune 6 that are 404 to the public (topkoa splitscreen/stems, OmikronApex
  tuner, Jafz2001 nam-rig-builder, DeathlySin song-preview, Erikcb91 shuffle).
- Fix a pre-existing Guitar Theory clone-command typo (nam-tone -> guitar-theory).

Verified: env_compat 7/7, JS 819/819, pytest 1690 collected / 0 import errors,
rename-sensitive + startup suites green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: byrongamatos <xasiklas@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-23 11:03:01 +02:00

4009 lines
162 KiB
Python

"""Tests for plugins/__init__.py — namespace isolation for sibling
modules and startup-time collision detection (feedBack#33).
The plugin loader used to insert each plugin directory onto `sys.path`,
which made bare `import sibling` fall through Python's per-name cache
in `sys.modules`. Two plugins shipping a same-named top-level module
(`extractor.py`, `util.py`, …) would step on each other. The loader
now exposes `context['load_sibling'](name)` that loads the sibling
under a namespaced module name `plugin_<plugin_id>.<name>` (with `.`
in plugin_id bijectively encoded — `_` -> `_5f_`, `.` -> `_2e_`),
plus a warning at startup so
existing colliding plugins are visible.
"""
import contextlib
import concurrent.futures
import importlib
import json
import logging
import sys
from pathlib import Path
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
@contextlib.contextmanager
def capture_logger(caplog, logger_name, level=logging.WARNING):
"""Context manager that attaches *caplog.handler* directly to the named
logger for the duration of the ``with`` block, then restores its original
level and ``propagate`` flag. Bypasses pytest's default root-logger
attachment so tests that set ``propagate=False`` on their logger still
capture records even when the ``feedBack`` hierarchy hasn't set up
handlers yet."""
logger = logging.getLogger(logger_name)
orig_level, orig_propagate = logger.level, logger.propagate
logger.addHandler(caplog.handler)
logger.setLevel(level)
logger.propagate = False
try:
yield logger
finally:
logger.removeHandler(caplog.handler)
logger.setLevel(orig_level)
logger.propagate = orig_propagate
# Bare module names that this test module pre-populates into
# sys.modules to simulate the bare-import path. Saved/restored by
# the reset_plugin_state fixture so they don't leak to other test
# files. Codex / Copilot review on PR for feedBack#33.
_BARE_NAMES_USED = ("util", "extractor")
@pytest.fixture()
def reset_plugin_state(monkeypatch):
"""Clear loader module-level state and restore on teardown.
Saves and restores:
* `plugins.LOADED_PLUGINS`
* any `plugin_*` keys we add to `sys.modules`
* the bare names this module simulates (`util`, `extractor`)
* `sys.path` — `plugins.load_plugins()` mutates it
Also unsets `FEEDBACK_PLUGINS_DIR` for the test's duration
(via monkeypatch) so a CI env that pre-sets it can't leak
real user plugins into a tmp_path-driven test. Per-module
locks are owned by the standard import system
(`importlib._bootstrap._module_locks`) and are not our
responsibility to reset.
"""
monkeypatch.delenv("FEEDBACK_PLUGINS_DIR", raising=False)
plugins = importlib.import_module("plugins")
saved_loaded = list(plugins.LOADED_PLUGINS)
saved_pending = dict(plugins.PENDING_PLUGINS)
saved_modules = {k: v for k, v in sys.modules.items() if k.startswith("plugin_")}
saved_bare = {k: sys.modules[k] for k in _BARE_NAMES_USED if k in sys.modules}
saved_path = list(sys.path)
plugins.LOADED_PLUGINS.clear()
plugins.PENDING_PLUGINS.clear()
for k in list(sys.modules):
if k.startswith("plugin_") or k in _BARE_NAMES_USED:
del sys.modules[k]
try:
yield plugins
finally:
plugins.LOADED_PLUGINS.clear()
plugins.LOADED_PLUGINS.extend(saved_loaded)
plugins.PENDING_PLUGINS.clear()
plugins.PENDING_PLUGINS.update(saved_pending)
for k in list(sys.modules):
if k.startswith("plugin_") or k in _BARE_NAMES_USED:
del sys.modules[k]
sys.modules.update(saved_modules)
sys.modules.update(saved_bare)
sys.path[:] = saved_path
def _make_plugin(plugin_root, plugin_id, *, sibling_files=None, routes_body=None):
"""Create a minimal plugin directory under `plugin_root`.
`sibling_files` is a dict of `{module_name: file_body}` written as
`{module_name}.py` next to routes. `routes_body` is the contents of
routes.py — defaults to a no-op `setup` so the plugin loads cleanly.
"""
plugin_dir = plugin_root / plugin_id
plugin_dir.mkdir(parents=True)
(plugin_dir / "plugin.json").write_text(
json.dumps({"id": plugin_id, "name": plugin_id, "routes": "routes.py"})
)
(plugin_dir / "routes.py").write_text(
routes_body if routes_body is not None else "def setup(app, ctx):\n pass\n"
)
for name, body in (sibling_files or {}).items():
(plugin_dir / f"{name}.py").write_text(body)
return plugin_dir
def _run_load_plugins(plugins, app, tmp_path, context=None):
"""Drive load_plugins against a tmp plugin root, restoring module
state on the way out so each test is isolated."""
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = tmp_path
try:
plugins.load_plugins(app, context if context is not None else {})
finally:
plugins.PLUGINS_DIR = saved_dir
def test_load_sibling_returns_per_plugin_namespaced_modules(tmp_path, reset_plugin_state):
"""Two plugins shipping `extractor.py` with different exports must
each see their OWN file via load_sibling — no cross-contamination."""
plugins = reset_plugin_state
_make_plugin(
tmp_path, "alpha",
sibling_files={"extractor": "MANIFEST_DIR = 'alpha-manifest'\n"},
routes_body=(
"def setup(app, ctx):\n"
" extractor = ctx['load_sibling']('extractor')\n"
" app.state.alpha_manifest = extractor.MANIFEST_DIR\n"
),
)
_make_plugin(
tmp_path, "beta",
sibling_files={"extractor": "BETA_VALUE = 42\n"},
routes_body=(
"def setup(app, ctx):\n"
" extractor = ctx['load_sibling']('extractor')\n"
" app.state.beta_value = extractor.BETA_VALUE\n"
),
)
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
_run_load_plugins(plugins, fake_app, tmp_path)
assert fake_app.state.alpha_manifest == "alpha-manifest"
assert fake_app.state.beta_value == 42
# The two extractors are namespaced into distinct sys.modules
# entries. `.` separates id and name to disambiguate when either
# contains underscores.
alpha_mod = sys.modules["plugin_alpha.extractor"]
beta_mod = sys.modules["plugin_beta.extractor"]
assert alpha_mod is not beta_mod
assert getattr(alpha_mod, "MANIFEST_DIR", None) == "alpha-manifest"
assert getattr(beta_mod, "BETA_VALUE", None) == 42
# Negative cross-check: alpha's extractor must NOT carry beta's exports.
assert not hasattr(alpha_mod, "BETA_VALUE")
assert not hasattr(beta_mod, "MANIFEST_DIR")
def test_register_library_provider_context_is_scoped_to_plugin_id(tmp_path, reset_plugin_state):
plugins = reset_plugin_state
captured = []
_make_plugin(
tmp_path, "remote_source",
routes_body=(
"def setup(app, ctx):\n"
" provider = {'id': 'remote:source', 'label': 'Remote Source', 'capabilities': ['library.read']}\n"
" ctx['register_library_provider'](provider)\n"
),
)
def register_library_provider(provider, *, replace=False, owner_plugin_id=None):
captured.append({"provider": provider, "replace": replace, "owner_plugin_id": owner_plugin_id})
return provider
_run_load_plugins(
plugins,
type("FakeApp", (), {})(),
tmp_path,
context={"register_library_provider": register_library_provider},
)
assert captured == [{
"provider": {"id": "remote:source", "label": "Remote Source", "capabilities": ["library.read"]},
"replace": False,
"owner_plugin_id": "remote_source",
}]
loaded = next(entry for entry in plugins.LOADED_PLUGINS if entry["id"] == "remote_source")
assert not any(entry.get("capability") == "library" for entry in loaded["compatibility_shims"])
def test_load_sibling_caches_repeat_calls(tmp_path, reset_plugin_state):
"""Two `load_sibling('util')` calls within the same plugin return
the identical module object — no double exec_module."""
plugins = reset_plugin_state
_make_plugin(
tmp_path, "cached",
sibling_files={"util": "INSTANCE = object()\n"},
routes_body=(
"def setup(app, ctx):\n"
" a = ctx['load_sibling']('util')\n"
" b = ctx['load_sibling']('util')\n"
" app.state.same = a is b\n"
" app.state.instance = a.INSTANCE\n"
),
)
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
_run_load_plugins(plugins, fake_app, tmp_path)
assert fake_app.state.same is True
assert fake_app.state.instance is sys.modules["plugin_cached.util"].INSTANCE
def test_load_sibling_missing_module_raises_import_error(tmp_path, reset_plugin_state):
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "bare")
with pytest.raises(ImportError):
plugins._load_plugin_sibling("bare", plugin_dir, "does_not_exist")
def test_load_sibling_rejects_traversal_and_suffix(tmp_path, reset_plugin_state):
"""The helper takes a bare module name; reject anything that could
traverse paths or carry a redundant .py suffix."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "p")
# Reject:
# - empty / non-string (bare module name required)
# - path traversal (`/`, `\`, `../`)
# - redundant `.py` suffix
# - any `.` (used as separator in the cache key — would
# otherwise allow ambiguous keys)
for bad in ("", "../etc", "sub/util", "util.py", "pkg.helper", 123, None):
with pytest.raises((ValueError, TypeError)):
plugins._load_plugin_sibling("p", plugin_dir, bad)
def test_collision_warning_fires_for_shared_module_name(tmp_path, reset_plugin_state, caplog):
"""Two plugins both shipping extractor.py must trigger the warning."""
plugins = reset_plugin_state
_make_plugin(tmp_path, "rs1extract", sibling_files={"extractor": "X = 1\n"})
_make_plugin(tmp_path, "discextract", sibling_files={"extractor": "Y = 2\n"})
with capture_logger(caplog, "feedBack.plugins"):
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
assert "Module-name collision" in caplog.text
assert "'extractor' (module)" in caplog.text
assert "rs1extract" in caplog.text
assert "discextract" in caplog.text
def test_collision_warning_silent_when_names_unique(tmp_path, reset_plugin_state, caplog):
plugins = reset_plugin_state
_make_plugin(tmp_path, "alpha", sibling_files={"alpha_helper": "A = 1\n"})
_make_plugin(tmp_path, "beta", sibling_files={"beta_helper": "B = 2\n"})
with capture_logger(caplog, "feedBack.plugins"):
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
assert "Module-name collision" not in caplog.text
def test_collision_warning_excludes_routes_and_dunders(tmp_path, reset_plugin_state, caplog):
"""routes.py is already namespaced by the loader; __init__.py
belongs to a plugin that opted into being a package and namespaces
itself. Neither should trip the collision warning even when both
plugins ship one."""
plugins = reset_plugin_state
p1 = _make_plugin(tmp_path, "one", sibling_files={"unique_one": "V = 1\n"})
p2 = _make_plugin(tmp_path, "two", sibling_files={"unique_two": "V = 2\n"})
(p1 / "__init__.py").write_text("")
(p2 / "__init__.py").write_text("")
with capture_logger(caplog, "feedBack.plugins"):
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
assert "Module-name collision" not in caplog.text
def test_collision_warning_dedupes_per_plugin(tmp_path, reset_plugin_state, caplog):
"""A single plugin shipping BOTH `extractor.py` and
`extractor/__init__.py` is a supported intra-plugin layout
(load_sibling deterministically prefers the package form,
matching CPython's import precedence). The warning must NOT
count it as a 2-plugin collision and emit a bogus message
listing the same plugin id twice. Codex round 5."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "lonely")
(plugin_dir / "extractor.py").write_text("FROM = 'file'\n")
pkg_dir = plugin_dir / "extractor"
pkg_dir.mkdir()
(pkg_dir / "__init__.py").write_text("FROM = 'package'\n")
with capture_logger(caplog, "feedBack.plugins"):
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
# Only one plugin is involved, so no cross-plugin warning fires.
assert "Module-name collision" not in caplog.text
def test_collision_warning_still_fires_when_two_plugins_each_have_both_forms(
tmp_path, reset_plugin_state, caplog
):
"""Two plugins each shipping both forms of `extractor` IS a
real cross-plugin collision and must be reported. Codex round 5
sanity check on the dedup logic."""
plugins = reset_plugin_state
for pid in ("alpha", "beta"):
plugin_dir = _make_plugin(tmp_path, pid)
(plugin_dir / "extractor.py").write_text(f"OWNER = '{pid}-file'\n")
pkg_dir = plugin_dir / "extractor"
pkg_dir.mkdir()
(pkg_dir / "__init__.py").write_text(f"OWNER = '{pid}-package'\n")
with capture_logger(caplog, "feedBack.plugins"):
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
collision_records = [r for r in caplog.records if "Module-name collision" in r.getMessage()]
assert len(collision_records) == 1
warning = collision_records[0].getMessage()
assert "alpha" in warning
assert "beta" in warning
# The warning text should list each plugin id ONCE, even though
# both plugins ship two forms of `extractor`.
assert warning.count("'alpha'") == 1
assert warning.count("'beta'") == 1
# Both forms reported in the kind label.
assert "module/package" in warning
def test_load_sibling_does_not_alias_bare_imported_package(tmp_path, reset_plugin_state):
"""A bare-imported package keeps `__package__` and
`__spec__.name` as the un-namespaced bare name, so lazy
relative imports inside it would still resolve through the
global cache. To avoid that, load_sibling does NOT reuse a
bare-imported package — it re-executes under the namespaced
spec instead. Two copies of the package coexist (one bare,
one namespaced); module-level state diverges. This is
documented as the trade-off; the alternative would silently
leak submodule cross-loads. Codex round 5."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "pkgsafe")
pkg_dir = plugin_dir / "extractor"
pkg_dir.mkdir()
(pkg_dir / "__init__.py").write_text("MARK = object()\n")
(pkg_dir / "child.py").write_text("FROM = 'leaf'\n")
# Pre-populate sys.modules['extractor'] as if a bare import had
# already pulled in the package.
spec = importlib.util.spec_from_file_location(
"extractor",
str(pkg_dir / "__init__.py"),
submodule_search_locations=[str(pkg_dir)],
)
bare_pkg = importlib.util.module_from_spec(spec)
sys.modules["extractor"] = bare_pkg
spec.loader.exec_module(bare_pkg)
bare_mark = bare_pkg.MARK
# load_sibling re-executes under the namespaced spec rather
# than aliasing the bare package.
via_helper = plugins._load_plugin_sibling("pkgsafe", plugin_dir, "extractor")
assert via_helper is not bare_pkg
# Different MARK objects confirm the namespaced version was
# actually re-executed.
assert via_helper.MARK is not bare_mark
# The namespaced submodule resolves through the namespaced
# package, NOT through `extractor.child`.
child = importlib.import_module("plugin_pkgsafe.extractor.child")
assert child.FROM == "leaf"
def test_load_sibling_handles_dotted_plugin_id_via_escape(tmp_path, reset_plugin_state):
"""Plugins with reverse-DNS-style ids (`foo.bar`) must still be
able to use load_sibling — the helper escapes `.` in the
plugin_id portion of the cache key so the synthetic parent
package is still well-formed. Spotted across codex review
rounds on PR for feedBack#33."""
plugins = reset_plugin_state
plugin_dir = tmp_path / "rdns"
plugin_dir.mkdir()
(plugin_dir / "util.py").write_text("VALUE = 'reverse-dns'\n")
util = plugins._load_plugin_sibling("com.example.foo", plugin_dir, "util")
assert util.VALUE == "reverse-dns"
# The cache key uses the bijectively-encoded form so it doesn't
# fight with Python's package resolution. `.` -> `_2e_`.
assert "plugin_com_2e_example_2e_foo.util" in sys.modules
assert sys.modules["plugin_com_2e_example_2e_foo.util"] is util
def test_load_sibling_rejects_empty_plugin_id(tmp_path, reset_plugin_state):
"""Empty / non-string plugin_id is still rejected — the helper
needs SOMETHING to namespace under."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "valid_id", sibling_files={"util": "X = 1\n"})
for bad in ("", None, 123):
with pytest.raises((ValueError, TypeError)):
plugins._load_plugin_sibling(bad, plugin_dir, "util")
def test_load_sibling_exposes_child_as_parent_attribute(tmp_path, reset_plugin_state):
"""After load_sibling caches a child, Python's package-style
relative imports (`from . import sibling`, `from .. import
sibling`) need to find the child as an ATTRIBUTE on the parent
package — not just in sys.modules. The standard import
machinery sets that attribute; load_sibling must mimic the
behavior. Codex round 9."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "expose")
(plugin_dir / "extractor.py").write_text("VAL = 'extr'\n")
# Another sibling does `from . import extractor` — pure
# attribute lookup on the synthetic parent.
(plugin_dir / "consumer.py").write_text(
"from . import extractor\n"
"GOT = extractor.VAL\n"
)
# Load the consumer first; while it's executing, the
# `from . import extractor` triggers extractor's import
# through the parent package's __path__. After it loads,
# extractor must be visible as an attribute on the parent.
consumer = plugins._load_plugin_sibling("expose", plugin_dir, "consumer")
assert consumer.GOT == "extr"
parent = sys.modules["plugin_expose"]
assert hasattr(parent, "extractor")
assert parent.extractor is sys.modules["plugin_expose.extractor"]
# And consumer is exposed on the parent the same way.
assert hasattr(parent, "consumer")
assert parent.consumer is consumer
def test_load_sibling_supports_relative_imports_between_siblings(tmp_path, reset_plugin_state):
"""A sibling loaded via load_sibling that does `from .shared
import X` (relative import to another top-level sibling) must
resolve. The synthetic parent's __path__ points at the plugin
directory so the import machinery can find sibling files via
the standard relative-import path. Codex round 7."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "rel")
(plugin_dir / "shared.py").write_text("SHARED_VALUE = 'shared'\n")
(plugin_dir / "extractor.py").write_text(
"from .shared import SHARED_VALUE\n"
"RE_EXPORT = SHARED_VALUE\n"
)
extractor = plugins._load_plugin_sibling("rel", plugin_dir, "extractor")
assert extractor.RE_EXPORT == "shared"
# The relatively-imported sibling is registered under the
# namespaced key, NOT polluted into the global `shared` slot
# (collision risk with other plugins' `shared.py`).
assert "plugin_rel.shared" in sys.modules
def test_load_sibling_package_relative_import_to_outside_sibling(tmp_path, reset_plugin_state):
"""A package-form sibling whose __init__.py does
`from ..shared import X` reaches the parent and finds another
sibling. Verifies the package + parent-__path__ wiring works
end-to-end. Codex round 7."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "pkgrel")
(plugin_dir / "shared.py").write_text("VAL = 42\n")
pkg_dir = plugin_dir / "extractor"
pkg_dir.mkdir()
(pkg_dir / "__init__.py").write_text(
"from ..shared import VAL\n"
"VALUE = VAL\n"
)
extractor = plugins._load_plugin_sibling("pkgrel", plugin_dir, "extractor")
assert extractor.VALUE == 42
def test_load_sibling_package_relative_import_works(tmp_path, reset_plugin_state):
"""A package-form sibling whose __init__.py uses `from .child
import X` must load. Without registering the synthetic parent
package `plugin_<id>` in sys.modules first, Python can't resolve
the relative import. Codex round 3 caught this."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "relpkg")
pkg_dir = plugin_dir / "extractor"
pkg_dir.mkdir()
(pkg_dir / "child.py").write_text("CHILD_VALUE = 99\n")
(pkg_dir / "__init__.py").write_text(
"from .child import CHILD_VALUE\n"
"RE_EXPORT = CHILD_VALUE\n"
)
extractor = plugins._load_plugin_sibling("relpkg", plugin_dir, "extractor")
assert extractor.RE_EXPORT == 99
# Parent package was registered as a synthetic ModuleType.
assert "plugin_relpkg" in sys.modules
def test_load_sibling_does_not_alias_bare_imported_file_module(tmp_path, reset_plugin_state):
"""Mixed migration: bare `import util` already cached this
plugin's util.py under the global `util` name. load_sibling
does NOT alias the bare module into the namespaced cache —
it re-executes under the namespaced spec. The bare module's
`__package__` / `__name__` / `__spec__` would otherwise stay
set to the un-namespaced bare name, and any later relative
import inside util.py (`from .shared import X` in a function
body) would route through the bare global cache, undoing the
isolation. Trade-off: module-level state in util splits across
two copies until the plugin removes its bare imports. Spotted
by codex review on PR for feedBack#33 round 8."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "mixmig")
util_path = plugin_dir / "util.py"
util_path.write_text("MARK = object()\n")
spec = importlib.util.spec_from_file_location("util", str(util_path))
bare_mod = importlib.util.module_from_spec(spec)
sys.modules["util"] = bare_mod
spec.loader.exec_module(bare_mod)
bare_mark = bare_mod.MARK
via_helper = plugins._load_plugin_sibling("mixmig", plugin_dir, "util")
# Different module objects — the helper re-executed instead
# of aliasing.
assert via_helper is not bare_mod
assert via_helper.MARK is not bare_mark
# Namespaced key has the namespaced object; bare key still has
# the bare-imported object.
assert sys.modules["plugin_mixmig.util"] is via_helper
assert sys.modules["util"] is bare_mod
# Critically, via_helper has the correct namespaced metadata so
# later relative imports inside it would route through the
# synthetic parent.
assert via_helper.__name__ == "plugin_mixmig.util"
assert via_helper.__package__ == "plugin_mixmig"
def test_safe_plugin_id_encoding_is_collision_free(reset_plugin_state):
"""Distinct plugin_ids must always map to distinct encoded
forms. The previous `.` -> `_x2e_` (only when `.` was present)
was not bijective: ids `foo.bar` and `foo_x2e_bar` both produced
`foo_x2e_bar`. With the bijective `_` -> `_5f_`, `.` -> `_2e_`
encoding (in that order), no two distinct plugin_ids map to the
same output. Copilot review on PR #105 round 3."""
plugins = reset_plugin_state
samples = [
"foo",
"foo_bar",
"foo.bar",
"foo_2e_bar",
"foo_5f_bar",
"foo_5f_2e_5f_bar",
"com.example.foo",
"com_example_foo",
"com_2e_example_2e_foo",
"", # empty edge — empty maps to empty, distinct from all others
"_",
".",
"._",
"_.",
]
encoded = [plugins._safe_plugin_id_for_module_name(s) for s in samples]
# Bijective: distinct inputs -> distinct outputs.
assert len(set(encoded)) == len(samples), dict(zip(samples, encoded))
def test_load_plugins_skips_non_string_id(tmp_path, reset_plugin_state, caplog):
"""A malformed manifest with a non-string id (e.g. number) is
skipped with a clear message rather than crashing later inside
`_safe_plugin_id_for_module_name`'s `.replace()` call. Copilot
review on PR #105 round 3."""
plugins = reset_plugin_state
bad_dir = tmp_path / "bad"
bad_dir.mkdir()
(bad_dir / "plugin.json").write_text('{"id": 42, "name": "bad"}')
_make_plugin(tmp_path, "good", sibling_files={"util": "X = 1\n"})
fake_app = type("FakeApp", (), {})()
with capture_logger(caplog, "feedBack.plugins"):
_run_load_plugins(plugins, fake_app, tmp_path)
assert "must be a string" in caplog.text
assert "int" in caplog.text # type name surfaced
loaded_ids = {p["id"] for p in plugins.LOADED_PLUGINS}
assert 42 not in loaded_ids
assert "good" in loaded_ids
def test_load_plugins_warns_on_falsy_non_string_id(tmp_path, reset_plugin_state, caplog):
"""`{"id": 0}` and `{"id": []}` are falsy non-strings. The
type-check must run BEFORE the falsy-empty check so the user
gets the explicit "must be a string" warning instead of a
silent skip. Copilot review on PR #105 round 4."""
plugins = reset_plugin_state
for i, bad_value in enumerate(("0", "[]", "false")): # JSON literals
bad_dir = tmp_path / f"bad{i}"
bad_dir.mkdir()
(bad_dir / "plugin.json").write_text(f'{{"id": {bad_value}, "name": "x"}}')
with capture_logger(caplog, "feedBack.plugins"):
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
# Each malformed manifest produces a "must be a string" warning;
# none are silently dropped.
assert caplog.text.count("must be a string") == 3
assert "int" in caplog.text # for {"id": 0}
assert "list" in caplog.text # for {"id": []}
assert "bool" in caplog.text # for {"id": false}
def test_load_plugins_escapes_dotted_id_in_routes_module_name(tmp_path, reset_plugin_state):
"""A plugin with a reverse-DNS id like `com.example.foo` must
have its routes module registered under a `.`-free name, or
Python would treat the cache key as a dotted package path and
set `__package__` to an unintended parent (relative imports in
routes.py would then resolve against something else entirely).
The same `.` -> `_2e_` encoding used by load_sibling now
applies to routes too. Copilot review on PR #105 round 2."""
plugins = reset_plugin_state
plugin_dir = tmp_path / "rdns_routes"
plugin_dir.mkdir()
(plugin_dir / "plugin.json").write_text(
json.dumps({"id": "com.example.foo", "name": "rdns", "routes": "routes.py"})
)
(plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n app.state.routes_loaded = True\n"
)
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
_run_load_plugins(plugins, fake_app, tmp_path)
assert fake_app.state.routes_loaded is True
# Routes module is registered under the escaped name and is a
# single identifier-shaped key — NOT a dotted path that Python
# would try to resolve as a real package.
assert "plugin_com_2e_example_2e_foo_routes" in sys.modules
routes_mod = sys.modules["plugin_com_2e_example_2e_foo_routes"]
# __package__ is empty (top-level module), not a dotted parent.
assert (routes_mod.__package__ or "") == ""
def test_load_sibling_parent_registration_is_atomic(tmp_path, reset_plugin_state):
"""Two threads loading DIFFERENT siblings for the same plugin
must agree on the synthetic parent. If they each constructed a
fresh ModuleType and assigned to sys.modules[parent_name]
without coordination, the second assignment could replace the
first — and child attributes already attached to the
first parent would disappear, breaking `from . import sibling`.
setdefault makes the registration atomic. Copilot round 2."""
import threading
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "atomic")
# Two slow siblings so the threads have time to overlap.
(plugin_dir / "alpha.py").write_text("import time\ntime.sleep(0.05)\nVALUE = 'a'\n")
(plugin_dir / "beta.py").write_text("import time\ntime.sleep(0.05)\nVALUE = 'b'\n")
errors: list = []
def worker(name):
try:
plugins._load_plugin_sibling("atomic", plugin_dir, name)
except BaseException as e: # pragma: no cover
errors.append(e)
threads = [
threading.Thread(target=worker, args=("alpha",)),
threading.Thread(target=worker, args=("beta",)),
]
for t in threads:
t.start()
for t in threads:
t.join()
assert not errors
parent = sys.modules["plugin_atomic"]
# Both children are exposed as attributes on the SAME parent —
# neither was lost to a parent-replacement race.
assert hasattr(parent, "alpha")
assert hasattr(parent, "beta")
assert parent.alpha.VALUE == "a"
assert parent.beta.VALUE == "b"
def test_load_sibling_concurrent_first_call_returns_fully_initialized(tmp_path, reset_plugin_state):
"""Two threads racing on the same first-time load_sibling call
should both receive a fully-initialized module object — neither
can observe the half-built module that's briefly registered in
sys.modules between `module_from_spec` and the end of
`exec_module`. The per-module lock added in round 8 enforces
this."""
import threading
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "racy")
# The sibling's __init__ does meaningful work BEFORE setting
# `READY = True`, so a partially-initialized module would lack
# the attribute even though the module object exists in
# sys.modules.
(plugin_dir / "slow.py").write_text(
"import time\n"
"time.sleep(0.05)\n"
"READY = True\n"
"VALUE = 'done'\n"
)
results: list = []
errors: list = []
def worker():
try:
mod = plugins._load_plugin_sibling("racy", plugin_dir, "slow")
results.append((mod, getattr(mod, "READY", None), getattr(mod, "VALUE", None)))
except BaseException as e: # pragma: no cover - bug path
errors.append(e)
threads = [threading.Thread(target=worker) for _ in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
assert not errors
assert len(results) == 8
# Every caller sees the same fully-initialized module.
first_mod, _, _ = results[0]
for mod, ready, value in results:
assert mod is first_mod
assert ready is True
assert value == "done"
def test_load_sibling_does_not_reuse_other_plugins_bare_import(tmp_path, reset_plugin_state):
"""If sys.path has already cached a util.py from PLUGIN A under
the bare name `util`, plugin B's load_sibling('util') must NOT
return plugin A's module — it has to load plugin B's own copy
under the namespaced key. This is the whole point of the
isolation fix; the reuse path can't accidentally undo it."""
plugins = reset_plugin_state
plugin_a = _make_plugin(tmp_path, "plug_a")
plugin_b = _make_plugin(tmp_path, "plug_b")
(plugin_a / "util.py").write_text("OWNER = 'a'\n")
(plugin_b / "util.py").write_text("OWNER = 'b'\n")
# Simulate plugin A's bare import landing in sys.modules['util'].
spec_a = importlib.util.spec_from_file_location("util", str(plugin_a / "util.py"))
bare_a = importlib.util.module_from_spec(spec_a)
sys.modules["util"] = bare_a
spec_a.loader.exec_module(bare_a)
assert bare_a.OWNER == "a"
# Plugin B's load_sibling must give plugin B's util, NOT plugin A's.
b_util = plugins._load_plugin_sibling("plug_b", plugin_b, "util")
assert b_util is not bare_a
assert b_util.OWNER == "b"
def test_load_sibling_loads_package_form(tmp_path, reset_plugin_state):
"""A plugin shipping a sibling as a package directory
(`extractor/__init__.py`) should be loadable through
load_sibling exactly like a single-file `.py` sibling. The
collision-warning scanner directs maintainers of package-form
plugins toward load_sibling, so the helper has to actually
support them. Codex review on PR for feedBack#33."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "pkgplugin")
pkg_dir = plugin_dir / "extractor"
pkg_dir.mkdir()
(pkg_dir / "__init__.py").write_text("ROOT_VALUE = 7\n")
(pkg_dir / "child.py").write_text("CHILD_VALUE = 8\n")
extractor = plugins._load_plugin_sibling("pkgplugin", plugin_dir, "extractor")
assert extractor.ROOT_VALUE == 7
# Submodule lookup works because spec carried submodule_search_locations.
child = importlib.import_module("plugin_pkgplugin.extractor.child")
assert child.CHILD_VALUE == 8
def test_load_sibling_prefers_package_over_file_when_both_exist(tmp_path, reset_plugin_state):
"""If a plugin ships BOTH `extractor.py` and `extractor/__init__.py`
in the same directory, the package form wins — matches CPython's
own import-resolution precedence so bare `import extractor` and
`load_sibling('extractor')` always run the same code path.
Spotted by codex review on PR for feedBack#33."""
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "both")
(plugin_dir / "extractor.py").write_text("FROM = 'file'\n")
pkg_dir = plugin_dir / "extractor"
pkg_dir.mkdir()
(pkg_dir / "__init__.py").write_text("FROM = 'package'\n")
extractor = plugins._load_plugin_sibling("both", plugin_dir, "extractor")
assert extractor.FROM == "package"
def test_load_sibling_missing_in_both_forms_raises_with_useful_message(tmp_path, reset_plugin_state):
plugins = reset_plugin_state
plugin_dir = _make_plugin(tmp_path, "empty")
with pytest.raises(ImportError) as exc:
plugins._load_plugin_sibling("empty", plugin_dir, "missing")
msg = str(exc.value)
# Error message should mention BOTH probed locations so a
# confused author sees "I checked here AND here" not "I checked
# only the .py form".
assert "missing.py" in msg
assert "missing" in msg and "__init__.py" in msg
def test_load_sibling_disambiguates_underscored_ids_and_names(tmp_path, reset_plugin_state):
"""`(plugin_id='a_b', name='c')` and `(plugin_id='a', name='b_c')`
must NOT collide in sys.modules. The `.` separator + bijective
`_` -> `_5f_` encoding of plugin_id make the cache key
unambiguous (the old `_` separator collapsed both to
`plugin_a_b_c`). Codex review on PR for feedBack#33."""
plugins = reset_plugin_state
p1 = _make_plugin(tmp_path, "a_b", sibling_files={"c": "WHO = 'a_b/c'\n"})
p2 = _make_plugin(tmp_path, "a", sibling_files={"b_c": "WHO = 'a/b_c'\n"})
m1 = plugins._load_plugin_sibling("a_b", p1, "c")
m2 = plugins._load_plugin_sibling("a", p2, "b_c")
assert m1 is not m2
assert m1.WHO == "a_b/c"
assert m2.WHO == "a/b_c"
# Both keys exist independently in sys.modules. plugin_id `a_b`
# encodes to `a_5f_b` so the parent is `plugin_a_5f_b`. The
# NAME portion is not encoded (it's only the plugin_id that
# could be confused with the `.` separator). So:
# id='a_b', name='c' -> plugin_a_5f_b.c
# id='a', name='b_c' -> plugin_a.b_c
# The old `_` separator collapsed both to `plugin_a_b_c`.
assert "plugin_a_5f_b.c" in sys.modules
assert "plugin_a.b_c" in sys.modules
assert sys.modules["plugin_a_5f_b.c"] is m1
assert sys.modules["plugin_a.b_c"] is m2
def test_collision_warning_detects_package_form(tmp_path, reset_plugin_state, caplog):
"""A plugin shipping `extractor/__init__.py` collides with another
plugin's `extractor.py` the same way two `.py` files would. The
scanner picks up packages too. Codex review on PR for feedBack#33."""
plugins = reset_plugin_state
# Plugin one: extractor.py
_make_plugin(tmp_path, "as_module", sibling_files={"extractor": "X = 1\n"})
# Plugin two: extractor/ (package form)
plugin_pkg = _make_plugin(tmp_path, "as_package")
pkg_dir = plugin_pkg / "extractor"
pkg_dir.mkdir()
(pkg_dir / "__init__.py").write_text("Y = 2\n")
with capture_logger(caplog, "feedBack.plugins"):
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
assert "Module-name collision" in caplog.text
assert "extractor" in caplog.text
assert "as_module" in caplog.text
assert "as_package" in caplog.text
# The mixed-form label should also appear so the maintainer knows
# to look for both shapes.
assert "module/package" in caplog.text
def test_collision_warning_detects_two_packages(tmp_path, reset_plugin_state, caplog):
"""Two plugins each shipping the SAME package directory form."""
plugins = reset_plugin_state
for pid in ("plug_a", "plug_b"):
plugin_dir = _make_plugin(tmp_path, pid)
pkg = plugin_dir / "shared_pkg"
pkg.mkdir()
(pkg / "__init__.py").write_text(f"# {pid}\n")
with capture_logger(caplog, "feedBack.plugins"):
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
assert "Module-name collision" in caplog.text
assert "shared_pkg" in caplog.text
assert "plug_a" in caplog.text
assert "plug_b" in caplog.text
def test_per_plugin_context_does_not_leak_load_sibling_across_plugins(tmp_path, reset_plugin_state):
"""Plugin A's `load_sibling` must close over plugin A's id+dir.
If both plugins received the SAME closure (the bug we are
preventing), plugin A calling `load_sibling('thing')` would
load whatever the loop's last-iteration closure pointed at —
typically the alphabetically-last plugin's directory."""
plugins = reset_plugin_state
_make_plugin(
tmp_path, "aaa",
sibling_files={"thing": "ORIGIN = 'aaa'\n"},
routes_body=(
"def setup(app, ctx):\n"
" app.state.aaa_origin = ctx['load_sibling']('thing').ORIGIN\n"
),
)
_make_plugin(
tmp_path, "zzz",
sibling_files={"thing": "ORIGIN = 'zzz'\n"},
routes_body=(
"def setup(app, ctx):\n"
" app.state.zzz_origin = ctx['load_sibling']('thing').ORIGIN\n"
),
)
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
_run_load_plugins(plugins, fake_app, tmp_path)
assert fake_app.state.aaa_origin == "aaa"
assert fake_app.state.zzz_origin == "zzz"
# ── Bundled plugins always win over user-installed copies ────────────────────
def test_bundled_plugin_always_wins_over_feedBack_plugins_dir_copy(
tmp_path, reset_plugin_state, monkeypatch, caplog
):
"""Bundled plugins always win over user-installed copies in FEEDBACK_PLUGINS_DIR.
Even though FEEDBACK_PLUGINS_DIR is scanned first, a user-installed copy
with the same id as a bundled plugin is evicted in favour of the bundled
version. The loader emits a warning naming the ignored user copy.
"""
plugins = reset_plugin_state
# Simulate the in-tree (bundled) plugins directory.
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
_make_plugin(
bundled_dir, "highway_3d",
routes_body=(
"def setup(app, ctx):\n"
" app.state.origin = 'bundled'\n"
),
)
# Mark the in-tree plugin as bundled — required for override detection.
(bundled_dir / "highway_3d" / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "highway_3d", "routes": "routes.py", "bundled": True})
)
# Simulate a user-installed plugins directory (FEEDBACK_PLUGINS_DIR).
user_dir = tmp_path / "user"
user_dir.mkdir()
_make_plugin(
user_dir, "highway_3d",
routes_body=(
"def setup(app, ctx):\n"
" app.state.origin = 'user'\n"
),
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
# Use bundled_dir as the in-tree PLUGINS_DIR root.
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
with capture_logger(caplog, "feedBack.plugins"):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# Bundled copy must win — setup() from bundled_dir ran, not user_dir.
assert fake_app.state.origin == "bundled"
# Exactly one highway_3d entry registered.
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
# The loader emits a warning about the ignored user copy, naming its path.
assert (
"User-installed copy of bundled plugin 'highway_3d'" in caplog.text
and "ignored" in caplog.text
and str(user_dir / "highway_3d") in caplog.text
)
def test_bundled_plugin_wins_over_user_copy_and_logs_warning(
tmp_path, reset_plugin_state, monkeypatch, caplog
):
"""Bundled plugin wins over a user-installed copy with the same id.
The loader emits a warning naming the ignored user copy. The kept entry
is the bundled version (bundled=True, dir matches plugin id).
"""
plugins = reset_plugin_state
# In-tree bundled plugin — plugin.json carries ``"bundled": true``.
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
plugin_dir = bundled_dir / "highway_3d"
plugin_dir.mkdir()
(plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "bundled": True})
)
# User-installed copy with the same id in a differently-named directory.
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "3dhighway" # different directory name
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)"})
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
with capture_logger(caplog, "feedBack.plugins"):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# Exactly one entry registered.
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
# The kept copy must be the bundled version.
kept = hw3d_entries[0]
assert str(kept["_dir"]) == str(plugin_dir)
assert kept.get("bundled") is True
# The loader must emit the specific warning about the ignored user copy, naming its path.
assert (
"User-installed copy of bundled plugin 'highway_3d'" in caplog.text
and "ignored" in caplog.text
and str(user_plugin_dir) in caplog.text
)
def test_bundled_plugin_wins_over_verbatim_user_copy(
tmp_path, reset_plugin_state, monkeypatch, caplog
):
"""Bundled plugin wins even when the user copy verbatim-copied ``"bundled": true``.
The three-part ``_is_bundled()`` check — in-tree directory, manifest field,
AND dir name == plugin id — correctly identifies the in-tree copy as
bundled and ignores the user copy regardless of its manifest contents.
"""
plugins = reset_plugin_state
# In-tree bundled plugin.
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
plugin_dir = bundled_dir / "highway_3d"
plugin_dir.mkdir()
(plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "bundled": True})
)
# User copy that was copied verbatim from in-tree — still carries bundled=true.
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d_custom"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (custom)", "bundled": True})
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
with capture_logger(caplog, "feedBack.plugins"):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# Exactly one entry registered.
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
# The kept copy must be the real bundled version.
kept = hw3d_entries[0]
assert str(kept["_dir"]) == str(plugin_dir)
assert kept.get("bundled") is True
# The loader emits the warning about the ignored user copy, naming its path.
assert (
"User-installed copy of bundled plugin 'highway_3d'" in caplog.text
and "ignored" in caplog.text
and str(user_plugin_dir) in caplog.text
)
def test_bundled_plugin_wins_over_copy_in_same_plugins_dir(
tmp_path, reset_plugin_state, caplog
):
"""Bundled plugin wins even when a user fork sorts alphabetically first.
Both plugins live under the same PLUGINS_DIR, so override detection cannot
use the directory parent alone. The three-part ``_is_bundled()`` check —
``(in-tree dir) AND manifest.get("bundled") AND (dir name == plugin id)`` —
identifies the bundled copy and discards the user fork.
Layout mirroring the canonical example from the PR description:
plugins/3dhighway/ ← user-installed fork (no "bundled" field)
plugins/highway_3d/ ← bundled core (has "bundled": true, dir name == id)
``3dhighway`` sorts before ``highway_3d`` alphabetically and is registered
first; when ``highway_3d`` arrives it evicts the user fork and wins.
"""
plugins = reset_plugin_state
plugins_dir = tmp_path / "plugins"
plugins_dir.mkdir()
# User-installed fork — comes first alphabetically.
user_plugin_dir = plugins_dir / "3dhighway"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user fork)"})
)
# Bundled core — sorts after "3dhighway"; evicts the user fork when encountered.
bundled_plugin_dir = plugins_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "bundled": True})
)
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = plugins_dir
try:
with capture_logger(caplog, "feedBack.plugins"):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
# The bundled copy must win, regardless of alphabetical order.
kept = hw3d_entries[0]
assert str(kept["_dir"]) == str(bundled_plugin_dir)
# Bundled flag set on the kept copy.
assert kept.get("bundled") is True
# The loader emits the specific warning about the ignored user copy, naming its path.
assert (
"User-installed copy of bundled plugin 'highway_3d'" in caplog.text
and "ignored" in caplog.text
and str(user_plugin_dir) in caplog.text
)
def test_bundled_plugin_wins_over_copy_in_same_plugins_dir_bundled_sorts_first(
tmp_path, reset_plugin_state, caplog
):
"""Bundled plugin wins when it sorts *before* the user directory.
When the bundled copy is encountered first it is registered; the later
user copy is discarded with a warning.
Layout where bundled sorts before user:
plugins/highway_3d/ ← bundled core (has "bundled": true), sorts first
plugins/zzz-highway/ ← user-installed fork (no "bundled" field), sorts last
"""
plugins = reset_plugin_state
plugins_dir = tmp_path / "plugins"
plugins_dir.mkdir()
# Bundled core — sorts first alphabetically.
bundled_plugin_dir = plugins_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "bundled": True})
)
# User-installed fork — sorts after "highway_3d", will be encountered second.
user_plugin_dir = plugins_dir / "zzz-highway"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user fork)"})
)
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = plugins_dir
try:
with capture_logger(caplog, "feedBack.plugins"):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
# The bundled copy must win; user fork is discarded.
kept = hw3d_entries[0]
assert str(kept["_dir"]) == str(bundled_plugin_dir)
# Bundled flag set on the kept copy.
assert kept.get("bundled") is True
# The loader emits the specific warning about the ignored user copy, naming its path.
assert (
"User-installed copy of bundled plugin 'highway_3d'" in caplog.text
and "ignored" in caplog.text
and str(user_plugin_dir) in caplog.text
)
def test_bundled_plugin_wins_over_verbatim_copy_in_same_plugins_dir(
tmp_path, reset_plugin_state, caplog
):
"""Bundled plugin wins over a verbatim user copy that carries ``"bundled": true``.
The three-part ``_is_bundled()`` check — in-tree directory, manifest field,
AND directory-name-matches-plugin-id — correctly identifies the real core
copy and rejects the verbatim clone (dir name ≠ plugin id).
Layout:
plugins/highway_3d/ ← bundled core (has "bundled": true, dir name == id)
plugins/zzz-highway/ ← verbatim user copy (has "bundled": true, dir name ≠ id)
``highway_3d`` sorts before ``zzz-highway``; the bundled copy is encountered
first. When the user copy arrives it is not truly bundled (dir name mismatch)
and is discarded with a warning.
"""
plugins = reset_plugin_state
plugins_dir = tmp_path / "plugins"
plugins_dir.mkdir()
# Bundled core — dir name == plugin id, has "bundled": true.
bundled_plugin_dir = plugins_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "bundled": True})
)
# Verbatim user copy — different dir name, but manifest is identical (including "bundled": true).
user_plugin_dir = plugins_dir / "zzz-highway"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "bundled": True})
)
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = plugins_dir
try:
with capture_logger(caplog, "feedBack.plugins"):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
# The real bundled copy (highway_3d) must win; the verbatim clone is discarded.
kept = hw3d_entries[0]
assert str(kept["_dir"]) == str(bundled_plugin_dir)
# Bundled flag set on the kept copy.
assert kept.get("bundled") is True
# Warning about the ignored user copy must be emitted, naming its path.
assert (
"User-installed copy of bundled plugin 'highway_3d'" in caplog.text
and "ignored" in caplog.text
and str(user_plugin_dir) in caplog.text
)
def test_bundled_wins_with_multiple_stale_copies(
tmp_path, reset_plugin_state, monkeypatch, caplog
):
"""Bundled plugin wins when multiple stale copies exist simultaneously.
Exercises the exact #181 layout: the user has both an external
``FEEDBACK_PLUGINS_DIR/highway_3d`` copy AND a stale in-tree clone at
``plugins/3dhighway``, alongside the real bundled ``plugins/highway_3d``.
Scan order:
1. FEEDBACK_PLUGINS_DIR scanned first → user_dir/highway_3d registered.
2. PLUGINS_DIR scanned next (sorted):
- ``3dhighway`` sorts before ``highway_3d`` → duplicate, neither is bundled
→ discarded with a warning naming the specific plugin_dir.
- ``highway_3d`` → bundled; evicts the FEEDBACK_PLUGINS_DIR copy; wins.
Both stale paths must be named in warning log messages.
"""
plugins = reset_plugin_state
# Simulate the in-tree (bundled) plugins directory.
plugins_dir = tmp_path / "plugins"
plugins_dir.mkdir()
# Stale in-tree clone (different dir name; not bundled because dir ≠ id).
stale_intree_dir = plugins_dir / "3dhighway"
stale_intree_dir.mkdir()
(stale_intree_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (stale clone)"})
)
# Real bundled copy.
bundled_plugin_dir = plugins_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "bundled": True})
)
# Simulate a user-installed plugins directory (FEEDBACK_PLUGINS_DIR).
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)"})
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = plugins_dir
try:
with capture_logger(caplog, "feedBack.plugins"):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# Exactly one entry registered.
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
# The bundled copy must win.
kept = hw3d_entries[0]
assert str(kept["_dir"]) == str(bundled_plugin_dir)
assert kept.get("bundled") is True
# The stale in-tree clone path must be named in the log.
assert str(stale_intree_dir) in caplog.text
# The user-installed copy path must be named in the log (bundled-wins warning).
assert (
"User-installed copy of bundled plugin 'highway_3d'" in caplog.text
and "ignored" in caplog.text
and str(user_plugin_dir) in caplog.text
)
def test_fallback_to_user_copy_when_bundled_routes_fail(
tmp_path, reset_plugin_state, monkeypatch, caplog
):
"""If a bundled plugin's routes.setup() raises, the loader falls back to
the previously-evicted user-installed copy so the server keeps working.
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
# Bundled copy whose routes.setup() will raise.
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('bundled broken')\n"
)
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n app.state.origin = 'user_fallback'\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
with capture_logger(caplog, "feedBack.plugins", level=logging.WARNING):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# The fallback user copy should be registered.
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
assert hw3d_entries[0].get("bundled") is False
assert str(hw3d_entries[0]["_dir"]) == str(user_plugin_dir)
# The fallback's routes.setup() should have run.
assert getattr(fake_app.state, "origin", None) == "user_fallback"
# The fallback warning must be logged.
assert "falling back to user-installed copy" in caplog.text
assert str(user_plugin_dir) in caplog.text
def test_fallback_when_bundled_sorts_first_and_routes_fail(
tmp_path, reset_plugin_state, caplog
):
"""Fallback works even when the bundled plugin is discovered BEFORE the user copy.
Thread 6 regression: the `elif kept_is_bundled:` branch previously
discarded the user copy without storing it in `_pending_evictions`, so
a bundled plugin discovered first had no fallback available.
Layout (both in the same plugins_dir, bundled sorts first):
plugins/highway_3d/ ← bundled (broken routes)
plugins/zzz-highway/ ← user copy (working routes)
"""
plugins = reset_plugin_state
plugins_dir = tmp_path / "plugins"
plugins_dir.mkdir()
# Bundled copy — sorts first alphabetically; routes.setup() will raise.
bundled_plugin_dir = plugins_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('bundled broken')\n"
)
# User copy — sorts after bundled; working routes.
user_plugin_dir = plugins_dir / "zzz-highway"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n app.state.origin = 'user_fallback'\n"
)
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = plugins_dir
try:
with capture_logger(caplog, "feedBack.plugins", level=logging.WARNING):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# The fallback user copy should be registered.
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
assert hw3d_entries[0].get("bundled") is False
assert str(hw3d_entries[0]["_dir"]) == str(user_plugin_dir)
# The fallback's routes.setup() should have run.
assert getattr(fake_app.state, "origin", None) == "user_fallback"
# The fallback warning must be logged.
assert "falling back to user-installed copy" in caplog.text
assert str(user_plugin_dir) in caplog.text
def test_plugin_absent_when_both_routes_fail(
tmp_path, reset_plugin_state, monkeypatch, caplog
):
"""When both bundled routes AND fallback routes fail, plugin is absent from LOADED_PLUGINS.
Thread 5 regression: the old code always appended to _loaded_batch even when
the fallback routes also failed, causing the plugin to appear in /api/plugins
despite having no working routes — contradicting the 'plugin unavailable' log.
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('bundled broken')\n"
)
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
# User copy routes also raise.
(user_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('user also broken')\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
with capture_logger(caplog, "feedBack.plugins", level=logging.WARNING):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# Plugin must be absent — neither bundled nor user copy registered.
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 0
# Both failures must be logged.
assert "falling back to user-installed copy" in caplog.text
assert "also failed to load routes" in caplog.text
def test_partial_route_registration_warning(
tmp_path, reset_plugin_state, monkeypatch, caplog
):
"""When bundled setup() registers routes before raising, a specific
warning should name the count so maintainers can identify the partial
registration in the server log (Thread 1, review-4226318201).
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
# Routes that register one endpoint before raising.
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n"
" app.routes.append('stub')\n"
" raise RuntimeError('partial fail')\n"
)
fake_app = type("FakeApp", (), {"routes": []})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
with capture_logger(caplog, "feedBack.plugins", level=logging.WARNING):
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
assert "registered 1 route" in caplog.text
assert "cannot be removed" in caplog.text
def test_sibling_module_cache_purged_before_fallback(
tmp_path, reset_plugin_state, monkeypatch
):
"""Sibling modules from the failed bundled copy are purged from sys.modules
before the fallback copy's routes are loaded, so the fallback gets a clean
slate and doesn't accidentally resolve to bundled helper code
(Thread 2, review-4226318201).
"""
import sys
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
# Bundled helper reports 'bundled'; bundled routes load it via load_sibling
# then fail — this caches the bundled helper in sys.modules.
(bundled_plugin_dir / "helper.py").write_text("ORIGIN = 'bundled'\n")
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n"
" ctx['load_sibling']('helper') # caches bundled helper\n"
" raise RuntimeError('bundled broken')\n"
)
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
# User copy has its own helper with a different ORIGIN; routes use load_sibling.
(user_plugin_dir / "helper.py").write_text("ORIGIN = 'user'\n")
(user_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n"
" helper = ctx['load_sibling']('helper')\n"
" app.state.helper_origin = helper.ORIGIN\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# The fallback routes ran and resolved the user copy's helper, not bundled's.
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 1
assert getattr(fake_app.state, "helper_origin", None) == "user"
def test_fallback_success_clears_error_in_progress_events(
tmp_path, reset_plugin_state, monkeypatch
):
"""When fallback succeeds, the progress event should carry explicit
error=None (clear_error=True) so downstream startup-status handlers can
clear the bundled-failure error text (Thread 3, review-4226318201).
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('bundled broken')\n"
)
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n pass\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
events = []
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {}, progress_cb=events.append)
finally:
plugins.PLUGINS_DIR = saved_dir
# After fallback success, a "plugin-registered" event with explicit
# "error": None should clear the bundled-failure error in the handler.
registered_clear_events = [
e for e in events
if e.get("phase") == "plugin-registered"
and e.get("plugin_id") == "highway_3d"
and "error" in e
and e["error"] is None
]
assert registered_clear_events, (
"Expected a plugin-registered event with explicit error=None after fallback success"
)
def test_plugins_complete_loaded_count_when_both_routes_fail(
tmp_path, reset_plugin_state, monkeypatch
):
"""When both bundled and fallback routes fail, plugins-complete must
report loaded = actual registered count, not len(plugin_load_specs)
(Thread 4, review-4226318201).
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
# One healthy plugin + one that will fail completely.
healthy_plugin_dir = bundled_dir / "healthy"
healthy_plugin_dir.mkdir()
(healthy_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "healthy", "name": "Healthy Plugin", "bundled": True})
)
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('bundled broken')\n"
)
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('user also broken')\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
events = []
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {}, progress_cb=events.append)
finally:
plugins.PLUGINS_DIR = saved_dir
complete = next(e for e in events if e["phase"] == "plugins-complete")
# "healthy" loaded OK, "highway_3d" failed completely → actual loaded == 1
assert complete["loaded"] == 1, (
f"plugins-complete loaded should be 1 (only healthy), got {complete['loaded']}"
)
# total still reflects how many specs were discovered
assert complete["total"] == 2
def test_fallback_preserves_plugin_order(
tmp_path, reset_plugin_state, monkeypatch
):
"""When a bundled copy evicts a user-installed copy and then fails to load
its routes, the fallback (user copy) must occupy the SAME slot the user
copy had at discovery time — not the position the bundled copy would have
been appended at (Thread 1, review-4227852922; Thread 3, review-4227201020).
Layout:
user dir: highway_3d (working fallback, has routes)
bundled: alpha (healthy), highway_3d (broken routes), omega (healthy)
Scan order: user dir first, then bundled. user/highway_3d lands at slot 0.
When bundled/highway_3d is scanned it replaces the user copy IN-PLACE at
slot 0 (review-4227852922 fix). After route failure the fallback is
inserted back at slot 0.
Expected order after fallback: [highway_3d (fallback), alpha, omega]
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
for name, content in [
("alpha", {"id": "alpha", "name": "Alpha", "bundled": True}),
("highway_3d", {"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True}),
("omega", {"id": "omega", "name": "Omega", "bundled": True}),
]:
d = bundled_dir / name
d.mkdir()
(d / "plugin.json").write_text(json.dumps(content))
if name == "highway_3d":
(d / "routes.py").write_text("def setup(app, ctx):\n raise RuntimeError('broken')\n")
user_dir = tmp_path / "user"
user_dir.mkdir()
user_hw = user_dir / "highway_3d"
user_hw.mkdir()
# User copy must have a routes file; without one fallback_routes_ok is
# False and the entry is never added to LOADED_PLUGINS (review-4227852922).
(user_hw / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_hw / "routes.py").write_text("def setup(app, ctx): pass\n")
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
ids = [p["id"] for p in plugins.LOADED_PLUGINS]
assert ids == ["highway_3d", "alpha", "omega"], (
f"Fallback entry must occupy the user copy's original discovery slot, got order: {ids}"
)
# The entry at position 0 (user copy's original slot) is the fallback.
assert plugins.LOADED_PLUGINS[0].get("fallback") is True
def test_fallback_not_registered_when_user_copy_has_no_routes(
tmp_path, reset_plugin_state, monkeypatch
):
"""If the evicted user copy has no routes entry it cannot restore the
bundled plugin's backend endpoints. The fallback must be treated as
unsuccessful (fallback_routes_ok=False) so the plugin is absent from
LOADED_PLUGINS and the bundled-failure error stays in startup-status.
(Thread 2, review-4227852922)
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bd = bundled_dir / "highway_3d"
bd.mkdir()
(bd / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bd / "routes.py").write_text("def setup(app, ctx):\n raise RuntimeError('broken')\n")
user_dir = tmp_path / "user"
user_dir.mkdir()
user_hw = user_dir / "highway_3d"
user_hw.mkdir()
# No routes in user copy — cannot restore bundled endpoints.
(user_hw / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)"})
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
ids = [p["id"] for p in plugins.LOADED_PLUGINS]
assert "highway_3d" not in ids, (
"Plugin with no-routes user copy must NOT appear in LOADED_PLUGINS"
)
def test_fallback_skipped_when_setup_was_mid_flight_on_timeout(
tmp_path, reset_plugin_state, monkeypatch, caplog
):
"""When bundled setup() times out mid-flight (already executing), the loader
must NOT activate the user-copy fallback — the original setup() may still be
mutating the router concurrently (Thread 4, review-4227201020).
Simulated by injecting an exception with ``setup_mid_flight=True`` (the
attribute _route_setup_on_main attaches when it detects _started is set at
timeout).
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bundled_plugin_dir / "routes.py").write_text("def setup(app, ctx): pass\n")
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n app.state.fallback_ran = True\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
# Inject a route_setup_fn that simulates a mid-flight timeout by raising
# a TimeoutError with setup_mid_flight=True.
def _mid_flight_timeout_setup_fn(fn):
e = concurrent.futures.TimeoutError()
e.setup_mid_flight = True
raise e
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
with capture_logger(caplog, "feedBack.plugins", level=logging.WARNING):
plugins.load_plugins(fake_app, {}, route_setup_fn=_mid_flight_timeout_setup_fn)
finally:
plugins.PLUGINS_DIR = saved_dir
# Fallback must NOT have run — original setup() was mid-flight.
assert not getattr(fake_app.state, "fallback_ran", False), (
"Fallback setup() must not run when bundled setup() was mid-flight at timeout"
)
# Plugin must be absent from LOADED_PLUGINS (both copies failed).
hw3d_entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(hw3d_entries) == 0, (
"Plugin must be absent when bundled timed out mid-flight and fallback was skipped"
)
# Warning about skipping fallback must appear in log.
assert "Skipping fallback" in caplog.text or "mid-flight" in caplog.text
def test_fallback_entry_has_fallback_true_field(
tmp_path, reset_plugin_state, monkeypatch
):
"""When a user-copy fallback is activated because bundled routes fail,
the registered LOADED_PLUGINS entry must include ``"fallback": True``.
The /api/plugins endpoint exposes this field so the settings UI can
show a warning badge (Thread 4, review-4226937699).
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('bundled broken')\n"
)
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_plugin_dir / "routes.py").write_text("def setup(app, ctx):\n pass\n")
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert len(entries) == 1
entry = entries[0]
assert entry.get("fallback") is True, (
"Fallback user copy entry must have fallback=True, got "
f"{entry.get('fallback')!r}"
)
assert entry.get("bundled") is False, (
"Fallback user copy must not be marked bundled"
)
def test_fallback_proceeds_when_install_requirements_returns_false(
tmp_path, reset_plugin_state, monkeypatch
):
"""_install_requirements returning False in the fallback path must be
non-fatal — the fallback must still load routes, matching the main loop
behaviour which only emits a plugin-error but does not abort.
Previously the fallback block did ``if not _install_requirements(...): continue``
which skipped the user copy entirely, leaving the bundled-failure error in
startup-status even when the user copy could have run without those deps.
(Thread 1, review-4227643820)
A plugin-error must also be emitted when requirements fail, mirroring the
main loop — without it the later clear_error=True makes startup look healthy
even though the running plugin may be missing dependencies.
(Thread 2, review-4228077246)
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('bundled broken')\n"
)
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n app.state.fallback_ran = True\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
# Simulate _install_requirements failing (read-only filesystem, optional
# dep, etc.) — same scenario the main loop tolerates.
monkeypatch.setattr(plugins, "_install_requirements", lambda *a, **kw: False)
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
events: list = []
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {}, progress_cb=events.append)
finally:
plugins.PLUGINS_DIR = saved_dir
# Despite requirements failure the fallback routes.setup() must have run.
assert getattr(fake_app.state, "fallback_ran", False) is True, (
"Fallback routes.setup() must run even when _install_requirements returns False"
)
entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert entries, "highway_3d must be registered as a fallback"
assert entries[0].get("fallback") is True
# A plugin-error must have been emitted for the fallback requirements failure.
req_errors = [
e for e in events
if e.get("phase") == "plugin-error"
and e.get("plugin_id") == "highway_3d"
and e.get("error")
]
assert req_errors, (
"Expected at least one plugin-error event for highway_3d fallback requirements failure; "
f"events were: {events}"
)
# The req-failure event must include loaded/total so startup-status
# progress counters are not zeroed out.
# (Thread 2, review-4228421486)
for e in req_errors:
assert "loaded" in e and "total" in e, (
f"plugin-error event for req failure must include loaded/total; got: {e}"
)
# After successful route registration the plugin-registered event must
# NOT carry error=None when req install failed; that would wipe the req
# error from startup-status, making startup look clean despite degraded
# dependencies. (Thread 1, review-4228421486)
registered_clear_events = [
e for e in events
if e.get("phase") == "plugin-registered"
and e.get("plugin_id") == "highway_3d"
and "error" in e
and e["error"] is None
]
assert not registered_clear_events, (
"plugin-registered must NOT carry error=None when req install failed "
"(that would silently wipe the req-failure error from startup-status); "
f"events were: {events}"
)
def test_fallback_routes_failure_emits_plugin_error(
tmp_path, reset_plugin_state, monkeypatch
):
"""When the fallback user-copy's routes also fail, a plugin-error event
must be emitted so startup-status reflects the fallback's failure as the
root cause.
Without this, startup-status is left pointing at the earlier bundled-copy
error even though that is no longer the active failure, making it harder
for operators to identify the correct root cause.
(Thread 1, review-4228077246)
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bundled_plugin_dir = bundled_dir / "highway_3d"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
(bundled_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('bundled broken')\n"
)
user_dir = tmp_path / "user"
user_dir.mkdir()
user_plugin_dir = user_dir / "highway_3d"
user_plugin_dir.mkdir()
(user_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_plugin_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('fallback also broken')\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
events: list = []
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {}, progress_cb=events.append)
finally:
plugins.PLUGINS_DIR = saved_dir
# The plugin must NOT be registered.
entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "highway_3d"]
assert not entries, "highway_3d must not be registered when both copies fail"
# A plugin-error event must have been emitted for the fallback failure.
fallback_errors = [
e for e in events
if e.get("phase") == "plugin-error"
and e.get("plugin_id") == "highway_3d"
and e.get("error")
]
assert fallback_errors, (
"Expected at least one plugin-error event when fallback routes also fail; "
f"events were: {events}"
)
# The error message must mention 'Both' or 'fallback' to distinguish it
# from the original bundled-failure error.
assert any(
"fallback" in e["error"].lower() or "both" in e["error"].lower()
for e in fallback_errors
), f"plugin-error text should reference the fallback failure; got: {[e['error'] for e in fallback_errors]}"
# The fallback route-error event must include loaded/total so the
# startup-status progress counters are not zeroed mid-run.
# (Thread 3, review-4228421486)
for e in fallback_errors:
assert "loaded" in e and "total" in e, (
f"plugin-error event for fallback route failure must include loaded/total; got: {e}"
)
def test_normal_plugins_do_not_have_fallback_field_set(
tmp_path, reset_plugin_state, monkeypatch
):
"""Normal (non-fallback) plugins — both bundled and user-installed —
must not have ``fallback: True`` set. Only emergency user-copy
fallbacks that replaced a broken bundled plugin carry the flag.
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
bundled_plugin_dir = bundled_dir / "my_plugin"
bundled_plugin_dir.mkdir()
(bundled_plugin_dir / "plugin.json").write_text(
json.dumps({"id": "my_plugin", "name": "My Plugin", "bundled": True})
)
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
entries = [p for p in plugins.LOADED_PLUGINS if p["id"] == "my_plugin"]
assert len(entries) == 1
assert not entries[0].get("fallback"), (
"Normal bundled plugin must not have fallback=True"
)
def test_sibling_purge_does_not_affect_other_plugin_with_same_prefix(
tmp_path, reset_plugin_state, monkeypatch
):
"""The sibling module purge in the fallback block must only remove the
exact routes module key for the evicted plugin and its namespaced
load_sibling modules. It must NOT delete modules for other plugins
whose namespaced IDs share the same prefix.
Previously, ``k.startswith(f"{_parent_pkg}_")`` would match
``plugin_a_routes`` when purging plugin ``a``, and would also
incorrectly match ``plugin_a_5f_b_routes`` (routes for plugin ``a_b``).
The fix replaces the startswith check with an exact key match.
(Thread 1, review-4226937699)
"""
plugins = reset_plugin_state
import sys
# plugin "a" — bundled, fails routes.
# plugin "a_b" — also bundled, healthy.
# user copy of "a" — provides the fallback.
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
ba_dir = bundled_dir / "a"
ba_dir.mkdir()
(ba_dir / "plugin.json").write_text(
json.dumps({"id": "a", "name": "Plugin A", "routes": "routes.py", "bundled": True})
)
(ba_dir / "routes.py").write_text(
"def setup(app, ctx):\n raise RuntimeError('a broken')\n"
)
ab_dir = bundled_dir / "a_b"
ab_dir.mkdir()
# `a_b` routes capture a ref so we can assert they ran from the right module.
(ab_dir / "plugin.json").write_text(
json.dumps({"id": "a_b", "name": "Plugin A_B", "routes": "routes.py", "bundled": True})
)
(ab_dir / "routes.py").write_text(
"def setup(app, ctx):\n app.state.ab_setup_ran = True\n"
)
user_dir = tmp_path / "user"
user_dir.mkdir()
ua_dir = user_dir / "a"
ua_dir.mkdir()
(ua_dir / "plugin.json").write_text(
json.dumps({"id": "a", "name": "Plugin A (user)", "routes": "routes.py"})
)
(ua_dir / "routes.py").write_text(
"def setup(app, ctx):\n app.state.a_fallback_ran = True\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# plugin "a_b" routes should have run without being clobbered by
# the prefix-based purge.
assert getattr(fake_app.state, "ab_setup_ran", False) is True, (
"plugin 'a_b' setup() was not called — its module was incorrectly "
"purged during the 'a' fallback sibling purge"
)
# plugin "a" fallback also ran.
assert getattr(fake_app.state, "a_fallback_ran", False) is True, (
"plugin 'a' fallback setup() did not run"
)
def test_bundled_flag_requires_both_in_tree_directory_and_manifest_field(
tmp_path, reset_plugin_state, monkeypatch
):
"""``bundled`` requires ALL THREE: in-tree PLUGINS_DIR location, the
manifest's ``"bundled": true`` field, AND the directory name matching the
plugin id.
- In-tree plugin, dir name == id, ``"bundled": true`` → ``bundled: true``.
- In-tree plugin, dir name == id, no ``"bundled"`` field → ``bundled: false``
(user-installed plugin cloned into plugins/).
- In-tree plugin, dir name ≠ id, ``"bundled": true`` → ``bundled: false``
(verbatim user copy under a different folder name).
- User-dir plugin, ``"bundled": true`` → ``bundled: false``
(not in-tree; manifest field alone is not sufficient).
This ensures a user-installed plugin cannot forge core status by adding
``"bundled": true`` to its own plugin.json, while correctly identifying
legitimate in-tree bundled plugins.
"""
plugins = reset_plugin_state
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
# In-tree plugin WITH "bundled": true AND dir name == id — the real bundled core plugin.
(bundled_dir / "real_core").mkdir()
(bundled_dir / "real_core" / "plugin.json").write_text(
json.dumps({"id": "real_core", "name": "Real Core Plugin", "bundled": True})
)
# In-tree plugin WITHOUT "bundled" field — a user plugin cloned into plugins/.
(bundled_dir / "user_in_tree").mkdir()
(bundled_dir / "user_in_tree" / "plugin.json").write_text(
json.dumps({"id": "user_in_tree", "name": "User Plugin (in plugins/)"})
)
# In-tree plugin WITH "bundled": true but dir name ≠ id — verbatim user copy
# of a bundled plugin placed under a different folder name.
(bundled_dir / "other_name").mkdir()
(bundled_dir / "other_name" / "plugin.json").write_text(
json.dumps({"id": "real_core_copy", "name": "Verbatim Copy", "bundled": True})
)
user_dir = tmp_path / "user"
user_dir.mkdir()
# User plugin in FEEDBACK_PLUGINS_DIR that forges "bundled": true.
(user_dir / "fake_bundled").mkdir()
(user_dir / "fake_bundled" / "plugin.json").write_text(
json.dumps({"id": "fake_bundled", "name": "Fake Bundled", "bundled": True})
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
by_id = {p["id"]: p for p in plugins.LOADED_PLUGINS}
# In-tree plugin with "bundled": true AND dir name == id → bundled=True.
assert by_id["real_core"]["bundled"] is True
# In-tree plugin without "bundled" field → bundled=False (user clone in plugins/).
assert by_id["user_in_tree"]["bundled"] is False
# In-tree plugin with "bundled": true but dir name ≠ id → bundled=False.
# (verbatim copy under a different folder; dir name "other_name" ≠ id "real_core_copy")
assert by_id["real_core_copy"]["bundled"] is False
# Plugin from user dir with "bundled": true → bundled=False (manifest field alone insufficient).
assert by_id["fake_bundled"]["bundled"] is False
def test_bundled_returned_by_api(tmp_path, reset_plugin_state):
"""GET /api/plugins must include a ``bundled`` boolean field for each plugin.
- A plugin loaded from a manifest with ``"bundled": true`` (in-tree, dir name
matches id) must surface ``bundled: true``.
- A plain user plugin must surface ``bundled: false``.
"""
plugins_mod = reset_plugin_state
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
# Bundled plugin entry.
plugins_mod.LOADED_PLUGINS.append({
"id": "core_viz",
"name": "Core Viz",
"nav": None,
"type": None,
"bundled": True,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False,
"_dir": plugin_dir,
"_manifest": {},
})
# Plain user plugin.
plugins_mod.LOADED_PLUGINS.append({
"id": "my_plugin",
"name": "My Plugin",
"nav": None,
"type": None,
"bundled": False,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False,
"_dir": plugin_dir,
"_manifest": {},
})
client = _make_api_client(plugins_mod)
try:
r = client.get("/api/plugins")
assert r.status_code == 200
ids = {p["id"]: p for p in r.json()}
assert ids["core_viz"]["bundled"] is True
assert "overrides_bundled" not in ids["core_viz"]
assert ids["my_plugin"]["bundled"] is False
assert "overrides_bundled" not in ids["my_plugin"]
finally:
client.close()
def test_fallback_field_is_surfaced_in_api_response(tmp_path, reset_plugin_state):
"""GET /api/plugins must include a ``fallback`` boolean field for each plugin.
- A fallback user-copy entry (bundled routes failed) must surface ``fallback: true``.
- A normal plugin must surface ``fallback: false``.
This ensures the frontend badge rendering is reliable — the badge depends
on this field being present in the JSON response from the endpoint, not
just in the internal LOADED_PLUGINS list (Thread 6, review-4227201020).
"""
plugins_mod = reset_plugin_state
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
# Fallback user-copy entry (bundled routes failed, user copy active).
plugins_mod.LOADED_PLUGINS.append({
"id": "highway_3d",
"name": "3D Highway (user fallback)",
"nav": None,
"type": None,
"bundled": False,
"fallback": True,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False,
"_dir": plugin_dir,
"_manifest": {},
})
# Normal user plugin — must NOT have fallback=true.
plugins_mod.LOADED_PLUGINS.append({
"id": "normal_plugin",
"name": "Normal Plugin",
"nav": None,
"type": None,
"bundled": False,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False,
"_dir": plugin_dir,
"_manifest": {},
})
client = _make_api_client(plugins_mod)
try:
r = client.get("/api/plugins")
assert r.status_code == 200
ids = {p["id"]: p for p in r.json()}
# Fallback entry must have fallback=true in the response.
assert "fallback" in ids["highway_3d"], (
"/api/plugins response must include the 'fallback' key"
)
assert ids["highway_3d"]["fallback"] is True
# Normal plugin must have fallback=false in the response.
assert "fallback" in ids["normal_plugin"], (
"/api/plugins response must include the 'fallback' key for all plugins"
)
assert ids["normal_plugin"]["fallback"] is False
finally:
client.close()
def test_api_plugins_exposes_capability_validation_and_shim_metadata(reset_plugin_state):
plugins_mod = reset_plugin_state
fixture_dir = Path(__file__).resolve().parents[1] / "tests" / "fixtures" / "plugin_capabilities"
valid = json.loads((fixture_dir / "valid_owner_provider.json").read_text(encoding="utf-8"))
invalid = json.loads((fixture_dir / "invalid_capability_metadata.json").read_text(encoding="utf-8"))
unsupported = json.loads((fixture_dir / "unsupported_capability_version.json").read_text(encoding="utf-8"))
valid_caps, _, _ = plugins_mod._capability_warnings(valid, "stems")
invalid_caps, invalid_warnings, _ = plugins_mod._capability_warnings(invalid, "broken_capability_metadata")
unsupported_caps, _, unsupported_versions = plugins_mod._capability_warnings(unsupported, "future_capabilities")
plugins_mod.LOADED_PLUGINS.extend([
{
"id": "stems", "name": "Stems", "nav": None, "type": None,
"bundled": False, "has_screen": False, "has_script": False,
"has_settings": False, "has_tour": False, "standards": valid["standards"],
"capabilities": valid_caps, "settings_schema": {},
"ui_contributions": plugins_mod._normalize_ui_contributions(valid),
"runtime_domains": plugins_mod._normalize_runtime_domains(valid),
"compatibility_shims": plugins_mod._compatibility_shims_from_manifest(valid, "stems"),
"capability_validation_warnings": [], "capability_unsupported_versions": [],
"_manifest": valid,
},
{
"id": "broken_capability_metadata", "name": "Broken Capability Metadata",
"nav": invalid["nav"], "type": None, "bundled": False,
"has_screen": True, "has_script": False, "has_settings": True,
"has_tour": False, "standards": invalid["standards"],
"capabilities": invalid_caps, "settings_schema": {},
"ui_contributions": plugins_mod._normalize_ui_contributions(invalid),
"runtime_domains": plugins_mod._normalize_runtime_domains(invalid),
"compatibility_shims": plugins_mod._compatibility_shims_from_manifest(invalid, "broken_capability_metadata"),
"capability_validation_warnings": invalid_warnings,
"capability_unsupported_versions": [], "_manifest": invalid,
},
{
"id": "future_capabilities", "name": "Future Capabilities",
"nav": None, "type": None, "bundled": False, "has_screen": False,
"has_script": False, "has_settings": False, "has_tour": False,
"standards": unsupported["standards"], "capabilities": unsupported_caps,
"settings_schema": {}, "ui_contributions": plugins_mod._normalize_ui_contributions(unsupported),
"runtime_domains": plugins_mod._normalize_runtime_domains(unsupported),
"compatibility_shims": plugins_mod._compatibility_shims_from_manifest(unsupported, "future_capabilities"),
"capability_validation_warnings": [],
"capability_unsupported_versions": unsupported_versions,
"_manifest": unsupported,
},
])
client = _make_api_client(plugins_mod)
try:
listing = {entry["id"]: entry for entry in client.get("/api/plugins").json()}
finally:
client.close()
assert listing["stems"]["capabilities"]["stems"]["roles"] == ["owner", "provider"]
assert listing["broken_capability_metadata"]["capabilities"] == {}
assert listing["broken_capability_metadata"]["capability_validation_warnings"]
assert listing["broken_capability_metadata"]["ui_contributions"]["legacy"]
assert listing["broken_capability_metadata"]["compatibility_shims"] == []
assert listing["future_capabilities"]["capability_unsupported_versions"]
assert listing["future_capabilities"]["capabilities"]["stems"]["incompatible"] is True
def _run_load_plugins_with_cb(plugins, app, tmp_path, progress_cb, context=None):
"""Like _run_load_plugins but passes a progress_cb spy."""
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = tmp_path
try:
plugins.load_plugins(app, context if context is not None else {}, progress_cb=progress_cb)
finally:
plugins.PLUGINS_DIR = saved_dir
def test_progress_cb_receives_phases_and_counts(tmp_path, reset_plugin_state):
"""progress_cb spy should receive structured events including
plugins-discovered, plugins-complete, loaded/total counters, and
the complete event should have loaded == total == number of plugins."""
plugins = reset_plugin_state
_make_plugin(tmp_path, "alpha")
_make_plugin(tmp_path, "beta")
events = []
_run_load_plugins_with_cb(plugins, type("FakeApp", (), {})(), tmp_path, events.append)
phases = [e["phase"] for e in events]
assert "plugins-discovered" in phases
assert "plugins-complete" in phases
# All events carry int counters.
for e in events:
assert isinstance(e["loaded"], int), f"loaded not int in {e}"
assert isinstance(e["total"], int), f"total not int in {e}"
# The complete event has loaded == total == 2.
complete = next(e for e in events if e["phase"] == "plugins-complete")
assert complete["loaded"] == 2
assert complete["total"] == 2
def test_progress_cb_errors_do_not_break_plugin_startup(tmp_path, reset_plugin_state):
"""A progress_cb that raises must not abort plugin loading — the
_emit_progress guard must swallow the exception."""
plugins = reset_plugin_state
_make_plugin(tmp_path, "safe")
def bad_cb(event):
raise RuntimeError("spy exploded")
_run_load_plugins_with_cb(plugins, type("FakeApp", (), {})(), tmp_path, bad_cb)
# Plugin still registered despite cb explosion.
assert any(p["id"] == "safe" for p in plugins.LOADED_PLUGINS)
def test_progress_cb_emits_plugin_error_on_requirements_failure(tmp_path, reset_plugin_state, monkeypatch):
"""When _install_requirements returns False, a plugin-error event
with a non-empty error field must be emitted via progress_cb, and
plugin loading must still continue (non-fatal)."""
plugins = reset_plugin_state
_make_plugin(tmp_path, "req_fail")
# Force _install_requirements to report failure without actually running pip.
monkeypatch.setattr(plugins, "_install_requirements", lambda *a, **kw: False)
events = []
_run_load_plugins_with_cb(plugins, type("FakeApp", (), {})(), tmp_path, events.append)
error_events = [e for e in events if e["phase"] == "plugin-error" and e.get("plugin_id") == "req_fail"]
assert error_events, "Expected at least one plugin-error event for req_fail"
assert all(e["error"] for e in error_events), "plugin-error events must carry a non-empty error field"
# Loading continued: req_fail is still registered.
assert any(p["id"] == "req_fail" for p in plugins.LOADED_PLUGINS)
# ── Async loading: pending registry, incremental publish, failed status ─────────
def test_pending_then_incremental_publish_observed_during_setup(tmp_path, reset_plugin_state):
"""Every discovered plugin is recorded in PENDING_PLUGINS as "installing"
up front, then GRADUATES into LOADED_PLUGINS as it becomes ready — one at a
time, not all-at-once at the end (issue #421 incremental publish).
Each plugin's setup() records the live LOADED/PENDING state under its own
plugin id (read from ctx['log'].name):
* During "aaa".setup(): both plugins are still pending+installing and
nothing has graduated yet (LOADED is empty).
* During "zzz".setup(): "aaa" has ALREADY graduated into LOADED while
"zzz" is still pending — proving plugins publish incrementally.
"""
plugins = reset_plugin_state
body = (
"import plugins\n"
"def setup(app, ctx):\n"
" pid = ctx['log'].name.rsplit('.', 1)[-1]\n"
" app.state.snaps = getattr(app.state, 'snaps', {})\n"
" app.state.snaps[pid] = {\n"
" 'pending': {k: v['status'] for k, v in plugins.PENDING_PLUGINS.items()},\n"
" 'loaded': [p['id'] for p in plugins.LOADED_PLUGINS],\n"
" }\n"
)
_make_plugin(tmp_path, "aaa", routes_body=body)
_make_plugin(tmp_path, "zzz", routes_body=body)
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
_run_load_plugins(plugins, fake_app, tmp_path)
snaps = fake_app.state.snaps
# During aaa.setup(): both pending+installing, nothing graduated yet.
assert snaps["aaa"]["pending"] == {"aaa": "installing", "zzz": "installing"}
assert snaps["aaa"]["loaded"] == []
# During zzz.setup(): aaa already graduated (incremental publish); zzz pending.
assert snaps["zzz"]["loaded"] == ["aaa"]
assert "zzz" in snaps["zzz"]["pending"]
assert "aaa" not in snaps["zzz"]["pending"]
# Final: both ready, none pending.
assert sorted(p["id"] for p in plugins.LOADED_PLUGINS) == ["aaa", "zzz"]
assert plugins.PENDING_PLUGINS == {}
def test_stale_load_pass_cannot_republish_after_newer_pass(tmp_path, reset_plugin_state):
"""A still-running load pass must NOT republish into the registries after a
NEWER load_plugins() pass has cleared them (re-entrancy race: a "reload
plugins" action, FEEDBACK_SYNC_STARTUP hot-reload, or test teardown firing
while the first pass's background install thread is mid-flight).
The loader bumps a generation token at the start of every pass; _graduate()
and _mark_failed() bail when the generation has advanced. Without the guard,
the older pass's _graduate() would re-insert a plugin the newer pass already
cleared, leaving a cross-generation / duplicate entry in LOADED_PLUGINS.
"""
import threading
plugins = reset_plugin_state
setup_entered = threading.Event()
release_setup = threading.Event()
# Stash the events on the module so the file-exec'd plugin setup() can reach
# them without globals plumbing. Cleaned up at the end of the test.
plugins._TEST_setup_entered = setup_entered
plugins._TEST_release_setup = release_setup
try:
root1 = tmp_path / "gen1"
root2 = tmp_path / "gen2"
root1.mkdir()
root2.mkdir()
# Pass 1's plugin blocks inside setup() — past discovery and the PENDING
# seed, just before it would graduate — until the test releases it.
_make_plugin(
root1, "slow",
routes_body=(
"import plugins\n"
"def setup(app, ctx):\n"
" plugins._TEST_setup_entered.set()\n"
" plugins._TEST_release_setup.wait(5)\n"
),
)
_make_plugin(root2, "fast") # no-op setup, graduates immediately
errors = []
def _pass1():
saved = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = root1
try:
plugins.load_plugins(type("FakeApp", (), {})(), {})
except Exception as e: # pragma: no cover - failure path
errors.append(e)
finally:
plugins.PLUGINS_DIR = saved
t1 = threading.Thread(target=_pass1)
t1.start()
# Wait until pass 1 is parked inside slow.setup() (discovery done,
# PENDING seeded, about to eventually graduate "slow").
assert setup_entered.wait(5), "pass 1 never entered slow.setup()"
# Pass 2 runs to completion on the main thread: bumps the generation,
# clears both registries, graduates "fast".
saved = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = root2
try:
plugins.load_plugins(type("FakeApp", (), {})(), {})
finally:
plugins.PLUGINS_DIR = saved
# Release pass 1; it resumes and attempts to graduate "slow" — now stale.
release_setup.set()
t1.join(5)
assert not t1.is_alive(), "pass 1 thread did not finish"
assert not errors, f"pass 1 raised: {errors}"
# Only pass 2's output survives. The stale pass's "slow" must NOT appear
# in either registry — its _graduate() was a no-op.
assert [p["id"] for p in plugins.LOADED_PLUGINS] == ["fast"]
assert "slow" not in plugins.PENDING_PLUGINS
assert "slow" not in {p["id"] for p in plugins.LOADED_PLUGINS}
finally:
delattr(plugins, "_TEST_setup_entered")
delattr(plugins, "_TEST_release_setup")
def test_failed_plugin_shows_failed_status_and_stays_visible(tmp_path, reset_plugin_state):
"""A plugin whose routes fail to load is NOT added to LOADED_PLUGINS
(ready-only) but STAYS visible as a disabled "failed" entry in
PENDING_PLUGINS and on /api/plugins, with its error text — so the nav can
render it disabled with a tooltip rather than dropping it (ADR 0001)."""
plugins = reset_plugin_state
_make_plugin(tmp_path, "okp") # healthy, no-op setup
_make_plugin(
tmp_path, "broken",
routes_body="def setup(app, ctx):\n raise RuntimeError('boom')\n",
)
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
# LOADED_PLUGINS stays ready-only: broken absent, okp present.
loaded_ids = {p["id"] for p in plugins.LOADED_PLUGINS}
assert "broken" not in loaded_ids
assert "okp" in loaded_ids
# broken remains a visible "failed" pending entry with error text.
assert plugins.PENDING_PLUGINS["broken"]["status"] == "failed"
assert plugins.PENDING_PLUGINS["broken"]["error"]
assert "okp" not in plugins.PENDING_PLUGINS # graduated, no longer pending
# /api/plugins exposes the union with per-plugin status + error.
client = _make_api_client(plugins)
try:
rows = {p["id"]: p for p in client.get("/api/plugins").json()}
assert rows["broken"]["status"] == "failed"
assert rows["broken"]["error"]
assert rows["okp"]["status"] == "ready"
assert rows["okp"]["error"] is None
finally:
client.close()
def test_api_plugins_surfaces_installing_pending_entry(tmp_path, reset_plugin_state):
"""An "installing" pending entry surfaces on /api/plugins immediately with
its manifest-derived nav fields and status="installing" (no error), so the
frontend can render a disabled "installing…" nav slot before the (possibly
very slow) dependency install finishes."""
plugins = reset_plugin_state
pdir = tmp_path / "heavy"
pdir.mkdir()
plugins.PENDING_PLUGINS["heavy"] = {
"id": "heavy", "name": "Heavy ML Plugin", "nav": "/heavy",
"type": "visualization", "bundled": True, "version": "2.0.0",
"has_screen": True, "has_script": False, "has_settings": False,
"has_tour": False, "_order": 0, "status": "installing", "error": None,
}
client = _make_api_client(plugins)
try:
rows = {p["id"]: p for p in client.get("/api/plugins").json()}
h = rows["heavy"]
assert h["status"] == "installing"
assert h["error"] is None
assert h["nav"] == "/heavy"
assert h["type"] == "visualization"
assert h["bundled"] is True
assert h["version"] == "2.0.0"
assert h["has_screen"] is True
assert h["fallback"] is False
finally:
client.close()
def test_api_plugins_union_sorted_by_discovery_order(tmp_path, reset_plugin_state):
"""The /api/plugins union is re-sorted by discovery order so an installing
plugin occupies its eventual nav slot regardless of whether it currently
lives in LOADED_PLUGINS (ready) or PENDING_PLUGINS (installing/failed)."""
plugins = reset_plugin_state
pdir = tmp_path / "d"
pdir.mkdir()
# Ready plugin discovered SECOND (order 1) lives in LOADED_PLUGINS.
plugins.LOADED_PLUGINS.append({
"id": "ready_second", "name": "Ready", "nav": None, "type": None,
"bundled": False, "has_screen": False, "has_script": False,
"has_settings": False, "has_tour": False, "status": "ready",
"_order": 1, "_dir": pdir, "_manifest": {},
})
# Installing plugin discovered FIRST (order 0) lives in PENDING_PLUGINS.
plugins.PENDING_PLUGINS["installing_first"] = {
"id": "installing_first", "name": "Installing", "nav": None, "type": None,
"bundled": False, "version": None, "has_screen": False, "has_script": False,
"has_settings": False, "has_tour": False, "_order": 0,
"status": "installing", "error": None,
}
client = _make_api_client(plugins)
try:
order = [p["id"] for p in client.get("/api/plugins").json()]
assert order == ["installing_first", "ready_second"]
finally:
client.close()
def test_pending_plugin_assets_not_served(tmp_path, reset_plugin_state):
"""Asset endpoints serve only ready plugins. A plugin still installing
(present in PENDING_PLUGINS, absent from LOADED_PLUGINS) returns 404 for
its screen.js / screen.html even though its manifest advertises them."""
plugins = reset_plugin_state
pdir = tmp_path / "heavy"
pdir.mkdir()
(pdir / "screen.js").write_text("console.log('x');\n")
(pdir / "screen.html").write_text("<p>hi</p>\n")
plugins.PENDING_PLUGINS["heavy"] = {
"id": "heavy", "name": "Heavy", "nav": None, "type": None,
"bundled": False, "version": None, "has_screen": True, "has_script": True,
"has_settings": False, "has_tour": False, "_order": 0,
"status": "installing", "error": None, "_dir": pdir, "_manifest": {},
}
client = _make_api_client(plugins)
try:
assert client.get("/api/plugins/heavy/screen.js").status_code == 404
assert client.get("/api/plugins/heavy/screen.html").status_code == 404
finally:
client.close()
# ── Tour API tests ─────────────────────────────────────────────────────────────
def _make_tour_plugin(plugin_root, plugin_id, *, tour_file_name="tour.json", tour_content=None):
"""Create a minimal plugin directory with a tour field in plugin.json."""
plugin_dir = plugin_root / plugin_id
plugin_dir.mkdir(parents=True)
manifest = {"id": plugin_id, "name": plugin_id, "tour": tour_file_name}
(plugin_dir / "plugin.json").write_text(json.dumps(manifest))
if tour_content is not None:
(plugin_dir / tour_file_name).write_text(json.dumps(tour_content))
return plugin_dir
def _make_api_client(plugins):
"""Register the plugin API on a fresh FastAPI app and return a TestClient."""
app = FastAPI()
plugins.register_plugin_api(app)
return TestClient(app)
@pytest.fixture()
def tour_client(tmp_path, reset_plugin_state):
"""A TestClient with two plugins: one with a tour, one without."""
plugins = reset_plugin_state
# Plugin with tour
tour_content = {"tour": [{"id": "step1", "title": "Hello", "content": "World"}]}
plugin_dir = _make_tour_plugin(tmp_path, "with_tour", tour_content=tour_content)
# Plugin without tour
no_tour_dir = tmp_path / "no_tour"
no_tour_dir.mkdir()
(no_tour_dir / "plugin.json").write_text(json.dumps({"id": "no_tour", "name": "No Tour"}))
# Stub LOADED_PLUGINS with both plugins
plugins.LOADED_PLUGINS.append({
"id": "with_tour",
"name": "With Tour",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": True,
"_dir": plugin_dir,
"_manifest": {"tour": "tour.json"},
})
plugins.LOADED_PLUGINS.append({
"id": "no_tour",
"name": "No Tour",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False,
"_dir": no_tour_dir,
"_manifest": {},
})
client = _make_api_client(plugins)
try:
yield client, plugin_dir
finally:
client.close()
def test_list_plugins_includes_has_tour(tour_client):
"""GET /api/plugins must include a has_tour boolean for each plugin."""
client, _ = tour_client
r = client.get("/api/plugins")
assert r.status_code == 200
plugins_list = r.json()
ids = {p["id"]: p for p in plugins_list}
assert "has_tour" in ids["with_tour"]
assert ids["with_tour"]["has_tour"] is True
assert "has_tour" in ids["no_tour"]
assert ids["no_tour"]["has_tour"] is False
def test_nav_entry_derives_has_styles_from_manifest(tmp_path, reset_plugin_state):
"""The real loader must derive has_styles + carry the styles path from the
manifest WITHOUT importing plugin code (mirrors has_screen/has_tour)."""
plugins = reset_plugin_state
styled = _make_plugin(tmp_path, "styled")
(styled / "plugin.json").write_text(json.dumps(
{"id": "styled", "name": "Styled", "routes": "routes.py",
"styles": "assets/plugin.css"}))
_make_plugin(tmp_path, "plain") # no styles key
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
by_id = {p["id"]: p for p in plugins.LOADED_PLUGINS}
assert by_id["styled"]["has_styles"] is True
assert by_id["styled"]["styles"] == "assets/plugin.css"
assert by_id["plain"]["has_styles"] is False
assert by_id["plain"]["styles"] is None
def test_list_plugins_surfaces_has_styles_and_path(tmp_path, reset_plugin_state):
"""GET /api/plugins must surface has_styles + styles for each plugin so the
frontend can build the <link> URL; a plugin without styles → False/None."""
plugins = reset_plugin_state
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "with_styles", "name": "With Styles", "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "has_styles": True, "styles": "assets/plugin.css",
"_dir": plugin_dir, "_manifest": {},
})
plugins.LOADED_PLUGINS.append({
"id": "no_styles", "name": "No Styles", "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "_dir": plugin_dir, "_manifest": {},
})
client = _make_api_client(plugins)
try:
r = client.get("/api/plugins")
assert r.status_code == 200
ids = {p["id"]: p for p in r.json()}
assert ids["with_styles"]["has_styles"] is True
assert ids["with_styles"]["styles"] == "assets/plugin.css"
# A stubbed entry without the keys falls back cleanly (no KeyError).
assert ids["no_styles"]["has_styles"] is False
assert ids["no_styles"]["styles"] is None
finally:
client.close()
def test_nav_entry_surfaces_pedalboard_metadata(tmp_path, reset_plugin_state):
"""The loader must carry description/category/icon from the manifest WITHOUT
importing plugin code, and auto-detect assets/thumb.png when `icon` is
omitted (v3 Pedalboard page)."""
plugins = reset_plugin_state
full = _make_plugin(tmp_path, "full")
(full / "plugin.json").write_text(json.dumps(
{"id": "full", "name": "Full", "routes": "routes.py",
"description": "A short blurb.", "category": "audio",
"icon": "assets/custom.png"}))
# No `icon` in the manifest, but ships assets/thumb.png → auto-detected.
probed = _make_plugin(tmp_path, "probed")
(probed / "plugin.json").write_text(json.dumps(
{"id": "probed", "name": "Probed", "routes": "routes.py"}))
(probed / "assets").mkdir(exist_ok=True)
(probed / "assets" / "thumb.png").write_bytes(b"\x89PNG\r\n")
_make_plugin(tmp_path, "bare") # no pedalboard fields, no thumb
_run_load_plugins(plugins, type("FakeApp", (), {})(), tmp_path)
by_id = {p["id"]: p for p in plugins.LOADED_PLUGINS}
assert by_id["full"]["description"] == "A short blurb."
assert by_id["full"]["category"] == "audio"
assert by_id["full"]["icon"] == "assets/custom.png"
# Auto-detected convention thumbnail.
assert by_id["probed"]["icon"] == "assets/thumb.png"
assert by_id["probed"]["description"] is None
assert by_id["probed"]["category"] is None
# Nothing declared and nothing on disk → all None (no default injected here;
# the frontend supplies the default pedal graphic).
assert by_id["bare"]["icon"] is None
assert by_id["bare"]["description"] is None
assert by_id["bare"]["category"] is None
def test_list_plugins_surfaces_pedalboard_metadata(tmp_path, reset_plugin_state):
"""GET /api/plugins must surface description/category/icon; a stubbed entry
missing the keys falls back to None without raising."""
plugins = reset_plugin_state
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "decorated", "name": "Decorated", "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "description": "Blurb", "category": "tools",
"icon": "assets/thumb.png", "_dir": plugin_dir, "_manifest": {},
})
plugins.LOADED_PLUGINS.append({
"id": "plain", "name": "Plain", "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "_dir": plugin_dir, "_manifest": {},
})
client = _make_api_client(plugins)
try:
r = client.get("/api/plugins")
assert r.status_code == 200
ids = {p["id"]: p for p in r.json()}
assert ids["decorated"]["description"] == "Blurb"
assert ids["decorated"]["category"] == "tools"
assert ids["decorated"]["icon"] == "assets/thumb.png"
assert ids["plain"]["description"] is None
assert ids["plain"]["category"] is None
assert ids["plain"]["icon"] is None
finally:
client.close()
def test_list_plugins_version_field(tmp_path, reset_plugin_state):
"""GET /api/plugins must include a `version` field for each plugin.
A plugin that declares ``version`` in its manifest must surface it;
a plugin that omits the field must return ``None`` (not raise).
"""
plugins_mod = reset_plugin_state
versioned_dir = tmp_path / "versioned"
versioned_dir.mkdir()
(versioned_dir / "plugin.json").write_text(
json.dumps({"id": "versioned", "name": "Versioned", "version": "1.2.3"})
)
unversioned_dir = tmp_path / "unversioned"
unversioned_dir.mkdir()
(unversioned_dir / "plugin.json").write_text(
json.dumps({"id": "unversioned", "name": "Unversioned"})
)
plugins_mod.LOADED_PLUGINS.append({
"id": "versioned",
"name": "Versioned",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False,
"_dir": versioned_dir,
"_manifest": {"version": "1.2.3"},
})
plugins_mod.LOADED_PLUGINS.append({
"id": "unversioned",
"name": "Unversioned",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False,
"_dir": unversioned_dir,
"_manifest": {},
})
client = _make_api_client(plugins_mod)
try:
r = client.get("/api/plugins")
assert r.status_code == 200
plugins_list = r.json()
ids = {p["id"]: p for p in plugins_list}
assert "version" in ids["versioned"], "version key must be present"
assert ids["versioned"]["version"] == "1.2.3"
assert "version" in ids["unversioned"], "version key must be present even when absent from manifest"
assert ids["unversioned"]["version"] is None
finally:
client.close()
def test_list_plugins_exposes_profile_domain_manifest_metadata(tmp_path, reset_plugin_state):
plugins_mod = reset_plugin_state
plugin_dir = tmp_path / "declared"
plugin_dir.mkdir()
manifest = {
"id": "declared",
"name": "Declared",
"nav": "Declared",
"screen": "screen.html",
"settings": "settings.html",
"type": "visualization",
"routes": "routes.py",
"standards": ["capability-pipelines.v1", "profiles.v1", 42, ""],
"capabilities": {"profiles": {"roles": ["observer"]}},
"settings_schema": {"schema_version": "1", "packable_keys": ["gain"]},
"ui_contributions": {"ui.player-panels": [{"id": "declared-panel"}]},
"ui": {"ui.player-controls": [{"id": "declared-control"}]},
"runtime_domains": {"midi-control": {"role": "observer"}, "custom-domain": {"role": "owner"}},
"domains": {"jobs": [{"id": "declared-job"}], "tempo-clock": {"role": "observer"}},
}
plugins_mod.LOADED_PLUGINS.append({
"id": "declared",
"name": "Declared",
"nav": "Declared",
"type": "visualization",
"has_screen": True,
"has_script": False,
"has_settings": True,
"has_tour": False,
"_dir": plugin_dir,
"_manifest": manifest,
})
client = _make_api_client(plugins_mod)
try:
response = client.get("/api/plugins")
assert response.status_code == 200
plugin = response.json()[0]
assert plugin["standards"] == ["capability-pipelines.v1", "profiles.v1"]
assert plugin["capabilities"] == {"profiles": {"roles": ["observer"]}}
assert plugin["settings_schema"] == {"schema_version": "1", "packable_keys": ["gain"]}
assert plugin["ui_contributions"]["declared"] == {
"ui.player-controls": [{"id": "declared-control"}],
"ui.player-panels": [{"id": "declared-panel"}],
}
assert {item["legacy_source"] for item in plugin["ui_contributions"]["legacy"]} == {"nav", "screen", "settings", "type"}
assert plugin["runtime_domains"] == {
"custom-domain": {"role": "owner"},
"jobs": [{"id": "declared-job"}],
"midi-control": {"role": "observer"},
"tempo-clock": {"role": "observer"},
}
finally:
client.close()
def test_tour_json_serves_file(tour_client):
"""GET /api/plugins/{id}/tour.json returns file content as JSON for a plugin with a tour."""
client, _ = tour_client
r = client.get("/api/plugins/with_tour/tour.json")
assert r.status_code == 200
assert r.headers["content-type"].startswith("application/json")
data = r.json()
assert "tour" in data
assert data["tour"][0]["id"] == "step1"
def test_tour_json_returns_404_for_missing_plugin(tour_client):
"""GET /api/plugins/{id}/tour.json returns 404 JSON for an unknown plugin."""
client, _ = tour_client
r = client.get("/api/plugins/nonexistent/tour.json")
assert r.status_code == 404
assert r.headers["content-type"].startswith("application/json")
def test_tour_json_returns_404_for_plugin_without_tour(tour_client):
"""GET /api/plugins/{id}/tour.json returns 404 for a plugin with no tour manifest entry."""
client, _ = tour_client
r = client.get("/api/plugins/no_tour/tour.json")
assert r.status_code == 404
assert r.headers["content-type"].startswith("application/json")
def test_tour_json_rejects_path_traversal(tmp_path, reset_plugin_state):
"""tour field with `../` path traversal must be rejected (returns 404)."""
plugins = reset_plugin_state
plugin_dir = tmp_path / "evil_plugin"
plugin_dir.mkdir()
# Write a 'secret' file outside the plugin dir that the traversal targets
(tmp_path / "secret.json").write_text(json.dumps({"secret": "data"}))
plugins.LOADED_PLUGINS.append({
"id": "evil_plugin",
"name": "Evil Plugin",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": True,
"_dir": plugin_dir,
"_manifest": {"tour": "../secret.json"},
})
app = FastAPI()
plugins.register_plugin_api(app)
client = TestClient(app)
try:
r = client.get("/api/plugins/evil_plugin/tour.json")
assert r.status_code == 404
finally:
client.close()
def test_tour_json_handles_non_string_tour_manifest(tmp_path, reset_plugin_state):
"""A plugin whose `tour` manifest field is truthy but not a string or dict
(e.g. ``true``, ``1``) must return 404 without raising AttributeError."""
plugins = reset_plugin_state
plugin_dir = tmp_path / "bool_tour_plugin"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "bool_tour_plugin",
"name": "Bool Tour Plugin",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False, # _is_valid_tour_manifest(True) → False
"_dir": plugin_dir,
"_manifest": {"tour": True}, # boolean true in manifest
})
app = FastAPI()
plugins.register_plugin_api(app)
client = TestClient(app)
try:
r = client.get("/api/plugins/bool_tour_plugin/tour.json")
assert r.status_code == 404
assert r.headers["content-type"].startswith("application/json")
finally:
client.close()
def test_tour_json_rejects_directory_path(tmp_path, reset_plugin_state):
"""A `tour` field of `"."` resolves to the plugin dir itself. The route
must return 404 (not IsADirectoryError) because is_file() gates read_text."""
plugins = reset_plugin_state
plugin_dir = tmp_path / "dot_tour_plugin"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "dot_tour_plugin",
"name": "Dot Tour Plugin",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": True,
"_dir": plugin_dir,
"_manifest": {"tour": "."},
})
app = FastAPI()
plugins.register_plugin_api(app)
client = TestClient(app)
try:
r = client.get("/api/plugins/dot_tour_plugin/tour.json")
assert r.status_code == 404
assert r.headers["content-type"].startswith("application/json")
finally:
client.close()
def test_tour_json_rejects_null_file_key(tmp_path, reset_plugin_state):
"""`{"tour": {"file": null}}` has an explicitly invalid `file` value.
_is_valid_tour_manifest must reject it (has_tour=False) and the route
must return 404 without an AttributeError."""
import plugins as _plugins_mod
# Directly verify the validation function rejects {"file": None}
assert _plugins_mod._is_valid_tour_manifest({"file": None}) is False
assert _plugins_mod._is_valid_tour_manifest({"file": ""}) is False
assert _plugins_mod._is_valid_tour_manifest({}) is True # bare dict defaults to tour.json
plugins = reset_plugin_state
plugin_dir = tmp_path / "null_file_plugin"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "null_file_plugin",
"name": "Null File Plugin",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False, # _is_valid_tour_manifest({"file": None}) → False
"_dir": plugin_dir,
"_manifest": {"tour": {"file": None}},
})
app = FastAPI()
plugins.register_plugin_api(app)
client = TestClient(app)
try:
# Confirm the listing endpoint reflects has_tour=False for this plugin
listing = client.get("/api/plugins").json()
p_entry = next(p for p in listing if p["id"] == "null_file_plugin")
assert p_entry["has_tour"] is False
# Route must also return 404 (defensive: route validates independently)
r = client.get("/api/plugins/null_file_plugin/tour.json")
assert r.status_code == 404
assert r.headers["content-type"].startswith("application/json")
finally:
client.close()
def test_install_requirements_marker_is_stable_across_calls(tmp_path, reset_plugin_state, monkeypatch):
"""The install marker must be deterministic so plugins don't reinstall
every boot. Regression for a bug where ``hash(req_file.read_text())``
returned a different integer per process (PYTHONHASHSEED randomisation),
so ``_install_requirements`` re-ran pip on every restart.
"""
import hashlib
import subprocess as _sp
plugins = reset_plugin_state
plugin_dir = tmp_path / "myplugin"
plugin_dir.mkdir()
req_file = plugin_dir / "requirements.txt"
req_file.write_text("librosa>=0.10.1\nnumpy>=1.24\n")
pip_target = tmp_path / "pip_packages"
monkeypatch.setattr(plugins, "_PIP_TARGET", pip_target)
pip_calls: list = []
def fake_run(cmd, **kwargs):
pip_calls.append(cmd)
return _sp.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
monkeypatch.setattr(plugins.subprocess, "run", fake_run)
# First call — pip runs once and writes the marker.
assert plugins._install_requirements(plugin_dir, "myplugin") is True
assert len(pip_calls) == 1, "first call must invoke pip"
marker = pip_target / ".installed_myplugin"
assert marker.exists(), "marker must be written after successful install"
first_marker_contents = marker.read_text().strip()
# Second call in the same process — must skip pip.
assert plugins._install_requirements(plugin_dir, "myplugin") is True
assert len(pip_calls) == 1, "second in-process call must not re-invoke pip"
# The crucial regression check: the marker must be derivable purely
# from the requirements.txt bytes — no process-local state. If a
# future change reintroduces ``hash()`` (or any other randomised
# digest) the freshly-computed expected value below differs from
# what was written, and the marker would never match on restart.
expected = hashlib.sha256(req_file.read_bytes()).hexdigest()
assert first_marker_contents == expected, (
"marker must contain a deterministic digest of requirements.txt; "
f"got {first_marker_contents!r}, expected {expected!r}"
)
# ── settings.html encoding ─────────────────────────────────────────────────────
def test_settings_html_served_as_utf8_regardless_of_host_locale(
tmp_path, reset_plugin_state, monkeypatch
):
"""Regression for feedBack-desktop#166 — mojibake in the Settings UI.
The ``settings.html`` plugin endpoint must read its file with an
explicit ``encoding="utf-8"``. Without it, ``Path.read_text()`` falls
back to the host's locale encoding — cp1252 on a Windows host — which
decodes UTF-8 punctuation (em-dash ``—``, arrow ``→``) into mojibake
(``â€``, ``â†``). Its sibling endpoints ``screen.html`` / ``screen.js``
already pass ``encoding="utf-8"``; ``settings.html`` was the lone
outlier. This test simulates the Windows locale default so the bug is
caught even on a UTF-8 CI host.
"""
plugins = reset_plugin_state
plugin_dir = tmp_path / "fx"
plugin_dir.mkdir()
settings_html = (
"<p>Audio Quality (Guitar → audio rendering)</p>"
"<p>Default — GeneralUser GS (~32 MB, bundled)</p>"
)
# The on-disk file is genuine, correct UTF-8 — the strings are not the bug.
(plugin_dir / "settings.html").write_text(settings_html, encoding="utf-8")
(plugin_dir / "plugin.json").write_text(
json.dumps({"id": "fx", "name": "FX", "settings": {"html": "settings.html"}})
)
plugins.LOADED_PLUGINS.append({
"id": "fx",
"name": "FX",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": True,
"has_tour": False,
"_dir": plugin_dir,
"_manifest": {"settings": {"html": "settings.html"}},
})
# Simulate a Windows host: a ``read_text()`` call with no explicit
# encoding falls back to the cp1252 locale default. An explicit
# encoding (the fix) is honoured unchanged.
real_read_text = Path.read_text
def cp1252_default_read_text(self, encoding=None, *args, **kwargs):
if encoding is None:
return self.read_bytes().decode("cp1252")
return real_read_text(self, encoding, *args, **kwargs)
monkeypatch.setattr(Path, "read_text", cp1252_default_read_text)
client = _make_api_client(plugins)
try:
r = client.get("/api/plugins/fx/settings.html")
assert r.status_code == 200
# The served body must byte-for-byte match the on-disk UTF-8 file.
assert r.content == settings_html.encode("utf-8")
# And must contain the real characters, not their mojibake forms.
assert "Guitar → audio rendering" in r.text
assert "Default — GeneralUser GS" in r.text
assert "â€" not in r.text # 'â€' — em-dash decoded as cp1252
assert "â†" not in r.text # 'â†' — arrow decoded as cp1252
finally:
client.close()
@pytest.fixture()
def asset_client(tmp_path, reset_plugin_state):
"""A TestClient with one plugin that bundles files under ``assets/``."""
plugins = reset_plugin_state
plugin_dir = tmp_path / "stems"
(plugin_dir / "assets" / "sub").mkdir(parents=True)
(plugin_dir / "assets" / "worklet.js").write_text("registerProcessor('x', class {});\n")
(plugin_dir / "assets" / "plugin.css").write_text(".x-[11px]{font-size:11px}\n")
(plugin_dir / "assets" / "sub" / "data.bin").write_bytes(b"\x00\x01\x02")
# A sibling Python module OUTSIDE assets/ that must never be reachable.
(plugin_dir / "routes.py").write_text("SECRET = 1\n")
plugins.LOADED_PLUGINS.append({
"id": "stems",
"name": "Stems",
"nav": None,
"type": None,
"has_screen": False,
"has_script": False,
"has_settings": False,
"has_tour": False,
"_dir": plugin_dir,
"_manifest": {},
})
client = _make_api_client(plugins)
try:
yield client
finally:
client.close()
def test_plugin_asset_serves_bundled_file(asset_client):
"""A real file under <plugin>/assets/ is served with a JS media type."""
r = asset_client.get("/api/plugins/stems/assets/worklet.js")
assert r.status_code == 200
assert r.text == "registerProcessor('x', class {});\n"
assert "javascript" in r.headers["content-type"]
def test_plugin_asset_serves_css_as_stylesheet(asset_client):
"""A plugin's `styles` CSS under assets/ must serve as text/css so a
<link rel=stylesheet> (the styles capability) is honoured by the browser."""
r = asset_client.get("/api/plugins/stems/assets/plugin.css")
assert r.status_code == 200
assert "text/css" in r.headers["content-type"]
def test_plugin_asset_serves_nested_file(asset_client):
"""Nested asset paths (assets/sub/...) resolve via the {path} match."""
r = asset_client.get("/api/plugins/stems/assets/sub/data.bin")
assert r.status_code == 200
assert r.content == b"\x00\x01\x02"
def test_plugin_asset_missing_file_404(asset_client):
"""A path under assets/ that does not exist returns 404, not 500."""
r = asset_client.get("/api/plugins/stems/assets/nope.js")
assert r.status_code == 404
def test_plugin_asset_rejects_traversal(asset_client):
"""`..` traversal must not escape assets/ to reach plugin Python modules."""
# FastAPI normalises some `..` segments at the routing layer; hit the
# handler's own safe_join guard with an encoded traversal too.
for attack in (
"/api/plugins/stems/assets/../routes.py",
"/api/plugins/stems/assets/..%2froutes.py",
"/api/plugins/stems/assets/sub/../../routes.py",
):
r = asset_client.get(attack)
assert r.status_code == 404, attack
assert "SECRET" not in r.text
def test_plugin_asset_unknown_plugin_404(asset_client):
"""An asset request for an unloaded plugin id returns 404."""
r = asset_client.get("/api/plugins/ghost/assets/worklet.js")
assert r.status_code == 404
# --- Plugin enable/disable (v3 Pedalboard footswitch) ---------------------
#
# Backend contract: every /api/plugins entry carries `enabled` (default true);
# POST /api/plugins/{id}/enabled persists the choice to
# CONFIG_DIR/plugin_state.json; the loader skips disabled plugins at startup but
# still surfaces them as "off" pedals; a disabled plugin is excluded from the
# capability pipeline (its capability metadata is suppressed in /api/plugins).
def test_list_plugins_enabled_default_true(tmp_path, reset_plugin_state):
"""A plugin entry with no `enabled` key surfaces enabled:true by default."""
plugins = reset_plugin_state
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "demo", "name": "Demo", "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "_dir": plugin_dir, "_manifest": {},
})
client = _make_api_client(plugins)
try:
r = client.get("/api/plugins")
assert r.status_code == 200
ids = {p["id"]: p for p in r.json()}
assert ids["demo"]["enabled"] is True
finally:
client.close()
def test_set_plugin_enabled_persists_and_suppresses_caps(tmp_path, reset_plugin_state, monkeypatch):
"""Disabling persists to plugin_state.json, returns {id, enabled:false}, and
a subsequent GET shows enabled:false AND empty capability metadata so the
plugin no longer occupies the capability pipeline graph."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "demo", "name": "Demo", "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "_dir": plugin_dir, "_manifest": {},
"capabilities": {"example.domain": {"role": "owner"}},
"standards": ["capability-pipelines.v1"],
})
client = _make_api_client(plugins)
try:
r = client.post("/api/plugins/demo/enabled", json={"enabled": False})
assert r.status_code == 200
assert r.json() == {"id": "demo", "enabled": False}
# Persisted (only the non-default entry is stored).
state = json.loads((config_dir / "plugin_state.json").read_text())
assert state == {"demo": {"enabled": False}}
# Reflected immediately, with capabilities suppressed.
r = client.get("/api/plugins")
ids = {p["id"]: p for p in r.json()}
assert ids["demo"]["enabled"] is False
assert ids["demo"]["capabilities"] == {}
assert ids["demo"]["standards"] == []
finally:
client.close()
def test_set_plugin_enabled_re_enable_clears_state(tmp_path, reset_plugin_state, monkeypatch):
"""Re-enabling drops the key from plugin_state.json (only non-defaults are
stored) and restores the plugin's capability metadata in /api/plugins."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
(config_dir / "plugin_state.json").write_text(json.dumps({"demo": {"enabled": False}}))
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "demo", "name": "Demo", "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "_dir": plugin_dir, "_manifest": {}, "enabled": False,
"capabilities": {"example.domain": {"role": "owner"}},
"standards": ["capability-pipelines.v1"],
})
client = _make_api_client(plugins)
try:
r = client.post("/api/plugins/demo/enabled", json={"enabled": True})
assert r.status_code == 200
assert r.json() == {"id": "demo", "enabled": True}
assert json.loads((config_dir / "plugin_state.json").read_text()) == {}
r = client.get("/api/plugins")
ids = {p["id"]: p for p in r.json()}
assert ids["demo"]["enabled"] is True
assert ids["demo"]["capabilities"] == {"example.domain": {"role": "owner"}}
assert ids["demo"]["standards"] == ["capability-pipelines.v1"]
finally:
client.close()
def test_set_plugin_enabled_unknown_id_404(tmp_path, reset_plugin_state, monkeypatch):
"""Toggling a plugin id that isn't loaded or pending → 404."""
plugins = reset_plugin_state
monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "config"))
client = _make_api_client(plugins)
try:
r = client.post("/api/plugins/ghost/enabled", json={"enabled": False})
assert r.status_code == 404
assert "error" in r.json()
finally:
client.close()
def test_set_plugin_enabled_bad_body_400(tmp_path, reset_plugin_state, monkeypatch):
"""A missing or non-boolean `enabled` field → 400 (not FastAPI's 422)."""
plugins = reset_plugin_state
monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "config"))
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "demo", "name": "Demo", "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "_dir": plugin_dir, "_manifest": {},
})
client = _make_api_client(plugins)
try:
for bad in ({}, {"enabled": "yes"}, {"enabled": 1}, {"other": True}):
r = client.post("/api/plugins/demo/enabled", json=bad)
assert r.status_code == 400, bad
assert "error" in r.json()
# A non-JSON body is rejected too.
r = client.post("/api/plugins/demo/enabled", content=b"not json")
assert r.status_code == 400
finally:
client.close()
def test_set_plugin_enabled_guardrails_block_always_on(tmp_path, reset_plugin_state, monkeypatch):
"""A LOADED capability_inspector / app_tour_* may never be disabled → 400,
and nothing is written to plugin_state.json."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
# The guard rail runs AFTER the existence check, so the always-on plugins
# must actually be present for the 400 to fire (an absent id → 404 instead).
for pid in ("capability_inspector", "app_tour_library"):
plugins.LOADED_PLUGINS.append({
"id": pid, "name": pid, "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "_dir": plugin_dir, "_manifest": {},
})
client = _make_api_client(plugins)
try:
for pid in ("capability_inspector", "app_tour_library"):
r = client.post(f"/api/plugins/{pid}/enabled", json={"enabled": False})
assert r.status_code == 400, pid
assert "error" in r.json()
assert not (config_dir / "plugin_state.json").exists()
finally:
client.close()
def test_set_plugin_enabled_unknown_always_on_id_is_404_not_400(tmp_path, reset_plugin_state, monkeypatch):
"""An UNKNOWN id that happens to match the always-on pattern (e.g.
app_tour_ghost) is "not found" (404), not "cannot be disabled" (400) — the
response must not depend on the spelling of a nonexistent id (Codex review)."""
plugins = reset_plugin_state
monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "config"))
client = _make_api_client(plugins)
try:
r = client.post("/api/plugins/app_tour_ghost/enabled", json={"enabled": False})
assert r.status_code == 404
finally:
client.close()
def test_set_plugin_enabled_persist_failure_returns_500(tmp_path, reset_plugin_state, monkeypatch):
"""A filesystem write failure when persisting is a controlled 500, and the
in-memory flag is NOT flipped (memory and disk stay consistent)."""
plugins = reset_plugin_state
monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "config"))
plugin_dir = tmp_path / "dummy"
plugin_dir.mkdir()
plugins.LOADED_PLUGINS.append({
"id": "demo", "name": "Demo", "nav": None, "type": None,
"has_screen": False, "has_script": False, "has_settings": False,
"has_tour": False, "_dir": plugin_dir, "_manifest": {},
})
def boom(plugin_id, enabled):
raise OSError("disk full")
monkeypatch.setattr(plugins, "_persist_plugin_enabled", boom)
client = _make_api_client(plugins)
try:
r = client.post("/api/plugins/demo/enabled", json={"enabled": False})
assert r.status_code == 500
assert "error" in r.json()
# In-memory flag untouched, so a later GET still shows enabled.
ids = {p["id"]: p for p in client.get("/api/plugins").json()}
assert ids["demo"]["enabled"] is True
finally:
client.close()
def test_disabled_plugin_skipped_by_loader(tmp_path, reset_plugin_state, monkeypatch):
"""A plugin persisted as enabled:false is NOT loaded (its routes.setup never
runs) but still surfaces in /api/plugins as a disabled entry; an enabled
sibling loads normally."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
(config_dir / "plugin_state.json").write_text(json.dumps({"off": {"enabled": False}}))
marker_body = (
"import os\n"
"from pathlib import Path\n"
"def setup(app, ctx):\n"
" Path(os.environ['CONFIG_DIR'], 'marker_{id}').write_text('1')\n"
)
plugins_root = tmp_path / "plugins"
_make_plugin(plugins_root, "off", routes_body=marker_body.format(id="off"))
_make_plugin(plugins_root, "on", routes_body=marker_body.format(id="on"))
_run_load_plugins(plugins, type("FakeApp", (), {})(), plugins_root)
loaded_ids = {p["id"] for p in plugins.LOADED_PLUGINS}
assert "on" in loaded_ids
assert "off" not in loaded_ids
# routes.setup() ran only for the enabled plugin.
assert (config_dir / "marker_on").exists()
assert not (config_dir / "marker_off").exists()
# The disabled plugin is still a visible, disabled pending entry.
assert plugins.PENDING_PLUGINS["off"]["status"] == "disabled"
assert plugins.PENDING_PLUGINS["off"]["enabled"] is False
client = _make_api_client(plugins)
try:
ids = {p["id"]: p for p in client.get("/api/plugins").json()}
assert ids["off"]["enabled"] is False
assert ids["off"]["status"] == "disabled"
assert ids["on"]["enabled"] is True
finally:
client.close()
def test_corrupt_plugin_state_does_not_crash_loader(tmp_path, reset_plugin_state, monkeypatch):
"""A corrupt plugin_state.json is ignored (all plugins enabled by default)
and loading does not raise."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
(config_dir / "plugin_state.json").write_text("{not valid json")
plugins_root = tmp_path / "plugins"
_make_plugin(plugins_root, "ok")
_run_load_plugins(plugins, type("FakeApp", (), {})(), plugins_root)
assert "ok" in {p["id"] for p in plugins.LOADED_PLUGINS}
def test_reenable_startup_disabled_plugin_keeps_caps_suppressed(tmp_path, reset_plugin_state, monkeypatch):
"""Re-enabling a plugin that was SKIPPED at startup (pending, status
"disabled") flips its `enabled` flag immediately but must NOT reintroduce
its capability metadata into /api/plugins — the plugin isn't mounted until
the next restart, so surfacing its capabilities would register a phantom
pipeline participant (Codex review)."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
(config_dir / "plugin_state.json").write_text(json.dumps({"off": {"enabled": False}}))
# A startup-skipped plugin lives in PENDING_PLUGINS with status "disabled".
plugins.PENDING_PLUGINS["off"] = {
"id": "off", "name": "Off", "nav": None, "type": None,
"status": "disabled", "enabled": False, "_order": 0,
"capabilities": {"example.domain": {"role": "owner"}},
"standards": ["capability-pipelines.v1"],
}
client = _make_api_client(plugins)
try:
r = client.post("/api/plugins/off/enabled", json={"enabled": True})
assert r.status_code == 200
assert r.json() == {"id": "off", "enabled": True}
ids = {p["id"]: p for p in client.get("/api/plugins").json()}
# Flag reflects intent immediately...
assert ids["off"]["enabled"] is True
# ...but it is still not mounted, so capabilities stay suppressed.
assert ids["off"]["status"] == "disabled"
assert ids["off"]["capabilities"] == {}
assert ids["off"]["standards"] == []
finally:
client.close()
def test_is_plugin_enabled_ignores_malformed_entry(reset_plugin_state):
"""Only a real boolean `enabled` is authoritative; missing or junk values
fall back to enabled (matches the endpoint's strict isinstance(bool) check),
so a partially-corrupt or hand-edited plugin_state.json can't silently
disable a plugin (Codex review)."""
plugins = reset_plugin_state
# A genuine ``{"enabled": false}`` disables.
assert plugins._is_plugin_enabled({"p": {"enabled": False}}, "p") is False
assert plugins._is_plugin_enabled({"p": {"enabled": True}}, "p") is True
# Junk / falsey-but-not-bool values are treated as enabled (default true).
for junk in (0, 1, "", "false", [], {}, None):
assert plugins._is_plugin_enabled({"p": {"enabled": junk}}, "p") is True, junk
# Missing key, non-dict entry, and absent plugin all default to enabled.
assert plugins._is_plugin_enabled({"p": {}}, "p") is True
assert plugins._is_plugin_enabled({"p": "nope"}, "p") is True
assert plugins._is_plugin_enabled({}, "p") is True
def test_disable_while_installing_is_carried_into_loaded_entry(tmp_path, reset_plugin_state, monkeypatch):
"""A disable that arrives while a plugin is still loading must survive
graduation. The ready entry is rebuilt from the startup _spec_entries
snapshot, so _graduate carries the live pending entry's `enabled` flag
forward — otherwise the runtime disable silently reverts until restart
(Codex review)."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
# routes.setup() runs after the pending-seed snapshot but before this
# plugin graduates, so flipping the live pending entry here simulates a
# runtime disable racing an in-flight load.
body = (
"import plugins\n"
"def setup(app, ctx):\n"
" plugins.PENDING_PLUGINS['demo']['enabled'] = False\n"
)
plugins_root = tmp_path / "plugins"
_make_plugin(plugins_root, "demo", routes_body=body)
_run_load_plugins(plugins, type("FakeApp", (), {})(), plugins_root)
by_id = {p["id"]: p for p in plugins.LOADED_PLUGINS}
assert "demo" in by_id, "plugin should still graduate (routes already mounting)"
# The ready entry reflects the runtime disable, not the stale startup value.
assert by_id["demo"]["enabled"] is False
def test_persist_does_not_clobber_on_unreadable_existing_state(tmp_path, reset_plugin_state, monkeypatch):
"""A read-time OSError on an existing plugin_state.json must propagate (→ a
controlled 500 at the endpoint) rather than reset to {} and overwrite — that
would silently clobber every other plugin's persisted disabled state (Codex
review)."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
# Make the state path unreadable as a file: a directory at that path makes
# read_text raise IsADirectoryError (a non-FileNotFound OSError).
(config_dir / "plugin_state.json").mkdir()
with pytest.raises(OSError):
plugins._persist_plugin_enabled("demo", False)
# Untouched — not replaced with a fresh file.
assert (config_dir / "plugin_state.json").is_dir()
def test_disable_before_route_setup_aborts_mount(tmp_path, reset_plugin_state, monkeypatch):
"""A disable that lands BEFORE a plugin's routes are mounted prevents the
mount entirely (no setup(), not graduated) — not just suppressed after the
fact. Plugins load alphabetically, so 'a_disabler' runs first and disables
'z_target' (simulating a runtime POST during an earlier install); the loader
must re-check the live flag and skip 'z_target' (Codex review)."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
plugins_root = tmp_path / "plugins"
_make_plugin(plugins_root, "a_disabler", routes_body=(
"import plugins\n"
"def setup(app, ctx):\n"
" plugins.PENDING_PLUGINS['z_target']['enabled'] = False\n"
))
_make_plugin(plugins_root, "z_target", routes_body=(
"import os\n"
"from pathlib import Path\n"
"def setup(app, ctx):\n"
" Path(os.environ['CONFIG_DIR'], 'z_marker').write_text('1')\n"
))
_run_load_plugins(plugins, type("FakeApp", (), {})(), plugins_root)
loaded_ids = {p["id"] for p in plugins.LOADED_PLUGINS}
assert "a_disabler" in loaded_ids
assert "z_target" not in loaded_ids
# Its routes.setup() never ran...
assert not (config_dir / "z_marker").exists()
# ...and it remains a visible, disabled pending entry.
assert plugins.PENDING_PLUGINS["z_target"]["status"] == "disabled"
assert plugins.PENDING_PLUGINS["z_target"]["enabled"] is False
def test_disable_during_requirements_install_aborts_mount(tmp_path, reset_plugin_state, monkeypatch):
"""A disable that lands WHILE this plugin's own requirements are installing
is caught by the final re-check before route setup, so the plugin is never
mounted (Codex review — closes the install-window gap)."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
plugins_root = tmp_path / "plugins"
_make_plugin(plugins_root, "demo", routes_body=(
"import os\n"
"from pathlib import Path\n"
"def setup(app, ctx):\n"
" Path(os.environ['CONFIG_DIR'], 'demo_marker').write_text('1')\n"
))
def fake_install(plugin_dir, plugin_id):
if plugin_id == "demo":
# Simulate a runtime POST /enabled {"enabled": false} arriving mid-install.
plugins.PENDING_PLUGINS["demo"]["enabled"] = False
return True
monkeypatch.setattr(plugins, "_install_requirements", fake_install)
_run_load_plugins(plugins, type("FakeApp", (), {})(), plugins_root)
assert "demo" not in {p["id"] for p in plugins.LOADED_PLUGINS}
assert not (config_dir / "demo_marker").exists(), "setup() must not have run"
assert plugins.PENDING_PLUGINS["demo"]["status"] == "disabled"
assert plugins.PENDING_PLUGINS["demo"]["enabled"] is False
def test_non_utf8_plugin_state_does_not_crash_loader(tmp_path, reset_plugin_state, monkeypatch):
"""Encoding-corrupt (non-UTF-8) plugin_state.json is tolerated like bad JSON:
the loader doesn't crash (all plugins enabled by default) and a subsequent
toggle self-heals the file rather than escaping its controlled path (Codex
review — UnicodeDecodeError is a ValueError, not OSError/JSONDecodeError)."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
(config_dir / "plugin_state.json").write_bytes(b"\xff\xfe\x00not utf-8")
# Load helper must not raise on the corrupt bytes.
assert plugins._load_plugin_state() == {}
plugins_root = tmp_path / "plugins"
_make_plugin(plugins_root, "ok")
_run_load_plugins(plugins, type("FakeApp", (), {})(), plugins_root)
assert "ok" in {p["id"] for p in plugins.LOADED_PLUGINS}
# Persist self-heals the encoding-corrupt file instead of raising.
plugins._persist_plugin_enabled("ok", False)
assert json.loads((config_dir / "plugin_state.json").read_text()) == {"ok": {"enabled": False}}
def test_disable_during_fallback_aborts_fallback_mount(tmp_path, reset_plugin_state, monkeypatch):
"""The bundled-plugin fallback path honors a runtime disable too: if a
disable lands before the fallback user copy mounts, the loader skips it
(no setup()), leaving a disabled entry — same contract as the main path
(Codex review)."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
bundled_dir = tmp_path / "bundled"
(bundled_dir / "highway_3d").mkdir(parents=True)
(bundled_dir / "highway_3d" / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway", "routes": "routes.py", "bundled": True})
)
# Bundled setup() disables the plugin (simulating a runtime POST) and then
# fails — so the disable is live by the time the fallback path runs.
(bundled_dir / "highway_3d" / "routes.py").write_text(
"import plugins\n"
"def setup(app, ctx):\n"
" plugins.PENDING_PLUGINS['highway_3d']['enabled'] = False\n"
" raise RuntimeError('bundled broken')\n"
)
user_dir = tmp_path / "user"
(user_dir / "highway_3d").mkdir(parents=True)
(user_dir / "highway_3d" / "plugin.json").write_text(
json.dumps({"id": "highway_3d", "name": "3D Highway (user)", "routes": "routes.py"})
)
(user_dir / "highway_3d" / "routes.py").write_text(
"def setup(app, ctx):\n app.state.origin = 'user_fallback'\n"
)
monkeypatch.setenv("FEEDBACK_PLUGINS_DIR", str(user_dir))
fake_app = type("FakeApp", (), {})()
fake_app.state = type("State", (), {})()
saved_dir = plugins.PLUGINS_DIR
plugins.PLUGINS_DIR = bundled_dir
try:
plugins.load_plugins(fake_app, {})
finally:
plugins.PLUGINS_DIR = saved_dir
# The fallback copy must NOT have been registered or mounted.
assert "highway_3d" not in {p["id"] for p in plugins.LOADED_PLUGINS}
assert getattr(fake_app.state, "origin", None) != "user_fallback"
# It remains a visible, disabled entry.
assert plugins.PENDING_PLUGINS["highway_3d"]["status"] == "disabled"
assert plugins.PENDING_PLUGINS["highway_3d"]["enabled"] is False
def test_loader_honors_persisted_disable_without_memory_flip(tmp_path, reset_plugin_state, monkeypatch):
"""The loader's _became_disabled check consults the PERSISTED state, not
just the in-memory flag — so a disable that has been written to disk (the
endpoint persists BEFORE flipping memory) is honored even if the in-memory
flag hasn't been flipped yet (Codex review — persist-then-flip window).
'a_disabler' persists a disable for 'z_target' WITHOUT touching memory."""
plugins = reset_plugin_state
config_dir = tmp_path / "config"
config_dir.mkdir()
monkeypatch.setenv("CONFIG_DIR", str(config_dir))
plugins_root = tmp_path / "plugins"
_make_plugin(plugins_root, "a_disabler", routes_body=(
"import plugins\n"
"def setup(app, ctx):\n"
" plugins._persist_plugin_enabled('z_target', False)\n" # disk only, no memory flip
))
_make_plugin(plugins_root, "z_target", routes_body=(
"import os\n"
"from pathlib import Path\n"
"def setup(app, ctx):\n"
" Path(os.environ['CONFIG_DIR'], 'z_marker').write_text('1')\n"
))
_run_load_plugins(plugins, type("FakeApp", (), {})(), plugins_root)
loaded_ids = {p["id"] for p in plugins.LOADED_PLUGINS}
assert "a_disabler" in loaded_ids
assert "z_target" not in loaded_ids
assert not (config_dir / "z_marker").exists()
assert plugins.PENDING_PLUGINS["z_target"]["status"] == "disabled"
assert plugins.PENDING_PLUGINS["z_target"]["enabled"] is False