157 lines
5.0 KiB
Python
157 lines
5.0 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import core.registry as registry_module
|
|
from core.registry import FrameworkRegistry
|
|
|
|
|
|
PROMPT = {"system": "Return JSON", "user_template": "Message: {user_input}"}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_global_registry():
|
|
registry_module.reset_registry()
|
|
yield
|
|
registry_module.reset_registry()
|
|
|
|
|
|
def test_discovery_is_alphabetical_ignores_private_packages_and_runs_once(monkeypatch):
|
|
loaded = []
|
|
|
|
def feature(name):
|
|
def register(target):
|
|
loaded.append(name)
|
|
target.describe(f"{name} feature")
|
|
target.register_command(
|
|
name,
|
|
lambda _context, _parsed: None,
|
|
PROMPT,
|
|
description=f"Handle {name}",
|
|
help_text=[f"use {name}"],
|
|
)
|
|
|
|
return SimpleNamespace(register=register)
|
|
|
|
fake_modules = {
|
|
"modules.alpha": feature("alpha"),
|
|
"modules.zeta": feature("zeta"),
|
|
}
|
|
|
|
def fake_import(name):
|
|
if name == "modules":
|
|
return SimpleNamespace(__path__=["unused"])
|
|
return fake_modules[name]
|
|
|
|
discovered = [
|
|
SimpleNamespace(name="zeta", ispkg=True),
|
|
SimpleNamespace(name="_private", ispkg=True),
|
|
SimpleNamespace(name="single_file", ispkg=False),
|
|
SimpleNamespace(name="alpha", ispkg=True),
|
|
]
|
|
monkeypatch.setattr(registry_module.importlib, "import_module", fake_import)
|
|
monkeypatch.setattr(
|
|
registry_module.pkgutil, "iter_modules", lambda _path: discovered
|
|
)
|
|
|
|
result = registry_module.discover_modules()
|
|
|
|
assert loaded == ["alpha", "zeta"]
|
|
assert list(result.modules) == ["alpha", "zeta"]
|
|
assert result.list_commands() == ["alpha", "zeta"]
|
|
assert registry_module.discover_modules() is result
|
|
assert loaded == ["alpha", "zeta"]
|
|
|
|
|
|
def test_duplicate_names_and_malformed_registrations_are_rejected():
|
|
target = FrameworkRegistry()
|
|
handler = lambda _context, _parsed: None
|
|
|
|
target.begin_module("first", "modules.first")
|
|
target.register_command("shared", handler, PROMPT)
|
|
target.register_job("shared.job", handler)
|
|
target.finish_module()
|
|
|
|
with pytest.raises(ValueError, match="Duplicate module name: first"):
|
|
target.begin_module("first", "modules.again")
|
|
|
|
target.begin_module("second", "modules.second")
|
|
with pytest.raises(ValueError, match="Duplicate command type: shared"):
|
|
target.register_command("shared", handler, PROMPT)
|
|
with pytest.raises(ValueError, match="Duplicate job type: shared.job"):
|
|
target.register_job("shared.job", handler)
|
|
with pytest.raises(TypeError, match="must be callable"):
|
|
target.register_command("not_callable", None, PROMPT)
|
|
with pytest.raises(TypeError, match="Validator .* must be callable"):
|
|
target.register_command("bad_validator", handler, PROMPT, validator="bad")
|
|
with pytest.raises(ValueError, match="system and user_template"):
|
|
target.register_command("bad_prompt", handler, {"system": "only one"})
|
|
target.finish_module()
|
|
|
|
with pytest.raises(RuntimeError, match="inside a module register"):
|
|
target.describe("orphan metadata")
|
|
|
|
|
|
def test_help_and_router_context_are_generated_from_sorted_metadata():
|
|
target = FrameworkRegistry()
|
|
handler = lambda _context, _parsed: None
|
|
target.begin_module("examples", "modules.examples")
|
|
target.register_command(
|
|
"zeta",
|
|
handler,
|
|
PROMPT,
|
|
description="Last command",
|
|
)
|
|
target.register_command(
|
|
"alpha",
|
|
handler,
|
|
PROMPT,
|
|
description="First command",
|
|
help_text=["say alpha", "ask alpha for help"],
|
|
)
|
|
target.finish_module()
|
|
|
|
assert target.router_context() == (
|
|
"- alpha: First command\n- zeta: Last command"
|
|
)
|
|
assert target.help_lines() == [
|
|
"- say alpha",
|
|
"- ask alpha for help",
|
|
"- zeta: Last command",
|
|
]
|
|
|
|
|
|
def test_failed_forced_discovery_clears_partial_state_and_can_retry(monkeypatch):
|
|
broken = SimpleNamespace(register=lambda target: target.describe("partial"))
|
|
monkeypatch.setattr(
|
|
registry_module.importlib,
|
|
"import_module",
|
|
lambda name: (
|
|
SimpleNamespace(__path__=["unused"]) if name == "modules" else broken
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
registry_module.pkgutil,
|
|
"iter_modules",
|
|
lambda _path: [SimpleNamespace(name="broken", ispkg=True)],
|
|
)
|
|
|
|
registry_module.discover_modules()
|
|
assert registry_module.registry.modules == {
|
|
"broken": {
|
|
"name": "broken",
|
|
"package": "modules.broken",
|
|
"description": "partial",
|
|
}
|
|
}
|
|
|
|
del broken.register
|
|
with pytest.raises(RuntimeError, match="must expose register"):
|
|
registry_module.discover_modules(force=True)
|
|
assert registry_module.registry.modules == {}
|
|
|
|
broken.register = lambda target: target.describe("recovered")
|
|
assert registry_module.discover_modules().modules["broken"][
|
|
"description"
|
|
] == "recovered"
|