354 lines
13 KiB
Python
354 lines
13 KiB
Python
"""Step definitions for A2A naming regression tests."""
|
|
|
|
import importlib
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
|
|
|
|
@given("the A2A module is properly installed")
|
|
def step_a2a_module_installed(context: Any) -> None:
|
|
"""Verify the A2A module is properly installed."""
|
|
try:
|
|
import cleveragents.a2a # noqa: F401
|
|
context.a2a_module = importlib.import_module("cleveragents.a2a")
|
|
except ImportError as e:
|
|
raise AssertionError(f"A2A module not installed: {e}") from e
|
|
|
|
|
|
@given("no ACP references exist in the codebase")
|
|
def step_no_acp_references(context: Any) -> None:
|
|
"""Verify no ACP references exist in the codebase."""
|
|
# This is verified by the linting step, but we can check here too
|
|
src_path = Path(__file__).parent.parent.parent / "src"
|
|
acp_found = False
|
|
|
|
for py_file in src_path.rglob("*.py"):
|
|
with open(py_file, encoding="utf-8") as f:
|
|
content = f.read()
|
|
if "acp" in content.lower():
|
|
# Check if it's actually an ACP reference (not just in comments)
|
|
for line in content.split("\n"):
|
|
if ("acp" in line.lower() and not line.strip().startswith("#") and
|
|
("from cleveragents.acp" in line or "import cleveragents.acp" in line)):
|
|
acp_found = True
|
|
break
|
|
|
|
if acp_found:
|
|
raise AssertionError("ACP references found in source code")
|
|
|
|
|
|
@when('I import the A2A facade from "cleveragents.a2a"')
|
|
def step_import_a2a_facade(context: Any) -> None:
|
|
"""Import the A2A facade from the new module path."""
|
|
try:
|
|
from cleveragents.a2a import A2aLocalFacade
|
|
context.a2a_facade = A2aLocalFacade
|
|
except ImportError as e:
|
|
raise AssertionError(f"Failed to import A2A facade: {e}") from e
|
|
|
|
|
|
@then("the import succeeds")
|
|
def step_import_succeeds(context: Any) -> None:
|
|
"""Verify the import succeeded."""
|
|
assert hasattr(context, "a2a_facade") or hasattr(context, "a2a_clients") or \
|
|
hasattr(context, "a2a_models") or hasattr(context, "a2a_errors") or \
|
|
hasattr(context, "a2a_events"), "Import did not succeed"
|
|
|
|
|
|
@then("the facade is an instance of A2aLocalFacade")
|
|
def step_facade_is_a2a_facade(context: Any) -> None:
|
|
"""Verify the facade is an instance of A2aLocalFacade."""
|
|
from cleveragents.a2a import A2aLocalFacade
|
|
assert context.a2a_facade is A2aLocalFacade, "Facade is not A2aLocalFacade"
|
|
|
|
|
|
@when('I import A2A clients from "cleveragents.a2a"')
|
|
def step_import_a2a_clients(context: Any) -> None:
|
|
"""Import A2A clients from the new module path."""
|
|
try:
|
|
from cleveragents.a2a import ServerClient
|
|
context.a2a_clients = ServerClient
|
|
except ImportError as e:
|
|
raise AssertionError(f"Failed to import A2A clients: {e}") from e
|
|
|
|
|
|
@then("the clients module contains ServerClient")
|
|
def step_clients_contains_factory(context: Any) -> None:
|
|
"""Verify the clients module contains ServerClient."""
|
|
from cleveragents.a2a import ServerClient
|
|
assert ServerClient is not None, \
|
|
"ServerClient not found in clients module"
|
|
|
|
|
|
@when('I import A2A models from "cleveragents.a2a"')
|
|
def step_import_a2a_models(context: Any) -> None:
|
|
"""Import A2A models from the new module path."""
|
|
try:
|
|
from cleveragents.a2a import A2aRequest
|
|
context.a2a_models = A2aRequest
|
|
except ImportError as e:
|
|
raise AssertionError(f"Failed to import A2A models: {e}") from e
|
|
|
|
|
|
@then("the models module contains A2aRequest")
|
|
def step_models_contains_message(context: Any) -> None:
|
|
"""Verify the models module contains A2aRequest."""
|
|
from cleveragents.a2a import A2aRequest
|
|
assert A2aRequest is not None, \
|
|
"A2aRequest not found in models module"
|
|
|
|
|
|
@when('I import A2A errors from "cleveragents.a2a"')
|
|
def step_import_a2a_errors(context: Any) -> None:
|
|
"""Import A2A errors from the new module path."""
|
|
try:
|
|
from cleveragents.a2a import A2aError
|
|
context.a2a_errors = A2aError
|
|
except ImportError as e:
|
|
raise AssertionError(f"Failed to import A2A errors: {e}") from e
|
|
|
|
|
|
@then("the errors module contains A2aError")
|
|
def step_errors_contains_error(context: Any) -> None:
|
|
"""Verify the errors module contains A2aError."""
|
|
from cleveragents.a2a import A2aError
|
|
assert A2aError is not None, \
|
|
"A2aError not found in errors module"
|
|
|
|
|
|
@when('I import A2A events from "cleveragents.a2a"')
|
|
def step_import_a2a_events(context: Any) -> None:
|
|
"""Import A2A events from the new module path."""
|
|
try:
|
|
from cleveragents.a2a import A2aEventQueue
|
|
context.a2a_events = A2aEventQueue
|
|
except ImportError as e:
|
|
raise AssertionError(f"Failed to import A2A events: {e}") from e
|
|
|
|
|
|
@then("the events module contains A2aEventQueue")
|
|
def step_events_contains_emitter(context: Any) -> None:
|
|
"""Verify the events module contains A2aEventQueue."""
|
|
from cleveragents.a2a import A2aEventQueue
|
|
assert A2aEventQueue is not None, \
|
|
"A2aEventQueue not found in events module"
|
|
|
|
|
|
@when('I attempt to import from old "cleveragents.acp" path')
|
|
def step_attempt_import_acp(context: Any) -> None:
|
|
"""Attempt to import from the old ACP path."""
|
|
context.import_error = None
|
|
try:
|
|
importlib.import_module("cleveragents.acp")
|
|
except ImportError as e:
|
|
context.import_error = e
|
|
|
|
|
|
@then("an ImportError is raised")
|
|
def step_import_error_raised(context: Any) -> None:
|
|
"""Verify an ImportError was raised."""
|
|
assert context.import_error is not None, "ImportError was not raised"
|
|
|
|
|
|
@then("the error message indicates the module does not exist")
|
|
def step_error_message_correct(context: Any) -> None:
|
|
"""Verify the error message indicates the module does not exist."""
|
|
error_msg = str(context.import_error).lower()
|
|
assert "no module" in error_msg or "cannot find" in error_msg or "acp" in error_msg, \
|
|
f"Error message does not indicate missing module: {context.import_error}"
|
|
|
|
|
|
@when('I import from "cleveragents.a2a"')
|
|
def step_import_from_a2a(context: Any) -> None:
|
|
"""Import from the A2A module."""
|
|
try:
|
|
context.a2a_module = importlib.import_module("cleveragents.a2a")
|
|
except ImportError as e:
|
|
raise AssertionError(f"Failed to import from cleveragents.a2a: {e}") from e
|
|
|
|
|
|
@then("the module exports A2aLocalFacade")
|
|
def step_module_exports_facade(context: Any) -> None:
|
|
"""Verify the module exports A2aLocalFacade."""
|
|
assert hasattr(context.a2a_module, "A2aLocalFacade"), \
|
|
"A2aLocalFacade not exported from cleveragents.a2a"
|
|
|
|
|
|
@then("the module exports ServerClient")
|
|
def step_module_exports_factory(context: Any) -> None:
|
|
"""Verify the module exports ServerClient."""
|
|
assert hasattr(context.a2a_module, "ServerClient"), \
|
|
"ServerClient not exported from cleveragents.a2a"
|
|
|
|
|
|
@then("the module exports A2aRequest")
|
|
def step_module_exports_message(context: Any) -> None:
|
|
"""Verify the module exports A2aRequest."""
|
|
assert hasattr(context.a2a_module, "A2aRequest"), \
|
|
"A2aRequest not exported from cleveragents.a2a"
|
|
|
|
|
|
@then("the module exports A2aError")
|
|
def step_module_exports_error(context: Any) -> None:
|
|
"""Verify the module exports A2aError."""
|
|
assert hasattr(context.a2a_module, "A2aError"), \
|
|
"A2aError not exported from cleveragents.a2a"
|
|
|
|
|
|
@then("the module exports A2aEventQueue")
|
|
def step_module_exports_emitter(context: Any) -> None:
|
|
"""Verify the module exports A2aEventQueue."""
|
|
assert hasattr(context.a2a_module, "A2aEventQueue"), \
|
|
"A2aEventQueue not exported from cleveragents.a2a"
|
|
|
|
|
|
@when("I scan the source code for ACP references")
|
|
def step_scan_acp_references(context: Any) -> None:
|
|
"""Scan the source code for ACP references."""
|
|
src_path = Path(__file__).parent.parent.parent / "src"
|
|
context.acp_imports = []
|
|
context.acp_from_imports = []
|
|
context.acp_import_statements = []
|
|
|
|
for py_file in src_path.rglob("*.py"):
|
|
with open(py_file, encoding="utf-8") as f:
|
|
for line_num, line in enumerate(f, 1):
|
|
if line.strip().startswith("#"):
|
|
continue
|
|
if "from cleveragents.acp" in line:
|
|
context.acp_from_imports.append((py_file, line_num, line.strip()))
|
|
if "import cleveragents.acp" in line:
|
|
context.acp_import_statements.append((py_file, line_num, line.strip()))
|
|
if " acp " in line.lower() and "a2a" not in line.lower():
|
|
context.acp_imports.append((py_file, line_num, line.strip()))
|
|
|
|
|
|
@then('no "acp" imports are found')
|
|
def step_no_acp_imports(context: Any) -> None:
|
|
"""Verify no ACP imports are found."""
|
|
assert len(context.acp_imports) == 0, \
|
|
f"ACP imports found: {context.acp_imports}"
|
|
|
|
|
|
@then('no "from cleveragents.acp" statements are found')
|
|
def step_no_from_acp_statements(context: Any) -> None:
|
|
"""Verify no 'from cleveragents.acp' statements are found."""
|
|
assert len(context.acp_from_imports) == 0, \
|
|
f"'from cleveragents.acp' statements found: {context.acp_from_imports}"
|
|
|
|
|
|
@then('no "import cleveragents.acp" statements are found')
|
|
def step_no_import_acp_statements(context: Any) -> None:
|
|
"""Verify no 'import cleveragents.acp' statements are found."""
|
|
assert len(context.acp_import_statements) == 0, \
|
|
f"'import cleveragents.acp' statements found: {context.acp_import_statements}"
|
|
|
|
|
|
@when("I inspect the A2A module structure")
|
|
def step_inspect_a2a_structure(context: Any) -> None:
|
|
"""Inspect the A2A module structure."""
|
|
a2a_path = Path(__file__).parent.parent.parent / "src" / "cleveragents" / "a2a"
|
|
context.a2a_files = set(f.name for f in a2a_path.glob("*.py"))
|
|
|
|
|
|
@then("the module contains __init__.py")
|
|
def step_module_contains_init(context: Any) -> None:
|
|
"""Verify the module contains __init__.py."""
|
|
assert "__init__.py" in context.a2a_files, "__init__.py not found"
|
|
|
|
|
|
@then("the module contains facade.py")
|
|
def step_module_contains_facade(context: Any) -> None:
|
|
"""Verify the module contains facade.py."""
|
|
assert "facade.py" in context.a2a_files, "facade.py not found"
|
|
|
|
|
|
@then("the module contains clients.py")
|
|
def step_module_contains_clients(context: Any) -> None:
|
|
"""Verify the module contains clients.py."""
|
|
assert "clients.py" in context.a2a_files, "clients.py not found"
|
|
|
|
|
|
@then("the module contains models.py")
|
|
def step_module_contains_models(context: Any) -> None:
|
|
"""Verify the module contains models.py."""
|
|
assert "models.py" in context.a2a_files, "models.py not found"
|
|
|
|
|
|
@then("the module contains errors.py")
|
|
def step_module_contains_errors(context: Any) -> None:
|
|
"""Verify the module contains errors.py."""
|
|
assert "errors.py" in context.a2a_files, "errors.py not found"
|
|
|
|
|
|
@then("the module contains events.py")
|
|
def step_module_contains_events(context: Any) -> None:
|
|
"""Verify the module contains events.py."""
|
|
assert "events.py" in context.a2a_files, "events.py not found"
|
|
|
|
|
|
@then("the module contains asgi.py")
|
|
def step_module_contains_asgi(context: Any) -> None:
|
|
"""Verify the module contains asgi.py."""
|
|
assert "asgi.py" in context.a2a_files, "asgi.py not found"
|
|
|
|
|
|
@then("the module contains transport.py")
|
|
def step_module_contains_transport(context: Any) -> None:
|
|
"""Verify the module contains transport.py."""
|
|
assert "transport.py" in context.a2a_files, "transport.py not found"
|
|
|
|
|
|
@then("the module contains versioning.py")
|
|
def step_module_contains_versioning(context: Any) -> None:
|
|
"""Verify the module contains versioning.py."""
|
|
assert "versioning.py" in context.a2a_files, "versioning.py not found"
|
|
|
|
|
|
@then("the module contains server_config.py")
|
|
def step_module_contains_server_config(context: Any) -> None:
|
|
"""Verify the module contains server_config.py."""
|
|
assert "server_config.py" in context.a2a_files, "server_config.py not found"
|
|
|
|
|
|
@then("the module contains cli_bootstrap.py")
|
|
def step_module_contains_cli_bootstrap(context: Any) -> None:
|
|
"""Verify the module contains cli_bootstrap.py."""
|
|
assert "cli_bootstrap.py" in context.a2a_files, "cli_bootstrap.py not found"
|
|
|
|
|
|
@when("I initialize the A2A facade")
|
|
def step_initialize_a2a_facade(context: Any) -> None:
|
|
"""Initialize the A2A facade."""
|
|
try:
|
|
from cleveragents.a2a import A2aLocalFacade
|
|
context.facade_instance = A2aLocalFacade()
|
|
except Exception as e:
|
|
context.facade_init_error = e
|
|
|
|
|
|
@then("the facade initializes without errors")
|
|
def step_facade_initializes(context: Any) -> None:
|
|
"""Verify the facade initializes without errors."""
|
|
assert not hasattr(context, "facade_init_error"), \
|
|
f"Facade initialization failed: {getattr(context, 'facade_init_error', None)}"
|
|
assert hasattr(context, "facade_instance"), "Facade instance not created"
|
|
|
|
|
|
@then("the facade has required methods")
|
|
def step_facade_has_methods(context: Any) -> None:
|
|
"""Verify the facade has required methods."""
|
|
facade = context.facade_instance
|
|
# Check for common facade methods
|
|
assert callable(getattr(facade, "__init__", None)), "Facade missing __init__"
|
|
|
|
|
|
@then("the facade can be used for A2A operations")
|
|
def step_facade_usable(context: Any) -> None:
|
|
"""Verify the facade can be used for A2A operations."""
|
|
facade = context.facade_instance
|
|
assert facade is not None, "Facade is None"
|
|
assert hasattr(facade, "__class__"), "Facade has no class"
|