Files
cleveragents-core/features/mocks/fast_init_test_helpers.py
T
CoreRasurae b21e0fedea
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 21s
CI / helm (pull_request) Successful in 24s
CI / typecheck (pull_request) Successful in 50s
CI / security (pull_request) Successful in 1m10s
CI / lint (pull_request) Successful in 3m20s
CI / quality (pull_request) Successful in 3m45s
CI / unit_tests (pull_request) Successful in 5m53s
CI / docker (pull_request) Successful in 1m32s
CI / coverage (pull_request) Successful in 12m27s
CI / e2e_tests (pull_request) Successful in 19m48s
CI / integration_tests (pull_request) Successful in 22m24s
CI / status-check (pull_request) Successful in 1s
CI / build (push) Successful in 16s
CI / helm (push) Successful in 22s
CI / lint (push) Successful in 3m17s
CI / quality (push) Successful in 3m39s
CI / typecheck (push) Successful in 3m53s
CI / security (push) Successful in 4m4s
CI / unit_tests (push) Successful in 9m13s
CI / docker (push) Successful in 20s
CI / coverage (push) Successful in 12m20s
CI / e2e_tests (push) Successful in 19m18s
CI / integration_tests (push) Successful in 24m36s
CI / status-check (push) Successful in 2s
CI / benchmark-regression (push) Has been skipped
CI / benchmark-regression (pull_request) Successful in 55m5s
CI / benchmark-publish (push) Successful in 28m26s
test(bdd): add direct test for _fast_init_or_upgrade early-return behavior
Add 8 BDD scenarios in features/fast_init_upgrade.feature that directly
exercise the _fast_init_or_upgrade closure installed by
_install_template_db_patch in features/environment.py.

Scenarios cover all code paths:
- Non-empty DB with matching prefix → early return (original NOT invoked)
- Non-existent DB with matching prefix → template copied (original NOT invoked)
- Existing empty DB with matching prefix → template copied (original NOT invoked)
- Non-matching prefix → delegates to original init_or_upgrade
- In-memory SQLite → delegates to original init_or_upgrade
- Non-SQLite URL → delegates to original init_or_upgrade
- Bare sqlite:// URL → delegates to original init_or_upgrade
- Delegation forwards runner instance and keyword arguments correctly

Test infrastructure:
- features/mocks/fast_init_test_helpers.py provides a context manager
  (patch_original_init_or_upgrade) that replaces the _original_init_or_upgrade
  reference inside the closure via cell_contents manipulation, enabling
  precise call-tracking without recreating the function under test.
- All scenarios tagged @mock_only to skip unnecessary DB setup.
- Cleanup registered via context._cleanup_handlers for temp file removal.
- Works in both sequential and parallel execution modes.

Review fix round:
- Added bare sqlite:// URL scenario covering the second branch of the
  in-memory disjunction (L1).
- Added scenario verifying runner instance and keyword argument forwarding
  through the delegation path (L2, L3).
- Replaced hardcoded test credentials with clearly synthetic pattern (I4).
- Fixed CHANGELOG wording that incorrectly claimed mktemp replacement (L5).

ISSUES CLOSED: #733
2026-04-01 23:52:20 +01:00

96 lines
3.1 KiB
Python

"""Test helpers for exercising ``_fast_init_or_upgrade`` closure logic.
This module provides utilities that safely swap the
``_original_init_or_upgrade`` reference captured inside the
``_fast_init_or_upgrade`` closure (installed by
``features.environment._install_template_db_patch``) with a
``unittest.mock.MagicMock`` so that step definitions can assert
whether the original migration runner was invoked.
All helpers live in ``features/mocks/`` per the project's mock-placement
rule.
"""
from __future__ import annotations
import contextlib
import types
from collections.abc import Generator
from typing import Any
from unittest.mock import MagicMock
def _find_closure_cell(
func: types.FunctionType,
var_name: str,
) -> tuple[int, types.CellType] | None:
"""Locate a closure cell by free-variable name.
Args:
func: The function whose closure to inspect.
var_name: The free-variable name to find.
Returns:
A ``(index, cell)`` tuple if found, otherwise ``None``.
Raises:
TypeError: If *func* is not a function with a closure.
"""
if not isinstance(func, types.FunctionType):
raise TypeError(f"Expected a function, got {type(func).__name__}")
freevars: tuple[str, ...] = func.__code__.co_freevars
closure: tuple[types.CellType, ...] | None = func.__closure__
if closure is None:
return None
for idx, name in enumerate(freevars):
if name == var_name:
return idx, closure[idx]
return None
@contextlib.contextmanager
def patch_original_init_or_upgrade() -> Generator[MagicMock]:
"""Context manager that replaces ``_original_init_or_upgrade`` inside
the ``_fast_init_or_upgrade`` closure with a :class:`~unittest.mock.MagicMock`.
Yields the mock so callers can assert on call counts and arguments.
Restores the original cell contents on exit.
Raises:
RuntimeError: If the fast-init patch is not installed on
``MigrationRunner.init_or_upgrade`` (e.g. template DB missing).
Example usage in a step definition::
with patch_original_init_or_upgrade() as mock_original:
runner.init_or_upgrade()
assert mock_original.called
"""
from cleveragents.infrastructure.database.migration_runner import (
MigrationRunner,
)
fast_fn: Any = MigrationRunner.init_or_upgrade
if not isinstance(fast_fn, types.FunctionType):
raise RuntimeError(
"MigrationRunner.init_or_upgrade is not a plain function — "
"the _fast_init_or_upgrade patch does not appear to be installed."
)
result = _find_closure_cell(fast_fn, "_original_init_or_upgrade")
if result is None:
raise RuntimeError(
"Cannot find '_original_init_or_upgrade' in the closure of "
f"{fast_fn.__qualname__}. Is the template-DB patch installed?"
)
_idx, cell = result
cell_ref: Any = cell
saved: Any = cell_ref.cell_contents
mock = MagicMock(name="_original_init_or_upgrade")
cell_ref.cell_contents = mock
try:
yield mock
finally:
cell_ref.cell_contents = saved