103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
"""Kompilacja definicji APM do obiektów agno.
|
|
|
|
Test nie woła modelu - sprawdza to, co da się sprawdzić bez GPU: czy każda definicja
|
|
agenta z pakietu APM daje się zbudować, czy jej narzędzia istnieją, czy skille się walidują
|
|
i czy schemat wyjścia jest znany. To bramka na literówkę w `.agent.md`.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import pytest
|
|
from agno.models.base import Model
|
|
|
|
from agentic_codemod.apm.agno_bridge import ToolResolutionError, build_agent, build_skills
|
|
from agentic_codemod.apm.loader import load_apm_context
|
|
from agentic_codemod.observability.audit import AuditLog
|
|
from agentic_codemod.tools import WorkspaceTools, build_tool_registry
|
|
from agentic_codemod.workflow.brains import SCHEMA_REGISTRY
|
|
|
|
|
|
@dataclass
|
|
class _StubModel(Model):
|
|
"""Model, który nigdy nie zostanie wywołany - potrzebny tylko po to,
|
|
|
|
by agno przyjęło konstrukcję agenta bez dostępu do endpointu.
|
|
"""
|
|
|
|
id: str = "stub"
|
|
name: str = "Stub"
|
|
provider: str = "Stub"
|
|
|
|
def invoke(self, *args, **kwargs):
|
|
raise NotImplementedError
|
|
|
|
async def ainvoke(self, *args, **kwargs):
|
|
raise NotImplementedError
|
|
|
|
def invoke_stream(self, *args, **kwargs):
|
|
raise NotImplementedError
|
|
|
|
async def ainvoke_stream(self, *args, **kwargs):
|
|
raise NotImplementedError
|
|
|
|
def _parse_provider_response(self, response, **kwargs):
|
|
raise NotImplementedError
|
|
|
|
def _parse_provider_response_delta(self, response):
|
|
raise NotImplementedError
|
|
|
|
|
|
class _StubModelFactory:
|
|
"""Zastępuje realny endpoint - budowa agenta nie może wymagać dostępnego LLM."""
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def for_profile(self, profile, temperature=None):
|
|
self.calls.append((profile, temperature))
|
|
return _StubModel(id=f"stub-{profile}")
|
|
|
|
|
|
@pytest.fixture()
|
|
def registry(acme_app, settings, tmp_path):
|
|
workspace = WorkspaceTools(acme_app, settings, AuditLog(run_dir=tmp_path / "audit"))
|
|
return build_tool_registry(workspace)
|
|
|
|
|
|
def test_kazda_definicja_agenta_daje_sie_zbudowac(repo_root, settings, registry):
|
|
ctx = load_apm_context(repo_root)
|
|
factory = _StubModelFactory()
|
|
registry = dict(registry)
|
|
registry["run_verification"] = lambda: "ok"
|
|
|
|
for name, definition in ctx.agents.items():
|
|
agent = build_agent(definition, ctx, settings, factory, registry, SCHEMA_REGISTRY)
|
|
assert agent.name == name
|
|
assert agent.instructions, "agent bez instrukcji to agent bez kontekstu"
|
|
assert agent.tool_call_limit and agent.tool_call_limit > 0
|
|
|
|
assert {profile for profile, _ in factory.calls} <= {"planner", "coder", "reviewer", "scribe"}
|
|
|
|
|
|
def test_schemat_wyjscia_jest_wiazany_z_rejestru(repo_root, settings, registry):
|
|
ctx = load_apm_context(repo_root)
|
|
registry = dict(registry)
|
|
registry["run_verification"] = lambda: "ok"
|
|
agent = build_agent(ctx.agent("planner"), ctx, settings, _StubModelFactory(), registry, SCHEMA_REGISTRY)
|
|
assert agent.output_schema is SCHEMA_REGISTRY["ChangePlan"]
|
|
|
|
|
|
def test_nieznane_narzedzie_w_definicji_konczy_sie_bledem(repo_root, settings, registry):
|
|
ctx = load_apm_context(repo_root)
|
|
definition = ctx.agent("coder")
|
|
definition.meta["tools"] = ["read_file", "rm_rf"]
|
|
with pytest.raises(ToolResolutionError, match="rm_rf"):
|
|
build_agent(definition, ctx, settings, _StubModelFactory(), registry, SCHEMA_REGISTRY)
|
|
|
|
|
|
def test_skille_agenta_sa_ladowane_z_katalogow_apm(repo_root):
|
|
ctx = load_apm_context(repo_root)
|
|
skills = build_skills(ctx, ctx.agent("coder").skill_names)
|
|
assert set(skills.get_skill_names()) == {"safe-code-edit", "sdk-version-upgrade"}
|
|
assert skills.get_skill("sdk-version-upgrade").references, "notatka migracyjna musi być widoczna dla agenta"
|