d19ee2eb76
CI / lint (pull_request) Successful in 29s
CI / quality (pull_request) Successful in 24s
CI / security (pull_request) Successful in 47s
CI / typecheck (pull_request) Successful in 47s
CI / build (pull_request) Successful in 23s
CI / helm (pull_request) Successful in 28s
CI / push-validation (pull_request) Successful in 21s
CI / e2e_tests (pull_request) Failing after 3m19s
CI / unit_tests (pull_request) Failing after 7m43s
CI / docker (pull_request) Has been skipped
CI / integration_tests (pull_request) Successful in 7m36s
CI / coverage (pull_request) Successful in 9m38s
CI / status-check (pull_request) Failing after 1s
The test assertion was too strict - it checked that all concurrent threads returned identical result sets. However, when clear() runs concurrently with list_actors(namespace=...), it's valid for some threads to see the full list while others see an empty list. Changed the assertion to verify that each result only contains actors from the requested namespace, allowing for empty results when clear() wins the race. Also stored the namespace in context during the When step so the Then step can reference it for validation. ISSUES CLOSED: #8588
469 lines
17 KiB
Python
469 lines
17 KiB
Python
"""Step definitions for actor_loading.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.actor.loader import ActorLoader
|
|
from cleveragents.actor.schema import ActorConfigSchema
|
|
from cleveragents.core.exceptions import ValidationError
|
|
from cleveragents.tool.registry import ToolRegistry
|
|
from cleveragents.tool.runtime import ToolSpec
|
|
|
|
|
|
def _minimal_actor_yaml(
|
|
name: str,
|
|
actor_type: str = "llm",
|
|
model: str = "gpt-4",
|
|
description: str = "Test actor",
|
|
tools: list[str] | None = None,
|
|
) -> str:
|
|
lines = [
|
|
f"name: {name}",
|
|
f"type: {actor_type}",
|
|
f"description: {description}",
|
|
'version: "1.0"',
|
|
]
|
|
if actor_type in ("llm", "graph"):
|
|
lines.append(f"model: {model}")
|
|
if tools:
|
|
lines.append("tools:")
|
|
for t in tools:
|
|
lines.append(f" - {t}")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _fresh_dir(context: Context, label: str = "default") -> Path:
|
|
"""Create a unique temporary directory per scenario invocation."""
|
|
d = Path(tempfile.mkdtemp(prefix=f"actload_{label}_"))
|
|
if not hasattr(context, "_tmp_dirs"):
|
|
context._tmp_dirs = []
|
|
context._tmp_dirs.append(d)
|
|
return d
|
|
|
|
|
|
# ──────────────────────────────────────────────────────
|
|
# Given steps
|
|
# ──────────────────────────────────────────────────────
|
|
|
|
|
|
@given("a temporary actors directory with files")
|
|
def step_given_temp_dir_with_files(context: Context) -> None:
|
|
d = _fresh_dir(context, "actors")
|
|
context._actor_dir = d
|
|
for row in context.table:
|
|
content = _minimal_actor_yaml(
|
|
name=row["name"],
|
|
actor_type=row["type"],
|
|
model=row.get("model", "gpt-4"),
|
|
)
|
|
(d / row["filename"]).write_text(content)
|
|
|
|
|
|
@given('a temporary actors directory "{dir_name}" with files')
|
|
def step_given_named_dir_with_files(context: Context, dir_name: str) -> None:
|
|
d = _fresh_dir(context, dir_name)
|
|
if not hasattr(context, "_actor_dirs"):
|
|
context._actor_dirs = []
|
|
context._actor_dirs.append(d)
|
|
for row in context.table:
|
|
content = _minimal_actor_yaml(
|
|
name=row["name"],
|
|
actor_type=row["type"],
|
|
model=row.get("model", "gpt-4"),
|
|
)
|
|
(d / row["filename"]).write_text(content)
|
|
|
|
|
|
@given('a non-YAML file "{filename}" in the directory')
|
|
def step_given_non_yaml_file(context: Context, filename: str) -> None:
|
|
d: Path = context._actor_dir
|
|
(d / filename).write_text("this is not yaml actor config\n")
|
|
|
|
|
|
@given("a temporary actors directory with nested structure")
|
|
def step_given_nested_dir(context: Context) -> None:
|
|
d = _fresh_dir(context, "nested")
|
|
context._actor_dir = d
|
|
for row in context.table:
|
|
subdir = d / row["subdir"]
|
|
subdir.mkdir(parents=True, exist_ok=True)
|
|
content = _minimal_actor_yaml(
|
|
name=row["name"],
|
|
actor_type=row["type"],
|
|
model=row.get("model", "gpt-4"),
|
|
)
|
|
(subdir / row["filename"]).write_text(content)
|
|
|
|
|
|
@given("a top-level actor file")
|
|
def step_given_top_level_file(context: Context) -> None:
|
|
d: Path = context._actor_dir
|
|
for row in context.table:
|
|
content = _minimal_actor_yaml(
|
|
name=row["name"],
|
|
actor_type=row["type"],
|
|
model=row.get("model", "gpt-4"),
|
|
)
|
|
(d / row["filename"]).write_text(content)
|
|
|
|
|
|
@given("a temporary actors directory with a nameless actor file")
|
|
def step_given_nameless_actor(context: Context) -> None:
|
|
d = _fresh_dir(context, "nameless")
|
|
context._actor_dir = d
|
|
for row in context.table:
|
|
content = _minimal_actor_yaml(
|
|
name=row["raw_name"],
|
|
actor_type=row["type"],
|
|
model=row.get("model", "gpt-4"),
|
|
)
|
|
(d / row["filename"]).write_text(content)
|
|
|
|
|
|
@given('a temporary actors directory with an invalid YAML file "{filename}"')
|
|
def step_given_invalid_yaml(context: Context, filename: str) -> None:
|
|
d = _fresh_dir(context, "invalid")
|
|
context._actor_dir = d
|
|
(d / filename).write_text("name: [\ninvalid: yaml: content:\n")
|
|
|
|
|
|
@given('a temporary actors directory with a schema-invalid file "{filename}"')
|
|
def step_given_schema_invalid(context: Context, filename: str) -> None:
|
|
d = _fresh_dir(context, "schemainvalid")
|
|
context._actor_dir = d
|
|
(d / filename).write_text("description: missing required fields\n")
|
|
|
|
|
|
@given("the project examples/actors directory exists")
|
|
def step_given_examples_dir(context: Context) -> None:
|
|
project_root = Path(__file__).resolve().parent.parent.parent
|
|
context._actor_dir = project_root / "examples" / "actors"
|
|
assert context._actor_dir.is_dir(), f"Expected {context._actor_dir} to exist"
|
|
|
|
|
|
@given('a tool registry with tool "{tool_name}"')
|
|
def step_given_tool_registry(context: Context, tool_name: str) -> None:
|
|
registry = ToolRegistry()
|
|
spec = ToolSpec(
|
|
name=tool_name,
|
|
description=f"Mock tool {tool_name}",
|
|
handler=lambda **kwargs: {"ok": True},
|
|
)
|
|
registry.register(spec)
|
|
context._tool_registry = registry
|
|
|
|
|
|
@given("an empty tool registry")
|
|
def step_given_empty_tool_registry(context: Context) -> None:
|
|
context._tool_registry = ToolRegistry()
|
|
|
|
|
|
@given("a temporary actors directory with a tool-actor file")
|
|
def step_given_tool_actor_file(context: Context) -> None:
|
|
d = _fresh_dir(context, "toolactors")
|
|
context._actor_dir = d
|
|
for row in context.table:
|
|
tools = [t.strip() for t in row["tools"].split(",")]
|
|
content = _minimal_actor_yaml(
|
|
name=row["name"],
|
|
actor_type=row["type"],
|
|
tools=tools,
|
|
)
|
|
(d / row["filename"]).write_text(content)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────
|
|
# When steps
|
|
# ──────────────────────────────────────────────────────
|
|
|
|
|
|
@when("I create an actor loader with that directory as search root")
|
|
def step_when_create_loader_single(context: Context) -> None:
|
|
tool_reg = getattr(context, "_tool_registry", None)
|
|
context._loader = ActorLoader(
|
|
search_roots=[context._actor_dir],
|
|
tool_registry=tool_reg,
|
|
)
|
|
|
|
|
|
@when("I create an actor loader with both directories as search roots")
|
|
def step_when_create_loader_multi(context: Context) -> None:
|
|
roots = context._actor_dirs
|
|
context._loader = ActorLoader(search_roots=roots)
|
|
|
|
|
|
@when("I run discovery")
|
|
def step_when_run_discovery(context: Context) -> None:
|
|
loader: ActorLoader = context._loader
|
|
context._discovered = loader.discover()
|
|
|
|
|
|
@when("I run discovery expecting errors")
|
|
def step_when_run_discovery_errors(context: Context) -> None:
|
|
loader: ActorLoader = context._loader
|
|
try:
|
|
loader.discover()
|
|
context._discovery_error = None
|
|
except (ValidationError, Exception) as exc:
|
|
context._discovery_error = exc
|
|
|
|
|
|
@when("I run discovery again")
|
|
def step_when_run_discovery_again(context: Context) -> None:
|
|
loader: ActorLoader = context._loader
|
|
context._discovered = loader.discover()
|
|
|
|
|
|
@when('I modify the actor file "{filename}" to change the model to "{new_model}"')
|
|
def step_when_modify_file(context: Context, filename: str, new_model: str) -> None:
|
|
d: Path = context._actor_dir
|
|
path = d / filename
|
|
text = path.read_text()
|
|
text = text.replace("model: gpt-4", f"model: {new_model}")
|
|
path.write_text(text)
|
|
|
|
|
|
@when("I clear the loader cache")
|
|
def step_when_clear_cache(context: Context) -> None:
|
|
loader: ActorLoader = context._loader
|
|
loader.clear()
|
|
|
|
|
|
@when('I delete the actor file "{filename}"')
|
|
def step_when_delete_file(context: Context, filename: str) -> None:
|
|
d: Path = context._actor_dir
|
|
(d / filename).unlink()
|
|
|
|
|
|
@when("I create an actor loader with the examples/actors directory")
|
|
def step_when_create_loader_examples(context: Context) -> None:
|
|
context._loader = ActorLoader(search_roots=[context._actor_dir])
|
|
|
|
|
|
@when("I create an actor loader with that directory and the tool registry")
|
|
def step_when_create_loader_with_tools(context: Context) -> None:
|
|
context._loader = ActorLoader(
|
|
search_roots=[context._actor_dir],
|
|
tool_registry=context._tool_registry,
|
|
)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────
|
|
# Then steps
|
|
# ──────────────────────────────────────────────────────
|
|
|
|
|
|
@then("the loader should find {count:d} actors")
|
|
def step_then_loader_count(context: Context, count: int) -> None:
|
|
loader: ActorLoader = context._loader
|
|
actors = loader.list_actors()
|
|
assert len(actors) == count, (
|
|
f"Expected {count} actors, got {len(actors)}: {[a.name for a in actors]}"
|
|
)
|
|
|
|
|
|
@then("the loader should find at least {count:d} actors")
|
|
def step_then_loader_min_count(context: Context, count: int) -> None:
|
|
loader: ActorLoader = context._loader
|
|
actors = loader.list_actors()
|
|
assert len(actors) >= count, f"Expected at least {count} actors, got {len(actors)}"
|
|
|
|
|
|
@then('the loader should contain actor "{name}"')
|
|
def step_then_loader_contains(context: Context, name: str) -> None:
|
|
loader: ActorLoader = context._loader
|
|
config = loader.get(name)
|
|
assert config is not None, f"Actor '{name}' not found in loader"
|
|
|
|
|
|
@then('the loader should raise a validation error mentioning "{text}"')
|
|
def step_then_validation_error_with_text(context: Context, text: str) -> None:
|
|
err = context._discovery_error
|
|
assert err is not None, "Expected a validation error but none was raised"
|
|
assert text in str(err), f"Expected '{text}' in error message: {err}"
|
|
|
|
|
|
@then("the loader should raise a validation error")
|
|
def step_then_validation_error(context: Context) -> None:
|
|
err = context._discovery_error
|
|
assert err is not None, "Expected a validation error but none was raised"
|
|
|
|
|
|
@then('I can retrieve actor "{name}" by name')
|
|
def step_then_retrieve_by_name(context: Context, name: str) -> None:
|
|
loader: ActorLoader = context._loader
|
|
config = loader.get(name)
|
|
assert config is not None, f"Actor '{name}' not found"
|
|
context._retrieved_actor = config
|
|
|
|
|
|
@then('the retrieved actor type should be "{actor_type}"')
|
|
def step_then_retrieved_type(context: Context, actor_type: str) -> None:
|
|
config: ActorConfigSchema = context._retrieved_actor
|
|
assert config.type.value == actor_type, (
|
|
f"Expected type '{actor_type}', got '{config.type.value}'"
|
|
)
|
|
|
|
|
|
@then('getting actor "{name}" should return None')
|
|
def step_then_get_none(context: Context, name: str) -> None:
|
|
loader: ActorLoader = context._loader
|
|
config = loader.get(name)
|
|
assert config is None, f"Expected None for '{name}', got {config}"
|
|
|
|
|
|
@then(
|
|
'listing loaded actors with namespace "{namespace}" should return {count:d} actors'
|
|
)
|
|
def step_then_list_by_namespace(context: Context, namespace: str, count: int) -> None:
|
|
loader: ActorLoader = context._loader
|
|
actors = loader.list_actors(namespace=namespace)
|
|
assert len(actors) == count, (
|
|
f"Expected {count} actors in namespace '{namespace}', got {len(actors)}: "
|
|
f"{[a.name for a in actors]}"
|
|
)
|
|
|
|
|
|
@then('the actor "{name}" should have been loaded only once')
|
|
def step_then_loaded_once(context: Context, name: str) -> None:
|
|
loader: ActorLoader = context._loader
|
|
count = loader.get_load_count(name)
|
|
assert count == 1, f"Expected load_count=1 for '{name}', got {count}"
|
|
|
|
|
|
@then('the actor "{name}" model should be "{model}"')
|
|
def step_then_actor_model(context: Context, name: str, model: str) -> None:
|
|
loader: ActorLoader = context._loader
|
|
config = loader.get(name)
|
|
assert config is not None, f"Actor '{name}' not found"
|
|
assert config.model == model, f"Expected model '{model}', got '{config.model}'"
|
|
|
|
|
|
@then('the actor "{name}" tools should include "{tool_name}"')
|
|
def step_then_actor_tools_include(context: Context, name: str, tool_name: str) -> None:
|
|
loader: ActorLoader = context._loader
|
|
config = loader.get(name)
|
|
assert config is not None, f"Actor '{name}' not found"
|
|
tool_names = [t if isinstance(t, str) else t.name for t in config.tools]
|
|
assert tool_name in tool_names, f"Tool '{tool_name}' not in {tool_names}"
|
|
|
|
|
|
@then('the loader should have a warning about unresolved tool "{tool_name}"')
|
|
def step_then_warning_unresolved(context: Context, tool_name: str) -> None:
|
|
loader: ActorLoader = context._loader
|
|
warnings = loader.warnings
|
|
matched = [w for w in warnings if tool_name in w]
|
|
assert matched, f"No warning about '{tool_name}' found in: {warnings}"
|
|
|
|
|
|
@then("the loader should have {count:d} warnings")
|
|
def step_then_warning_count(context: Context, count: int) -> None:
|
|
loader: ActorLoader = context._loader
|
|
warnings = loader.warnings
|
|
assert len(warnings) == count, (
|
|
f"Expected {count} warnings, got {len(warnings)}: {warnings}"
|
|
)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────
|
|
# Given steps for coverage edge cases
|
|
# ──────────────────────────────────────────────────────
|
|
|
|
|
|
@given("a non-existent directory path as search root")
|
|
def step_given_nonexistent_dir(context: Context) -> None:
|
|
context._actor_dir = Path("/tmp/does_not_exist_actorloader_test")
|
|
|
|
|
|
@given('a temporary actors directory with a list-content YAML file "{filename}"')
|
|
def step_given_list_yaml(context: Context, filename: str) -> None:
|
|
d = _fresh_dir(context, "listyaml")
|
|
context._actor_dir = d
|
|
(d / filename).write_text("- item_one\n- item_two\n")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────
|
|
# Concurrency test steps
|
|
# ──────────────────────────────────────────────────────
|
|
|
|
|
|
@when(
|
|
"{n:d} threads concurrently call list_actors with namespace "
|
|
'"{namespace}" while another thread calls clear'
|
|
)
|
|
def step_when_concurrent_list_actors_with_clear(
|
|
context: Context, n: int, namespace: str
|
|
) -> None:
|
|
"""Test race condition: list_actors with namespace filter under concurrent clear."""
|
|
import threading
|
|
|
|
loader: ActorLoader = context._loader
|
|
results: list[list[ActorConfigSchema]] = []
|
|
errors: list[Exception] = []
|
|
barrier = threading.Barrier(n + 1) # n list_actors threads + 1 clear thread
|
|
|
|
def list_actors_worker() -> None:
|
|
try:
|
|
barrier.wait() # All threads start simultaneously
|
|
result = loader.list_actors(namespace=namespace)
|
|
results.append(result)
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
def clear_worker() -> None:
|
|
try:
|
|
barrier.wait() # All threads start simultaneously
|
|
# Small delay to let list_actors threads acquire the lock first
|
|
import time
|
|
|
|
time.sleep(0.001)
|
|
loader.clear()
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
# Create n threads for list_actors calls
|
|
list_threads = [threading.Thread(target=list_actors_worker) for _ in range(n)]
|
|
# Create 1 thread for clear
|
|
clear_thread = threading.Thread(target=clear_worker)
|
|
|
|
# Start all threads
|
|
for t in list_threads:
|
|
t.start()
|
|
clear_thread.start()
|
|
|
|
# Wait for all threads to complete
|
|
for t in list_threads:
|
|
t.join()
|
|
clear_thread.join()
|
|
|
|
assert not errors, f"Threads raised exceptions: {errors}"
|
|
|
|
# Store results and namespace for assertion
|
|
context._concurrent_list_results = results
|
|
context._concurrent_namespace = namespace
|
|
|
|
|
|
@then("all concurrent list_actors calls should return consistent results")
|
|
def step_then_concurrent_results_consistent(context: Context) -> None:
|
|
"""Verify that all concurrent list_actors calls returned consistent results."""
|
|
results: list[list[ActorConfigSchema]] = context._concurrent_list_results
|
|
namespace: str = context._concurrent_namespace
|
|
|
|
# Each result should only contain actors from the requested namespace.
|
|
# When clear() runs concurrently, some threads may see the full list while
|
|
# others see an empty list, but no result should have stale/partial data.
|
|
if not results:
|
|
return
|
|
|
|
# Check that each result only contains actors from the requested namespace
|
|
for i, result in enumerate(results):
|
|
for actor_config in result:
|
|
assert actor_config.name.startswith(f"{namespace}/"), (
|
|
f"Result {i} contains actor from wrong namespace: {actor_config.name}"
|
|
)
|