Files
cleveragents-core/robot/helper_test_data_factory.py
hurui200320 0127b6f745
CI / lint (pull_request) Successful in 1m7s
CI / typecheck (pull_request) Successful in 1m13s
CI / security (pull_request) Successful in 1m13s
CI / benchmark-publish (pull_request) Has been skipped
CI / push-validation (pull_request) Successful in 33s
CI / helm (pull_request) Successful in 35s
CI / build (pull_request) Successful in 51s
CI / quality (pull_request) Successful in 56s
CI / e2e_tests (pull_request) Successful in 4m33s
CI / integration_tests (pull_request) Successful in 5m4s
CI / unit_tests (pull_request) Successful in 6m24s
CI / docker (pull_request) Successful in 1m29s
CI / coverage (pull_request) Successful in 11m42s
CI / status-check (pull_request) Successful in 5s
CI / build (push) Successful in 50s
CI / helm (push) Successful in 30s
CI / push-validation (push) Successful in 28s
CI / lint (push) Successful in 1m6s
CI / quality (push) Successful in 1m10s
CI / typecheck (push) Successful in 1m33s
CI / security (push) Successful in 1m37s
CI / benchmark-publish (push) Failing after 48s
CI / e2e_tests (push) Successful in 3m54s
CI / integration_tests (push) Successful in 4m3s
CI / unit_tests (push) Successful in 6m6s
CI / docker (push) Successful in 1m32s
CI / coverage (push) Successful in 12m3s
CI / status-check (push) Successful in 3s
fix(reactive): synthesise execution route for type:llm actors in ReactiveConfigParser
When _build_from_v3() creates agents for type:llm or type:tool actors, and
when _build() processes the nested actors: map format (cleveragents version
3.0 YAML), no RouteConfig is produced. run_single_shot() then falls through
to the RxPY stream path which has no subscribers, causing the LLM to never
be invoked and the command to silently return empty output.

Fix A: after creating the agent in _build_from_v3() for type:llm/tool, call
_synthesise_single_node_route() to add a graph route with a message_router
node (catch-all rule), an actor node, and an edge to "end".

Fix B: after the agent loop in _build(), if rc.agents is non-empty and
rc.routes is empty, synthesise a default route using the cleveragents
default_actor (or the first agent).

Fix C: in _build(), the nested actors: map path now translates the v3
actor: "provider/model" key into separate provider and model keys in the
agent config dict. SimpleLLMAgent._resolve_llm() expects these keys; without
this translation, the LLM provider defaults to None (OpenAI) regardless of
the configured actor reference.

13 BDD scenarios in actor_v3_route_synthesis.feature:
- Flat v3 LLM/tool builds produce non-empty routes
- Route structure has router + actor nodes with edge to end
- Nested actors: map format produces routes, respects default_actor
- Explicit routes are not duplicated
- run_single_shot returns non-empty output with synthesised routes
- Graph actor regression guard
- Nested actors: map with actor key infers provider and model
- Nested actors: map with actor key without slash sets model only
- Nested actors: map with explicit provider/model keeps them unchanged

ISSUES CLOSED: #10807
2026-04-28 03:22:06 +00:00

375 lines
11 KiB
Python

