freemo/cleveragents-core
0
0
Fork 0
forked from cleveragents/cleveragents-core
Code Pull Requests Activity

fix(test): harden TDD bug #647 crash coverage and isolation

Browse Source 
Strengthened Container.resolve() crash tests to prevent false positives and cross-scenario state bleed by adding strict AttributeError checks, per-run plan IDs, and in-memory engine cache cleanup. Reduced Robot timeouts for faster failure feedback and added a concise review-resolution note for PR discussion context.

ISSUES CLOSED: #648
This commit is contained in:
Aditya Chhabra
2026-03-11 07:48:25 +00:00
parent bbd0c18d65
commit 452e1ecda7
3 changed files with 106 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/steps/container_resolve_crash_steps.py
+38 -52
View File
@@ -20,13 +20,13 @@ from behave.runner import Context
from typer.testing import CliRunner
from ulid import ULID
from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
if TYPE_CHECKING:
from cleveragents.domain.models.core.decision import Decision
cli_runner = CliRunner()
_PLAN_ID = str(ULID())
# ---------------------------------------------------------------------------
# GIVEN — Real container with seeded decisions
@@ -71,9 +71,10 @@ def step_cr647_setup_container(context: Context) -> None:
# Create DecisionService and seed decisions
decision_svc = DecisionService(settings=mock_settings, unit_of_work=uow)
# Seed a three-level decision tree
# Seed the minimum data required for CLI invocation.
plan_id = str(ULID())
root: Decision = decision_svc.record_decision(
plan_id=_PLAN_ID,
plan_id=plan_id,
decision_type=DecisionType.PROMPT_DEFINITION,
question="What should we build?",
chosen_option="A REST API",
@@ -90,55 +91,21 @@ def step_cr647_setup_container(context: Context) -> None:
),
)
child: Decision = decision_svc.record_decision(
plan_id=_PLAN_ID,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Which framework?",
chosen_option="FastAPI",
parent_decision_id=root.decision_id,
alternatives_considered=["Flask", "Django"],
confidence_score=0.90,
rationale="FastAPI provides async support",
context_snapshot=ContextSnapshot(
hot_context_hash="sha256:child",
hot_context_ref="store://child",
relevant_resources=[
ResourceRef(resource_id=str(ULID()), path="requirements.txt"),
],
actor_state_ref="checkpoint://child",
),
)
grandchild: Decision = decision_svc.record_decision(
plan_id=_PLAN_ID,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Which database?",
chosen_option="PostgreSQL",
parent_decision_id=child.decision_id,
alternatives_considered=["SQLite", "MySQL"],
confidence_score=0.85,
rationale="PostgreSQL supports relational workloads",
context_snapshot=ContextSnapshot(
hot_context_hash="sha256:grandchild",
hot_context_ref="store://grandchild",
relevant_resources=[
ResourceRef(resource_id=str(ULID()), path="docker-compose.yml"),
],
actor_state_ref="checkpoint://grandchild",
),
)
# Store IDs for use in test steps
context.cr647_plan_id = _PLAN_ID
# Store IDs for use in test steps.
# explain/correct only need one valid decision_id to trigger the resolve() path.
context.cr647_plan_id = plan_id
context.cr647_root_id = root.decision_id
context.cr647_child_id = child.decision_id
context.cr647_grandchild_id = grandchild.decision_id
context.cr647_child_id = root.decision_id
context.cr647_grandchild_id = root.decision_id
# Get the real container (not mocked)
context.cr647_container = get_container()
# Store cleanup handler
def cleanup():
def cleanup() -> None:
engine = MEMORY_ENGINES.pop(database_url, None)
if engine is not None:
engine.dispose()
reset_container()
os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
@@ -242,24 +209,43 @@ def step_cr647_check_attribute_error(context: Context) -> None:
"""
result = context.cr647_result
# The command should exit with non-zero code
# The command should exit with non-zero code.
assert result.exit_code != 0, (
f"Expected non-zero exit code, got {result.exit_code}. "
f"Output: {result.output}\nException: {result.exception}"
)
# Check for AttributeError with 'resolve' in the exception or output
# Check explicit exception type first to avoid false positives.
assert result.exception is not None, (
f"Expected exception, but got none. Output: {result.output}"
)
assert isinstance(result.exception, AttributeError), (
"Expected AttributeError, got "
f"{type(result.exception).__name__}: {result.exception}"
)
# Verify this is specifically the missing resolve() attribute on the container object.
exception_str = str(result.exception) if result.exception else ""
output_str = result.output or ""
combined = exception_str + output_str
# The error should mention 'resolve' and indicate it's not an attribute
assert "resolve" in combined.lower(), (
f"Expected 'resolve' in error output. Got:\n"
f"Exception: {exception_str}\nOutput: {output_str}"
)
assert "attributeerror" in combined.lower() or "attribute" in combined.lower(), (
f"Expected AttributeError in output. Got:\n"
assert "no attribute" in combined.lower(), (
f"Expected missing attribute message in output. Got:\n"
f"Exception: {exception_str}\nOutput: {output_str}"
)
assert "container" in combined.lower(), (
f"Expected container-related source in output. Got:\n"
f"Exception: {exception_str}\nOutput: {output_str}"
)
# POST-FIX ASSERTION TEMPLATE:
# When bug #647 is fixed and @tdd_expected_fail is removed,
# replace these checks with success assertions:
# assert result.exit_code == 0
# assert result.exception is None
robot/container_resolve_crash.robot
+3 -3
View File
@@ -26,7 +26,7 @@ ${HELPER} ${CURDIR}/helper_container_resolve_crash.py
Plan Tree Command Crashes With Container.resolve()
[Documentation] TDD test: plan tree calls container.resolve() which doesn't exist
[Tags] tdd_bug tdd_bug_647 tdd_expected_fail
${result}= Run Process ${PYTHON} ${HELPER} plan-tree-crash cwd=${WORKSPACE} timeout=120s
${result}= Run Process ${PYTHON} ${HELPER} plan-tree-crash cwd=${WORKSPACE} timeout=30s
Log ${result.stdout}
Log ${result.stderr}
# Helper exits 0 when it successfully reproduces the AttributeError crash
@@ -36,7 +36,7 @@ Plan Tree Command Crashes With Container.resolve()
Plan Explain Command Crashes With Container.resolve()
[Documentation] TDD test: plan explain calls container.resolve() which doesn't exist
[Tags] tdd_bug tdd_bug_647 tdd_expected_fail
${result}= Run Process ${PYTHON} ${HELPER} plan-explain-crash cwd=${WORKSPACE} timeout=120s
${result}= Run Process ${PYTHON} ${HELPER} plan-explain-crash cwd=${WORKSPACE} timeout=30s
Log ${result.stdout}
Log ${result.stderr}
# Helper exits 0 when it successfully reproduces the AttributeError crash
@@ -46,7 +46,7 @@ Plan Explain Command Crashes With Container.resolve()
Plan Correct Command Crashes With Container.resolve()
[Documentation] TDD test: plan correct calls container.resolve() which doesn't exist
[Tags] tdd_bug tdd_bug_647 tdd_expected_fail
${result}= Run Process ${PYTHON} ${HELPER} plan-correct-crash cwd=${WORKSPACE} timeout=120s
${result}= Run Process ${PYTHON} ${HELPER} plan-correct-crash cwd=${WORKSPACE} timeout=30s
Log ${result.stdout}
Log ${result.stderr}
# Helper exits 0 when it successfully reproduces the AttributeError crash
robot/helper_container_resolve_crash.py
+65 -65
View File
@@ -49,16 +49,16 @@ from cleveragents.domain.models.core.decision import (
DecisionType,
ResourceRef,
)
from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
cli_runner = CliRunner()
_PLAN_ID = str(ULID())
class DecisionIDs(NamedTuple):
"""Container for decision IDs returned by setup."""
plan_id: str
root_id: str
child_id: str
grandchild_id: str
@@ -79,6 +79,9 @@ def _cleanup() -> None:
"""Clean up test resources."""
from cleveragents.application.container import reset_container
engine = MEMORY_ENGINES.pop("sqlite:///:memory:", None)
if engine is not None:
engine.dispose()
reset_container()
os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
@@ -114,9 +117,10 @@ def _setup_decisions() -> DecisionIDs:
# Create DecisionService and seed decisions
decision_svc = DecisionService(settings=mock_settings, unit_of_work=uow)
# Seed a three-level decision tree
# Seed minimum decision data required for CLI invocation.
plan_id = str(ULID())
root = decision_svc.record_decision(
plan_id=_PLAN_ID,
plan_id=plan_id,
decision_type=DecisionType.PROMPT_DEFINITION,
question="What should we build?",
chosen_option="A REST API",
@@ -133,52 +137,15 @@ def _setup_decisions() -> DecisionIDs:
),
)
child = decision_svc.record_decision(
plan_id=_PLAN_ID,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Which framework?",
chosen_option="FastAPI",
parent_decision_id=root.decision_id,
alternatives_considered=["Flask", "Django"],
confidence_score=0.90,
rationale="FastAPI provides async support",
context_snapshot=ContextSnapshot(
hot_context_hash="sha256:child",
hot_context_ref="store://child",
relevant_resources=[
ResourceRef(resource_id=str(ULID()), path="requirements.txt"),
],
actor_state_ref="checkpoint://child",
),
)
grandchild = decision_svc.record_decision(
plan_id=_PLAN_ID,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Which database?",
chosen_option="PostgreSQL",
parent_decision_id=child.decision_id,
alternatives_considered=["SQLite", "MySQL"],
confidence_score=0.85,
rationale="PostgreSQL supports relational workloads",
context_snapshot=ContextSnapshot(
hot_context_hash="sha256:grandchild",
hot_context_ref="store://grandchild",
relevant_resources=[
ResourceRef(resource_id=str(ULID()), path="docker-compose.yml"),
],
actor_state_ref="checkpoint://grandchild",
),
)
# Ensure the real container is initialized
# (The CLI commands will call get_container() and get this instance)
_ = get_container()
return DecisionIDs(
plan_id=plan_id,
root_id=root.decision_id,
child_id=child.decision_id,
grandchild_id=grandchild.decision_id,
child_id=root.decision_id,
grandchild_id=root.decision_id,
)
@@ -194,12 +161,12 @@ def plan_tree_crash() -> None:
which fails because Container has no resolve() method.
"""
try:
_ = _setup_decisions()
decision_ids = _setup_decisions()
# Invoke the real CLI command without mocking get_container()
result = cli_runner.invoke(
plan_app,
["tree", _PLAN_ID, "--format", "json"],
["tree", decision_ids.plan_id, "--format", "json"],
catch_exceptions=True,
)
@@ -210,7 +177,15 @@ def plan_tree_crash() -> None:
f"Output: {result.output}"
)
# Verify the error is about resolve()
if result.exception is None:
_fail(f"Expected exception but got none. Output: {result.output}")
if not isinstance(result.exception, AttributeError):
_fail(
"Expected AttributeError, got "
f"{type(result.exception).__name__}: {result.exception}"
)
# Verify the error is about resolve() on the container object.
exception_str = str(result.exception) if result.exception else ""
output_str = result.output or ""
combined = exception_str + output_str
@@ -221,12 +196,15 @@ def plan_tree_crash() -> None:
f"Exception: {exception_str}\nOutput: {output_str}"
)
if (
"attributeerror" not in combined.lower()
and "attribute" not in combined.lower()
):
if "no attribute" not in combined.lower():
_fail(
f"Expected AttributeError, but got different error:\n"
f"Expected missing attribute message, but got:\n"
f"Exception: {exception_str}\nOutput: {output_str}"
)
if "container" not in combined.lower():
_fail(
f"Expected container-related source, but got:\n"
f"Exception: {exception_str}\nOutput: {output_str}"
)
@@ -265,7 +243,15 @@ def plan_explain_crash() -> None:
f"Output: {result.output}"
)
# Verify the error is about resolve()
if result.exception is None:
_fail(f"Expected exception but got none. Output: {result.output}")
if not isinstance(result.exception, AttributeError):
_fail(
"Expected AttributeError, got "
f"{type(result.exception).__name__}: {result.exception}"
)
# Verify the error is about resolve() on the container object.
exception_str = str(result.exception) if result.exception else ""
output_str = result.output or ""
combined = exception_str + output_str
@@ -276,12 +262,15 @@ def plan_explain_crash() -> None:
f"Exception: {exception_str}\nOutput: {output_str}"
)
if (
"attributeerror" not in combined.lower()
and "attribute" not in combined.lower()
):
if "no attribute" not in combined.lower():
_fail(
f"Expected AttributeError, but got different error:\n"
f"Expected missing attribute message, but got:\n"
f"Exception: {exception_str}\nOutput: {output_str}"
)
if "container" not in combined.lower():
_fail(
f"Expected container-related source, but got:\n"
f"Exception: {exception_str}\nOutput: {output_str}"
)
@@ -310,7 +299,7 @@ def plan_correct_crash() -> None:
"--guidance",
"Use Django instead",
"--plan",
_PLAN_ID,
decision_ids.plan_id,
"--dry-run",
"--format",
"json",
@@ -325,7 +314,15 @@ def plan_correct_crash() -> None:
f"Output: {result.output}"
)
# Verify the error is about resolve()
if result.exception is None:
_fail(f"Expected exception but got none. Output: {result.output}")
if not isinstance(result.exception, AttributeError):
_fail(
"Expected AttributeError, got "
f"{type(result.exception).__name__}: {result.exception}"
)
# Verify the error is about resolve() on the container object.
exception_str = str(result.exception) if result.exception else ""
output_str = result.output or ""
combined = exception_str + output_str
@@ -336,12 +333,15 @@ def plan_correct_crash() -> None:
f"Exception: {exception_str}\nOutput: {output_str}"
)
if (
"attributeerror" not in combined.lower()
and "attribute" not in combined.lower()
):
if "no attribute" not in combined.lower():
_fail(
f"Expected AttributeError, but got different error:\n"
f"Expected missing attribute message, but got:\n"
f"Exception: {exception_str}\nOutput: {output_str}"
)
if "container" not in combined.lower():
_fail(
f"Expected container-related source, but got:\n"
f"Exception: {exception_str}\nOutput: {output_str}"
)