test: add TDD bug-capture test for #990 — automation_profile DI bypass
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 19s
CI / helm (pull_request) Successful in 21s
CI / lint (pull_request) Successful in 3m18s
CI / quality (pull_request) Successful in 3m41s
CI / typecheck (pull_request) Successful in 3m56s
CI / security (pull_request) Successful in 4m12s
CI / integration_tests (pull_request) Successful in 9m13s
CI / unit_tests (pull_request) Successful in 9m37s
CI / docker (pull_request) Successful in 1m16s
CI / coverage (pull_request) Successful in 12m57s
CI / e2e_tests (pull_request) Successful in 19m13s
CI / status-check (pull_request) Successful in 1s
CI / benchmark-regression (pull_request) Successful in 57m29s

Add a Behave BDD feature (features/tdd_automation_profile_di_bypass.feature)
with three scenarios and a Robot Framework integration test
(robot/tdd_automation_profile_di_bypass.robot) with two test cases that
capture the DI bypass bug described in issue #990.

Bug #990: The _get_service() function in automation_profile.py manually
constructs create_engine / sessionmaker / AutomationProfileRepository
instead of resolving AutomationProfileService through the DI container.
This is the only CLI command that bypasses DI.

Behave scenarios:
  1. Verifies _get_service() resolves from container.automation_profile_service()
  2. Verifies create_engine is NOT called directly
  3. Verifies sessionmaker is NOT called directly

Robot integration tests:
  1. Verifies _get_service() uses the DI container at subprocess level
  2. Verifies create_engine is not called directly at subprocess level

All tests are tagged @tdd_bug @tdd_bug_990 @tdd_issue @tdd_issue_990
@tdd_expected_fail. The @tdd_expected_fail tag inverts the result so CI
passes while the bug is unfixed. Both @tdd_bug and @tdd_issue tags are
included: @tdd_bug per reviewer convention, @tdd_issue to satisfy the
tag validation rules in features/environment.py.

