Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 71 additions & 0 deletions python/tests/codegen/test_get_flow.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import json
import textwrap
from pathlib import Path

Expand Down Expand Up @@ -144,6 +145,76 @@ def test_system_prompt_none(self, workspace):
config = _single_node(_flow(ws))["data"]["config"]
assert config["system_prompt"]["value"] is None

def test_voice_config_dict(self, workspace):
"""voice_config set in the source round-trips through get-flow."""
ws = workspace("""\
from timbal.core import Agent

agent = Agent(
name="a",
model="openai/gpt-4o-mini",
max_tokens=128,
voice_config={"voice": "CURRENT_VOICE", "tts_extra": {"auto_mode": True}},
)
""")
config = _single_node(_flow(ws))["data"]["config"]
assert config["voice_config"]["value"] == {
"voice": "CURRENT_VOICE",
"tts_extra": {"auto_mode": True},
}

def test_voice_config_absent(self, workspace):
"""Agents without voice_config still expose the key with a None value."""
ws = workspace("""\
from timbal.core import Agent

agent = Agent(name="a", model="openai/gpt-4o-mini", max_tokens=128)
""")
config = _single_node(_flow(ws))["data"]["config"]
assert "voice_config" in config
assert config["voice_config"]["value"] is None

def test_voice_config_callable(self, workspace):
"""A callable voice_config is rendered as an opaque placeholder."""
ws = workspace("""\
from timbal.core import Agent

def make_voice_config():
return {"voice": "CURRENT_VOICE"}

agent = Agent(
name="a",
model="openai/gpt-4o-mini",
max_tokens=128,
voice_config=make_voice_config,
)
""")
config = _single_node(_flow(ws))["data"]["config"]
assert config["voice_config"]["value"] == "<make_voice_config>"
# Schema must advertise the callable variant (mirrors system_prompt).
any_of_types = [v.get("type") for v in config["voice_config"]["anyOf"]]
assert "callable" in any_of_types
assert "object" in any_of_types

def test_voice_config_instance(self, workspace):
"""A VoiceConfig instance is dumped to a JSON-safe dict."""
ws = workspace("""\
from timbal.core import Agent
from timbal.voice.config import VoiceConfig

agent = Agent(
name="a",
model="openai/gpt-4o-mini",
max_tokens=128,
voice_config=VoiceConfig(voice="my-voice"),
)
""")
config = _single_node(_flow(ws))["data"]["config"]
value = config["voice_config"]["value"]
assert isinstance(value, dict)
assert value["voice"] == "my-voice"
json.dumps(value) # must stay JSON-serialisable

