test(cli): add TDD regression tests for automation_profile._get_service() DI bypass
CI / push-validation (pull_request) Successful in 27s
CI / helm (pull_request) Successful in 30s
CI / lint (pull_request) Successful in 39s
CI / build (pull_request) Successful in 52s
CI / quality (pull_request) Successful in 58s
CI / typecheck (pull_request) Successful in 1m19s
CI / security (pull_request) Successful in 1m20s
CI / integration_tests (pull_request) Successful in 3m43s
CI / unit_tests (pull_request) Successful in 4m33s
CI / docker (pull_request) Successful in 1m26s
CI / coverage (pull_request) Successful in 12m0s
CI / status-check (pull_request) Successful in 3s

Three BDD scenarios and two Robot Framework integration tests verifying that
_get_service() in automation_profile.py resolves AutomationProfileService
through the DI container rather than manually calling create_engine or
sessionmaker (bug #990).

Bug #990 was fixed by PR #1181 before this TDD test PR merged; the
@tdd_expected_fail tag is therefore absent and these tests serve as
permanent regression guards confirming the fix remains in place.

ISSUES CLOSED: #1031
This commit is contained in:
2026-05-29 15:26:47 -04:00
parent 0af6ec5395
commit 6a76ea894d
5 changed files with 366 additions and 0 deletions
+6
View File
@@ -6,6 +6,12 @@ Changed `wf10_batch.robot` to be less likely to create files, and
`plan_generation_graph.robot` to give more test answers.
## [Unreleased]
- **TDD regression tests for automation_profile DI bypass** (#1031): Added BDD scenarios
and Robot Framework integration tests verifying that ``_get_service()`` in
``automation_profile.py`` resolves ``AutomationProfileService`` through the DI container
rather than directly calling ``create_engine`` or ``sessionmaker``. Bug #990 was fixed by
PR #1181 before this TDD test PR merged; these tests serve as permanent regression guards
confirming the fix remains in place.
- Added team collaboration features: multi-user connection handling with
user identity tracking (TeamMember with owner/admin/member/viewer roles),
role-based access control (TeamPermission with read/write/admin/manage_members),
@@ -0,0 +1,176 @@
"""Step definitions for tdd_automation_profile_di_bypass.feature.
These steps verify that ``_get_service()`` in
``cleveragents.cli.commands.automation_profile`` delegates to
``container.automation_profile_service()`` rather than manually constructing
the service with ``create_engine`` / ``sessionmaker`` / repositories.
Bug #990 was fixed by PR #1181. The ``@tdd_expected_fail`` tag was omitted
because the fix landed on master before this TDD test PR merged; these
scenarios now run as permanent regression tests confirming the DI bypass
remains fixed.
Step namespacing: all steps use the ``ap990-`` prefix to avoid collisions
with other test suites. The defensive patching approach provides both a DI
container mock *and* infrastructure mocks so that a regression causes the
tests to fail for the right reason (the DI bypass itself) rather than an
unrelated import error or attribute lookup.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock, patch
from behave import given, then, when
from behave.runner import Context
# ---------------------------------------------------------------------------
# Constants — patch targets
# ---------------------------------------------------------------------------
_PATCH_GET_CONTAINER = "cleveragents.application.container.get_container"
_PATCH_CREATE_ENGINE = "sqlalchemy.create_engine"
_PATCH_SESSIONMAKER = "sqlalchemy.orm.sessionmaker"
# ---------------------------------------------------------------------------
# Scenario 1: _get_service delegates to container.automation_profile_service
# ---------------------------------------------------------------------------
@given("ap990- the DI container has an automation_profile_service provider")
def step_ap990_container_has_provider(context: Context) -> None:
"""Set up a mock container with an automation_profile_service() callable.
Defensive patching: both a DI container mock and silent sqlalchemy stubs
are set up so that if _get_service() regresses to the bug #990 pattern it
reaches the Then assertion (and fails there with a meaningful message)
rather than crashing on a missing DB URL.
"""
mock_service: MagicMock = MagicMock(name="MockAutomationProfileService")
mock_container: MagicMock = MagicMock(name="MockContainer")
mock_container.automation_profile_service.return_value = mock_service
context.ap990_mock_container: MagicMock = mock_container
context.ap990_mock_service: MagicMock = mock_service
# Shared slot used by the When step that is also shared with Scenarios 2/3.
context.ap990_di_mock_container: MagicMock = mock_container
context.ap990_returned_service: Any = None
@when("ap990- _get_service is called with the mocked container")
def step_ap990_call_get_service_with_mock(context: Context) -> None:
"""Call _get_service() with get_container() returning the mock container."""
from cleveragents.cli.commands.automation_profile import _get_service
with patch(_PATCH_GET_CONTAINER, return_value=context.ap990_mock_container):
context.ap990_returned_service = _get_service()
@then(
"ap990- the returned service should be the one from container.automation_profile_service"
)
def step_ap990_verify_service_identity(context: Context) -> None:
"""The returned service must be exactly what container.automation_profile_service() returns.
An identity check (``is``) catches regressions where _get_service()
bypasses the container and constructs a fresh AutomationProfileService
independently — the resulting object would differ from the mock sentinel.
"""
assert context.ap990_returned_service is context.ap990_mock_service, (
f"Expected the service returned by container.automation_profile_service(), "
f"but got {type(context.ap990_returned_service)!r}. "
"_get_service() is NOT delegating to the DI container."
)
# ---------------------------------------------------------------------------
# Scenario 2: _get_service does not call create_engine directly
# ---------------------------------------------------------------------------
@given("ap990- create_engine is patched to detect direct calls")
def step_ap990_patch_create_engine(context: Context) -> None:
"""Patch create_engine and set up a DI mock so _get_service() can succeed.
Defensive patching: the DI mock ensures the fixed code path (container
delegation) completes successfully. The create_engine mock catches any
regression where _get_service() rebuilds the session factory manually.
"""
create_engine_mock: MagicMock = MagicMock(name="MockCreateEngine")
patcher = patch(_PATCH_CREATE_ENGINE, create_engine_mock)
patcher.start()
context.add_cleanup(patcher.stop)
context.ap990_create_engine_mock: MagicMock = create_engine_mock
mock_service: MagicMock = MagicMock(name="MockAutomationProfileServiceForEngine")
mock_container: MagicMock = MagicMock(name="MockContainerForEngine")
mock_container.automation_profile_service.return_value = mock_service
context.ap990_di_mock_container: MagicMock = mock_container
@when("ap990- _get_service is called with a valid DI container")
def step_ap990_call_get_service_di_path(context: Context) -> None:
"""Call _get_service() via the DI path; the mock container is set in the Given step."""
from cleveragents.cli.commands.automation_profile import _get_service
with patch(_PATCH_GET_CONTAINER, return_value=context.ap990_di_mock_container):
context.ap990_result: Any = _get_service()
@then("ap990- create_engine should not have been called")
def step_ap990_verify_create_engine_not_called(context: Context) -> None:
"""create_engine must never be called; the DI container owns engine construction.
If this assertion fails, _get_service() has regressed to the bug #990
pattern of manually building the persistence layer with a raw SQLAlchemy
engine instead of delegating to container.automation_profile_service().
"""
mock: MagicMock = context.ap990_create_engine_mock
assert not mock.called, (
f"Bug #990 regression: _get_service() called create_engine() directly "
f"({mock.call_count} time(s)) instead of delegating to "
"container.automation_profile_service()."
)
# ---------------------------------------------------------------------------
# Scenario 3: _get_service does not call sessionmaker directly
# ---------------------------------------------------------------------------
@given("ap990- sessionmaker is patched to detect direct calls")
def step_ap990_patch_sessionmaker(context: Context) -> None:
"""Patch sessionmaker and set up a DI mock so _get_service() can succeed.
Defensive patching mirrors Scenario 2: the DI mock ensures the fixed
code path completes without error, and the sessionmaker mock detects any
regression where _get_service() builds the ORM session factory directly.
"""
sessionmaker_mock: MagicMock = MagicMock(name="MockSessionmaker")
patcher = patch(_PATCH_SESSIONMAKER, sessionmaker_mock)
patcher.start()
context.add_cleanup(patcher.stop)
context.ap990_sessionmaker_mock: MagicMock = sessionmaker_mock
mock_service: MagicMock = MagicMock(name="MockAutomationProfileServiceForSM")
mock_container: MagicMock = MagicMock(name="MockContainerForSM")
mock_container.automation_profile_service.return_value = mock_service
context.ap990_di_mock_container: MagicMock = mock_container
@then("ap990- sessionmaker should not have been called")
def step_ap990_verify_sessionmaker_not_called(context: Context) -> None:
"""sessionmaker must never be called; the DI container owns session factory setup.
If this assertion fails, _get_service() has regressed to the bug #990
pattern of manually constructing the ORM session factory with a raw
sessionmaker call instead of delegating to container.automation_profile_service().
"""
mock: MagicMock = context.ap990_sessionmaker_mock
assert not mock.called, (
f"Bug #990 regression: _get_service() called sessionmaker() directly "
f"({mock.call_count} time(s)) instead of delegating to "
"container.automation_profile_service()."
)
@@ -0,0 +1,32 @@
@tdd_bug @tdd_bug_990 @tdd_issue @tdd_issue_990
Feature: TDD Bug #990 — automation_profile._get_service() bypasses DI container
As a developer relying on the DI container for service wiring
I want _get_service() in automation_profile.py to resolve AutomationProfileService
through the DI container
So that service construction is consistent and dependencies are correctly injected
This feature captures bug #990. Previously, _get_service() bypassed the DI
container by directly constructing the persistence layer with create_engine()
and sessionmaker() rather than delegating to container.automation_profile_service().
This caused service wiring to diverge from the canonical DI path and broke
dependency injection for AutomationProfileService.
Bug #990 was fixed by PR #1181, which updated _get_service() to resolve the
service through the DI container. The @tdd_expected_fail tag has been omitted
because the fix landed on master before this TDD test PR merged; these scenarios
serve as permanent regression tests confirming the DI bypass remains fixed.
Scenario: Bug #990 — _get_service resolves AutomationProfileService through the DI container
Given ap990- the DI container has an automation_profile_service provider
When ap990- _get_service is called with the mocked container
Then ap990- the returned service should be the one from container.automation_profile_service
Scenario: Bug #990 — _get_service does not call create_engine directly
Given ap990- create_engine is patched to detect direct calls
When ap990- _get_service is called with a valid DI container
Then ap990- create_engine should not have been called
Scenario: Bug #990 — _get_service does not call sessionmaker directly
Given ap990- sessionmaker is patched to detect direct calls
When ap990- _get_service is called with a valid DI container
Then ap990- sessionmaker should not have been called
@@ -0,0 +1,121 @@
"""Helper script for tdd_automation_profile_di_bypass.robot smoke tests.
Each subcommand exercises ``_get_service()`` from
``cleveragents.cli.commands.automation_profile`` to verify that it
delegates to the DI container rather than manually constructing the
persistence layer with ``create_engine`` / ``sessionmaker``.
Bug #990 was fixed by PR #1181. These helpers report pass (exit 0 + sentinel)
when the DI delegation is correct, and fail (exit 1) when the bug regresses.
"""
from __future__ import annotations
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any
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)
_PATCH_GET_CONTAINER = "cleveragents.application.container.get_container"
_PATCH_CREATE_ENGINE = "sqlalchemy.create_engine"
_PATCH_SESSIONMAKER = "sqlalchemy.orm.sessionmaker"
# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------
def container_resolution() -> None:
"""Verify _get_service() returns the container's automation_profile_service.
Exits 0 with sentinel when the DI container is correctly used (bug fixed).
Exits 1 when _get_service() does not delegate to the container (bug regressed).
"""
from cleveragents.cli.commands.automation_profile import _get_service
mock_service: MagicMock = MagicMock(name="MockAutomationProfileService")
mock_container: MagicMock = MagicMock(name="MockContainer")
mock_container.automation_profile_service.return_value = mock_service
with patch(_PATCH_GET_CONTAINER, return_value=mock_container):
result: Any = _get_service()
if result is not mock_service:
print(
f"_get_service() returned {type(result)!r} instead of the "
"container service. Bug #990 regression: not delegating to DI container.",
file=sys.stderr,
)
sys.exit(1)
print("tdd-ap990-container-resolution-ok")
def di_bypass() -> None:
"""Verify _get_service() does not call create_engine or sessionmaker directly.
Exits 0 with sentinel when neither is called (bug fixed).
Exits 1 with an explanation when either is called (bug regressed).
"""
from cleveragents.cli.commands.automation_profile import _get_service
create_engine_mock: MagicMock = MagicMock(name="MockCreateEngine")
sessionmaker_mock: MagicMock = MagicMock(name="MockSessionmaker")
mock_service: MagicMock = MagicMock(name="MockAutomationProfileService")
mock_container: MagicMock = MagicMock(name="MockContainer")
mock_container.automation_profile_service.return_value = mock_service
with (
patch(_PATCH_CREATE_ENGINE, create_engine_mock),
patch(_PATCH_SESSIONMAKER, sessionmaker_mock),
patch(_PATCH_GET_CONTAINER, return_value=mock_container),
):
_get_service()
if create_engine_mock.called:
print(
f"Bug #990 regression: _get_service() called create_engine() directly "
f"({create_engine_mock.call_count} time(s)) instead of delegating to "
"container.automation_profile_service().",
file=sys.stderr,
)
sys.exit(1)
if sessionmaker_mock.called:
print(
f"Bug #990 regression: _get_service() called sessionmaker() directly "
f"({sessionmaker_mock.call_count} time(s)) instead of delegating to "
"container.automation_profile_service().",
file=sys.stderr,
)
sys.exit(1)
print("tdd-ap990-di-bypass-ok")
# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"container-resolution": container_resolution,
"di-bypass": di_bypass,
}
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)
_COMMANDS[sys.argv[1]]()
@@ -0,0 +1,31 @@
*** Settings ***
Documentation TDD Bug #990 — automation_profile._get_service() bypasses DI container
... Integration smoke tests verifying that _get_service() in
... automation_profile.py resolves AutomationProfileService through the
... DI container rather than calling create_engine or sessionmaker directly.
... Bug #990 was fixed by PR #1181; these tests serve as regression guards.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_tdd_automation_profile_di_bypass.py
*** Test Cases ***
TDD Automation Profile DI Container Resolution
[Documentation] Verify _get_service() returns the service from the DI container
[Tags] tdd_bug tdd_bug_990 tdd_issue tdd_issue_990
${result}= Run Process ${PYTHON} ${HELPER} container-resolution cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tdd-ap990-container-resolution-ok
TDD Automation Profile DI Bypass Check
[Documentation] Verify _get_service() does not call create_engine or sessionmaker directly
[Tags] tdd_bug tdd_bug_990 tdd_issue tdd_issue_990
${result}= Run Process ${PYTHON} ${HELPER} di-bypass cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tdd-ap990-di-bypass-ok