lambdagent/tests/test_fromconfig.py

"""Tests for lambdagent.fromconfig compiler."""
import os
import json
import tempfile
import pytest
import yaml
from lambdagent.fromconfig import from_config, lint_config, to_lambda_expr, describe_config
from lambdagent.fromconfig.errors import SchemaError
from lambdagent.fromconfig.schema import validate_schema
from lambdagent.primitives import Lam, Loop, Compose
from lambdagent.extensions import Memory, Guard, Route
from lambdagent.conversation import ConversationLam


def _write_config(config):
    f = tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False, encoding='utf-8')
    yaml.dump(config, f, allow_unicode=True)
    f.close()
    return f.name


def test_compile_simple():
    path = _write_config({"type": "simple", "systemPrompt": "Hello", "model": {"name": "test"}})
    try:
        term = from_config(path)
        # The compiler may emit either Lam (stateless) or ConversationLam
        # (history-aware) for a "simple" agent depending on provider config.
        # Both are valid Lambda abstractions — accept either.
        assert isinstance(term, (Lam, ConversationLam)), \
            f"expected Lam or ConversationLam, got {type(term).__name__}"
    finally:
        os.unlink(path)


def test_compile_chain():
    path = _write_config({
        "type": "chain", "model": {"name": "test"},
        "chain": {"steps": [
            {"name": "s1", "prompt": "Step 1"},
            {"name": "s2", "prompt": "Step 2"},
        ]}
    })
    try:
        term = from_config(path)
        assert isinstance(term, Compose)
        assert len(term.stages) == 2
    finally:
        os.unlink(path)


def test_compile_react():
    path = _write_config({
        "type": "react", "systemPrompt": "Think",
        "model": {"name": "test"},
        "react": {"maxSteps": 5},
        "mcp": {"localTools": ["terminate"]},
    })
    try:
        term = from_config(path)
        assert isinstance(term, Loop)
        assert term.max_steps == 5
    finally:
        os.unlink(path)


def test_compile_with_memory():
    path = _write_config({
        "type": "simple", "systemPrompt": "Hello",
        "model": {"name": "test"},
        "memory": {"enabled": True, "strategy": "local", "size": 10, "ttl": 3600},
    })
    try:
        term = from_config(path)
        assert isinstance(term, Memory)
    finally:
        os.unlink(path)


def test_compile_with_guard():
    path = _write_config({
        "type": "simple", "systemPrompt": "Hello",
        "model": {"name": "test"},
        "guard": {"validator": "len(x) > 5", "retry": 1},
    })
    try:
        term = from_config(path)
        assert isinstance(term, Guard)
    finally:
        os.unlink(path)


def test_schema_validation_missing_type():
    errors = validate_schema({})
    assert any(e[0] == "ERROR" for e in errors)


def test_lint_no_terminate():
    results = lint_config({
        "type": "react", "systemPrompt": "x",
        "react": {"maxSteps": 10},
        "mcp": {"localTools": []},
    })
    rules = [r.rule for r in results]
    assert any("L004" in r for r in rules)


def test_lint_clean():
    results = lint_config({
        "type": "react", "systemPrompt": "x",
        "model": {"name": "test"},
        "react": {"maxSteps": 10},
        "mcp": {"localTools": ["terminate"]},
    })
    errors = [r for r in results if r.level == "ERROR"]
    assert len(errors) == 0


def test_to_lambda_expr():
    expr = to_lambda_expr({"type": "simple", "name": "test", "systemPrompt": "hi"})
    assert "test" in expr
    assert "lambda" in expr


def test_describe_config():
    desc = describe_config({"type": "react", "name": "agent", "react": {"maxSteps": 10},
                            "mcp": {"onlineTool": {"s": ["search"]}, "localTools": ["terminate"]}})
    assert "Y_10" in desc or "Loop" in desc