m4
5a581c78a2
Build / build (push) Has been cancelled
Docker / build (push) Has been cancelled
Lint / ruff (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.11) (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.12) (push) Has been cancelled
Test / pytest (windows-latest, 3.11) (push) Has been cancelled
Test / pytest (windows-latest, 3.12) (push) Has been cancelled
Introduce provider, model, and invocation contracts with encrypted configuration persistence. Add web runtime fencing, route fallback, recovery middleware, workspace scoping, and comprehensive tests.
103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from EvoScientist.runtime_integrations import (
|
|
RuntimeIntegrationUnavailable,
|
|
configure_runtime_integrations,
|
|
get_app_connection,
|
|
get_image_backend,
|
|
get_session_connection,
|
|
get_session_dsn,
|
|
handle_knowledge_file,
|
|
record_service_usage,
|
|
reset_runtime_integrations,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_integrations():
|
|
reset_runtime_integrations()
|
|
yield
|
|
reset_runtime_integrations()
|
|
|
|
|
|
def test_core_package_does_not_import_gateway():
|
|
package_root = Path(__file__).resolve().parents[1] / "EvoScientist"
|
|
violations = []
|
|
for source_file in package_root.rglob("*.py"):
|
|
tree = ast.parse(
|
|
source_file.read_text(encoding="utf-8"), filename=str(source_file)
|
|
)
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
names = [alias.name for alias in node.names]
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if node.level:
|
|
continue
|
|
names = [node.module or ""]
|
|
else:
|
|
continue
|
|
if any(name == "gateway" or name.startswith("gateway.") for name in names):
|
|
violations.append(
|
|
f"{source_file.relative_to(package_root)}:{node.lineno}"
|
|
)
|
|
assert violations == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_optional_integrations_are_safe_without_web_runtime(tmp_path):
|
|
assert get_session_dsn() is None
|
|
await handle_knowledge_file(tmp_path / "result.md")
|
|
await record_service_usage("search", "query")
|
|
with pytest.raises(RuntimeIntegrationUnavailable):
|
|
await get_app_connection()
|
|
with pytest.raises(RuntimeIntegrationUnavailable):
|
|
await get_session_connection()
|
|
with pytest.raises(RuntimeIntegrationUnavailable):
|
|
get_image_backend()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_host_can_register_runtime_integrations(tmp_path):
|
|
app_connection = object()
|
|
session_connection = object()
|
|
knowledge_paths = []
|
|
usage = []
|
|
image_backend = object()
|
|
|
|
async def provide_app_connection():
|
|
return app_connection
|
|
|
|
async def provide_session_connection():
|
|
return session_connection
|
|
|
|
async def handle_file(path):
|
|
knowledge_paths.append(path)
|
|
|
|
async def record_usage(service, action):
|
|
usage.append((service, action))
|
|
|
|
configure_runtime_integrations(
|
|
app_connection_provider=provide_app_connection,
|
|
session_connection_provider=provide_session_connection,
|
|
session_dsn_provider=lambda: "postgresql://example/session",
|
|
knowledge_file_handler=handle_file,
|
|
usage_recorder=record_usage,
|
|
image_backend_factory=lambda: image_backend,
|
|
)
|
|
|
|
path = tmp_path / "result.md"
|
|
await handle_knowledge_file(path)
|
|
await record_service_usage("mineru", "parse")
|
|
|
|
assert await get_app_connection() is app_connection
|
|
assert await get_session_connection() is session_connection
|
|
assert get_session_dsn() == "postgresql://example/session"
|
|
assert get_image_backend() is image_backend
|
|
assert knowledge_paths == [path]
|
|
assert usage == [("mineru", "parse")]
|