"""Robot Framework test data factory helper using Faker for realistic data generation.
This module provides a CLI-style interface for Robot Framework to generate realistic
test data using Faker. It replaces hardcoded test data with dynamically generated,
varied, and realistic values.
Usage:
python robot/helper_test_data_factory.py generate-fragment
python robot/helper_test_data_factory.py generate-fragments 5
python robot/helper_test_data_factory.py generate-project-name
python robot/helper_test_data_factory.py generate-code
python robot/helper_test_data_factory.py generate-budget
"""
from __future__ import annotations
import json
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any
from faker import Faker
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
ContextBudget,
ContextFragment,
FragmentProvenance,
)
_faker = Faker()
class RobotTestDataGenerator:
"""Generate realistic test data for Robot Framework tests."""
@staticmethod
def project_name() -> str:
"""Generate a realistic project name."""
return _faker.word() + "-" + _faker.word()
@staticmethod
def skill_name() -> str:
"""Generate a realistic skill name."""
return _faker.word() + "_" + _faker.word()
@staticmethod
def tool_name() -> str:
"""Generate a realistic tool name."""
return _faker.word() + "-tool"
@staticmethod
def actor_name() -> str:
"""Generate a realistic actor name."""
return _faker.first_name() + _faker.last_name()
@staticmethod
def python_code() -> str:
"""Generate a realistic Python code snippet."""
var_name = _faker.word()
value = _faker.random_int(min=1, max=1000)
return f"{var_name} = {value}"
@staticmethod
def python_prose() -> str:
"""Generate realistic Python-like prose."""
return _faker.sentence(nb_words=10)
@staticmethod
def file_path() -> str:
"""Generate a realistic file path."""
return f"src/{_faker.word()}/{_faker.word()}.py"
@staticmethod
def uko_node_uri() -> str:
"""Generate a realistic UKO node URI."""
return f"p://{_faker.word()}/{_faker.word()}.py"
@staticmethod
def resource_uri() -> str:
"""Generate a realistic resource URI."""
return f"test://{_faker.word()}-{_faker.word()}"
@staticmethod
def relevance_score() -> float:
"""Generate a realistic relevance score (0.0 to 1.0)."""
return round(_faker.pyfloat(min_value=0.0, max_value=1.0, right_digits=2), 2)
@staticmethod
def token_count() -> int:
"""Generate a realistic token count."""
return _faker.random_int(min=10, max=500)
@staticmethod
def detail_depth() -> int:
"""Generate a realistic detail depth level."""
return _faker.random_int(min=1, max=9)
@staticmethod
def ulid() -> str:
"""Generate a valid ULID-like string (26 chars, base32)."""
valid_chars = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
return "".join(_faker.random.choice(valid_chars) for _ in range(26))
class RobotContextFragmentFactory:
"""Factory for creating realistic ContextFragment test objects for Robot."""
@staticmethod
def create_dict(**overrides: Any) -> dict[str, Any]:
"""Create a ContextFragment as a dictionary (for Robot serialization).
Args:
**overrides: Override any default values
Returns:
A dictionary representation of a ContextFragment
"""
defaults = {
"uko_node": RobotTestDataGenerator.uko_node_uri(),
"content": RobotTestDataGenerator.python_code(),
"token_count": RobotTestDataGenerator.token_count(),
"relevance_score": RobotTestDataGenerator.relevance_score(),
"detail_depth": RobotTestDataGenerator.detail_depth(),
"provenance": {
"resource_uri": RobotTestDataGenerator.resource_uri(),
},
}
defaults.update(overrides)
return defaults
@staticmethod
def create_object(**overrides: Any) -> ContextFragment:
"""Create a ContextFragment object.
Args:
**overrides: Override any default values
Returns:
A ContextFragment instance
"""
data = RobotContextFragmentFactory.create_dict(**overrides)
provenance_data = data.pop("provenance")
provenance = FragmentProvenance(**provenance_data)
return ContextFragment(provenance=provenance, **data)
@staticmethod
def create_batch_dicts(count: int, **overrides: Any) -> list[dict[str, Any]]:
"""Create multiple ContextFragments as dictionaries.
Args:
count: Number of fragments to create
**overrides: Override any default values
Returns:
A list of dictionary representations
"""
return [
RobotContextFragmentFactory.create_dict(**overrides) for _ in range(count)
]
@staticmethod
def create_batch_objects(count: int, **overrides: Any) -> list[ContextFragment]:
"""Create multiple ContextFragment objects.
Args:
count: Number of fragments to create
**overrides: Override any default values
Returns:
A list of ContextFragment instances
"""
return [
RobotContextFragmentFactory.create_object(**overrides) for _ in range(count)
]
class RobotContextBudgetFactory:
"""Factory for creating realistic ContextBudget test objects for Robot."""
@staticmethod
def create_dict(
max_tokens: int | None = None,
reserved_tokens: int | None = None,
) -> dict[str, Any]:
"""Create a ContextBudget as a dictionary.
Args:
max_tokens: Maximum tokens (default: random 100-1000)
reserved_tokens: Reserved tokens (default: 0)
Returns:
A dictionary representation of a ContextBudget
"""
if max_tokens is None:
max_tokens = _faker.random_int(min=100, max=1000)
if reserved_tokens is None:
reserved_tokens = 0
return {
"max_tokens": max_tokens,
"reserved_tokens": reserved_tokens,
}
@staticmethod
def create_object(
max_tokens: int | None = None,
reserved_tokens: int | None = None,
) -> ContextBudget:
"""Create a ContextBudget object.
Args:
max_tokens: Maximum tokens (default: random 100-1000)
reserved_tokens: Reserved tokens (default: 0)
Returns:
A ContextBudget instance
"""
data = RobotContextBudgetFactory.create_dict(max_tokens, reserved_tokens)
return ContextBudget(**data)
# ---------------------------------------------------------------------------
# CLI commands for Robot Framework
# ---------------------------------------------------------------------------
def _cmd_generate_fragment() -> int:
"""Generate a single context fragment as JSON."""
frag_dict = RobotContextFragmentFactory.create_dict()
print(json.dumps(frag_dict, indent=2))
return 0
def _cmd_generate_fragments() -> int:
"""Generate multiple context fragments as JSON."""
count = int(sys.argv[2]) if len(sys.argv) > 2 else 5
frags = RobotContextFragmentFactory.create_batch_dicts(count)
print(json.dumps(frags, indent=2))
return 0
def _cmd_generate_project_name() -> int:
"""Generate a realistic project name."""
print(RobotTestDataGenerator.project_name())
return 0
def _cmd_generate_skill_name() -> int:
"""Generate a realistic skill name."""
print(RobotTestDataGenerator.skill_name())
return 0
def _cmd_generate_tool_name() -> int:
"""Generate a realistic tool name."""
print(RobotTestDataGenerator.tool_name())
return 0
def _cmd_generate_actor_name() -> int:
"""Generate a realistic actor name."""
print(RobotTestDataGenerator.actor_name())
return 0
def _cmd_generate_code() -> int:
"""Generate a realistic Python code snippet."""
print(RobotTestDataGenerator.python_code())
return 0
def _cmd_generate_prose() -> int:
"""Generate realistic Python-like prose."""
print(RobotTestDataGenerator.python_prose())
return 0
def _cmd_generate_file_path() -> int:
"""Generate a realistic file path."""
print(RobotTestDataGenerator.file_path())
return 0
def _cmd_generate_uko_uri() -> int:
"""Generate a realistic UKO node URI."""
print(RobotTestDataGenerator.uko_node_uri())
return 0
def _cmd_generate_resource_uri() -> int:
"""Generate a realistic resource URI."""
print(RobotTestDataGenerator.resource_uri())
return 0
def _cmd_generate_relevance_score() -> int:
"""Generate a realistic relevance score."""
print(RobotTestDataGenerator.relevance_score())
return 0
def _cmd_generate_token_count() -> int:
"""Generate a realistic token count."""
print(RobotTestDataGenerator.token_count())
return 0
def _cmd_generate_detail_depth() -> int:
"""Generate a realistic detail depth."""
print(RobotTestDataGenerator.detail_depth())
return 0
def _cmd_generate_ulid() -> int:
"""Generate a valid ULID."""
print(RobotTestDataGenerator.ulid())
return 0
def _cmd_generate_budget() -> int:
"""Generate a context budget as JSON."""
budget_dict = RobotContextBudgetFactory.create_dict()
print(json.dumps(budget_dict, indent=2))
return 0
# ---------------------------------------------------------------------------
# CLI dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], int]] = {
"generate-fragment": _cmd_generate_fragment,
"generate-fragments": _cmd_generate_fragments,
"generate-project-name": _cmd_generate_project_name,
"generate-skill-name": _cmd_generate_skill_name,
"generate-tool-name": _cmd_generate_tool_name,
"generate-actor-name": _cmd_generate_actor_name,
"generate-code": _cmd_generate_code,
"generate-prose": _cmd_generate_prose,
"generate-file-path": _cmd_generate_file_path,
"generate-uko-uri": _cmd_generate_uko_uri,
"generate-resource-uri": _cmd_generate_resource_uri,
"generate-relevance-score": _cmd_generate_relevance_score,
"generate-token-count": _cmd_generate_token_count,
"generate-detail-depth": _cmd_generate_detail_depth,
"generate-ulid": _cmd_generate_ulid,
"generate-budget": _cmd_generate_budget,
}
def main() -> int:
"""Main entry point for CLI."""
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
print(
f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
file=sys.stderr,
)
return 1
try:
return _COMMANDS[sys.argv[1]]()
except Exception as exc:
print(f"FAIL: {exc}", file=sys.stderr)
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
sys.exit(main())