test(session): add TDD failing tests for session list DI error #653

Merged
brent.edwards merged 5 commits from tdd/session-list-di-error into master 2026-03-11 00:49:27 +00:00
7 changed files with 486 additions and 32 deletions
+77
View File
@@ -0,0 +1,77 @@
"""ASV benchmarks for TDD Bug #554 — session list CLI throughput.
Measures the performance of the session list CLI command path to establish
a baseline before and after the DI bug fix. Uses a mocked service so the
benchmark isolates CLI/rendering overhead from database I/O.
"""
from __future__ import annotations
import sys
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from typer.testing import CliRunner # noqa: E402
from ulid import ULID # noqa: E402
from cleveragents.cli.commands import session as session_mod # noqa: E402
from cleveragents.cli.commands.session import app as session_app # noqa: E402
from cleveragents.domain.models.core.session import ( # noqa: E402
Session,
SessionTokenUsage,
)
_runner = CliRunner()
def _mock_session(
session_id: str | None = None,
actor_name: str | None = None,
) -> Session:
return Session(
session_id=session_id or str(ULID()),
actor_name=actor_name,
namespace="local",
messages=[],
token_usage=SessionTokenUsage(),
created_at=datetime.now(),
updated_at=datetime.now(),
)
class TDDSessionListDISuite:
"""Benchmark session list command throughput (TDD bug #554 baseline)."""
def setup(self) -> None:
self._svc = MagicMock()
self._svc.list.return_value = [
_mock_session(actor_name=f"openai/gpt-{i}") for i in range(50)
]
session_mod._service = self._svc
def teardown(self) -> None:
session_mod._service = None
def time_list_rich(self) -> None:
"""Benchmark listing sessions with default rich format."""
_runner.invoke(session_app, ["list"])
def time_list_json(self) -> None:
"""Benchmark listing sessions with JSON format."""
_runner.invoke(session_app, ["list", "--format", "json"])
def time_list_empty(self) -> None:
"""Benchmark listing with no sessions."""
self._svc.list.return_value = []
_runner.invoke(session_app, ["list"])
# Restore for next benchmark
self._svc.list.return_value = [
_mock_session(actor_name=f"openai/gpt-{i}") for i in range(50)
]
+72 -30
View File
@@ -2,12 +2,16 @@
import contextlib
import os
import re
import shutil
import sys
import tempfile
from pathlib import Path
from typing import Any
from behave.model import Scenario
from behave.model_core import Status
LANGSMITH_ENV_VARS = [
"CLEVERAGENTS_LANGSMITH_ENABLED",
"CLEVERAGENTS_LANGSMITH_PROJECT",
@@ -305,38 +309,76 @@ def before_scenario(context, scenario):
pass # Container not needed for all tests
# ---------------------------------------------------------------------------
# TDD expected-fail tag handler (see CONTRIBUTING.md §TDD Bug Test Tags)
# ---------------------------------------------------------------------------
_TDD_BUG_N_RE = re.compile(r"^tdd_bug_\d+$")
def _handle_tdd_expected_fail(scenario: Scenario) -> None:
"""Invert pass/fail for scenarios tagged ``@tdd_expected_fail``.
When a TDD bug-capture test carries ``@tdd_expected_fail``, the test is
*expected* to fail because the bug it captures is still present. This
hook inverts the status so CI stays green:
* **failed → passed** — the bug still triggers, which is expected.
* **passed → failed** — the bug was fixed but the tag was not removed;
this is an error that must be caught.
Tag validation (per CONTRIBUTING.md):
* ``@tdd_expected_fail`` requires ``@tdd_bug`` **and** at least one
``@tdd_bug_<N>`` tag. Missing companion tags cause the scenario to
fail unconditionally.
"""
tags = set(scenario.effective_tags)
if "tdd_expected_fail" not in tags:
return
# --- tag validation ---------------------------------------------------
if "tdd_bug" not in tags:
scenario.set_status(Status.failed)
sys.stderr.write(
f"TDD TAG ERROR: {scenario.name!r}"
"@tdd_expected_fail requires @tdd_bug tag\n"
)
return
if not any(_TDD_BUG_N_RE.match(t) for t in tags):
scenario.set_status(Status.failed)
sys.stderr.write(
f"TDD TAG ERROR: {scenario.name!r}"
"@tdd_expected_fail requires at least one @tdd_bug_<N> tag\n"
)
return
# --- status inversion -------------------------------------------------
if scenario.status == Status.failed:
# Bug still present — expected. Mark scenario and its failed/skipped
# steps as passed so that summary counts are accurate.
scenario.clear_status()
scenario.set_status(Status.passed)
for step in scenario.steps:
if step.status in (Status.failed, Status.skipped):
step.status = Status.passed
elif scenario.status == Status.passed:
# Bug was fixed but @tdd_expected_fail was not removed — error.
scenario.set_status(Status.failed)
sys.stderr.write(
f"TDD TAG ERROR: {scenario.name!r}"
"scenario passed but still carries @tdd_expected_fail; "
"remove the tag now that the bug is fixed\n"
)
def after_scenario(context, scenario):
"""Clean up after each scenario."""
# ── @tdd_expected_fail inversion ──────────────────────────────────
# When a scenario is tagged @tdd_expected_fail the test captures a
# bug that has NOT yet been fixed. The assertions describe the
# *correct* (post-fix) behaviour, so the scenario is expected to
# FAIL while the bug exists. We invert the result so CI stays
# green:
# • scenario FAILED → mark PASSED (expected — bug still exists)
# • scenario PASSED → mark FAILED (unexpected — fix landed but
# the @tdd_expected_fail tag was not removed)
# See CONTRIBUTING.md § TDD Bug Test Tags for the full convention.
if "tdd_expected_fail" in scenario.tags:
from behave.model import Status
if scenario.status == Status.failed:
# Expected failure — reset all steps and the scenario so
# Behave counts this as a pass.
for step in scenario.steps:
step.status = Status.passed
step.error_message = None
scenario.clear_status()
scenario.set_status(Status.passed)
elif scenario.status == Status.passed:
# Unexpected pass — the bug appears fixed but the tag was
# not removed. Force a failure so the developer notices.
scenario.set_status(Status.failed)
scenario.error_message = (
"[tdd_expected_fail] Test passed but still has the "
"tdd_expected_fail tag. The bug appears to be fixed "
"— remove the tag."
)
# Handle TDD expected-fail inversion BEFORE cleanup (status is already set
# by step execution; cleanup does not change it).
_handle_tdd_expected_fail(scenario)
# Return to original directory first
if hasattr(context, "original_cwd"):
+113
View File
@@ -0,0 +1,113 @@
"""Step definitions for TDD Bug #554 — session list DI error.
These steps exercise the *real* DI path in ``_get_session_service()`` without
mocking, so the ``container.db()`` ``AttributeError`` is triggered. The
``@tdd_expected_fail`` tag on the scenarios inverts the result: these tests
**pass** CI while the bug is present and will **fail** once the bug is fixed
(signalling that the tag should be removed).
"""
from __future__ import annotations
import contextlib
import json
import os
import tempfile
from behave import given, then, when
from behave.runner import Context
from typer.testing import CliRunner
from cleveragents.cli.commands import session as session_mod
from cleveragents.cli.commands.session import app as session_app
@given("a CLI runner using the real session DI path")
def step_real_di_runner(context: Context) -> None:
"""Set up a CLI runner that does NOT mock the session service.
By ensuring ``session_mod._service`` is ``None``, the CLI will call
``_get_session_service()`` which hits the real DI container and
triggers the ``container.db()`` bug.
"""
context.runner = CliRunner()
# Ensure we go through the real DI path — no mock service.
session_mod._service = None
# Provide a database URL so the container can be constructed (the bug
# triggers before the URL is actually used).
fd, db_path = tempfile.mkstemp(suffix=".db")
os.close(fd)
context._tdd_db_path = db_path
os.environ["CLEVERAGENTS_DATABASE_URL"] = f"sqlite:///{db_path}"
def _cleanup() -> None:
session_mod._service = None
os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
try:
from cleveragents.application.container import reset_container
reset_container()
except ImportError:
pass
with contextlib.suppress(OSError):
os.unlink(db_path)
context.add_cleanup(_cleanup)
@when("I invoke the session list command")
def step_invoke_list(context: Context) -> None:
"""Invoke ``session list`` through the real CLI app."""
context.result = context.runner.invoke(session_app, ["list"])
@when("I request the session service from the DI container")
def step_request_service(context: Context) -> None:
"""Call ``_get_session_service()`` directly to test the DI wiring."""
try:
context.session_service = session_mod._get_session_service()
context.service_error = None
except AttributeError as exc:
context.session_service = None
context.service_error = exc
@when("I invoke the session list command with format json")
def step_invoke_list_json(context: Context) -> None:
"""Invoke ``session list --format json`` through the real CLI app."""
context.result = context.runner.invoke(session_app, ["list", "--format", "json"])
@then("the session list command should exit successfully")
def step_list_exits_ok(context: Context) -> None:
"""Assert the command exits with code 0."""
assert context.result.exit_code == 0, (
f"Expected exit code 0, got {context.result.exit_code}.\n"
f"Output:\n{context.result.output}"
)
@then("the session service should be a valid SessionService instance")
def step_service_is_valid(context: Context) -> None:
"""Assert that ``_get_session_service()`` returned a usable service."""
from cleveragents.domain.models.core.session import SessionService
assert context.service_error is None, (
f"_get_session_service() raised {context.service_error!r}"
)
assert isinstance(context.session_service, SessionService), (
f"Expected SessionService, got {type(context.session_service)}"
)
@then("the session list output should be valid JSON")
def step_list_output_json(context: Context) -> None:
"""Assert the output is parseable JSON."""
try:
json.loads(context.result.output)
except json.JSONDecodeError as exc:
raise AssertionError(
f"Output is not valid JSON:\n{context.result.output}"
) from exc
+29
View File
@@ -0,0 +1,29 @@
@tdd_bug @tdd_bug_554
Feature: TDD Bug #554 — session list DI container missing db provider
As a developer
I want to verify that `agents session list` fails due to the
DI container missing a `db` provider
So that the bug is captured and will be caught by a regression test
The root cause is that `_get_session_service()` in session.py calls
`container.db()`, but the Container class has no `db` provider, causing
an AttributeError at runtime.
@tdd_expected_fail
Scenario: Session list command succeeds via DI container
Given a CLI runner using the real session DI path
When I invoke the session list command
Then the session list command should exit successfully
@tdd_expected_fail
Scenario: Session list DI path resolves a SessionService
Given a CLI runner using the real session DI path
When I request the session service from the DI container
Then the session service should be a valid SessionService instance
@tdd_expected_fail
Scenario: Session list command produces structured output via DI
Given a CLI runner using the real session DI path
When I invoke the session list command with format json
Then the session list command should exit successfully
And the session list output should be valid JSON
+14 -2
View File
@@ -251,6 +251,18 @@ def _has_failures(total):
)
def _no_scenarios_ran(total):
"""Return True when the runner collected zero scenario results.
This catches runner-level crashes (e.g. ``before_all`` failure) that
prevent any scenario from executing. Without this guard the summary
would contain all-zero counters, ``_has_failures()`` would return
``False``, and CI would silently pass a broken suite.
"""
s = total["scenarios"]
return s["passed"] + s["failed"] + s["errors"] + s["skipped"] == 0
# ---------------------------------------------------------------------------
# Feature discovery
# ---------------------------------------------------------------------------
1
@@ -423,9 +435,9 @@ def main(argv=None):
# Safety net: if features were requested but zero scenarios ran, the
# runner crashed before executing any scenario (e.g. ``before_all``
# failure). Treat this as a failure so CI does not silently pass.
if feature_paths and total["scenarios"]["passed"] == 0 and total["scenarios"]["failed"] == 0:
if feature_paths and _no_scenarios_ran(total):
print(
"ERROR: features were requested but no scenarios ran -- "
"ERROR: features were requested but no scenarios ran "
"possible runner-level crash.",
file=sys.stderr,
)
+142
View File
@@ -0,0 +1,142 @@
"""Helper script for tdd_session_list_di.robot smoke tests.
Each subcommand exercises the real DI path (no mocks) to reproduce bug #554.
The helper reports the **real** outcome: it exits 0 and prints the sentinel
when the operation succeeds (bug is fixed), and exits 1 when the bug is
still present. The ``tdd_expected_fail_listener`` on the Robot side handles
pass/fail inversion while the bug remains open.
"""
from __future__ import annotations
import contextlib
import os
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path
# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from typer.testing import CliRunner # noqa: E402
from cleveragents.cli.commands import session as session_mod # noqa: E402
from cleveragents.cli.commands.session import app as session_app # noqa: E402
runner = CliRunner()
def _setup_real_di() -> str:
"""Prepare the environment for real DI resolution.
Returns the path to a temporary database file (caller must clean up).
"""
session_mod._service = None
fd, db_path = tempfile.mkstemp(suffix=".db")
os.close(fd)
os.environ["CLEVERAGENTS_DATABASE_URL"] = f"sqlite:///{db_path}"
return db_path
def _teardown(db_path: str) -> None:
"""Clean up after a test."""
session_mod._service = None
os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
try:
from cleveragents.application.container import reset_container
reset_container()
except ImportError:
pass
with contextlib.suppress(OSError):
os.unlink(db_path)
# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------
def list_di_error() -> None:
"""Invoke ``session list`` through the real DI path.
Exits 0 with sentinel when the command succeeds (bug fixed).
Exits 1 when the command fails (bug still present).
"""
db_path = _setup_real_di()
try:
result = runner.invoke(session_app, ["list"])
if result.exit_code == 0:
print("tdd-session-list-di-error-ok")
else:
print(
f"session list failed with exit code {result.exit_code}",
file=sys.stderr,
)
sys.exit(1)
finally:
_teardown(db_path)
def service_resolution() -> None:
"""Call ``_get_session_service()`` to verify DI resolution.
Exits 0 with sentinel when the service resolves (bug fixed).
Exits 1 when resolution raises AttributeError (bug still present).
"""
db_path = _setup_real_di()
try:
try:
session_mod._get_session_service()
except AttributeError as exc:
print(
f"_get_session_service() raised {exc!r}",
file=sys.stderr,
)
sys.exit(1)
print("tdd-session-list-service-resolution-ok")
finally:
_teardown(db_path)
def list_json() -> None:
"""Invoke ``session list --format json`` through the real DI path.
Exits 0 with sentinel when the command succeeds (bug fixed).
Exits 1 when the command fails (bug still present).
"""
db_path = _setup_real_di()
try:
result = runner.invoke(session_app, ["list", "--format", "json"])
if result.exit_code == 0:
print("tdd-session-list-json-ok")
else:
print(
f"session list --format json failed with exit code {result.exit_code}",
file=sys.stderr,
)
sys.exit(1)
finally:
_teardown(db_path)
# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"list-di-error": list_di_error,
"service-resolution": service_resolution,
"list-json": list_json,
}
if __name__ == "__main__":
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
sys.exit(1)
cmd = _COMMANDS[sys.argv[1]]
cmd()
+39
View File
@@ -0,0 +1,39 @@
*** Settings ***
Documentation TDD Bug #554 — session list DI container missing db provider
... Integration smoke tests verifying that the session list command
... succeeds once the DI container has a proper ``db`` provider.
... Tagged ``tdd_expected_fail`` until bug #554 is resolved.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_tdd_session_list_di.py
*** Test Cases ***
TDD Session List DI Error Via CLI
[Documentation] Verify that ``session list`` succeeds via the real DI path
[Tags] tdd_bug tdd_bug_554 tdd_expected_fail
${result}= Run Process ${PYTHON} ${HELPER} list-di-error cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tdd-session-list-di-error-ok
TDD Session List DI Service Resolution
[Documentation] Verify that ``_get_session_service()`` resolves a valid service
[Tags] tdd_bug tdd_bug_554 tdd_expected_fail
${result}= Run Process ${PYTHON} ${HELPER} service-resolution cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tdd-session-list-service-resolution-ok
TDD Session List DI JSON Output
[Documentation] Verify that ``session list --format json`` succeeds via the real DI path
[Tags] tdd_bug tdd_bug_554 tdd_expected_fail
${result}= Run Process ${PYTHON} ${HELPER} list-json cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tdd-session-list-json-ok