ISSUES CLOSED: #1031
This commit is contained in:
2026-03-31 01:49:36 +00:00
committed by HAL9000
parent 84b0c10dbf
commit c5ffcf8de4
5 changed files with 660 additions and 1598 deletions
+246 -1598
View File
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,213 @@
"""Step definitions for TDD Bug #990 — automation_profile DI bypass.
This module captures the bug described in issue #990: the
``automation_profile`` CLI module's ``_get_service()`` function manually
constructs ``create_engine`` / ``sessionmaker`` /
``AutomationProfileRepository`` instead of resolving
``AutomationProfileService`` through the DI container.
Every other CLI command resolves its service through the container
(e.g., ``container.session_service()``). These tests verify that
``_get_service()`` follows the same DI pattern. Until the bug is
fixed, the assertions will fail; the ``@tdd_expected_fail`` tag on
the feature inverts the result so CI passes.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
from behave import given, then, when
from behave.runner import Context
from cleveragents.application.services.automation_profile_service import (
AutomationProfileService,
)
# -----------------------------------------------------------------------
# Scenario 1: _get_service resolves from DI container
# -----------------------------------------------------------------------
@given("ap990- a DI container that provides automation_profile_service")
def step_given_container_with_provider(context: Context) -> None:
"""Set up a mock DI container with an ``automation_profile_service`` provider.
The mock container is configured so that calling
``container.automation_profile_service()`` returns a pre-built
``AutomationProfileService`` instance. If ``_get_service()`` uses
the DI container correctly, it will call this provider.
We also provide ``database_url`` so the current buggy code path
does not crash on infrastructure before reaching the assertion.
"""
mock_service = MagicMock(spec=AutomationProfileService)
mock_container = MagicMock()
mock_container.automation_profile_service.return_value = mock_service
mock_container.database_url.return_value = "sqlite:///:memory:"
context.ap990_mock_container = mock_container
context.ap990_expected_service = mock_service
@when("ap990- _get_service is called")
def step_when_get_service_called(context: Context) -> None:
"""Invoke ``_get_service()`` with the DI container patched.
Patches ``get_container`` at the source module so that the lazy
import inside ``_get_service()`` picks up the mock. Also patches
``create_engine`` and ``sessionmaker`` with safe stubs so the
current (buggy) code path can complete without infrastructure
errors — this ensures the assertion in the Then step fires as
an ``AssertionError`` rather than an unrelated crash.
"""
from cleveragents.cli.commands.automation_profile import _get_service
with (
patch(
"cleveragents.application.container.get_container",
return_value=context.ap990_mock_container,
),
patch("sqlalchemy.create_engine", return_value=MagicMock()),
patch("sqlalchemy.orm.sessionmaker", return_value=MagicMock()),
):
context.ap990_actual_service = _get_service()
@then("ap990- the service should have been resolved from the container")
def step_then_service_resolved_from_container(context: Context) -> None:
"""Assert that ``_get_service()`` returned the service provided by the container.
This proves the function delegates to the DI container rather than
constructing the service manually.
"""
context.ap990_mock_container.automation_profile_service.assert_called_once()
assert context.ap990_actual_service is context.ap990_expected_service, (
"Expected _get_service() to return the AutomationProfileService "
"from the DI container, but it returned a different instance. "
"This indicates _get_service() is constructing the service "
"manually instead of resolving through the container."
)
# -----------------------------------------------------------------------
# Scenario 2: _get_service does NOT manually call create_engine
# -----------------------------------------------------------------------
@given("ap990- a patched environment tracking create_engine calls")
def step_given_track_create_engine(context: Context) -> None:
"""Set up a spy on ``create_engine`` to detect manual construction.
Also patches ``get_container`` to provide a plausible mock container
(with ``database_url`` so the current buggy code can proceed, and
``automation_profile_service`` so fixed code has a provider to call).
"""
context.ap990_engine_spy = MagicMock(
name="create_engine_spy",
return_value=MagicMock(name="engine"),
)
mock_container = MagicMock()
mock_container.database_url.return_value = "sqlite:///:memory:"
mock_container.automation_profile_service.return_value = MagicMock(
spec=AutomationProfileService,
)
context.ap990_engine_container = mock_container
@when("ap990- _get_service is invoked through the patched environment")
def step_when_invoke_with_engine_patch(context: Context) -> None:
"""Call ``_get_service()`` while intercepting ``create_engine``."""
from cleveragents.cli.commands.automation_profile import _get_service
with (
patch(
"cleveragents.application.container.get_container",
return_value=context.ap990_engine_container,
),
patch(
"sqlalchemy.create_engine",
context.ap990_engine_spy,
),
patch(
"sqlalchemy.orm.sessionmaker",
return_value=MagicMock(name="session_factory"),
),
):
context.ap990_engine_result = _get_service()
@then("ap990- create_engine should not have been called directly")
def step_then_create_engine_not_called(context: Context) -> None:
"""Assert ``create_engine`` was never called by ``_get_service()``.
The DI container should own database engine creation. If
``_get_service()`` calls ``create_engine`` directly, the DI
container is being bypassed (bug #990).
"""
assert not context.ap990_engine_spy.called, (
"create_engine was called directly by _get_service(). "
"The function should resolve the service from the DI "
"container instead of manually constructing database "
"infrastructure (bug #990)."
)
# -----------------------------------------------------------------------
# Scenario 3: _get_service does NOT manually call sessionmaker
# -----------------------------------------------------------------------
@given("ap990- a patched environment tracking sessionmaker calls")
def step_given_track_sessionmaker(context: Context) -> None:
"""Set up a spy on ``sessionmaker`` to detect manual construction."""
context.ap990_sessionmaker_spy = MagicMock(
name="sessionmaker_spy",
return_value=MagicMock(name="session_factory"),
)
mock_container = MagicMock()
mock_container.database_url.return_value = "sqlite:///:memory:"
mock_container.automation_profile_service.return_value = MagicMock(
spec=AutomationProfileService,
)
context.ap990_sm_container = mock_container
@when("ap990- _get_service is invoked through the sessionmaker-patched environment")
def step_when_invoke_with_sessionmaker_patch(context: Context) -> None:
"""Call ``_get_service()`` while intercepting ``sessionmaker``."""
from cleveragents.cli.commands.automation_profile import _get_service
with (
patch(
"cleveragents.application.container.get_container",
return_value=context.ap990_sm_container,
),
patch(
"sqlalchemy.create_engine",
return_value=MagicMock(name="engine"),
),
patch(
"sqlalchemy.orm.sessionmaker",
context.ap990_sessionmaker_spy,
),
):
context.ap990_sm_result = _get_service()
@then("ap990- sessionmaker should not have been called directly")
def step_then_sessionmaker_not_called(context: Context) -> None:
"""Assert ``sessionmaker`` was never called by ``_get_service()``.
The DI container should own session factory creation. If
``_get_service()`` calls ``sessionmaker`` directly, the DI
container is being bypassed (bug #990).
"""
assert not context.ap990_sessionmaker_spy.called, (
"sessionmaker was called directly by _get_service(). "
"The function should resolve the service from the DI "
"container instead of manually constructing database "
"infrastructure (bug #990)."
)
@@ -0,0 +1,28 @@
@tdd_bug @tdd_bug_990 @tdd_issue @tdd_issue_990 @tdd_expected_fail
Feature: TDD Bug #990 — automation_profile._get_service() bypasses DI container
As a developer
I want to verify that the automation_profile CLI module resolves
AutomationProfileService through the DI container
So that database connections, middleware, and lifecycle management
are consistent with all other CLI commands
Bug #990: The `_get_service()` function in automation_profile.py manually
constructs `create_engine` / `sessionmaker` / `AutomationProfileRepository`
instead of resolving through the DI container. Every other CLI command
uses the container (e.g., `container.session_service()`). This test
captures the DI bypass so it will fail until the bug is fixed.
Scenario: _get_service resolves AutomationProfileService from the DI container
Given ap990- a DI container that provides automation_profile_service
When ap990- _get_service is called
Then ap990- the service should have been resolved from the container
Scenario: _get_service does not manually construct a SQLAlchemy engine
Given ap990- a patched environment tracking create_engine calls
When ap990- _get_service is invoked through the patched environment
Then ap990- create_engine should not have been called directly
Scenario: _get_service does not manually construct a sessionmaker
Given ap990- a patched environment tracking sessionmaker calls
When ap990- _get_service is invoked through the sessionmaker-patched environment
Then ap990- sessionmaker should not have been called directly
@@ -0,0 +1,131 @@
"""Helper script for tdd_automation_profile_di_bypass.robot smoke tests.
Each subcommand exercises the ``_get_service()`` function in the
``automation_profile`` CLI module to detect whether it resolves the
service from the DI container or manually constructs database
infrastructure (bug #990).
The helper reports the **real** outcome: it exits 0 and prints the
sentinel when the DI container is used correctly (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 sys
from collections.abc import Callable
from pathlib import Path
from unittest.mock import MagicMock, patch
# Ensure local source tree is importable.
_ROOT = Path(__file__).resolve().parents[1]
_SRC = str(_ROOT / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.application.services.automation_profile_service import ( # noqa: E402
AutomationProfileService,
)
# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------
def check_di_resolution() -> None:
"""Verify that ``_get_service()`` resolves from the DI container.
Sets up a mock container with an ``automation_profile_service``
provider and calls ``_get_service()``. If the returned service
matches the mock provider's return value, the DI path is used
(bug is fixed). Otherwise the bug is still present.
"""
from cleveragents.cli.commands.automation_profile import _get_service
mock_service = MagicMock(spec=AutomationProfileService)
mock_container = MagicMock()
mock_container.automation_profile_service.return_value = mock_service
mock_container.database_url.return_value = "sqlite:///:memory:"
with (
patch(
"cleveragents.application.container.get_container",
return_value=mock_container,
),
patch("sqlalchemy.create_engine", return_value=MagicMock()),
patch("sqlalchemy.orm.sessionmaker", return_value=MagicMock()),
):
actual = _get_service()
if actual is mock_service:
print("tdd-automation-profile-di-resolution-ok")
else:
print(
"_get_service() did not resolve from DI container",
file=sys.stderr,
)
sys.exit(1)
def check_no_create_engine() -> None:
"""Verify that ``_get_service()`` does NOT call ``create_engine``.
Patches ``create_engine`` as a spy. If the spy is called, the DI
container is being bypassed (bug #990 still present).
"""
from cleveragents.cli.commands.automation_profile import _get_service
engine_spy = MagicMock(
name="create_engine_spy",
return_value=MagicMock(name="engine"),
)
mock_container = MagicMock()
mock_container.database_url.return_value = "sqlite:///:memory:"
mock_container.automation_profile_service.return_value = MagicMock(
spec=AutomationProfileService,
)
with (
patch(
"cleveragents.application.container.get_container",
return_value=mock_container,
),
patch("sqlalchemy.create_engine", engine_spy),
patch(
"sqlalchemy.orm.sessionmaker",
return_value=MagicMock(name="session_factory"),
),
):
_get_service()
if not engine_spy.called:
print("tdd-automation-profile-no-create-engine-ok")
else:
print(
"create_engine was called directly by _get_service()",
file=sys.stderr,
)
sys.exit(1)
# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"check-di-resolution": check_di_resolution,
"check-no-create-engine": check_no_create_engine,
}
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()
@@ -0,0 +1,42 @@
*** Settings ***
Documentation TDD Bug #990 — automation_profile._get_service() bypasses DI container
... Integration smoke test verifying that the ``automation_profile``
... CLI module resolves ``AutomationProfileService`` through the DI
... container instead of manually constructing ``create_engine`` /
... ``sessionmaker`` / ``AutomationProfileRepository``.
...
... Bug #990: ``_get_service()`` is the only CLI command that bypasses
... DI. This test invokes ``_get_service()`` and checks whether
... ``create_engine`` is called directly (indicating the DI bypass).
... Tagged ``tdd_expected_fail`` so CI passes while the bug is unfixed.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
Force Tags tdd_bug tdd_bug_990 tdd_issue tdd_issue_990 tdd_expected_fail
*** Variables ***
${HELPER} ${CURDIR}/helper_tdd_automation_profile_di_bypass.py
*** Test Cases ***
TDD Automation Profile Get Service Uses DI Container
[Documentation] Verify that ``_get_service()`` resolves
... ``AutomationProfileService`` from the DI container
... provider rather than constructing it manually.
... The helper exits 0 with a sentinel when the service
... comes from the container (bug is fixed), and exits 1
... when ``_get_service()`` bypasses DI (bug present).
${result}= Run Process ${PYTHON} ${HELPER} check-di-resolution cwd=${WORKSPACE} timeout=30s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tdd-automation-profile-di-resolution-ok
TDD Automation Profile Get Service Does Not Call Create Engine
[Documentation] Verify that ``_get_service()`` does NOT call
... ``create_engine`` directly. If it does, the DI
... container is being bypassed (bug #990).
${result}= Run Process ${PYTHON} ${HELPER} check-no-create-engine cwd=${WORKSPACE} timeout=30s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tdd-automation-profile-no-create-engine-ok