def test_has_params_and_return(self, workspace):
ws = workspace("""\
from timbal.core import Agent
Expand Down
17 changes: 17 additions & 0 deletions python/tests/codegen/test_remove_edge.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,23 @@ def test_remove_one_from_multiple_depends_on(self, wf_workspace):
assert '"agent_a"' not in normalized.split("depends_on")[1] if "depends_on" in normalized else True


class TestIdempotency:
def test_remove_absent_edge_is_noop_success(self, wf_workspace):
"""Removing an edge that is already gone succeeds without changes."""
ws = wf_workspace("""\
from timbal import Agent, Workflow

agent_a = Agent(name="agent_a", model="openai/gpt-4o-mini")
agent_b = Agent(name="agent_b", model="openai/gpt-4o-mini")

workflow = Workflow(name="wf")
workflow.step(agent_a)
workflow.step(agent_b)
""")
output = _run(ws, source="agent_a", target="agent_b")
assert "workflow.step(agent_b)" in output


class TestRemoveDataFlowEdge:
def test_remove_param_lambda(self, wf_workspace):
ws = wf_workspace("""\
Expand Down
158 changes: 158 additions & 0 deletions python/tests/codegen/test_set_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,164 @@ def my_func():
assert tool.description == "Does something useful"


# ---------------------------------------------------------------------------
# Entry point shapes (annotated / aliased / factory assignments)
# ---------------------------------------------------------------------------


class TestEntryPointShapes:
def test_annotated_assignment(self, workspace):
"""``agent: Agent = Agent(...)`` is transformed like a plain assignment."""
ws = workspace("""\
from timbal.core import Agent

agent: Agent = Agent(name="a", model="openai/gpt-4o-mini")
""")
config = json.dumps({"voice_config": {"voice": "NEW_VOICE"}})
output = _run_dry(ws, "--config", config)
ns = _exec_agent(output)
assert ns["agent"].voice_config == {"voice": "NEW_VOICE"}

def test_annotated_assignment_set_model(self, workspace):
ws = workspace("""\
from timbal.core import Agent

agent: Agent = Agent(name="a", model="openai/gpt-4o-mini")
""")
config = json.dumps({"model": "openai/gpt-4o"})
output = _run_dry(ws, "--config", config)
ns = _exec_agent(output)
assert ns["agent"].model == "openai/gpt-4o"

def test_aliased_assignment_fails_loudly(self, workspace):
"""``agent = _agent`` cannot be transformed — must exit non-zero, not silently no-op."""
ws = workspace("""\
from timbal.core import Agent

_agent = Agent(name="a", model="openai/gpt-4o-mini")
agent = _agent
""")
config = json.dumps({"voice_config": {"voice": "NEW_VOICE"}})
result = _run_dry_fail(ws, "--config", config)
assert result.returncode != 0
assert "produced no changes" in result.stderr

def test_factory_call_fails_loudly(self, workspace):
"""``agent = build()`` must error instead of appending kwargs to build()."""
ws = workspace("""\
from timbal.core import Agent

def build():
return Agent(name="a", model="openai/gpt-4o-mini")

agent = build()
""")
config = json.dumps({"voice_config": {"voice": "NEW_VOICE"}})
result = _run_dry_fail(ws, "--config", config)
assert result.returncode != 0
assert "factory" in result.stderr

def test_imported_factory_call_fails_loudly(self, workspace):
"""An imported factory must also error, not get kwargs injected."""
ws = workspace("""\
from helpers import build

agent = build()
""")
config = json.dumps({"voice_config": {"voice": "NEW_VOICE"}})
result = _run_dry_fail(ws, "--config", config)
assert result.returncode != 0
assert "factory" in result.stderr

def test_aliased_agent_import_still_works(self, workspace):
"""``from timbal.core import Agent as A`` is recognized as a constructor."""
ws = workspace("""\
from timbal.core import Agent as A

agent = A(name="a", model="openai/gpt-4o-mini")
""")
config = json.dumps({"model": "openai/gpt-4o"})
output = _run_dry(ws, "--config", config)
ns = _exec_agent(output)
assert ns["agent"].model == "openai/gpt-4o"

def test_module_qualified_constructor_works(self, workspace):
"""``timbal.Agent(...)`` is recognized as a constructor."""
ws = workspace("""\
import timbal

agent = timbal.Agent(name="a", model="openai/gpt-4o-mini")
""")
config = json.dumps({"model": "openai/gpt-4o"})
output = _run_dry(ws, "--config", config)
ns = _exec_agent(output)
assert ns["agent"].model == "openai/gpt-4o"

def test_annotated_assignment_tool_config(self, workspace):
"""Tool-level set-config works when the agent uses an annotated assignment."""
ws = workspace("""\
from timbal.core import Agent
from timbal.core.tool import Tool

def my_func():
return "hello"

my_tool = Tool(handler=my_func, name="my_tool")

agent: Agent = Agent(name="a", model="openai/gpt-4o-mini", tools=[my_tool])
""")
config = json.dumps({"description": "Does something useful"})
output = _run_dry(ws, "--name", "my_tool", "--config", config)
ns = _exec_agent(output)
assert ns["my_tool"].description == "Does something useful"

def test_annotated_assignment_inline_tool_migration(self, workspace):
"""Inline tool migration inserts the variable before an annotated entry point."""
ws = workspace("""\
from timbal.core import Agent
from timbal.tools import WebSearch

agent: Agent = Agent(name="a", model="openai/gpt-4o-mini", tools=[WebSearch()])
""")
config = json.dumps({"allowed_domains": ["example.com"]})
output = _run_dry(ws, "--name", "web_search", "--config", config)
# exec fails with NameError if the variable lands after the entry point.
ns = _exec_agent(output)
tool = next(t for t in ns["agent"].tools if t.name == "web_search")
assert tool.allowed_domains == ["example.com"]

def test_noop_same_value_is_idempotent_success(self, workspace):
"""Setting a field to its current value is a success: the transformer
matched the entry point, the source was already in the desired state."""
ws = workspace("""\
from timbal.core import Agent

agent = Agent(name="a", model="openai/gpt-4o-mini")
""")
config = json.dumps({"model": "openai/gpt-4o-mini"})
output = _run_dry(ws, "--config", config)
ns = _exec_agent(output)
assert ns["agent"].model == "openai/gpt-4o-mini"

def test_write_mode_noop_leaves_file_untouched(self, workspace):
"""A failed no-op without --dry-run must not rewrite the source file."""
ws = workspace("""\
from timbal.core import Agent

_agent = Agent(name="a", model="openai/gpt-4o-mini")
agent = _agent
""")
original = (ws / "agent.py").read_text()
config = json.dumps({"voice_config": {"voice": "NEW_VOICE"}})
result = subprocess.run(
codegen_cmd("--path", str(ws), "set-config", "--config", config),
capture_output=True,
text=True,
)
assert result.returncode != 0
assert (ws / "agent.py").read_text() == original


# ---------------------------------------------------------------------------
# Workflow step config
# ---------------------------------------------------------------------------
Expand Down
13 changes: 13 additions & 0 deletions python/tests/codegen/test_set_param.py
Original file line number Diff line number Diff line change
Expand Up @@ -362,6 +362,19 @@ def test_remove_param_with_null(self, wf_workspace):
output = _run(ws, target="agent_b", name="prompt", param_type="value", source=None, key=None, value="null")
assert "prompt=" not in output

def test_set_same_value_is_idempotent_success(self, wf_workspace):
"""Setting a param to its current value succeeds without changes."""
ws = wf_workspace("""\
from timbal import Agent, Workflow

agent_a = Agent(name="agent_a", model="openai/gpt-4o-mini")

workflow = Workflow(name="wf")
workflow.step(agent_a, prompt="Hello world")
""")
output = _run(ws, target="agent_a", name="prompt", param_type="value", source=None, key=None, value='"Hello world"')
assert 'prompt="Hello world"' in output

def test_update_existing_value_param(self, wf_workspace):
ws = wf_workspace("""\
from timbal import Agent, Workflow
Expand Down
15 changes: 15 additions & 0 deletions python/tests/codegen/test_set_position.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,21 @@ def test_update_existing_position(self, workspace):
ns = _exec_agent(output)
assert ns["agent"].metadata["position"] == {"x": 300.0, "y": 400.0}

def test_same_position_is_idempotent_success(self, workspace):
"""Setting the current coordinates again succeeds without changes."""
ws = workspace("""\
from timbal.core import Agent

agent = Agent(
name="a",
model="openai/gpt-4o-mini",
metadata={"position": {"x": 10.0, "y": 20.0}},
)
""")
output = _run_dry(ws, "--x", "10", "--y", "20")
ns = _exec_agent(output)
assert ns["agent"].metadata["position"] == {"x": 10.0, "y": 20.0}

def test_preserves_other_kwargs(self, workspace):
ws = workspace("""\
from timbal.core import Agent
Expand Down
6 changes: 6 additions & 0 deletions python/tests/evals/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,12 @@ def test_regular_subdirectories_are_still_scanned(self, tmp_path):
b = touch(tmp_path / "more_evals" / "deep" / "b_eval.yaml")
assert discover_eval_files(tmp_path) == sorted([a, b])

def test_skips_evalconf(self, tmp_path):
"""evalconf.yaml matches the eval* glob but is the shared config, not a suite."""
touch(tmp_path / "evalconf.yaml", "runnable: ../agent.py::agent\n")
eval_file = touch(tmp_path / "eval_smoke.yaml")
assert discover_eval_files(tmp_path) == [eval_file]


class TestParseEvalFileRunnableResolution:
def test_runnable_relative_to_eval_file_dir(self, tmp_path):
Expand Down
Loading
Loading