m4
c2743251e9
Self-evolving AI scientist framework built on LangGraph/LangChain with CLI/TUI core, FastAPI gateway, and Next.js frontend. Co-Authored-By: Claude Opus 4 <noreply@anthropic.com>
101 lines
3.1 KiB
Python
101 lines
3.1 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):
|
|
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")]
|