generated from coulomb/repo-seed
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
from markitect_tool.extension import (
|
|
ExtensionDescriptor,
|
|
ExtensionExecutor,
|
|
ExtensionLifecycle,
|
|
ExtensionRegistry,
|
|
ExtensionRegistryError,
|
|
OptionalDependency,
|
|
ProcessingRequest,
|
|
ProcessingResult,
|
|
)
|
|
|
|
|
|
def test_extension_executor_runs_callbacks_in_order():
|
|
events: list[str] = []
|
|
|
|
def runner(request: ProcessingRequest) -> ProcessingResult:
|
|
events.append(f"run:{request.operation}")
|
|
return ProcessingResult(output={"ok": True})
|
|
|
|
lifecycle = ExtensionLifecycle()
|
|
lifecycle.on_before(lambda descriptor, request: events.append(f"before:{descriptor.id}"))
|
|
lifecycle.on_success(
|
|
lambda descriptor, request, result: events.append(f"success:{result.output['ok']}")
|
|
)
|
|
lifecycle.on_after(lambda descriptor, request, result: events.append("after"))
|
|
registry = ExtensionRegistry(
|
|
[ExtensionDescriptor(id="fake.runner", kind="test", factory=lambda: runner)]
|
|
)
|
|
|
|
result = ExtensionExecutor(registry, lifecycle=lifecycle).execute(
|
|
"fake.runner",
|
|
ProcessingRequest(operation="fake.run", input={}),
|
|
)
|
|
|
|
assert result.valid
|
|
assert result.trace[-1].event == "extension.executed"
|
|
assert events == ["before:fake.runner", "run:fake.run", "success:True", "after"]
|
|
|
|
|
|
def test_extension_executor_routes_failure_callbacks():
|
|
events: list[str] = []
|
|
|
|
def runner(request: ProcessingRequest) -> ProcessingResult:
|
|
return ProcessingResult.from_error(code="fake.error", message="Nope")
|
|
|
|
lifecycle = ExtensionLifecycle()
|
|
lifecycle.on_failure(lambda descriptor, request, result: events.append(result.diagnostics[0].code))
|
|
registry = ExtensionRegistry(
|
|
[ExtensionDescriptor(id="fake.runner", kind="test", factory=lambda: runner)]
|
|
)
|
|
|
|
result = ExtensionExecutor(registry, lifecycle=lifecycle).execute(
|
|
"fake.runner",
|
|
ProcessingRequest(operation="fake.run", input={}),
|
|
)
|
|
|
|
assert not result.valid
|
|
assert events == ["fake.error"]
|
|
|
|
|
|
def test_extension_executor_blocks_missing_required_dependency():
|
|
registry = ExtensionRegistry(
|
|
[
|
|
ExtensionDescriptor(
|
|
id="query.jsonpath",
|
|
kind="query-engine",
|
|
factory=lambda: lambda request: ProcessingResult(output=[]),
|
|
optional_dependencies=[
|
|
OptionalDependency(name="definitely_missing_markitect_dep", required=True)
|
|
],
|
|
)
|
|
]
|
|
)
|
|
|
|
result = ExtensionExecutor(registry).execute(
|
|
"query.jsonpath",
|
|
ProcessingRequest(operation="query.jsonpath", input={}),
|
|
)
|
|
|
|
assert not result.valid
|
|
assert result.diagnostics[0].code == "extension.missing_dependency"
|
|
assert "definitely_missing_markitect_dep" in result.diagnostics[0].details["missing"]
|
|
|
|
|
|
def test_extension_executor_rejects_non_result_return():
|
|
registry = ExtensionRegistry(
|
|
[ExtensionDescriptor(id="bad.runner", kind="test", factory=lambda: lambda request: {})]
|
|
)
|
|
|
|
try:
|
|
ExtensionExecutor(registry).execute(
|
|
"bad.runner",
|
|
ProcessingRequest(operation="bad.run", input={}),
|
|
)
|
|
except ExtensionRegistryError as exc:
|
|
assert "expected ProcessingResult" in str(exc)
|
|
else:
|
|
raise AssertionError("Expected ExtensionRegistryError")
|