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
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
96 lines
3.1 KiB
Python
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
|