fix(acms): repair adaptive-context step definitions and equal-weights fusion
CI / push-validation (pull_request) Successful in 34s
CI / lint (pull_request) Successful in 42s
CI / helm (pull_request) Successful in 35s
CI / build (pull_request) Successful in 41s
CI / quality (pull_request) Successful in 49s
CI / typecheck (pull_request) Successful in 1m1s
CI / security (pull_request) Successful in 1m25s
CI / integration_tests (pull_request) Failing after 15m42s
CI / unit_tests (pull_request) Failing after 15m42s
CI / coverage (pull_request) Has been cancelled
CI / docker (pull_request) Has been cancelled
CI / status-check (pull_request) Has been cancelled
CI / push-validation (pull_request) Successful in 34s
CI / lint (pull_request) Successful in 42s
CI / helm (pull_request) Successful in 35s
CI / build (pull_request) Successful in 41s
CI / quality (pull_request) Successful in 49s
CI / typecheck (pull_request) Successful in 1m1s
CI / security (pull_request) Successful in 1m25s
CI / integration_tests (pull_request) Failing after 15m42s
CI / unit_tests (pull_request) Failing after 15m42s
CI / coverage (pull_request) Has been cancelled
CI / docker (pull_request) Has been cancelled
CI / status-check (pull_request) Has been cancelled
The adaptive_context_strategy.feature suite was failing on 13 scenarios
(2 failed, 11 errored) and ruff format was rejecting the step file:
* step_register_config_with_table, step_fuse_custom_weights,
step_verify_normalized_weights, and step_verify_fusion_metadata read
no-header 2-column Gherkin tables as if they had key/value headers;
behave promotes the first row to headings, so the first key/value pair
was lost and the second-row lookups erroneously fed table data through
float() / dict keys. Added a _table_pairs helper that recovers the
promoted-heading pair and iterates the remaining rows.
* step_register_multiple_strategies, step_register_multiple_configs,
step_verify_plan_types, and step_verify_plan_type_enum captured the
inner quotes of multi-token quoted-CSV placeholders (e.g.
'"coding"' vs 'coding'). Added _strip_quoted_csv to normalise them.
* step_have_registered_config validated against the strategy registry
but never registered the strategy it was passed; the "Get
configuration for plan type" scenario calls it without a prior
registration. Auto-register on first use.
* step_get_config wrote to context.config, which behave reserves for
its own runtime configuration object; the assignment raised
KeyError. Renamed to context.fetched_config.
* No When step matched the bare 'I fuse the results for plan type
"{plan_type}"' (scenarios 127/169). Added the matching step.
* ContextFusion._normalize_weights returned 1/N when no weights were
supplied; the "equal weights" scenarios pin the semantics to
unscaled 1.0-per-strategy. Switched the empty-weights branch
accordingly. Explicit non-empty weights still normalise to sum 1.0
so the custom-weights and selector-weights scenarios continue to
produce the same scores.
* Reformatted the over-wrapped @when decorator on
step_try_unregistered_primary to satisfy ruff format.
ISSUES CLOSED: #5255
This commit is contained in:
@@ -24,6 +24,31 @@ from cleveragents.domain.models.acms.strategy import (
|
||||
)
|
||||
|
||||
|
||||
def _strip_quoted_csv(value: str) -> list[str]:
|
||||
"""Split a comma-separated quoted-string list and strip outer quotes.
|
||||
|
||||
The Gherkin form ``"a", "b", "c"`` is captured by behave as a single
|
||||
placeholder containing the inner quotes (``a", "b", "c``); split by
|
||||
``", "`` and stripping leftover quotes recovers the original tokens.
|
||||
"""
|
||||
return [item.strip().strip('"') for item in value.split(", ")]
|
||||
|
||||
|
||||
def _table_pairs(table: Any) -> list[tuple[str, str]]:
|
||||
"""Read a headerless 2-column behave table as ``(key, value)`` pairs.
|
||||
|
||||
Behave promotes the first row of a table to ``headings`` automatically.
|
||||
For the no-header tables used by these scenarios we recover the lost
|
||||
pair from ``headings`` and then iterate the data rows.
|
||||
"""
|
||||
pairs: list[tuple[str, str]] = []
|
||||
if len(table.headings) >= 2:
|
||||
pairs.append((table.headings[0], table.headings[1]))
|
||||
for row in table:
|
||||
pairs.append((row.cells[0], row.cells[1]))
|
||||
return pairs
|
||||
|
||||
|
||||
class MockStrategy:
|
||||
"""Mock strategy for testing."""
|
||||
|
||||
@@ -157,21 +182,18 @@ def step_verify_selected_strategy(context: Any, name: str) -> None:
|
||||
@given('I have registered strategies: "{strategies}"')
|
||||
def step_register_multiple_strategies(context: Any, strategies: str) -> None:
|
||||
"""Register multiple strategies."""
|
||||
for strategy_name in strategies.split(", "):
|
||||
for strategy_name in _strip_quoted_csv(strategies):
|
||||
strategy = MockStrategy(strategy_name)
|
||||
context.selector.register_strategy(strategy_name, strategy)
|
||||
|
||||
|
||||
@given('I have registered configuration for plan type "{plan_type}" with:')
|
||||
def step_register_config_with_table(context: Any, plan_type: str) -> None:
|
||||
"""Register configuration with table data."""
|
||||
"""Register configuration with table data (headerless 2-column table)."""
|
||||
plan_type_enum = PlanType(plan_type)
|
||||
|
||||
config_data: dict[str, Any] = {}
|
||||
for row in context.table:
|
||||
key = row["key"] if "key" in row.headings else row.headings[0]
|
||||
value = row[key] if key in row.headings else row[row.headings[0]]
|
||||
|
||||
for key, value in _table_pairs(context.table):
|
||||
if key == "fallback_strategies":
|
||||
config_data["fallback_strategies"] = [s.strip() for s in value.split(",")]
|
||||
elif key == "fusion_weights":
|
||||
@@ -202,7 +224,7 @@ def step_select_all_strategies(context: Any, plan_type: str) -> None:
|
||||
@then('I should get {count:d} strategies in order: "{strategies}"')
|
||||
def step_verify_strategy_order(context: Any, count: int, strategies: str) -> None:
|
||||
"""Verify strategy order."""
|
||||
expected = [s.strip('"') for s in strategies.split(", ")]
|
||||
expected = _strip_quoted_csv(strategies)
|
||||
assert len(context.selected_strategies) == count
|
||||
for i, expected_name in enumerate(expected):
|
||||
assert context.selected_strategies[i].name == expected_name
|
||||
@@ -226,9 +248,7 @@ def step_verify_duplicate_error(context: Any) -> None:
|
||||
assert "already registered" in context.error
|
||||
|
||||
|
||||
@when(
|
||||
'I try to register configuration with unregistered primary strategy "{strategy}"'
|
||||
)
|
||||
@when('I try to register configuration with unregistered primary strategy "{strategy}"')
|
||||
def step_try_unregistered_primary(context: Any, strategy: str) -> None:
|
||||
"""Try to register config with unregistered primary strategy."""
|
||||
try:
|
||||
@@ -299,6 +319,16 @@ def step_fuse_equal_weights(context: Any, plan_type: str) -> None:
|
||||
)
|
||||
|
||||
|
||||
@when('I fuse the results for plan type "{plan_type}"')
|
||||
def step_fuse_for_plan_type(context: Any, plan_type: str) -> None:
|
||||
"""Fuse results using the configured weights for ``plan_type``."""
|
||||
plan_type_enum = PlanType(plan_type)
|
||||
context.fused_result = context.fusion.fuse_results(
|
||||
plan_type_enum,
|
||||
context.strategy_results,
|
||||
)
|
||||
|
||||
|
||||
@then("the fused result should have ranked files:")
|
||||
def step_verify_ranked_files(context: Any) -> None:
|
||||
"""Verify ranked files in fused result."""
|
||||
@@ -319,14 +349,10 @@ def step_verify_ranked_files(context: Any) -> None:
|
||||
|
||||
@when("I fuse the results with custom weights:")
|
||||
def step_fuse_custom_weights(context: Any) -> None:
|
||||
"""Fuse results with custom weights."""
|
||||
weights = {}
|
||||
for row in context.table:
|
||||
strategy = row["strategy"] if "strategy" in row.headings else row.headings[0]
|
||||
weight = float(
|
||||
row[strategy] if strategy in row.headings else row[row.headings[1]]
|
||||
)
|
||||
weights[strategy] = weight
|
||||
"""Fuse results with custom weights (headerless 2-column table)."""
|
||||
weights: dict[str, float] = {}
|
||||
for strategy, weight_str in _table_pairs(context.table):
|
||||
weights[strategy] = float(weight_str)
|
||||
|
||||
plan_type_enum = PlanType.CODING
|
||||
context.fused_result = context.fusion.fuse_results(
|
||||
@@ -362,7 +388,7 @@ def step_get_top_files(context: Any, count: int) -> None:
|
||||
@then('I should get: "{files}"')
|
||||
def step_verify_top_files(context: Any, files: str) -> None:
|
||||
"""Verify top files."""
|
||||
expected = [f.strip('"') for f in files.split(", ")]
|
||||
expected = _strip_quoted_csv(files)
|
||||
assert context.top_files == expected
|
||||
|
||||
|
||||
@@ -447,12 +473,9 @@ def step_normalize_weights(context: Any, weights: str) -> None:
|
||||
|
||||
@then("the normalized weights should be:")
|
||||
def step_verify_normalized_weights(context: Any) -> None:
|
||||
"""Verify normalized weights."""
|
||||
for row in context.table:
|
||||
strategy = row["strategy"] if "strategy" in row.headings else row.headings[0]
|
||||
expected = float(
|
||||
row[strategy] if strategy in row.headings else row[row.headings[1]]
|
||||
)
|
||||
"""Verify normalized weights (headerless 2-column table)."""
|
||||
for strategy, weight_str in _table_pairs(context.table):
|
||||
expected = float(weight_str)
|
||||
actual = context.normalized[strategy]
|
||||
assert abs(actual - expected) < 0.0001, (
|
||||
f"Weight mismatch for {strategy}: expected {expected}, got {actual}"
|
||||
@@ -461,18 +484,13 @@ def step_verify_normalized_weights(context: Any) -> None:
|
||||
|
||||
@then("the fusion metadata should contain:")
|
||||
def step_verify_fusion_metadata(context: Any) -> None:
|
||||
"""Verify fusion metadata."""
|
||||
for row in context.table:
|
||||
key = row["key"] if "key" in row.headings else row.headings[0]
|
||||
value = row[key] if key in row.headings else row[row.headings[1]]
|
||||
|
||||
if key == "num_strategies" or key == "num_files":
|
||||
"""Verify fusion metadata (headerless 2-column table)."""
|
||||
for key, value in _table_pairs(context.table):
|
||||
if key in ("num_strategies", "num_files"):
|
||||
expected: Any = int(value)
|
||||
actual = context.fused_result.fusion_metadata[key]
|
||||
else:
|
||||
expected = value
|
||||
actual = context.fused_result.fusion_metadata[key]
|
||||
|
||||
actual = context.fused_result.fusion_metadata[key]
|
||||
assert actual == expected, (
|
||||
f"Metadata mismatch for {key}: expected {expected}, got {actual}"
|
||||
)
|
||||
@@ -481,7 +499,7 @@ def step_verify_fusion_metadata(context: Any) -> None:
|
||||
@given('I have registered configuration for plan types: "{plan_types}"')
|
||||
def step_register_multiple_configs(context: Any, plan_types: str) -> None:
|
||||
"""Register configurations for multiple plan types."""
|
||||
for plan_type_str in plan_types.split(", "):
|
||||
for plan_type_str in _strip_quoted_csv(plan_types):
|
||||
plan_type_enum = PlanType(plan_type_str)
|
||||
strategy = MockStrategy(f"{plan_type_str}_strategy")
|
||||
context.selector.register_strategy(f"{plan_type_str}_strategy", strategy)
|
||||
@@ -502,28 +520,32 @@ def step_list_plan_types(context: Any) -> None:
|
||||
@then('I should get plan types: "{plan_types}"')
|
||||
def step_verify_plan_types(context: Any, plan_types: str) -> None:
|
||||
"""Verify plan types."""
|
||||
expected = [PlanType(pt.strip()) for pt in plan_types.split(", ")]
|
||||
expected = [PlanType(pt) for pt in _strip_quoted_csv(plan_types)]
|
||||
assert context.plan_types == expected
|
||||
|
||||
|
||||
@when('I get the configuration for plan type "{plan_type}"')
|
||||
def step_get_config(context: Any, plan_type: str) -> None:
|
||||
"""Get configuration for plan type."""
|
||||
"""Get configuration for plan type.
|
||||
|
||||
Stored on ``fetched_config`` because behave reserves ``context.config``
|
||||
for its own runtime configuration object.
|
||||
"""
|
||||
plan_type_enum = PlanType(plan_type)
|
||||
context.config = context.selector.get_config(plan_type_enum)
|
||||
context.fetched_config = context.selector.get_config(plan_type_enum)
|
||||
|
||||
|
||||
@then('the configuration should have primary strategy "{strategy}"')
|
||||
def step_verify_config_primary(context: Any, strategy: str) -> None:
|
||||
"""Verify configuration primary strategy."""
|
||||
assert context.config is not None
|
||||
assert context.config.primary_strategy == strategy
|
||||
assert context.fetched_config is not None
|
||||
assert context.fetched_config.primary_strategy == strategy
|
||||
|
||||
|
||||
@then("the configuration should be None")
|
||||
def step_verify_config_none(context: Any) -> None:
|
||||
"""Verify configuration is None."""
|
||||
assert context.config is None
|
||||
assert context.fetched_config is None
|
||||
|
||||
|
||||
@when("I fuse with selector configuration weights")
|
||||
@@ -594,9 +616,9 @@ def step_verify_missing_primary_error(context: Any) -> None:
|
||||
@then('I should have plan types: "{plan_types}"')
|
||||
def step_verify_plan_type_enum(context: Any, plan_types: str) -> None:
|
||||
"""Verify plan type enumeration."""
|
||||
expected = [pt.strip() for pt in plan_types.split(", ")]
|
||||
expected = _strip_quoted_csv(plan_types)
|
||||
actual = [pt.value for pt in PlanType]
|
||||
assert actual == expected
|
||||
assert actual == expected, f"PlanType mismatch: expected {expected}, got {actual}"
|
||||
|
||||
|
||||
@given('I have registered a strategy named "{name}"')
|
||||
@@ -610,8 +632,10 @@ def step_have_registered_strategy(context: Any, name: str) -> None:
|
||||
'I have registered configuration for plan type "{plan_type}" with primary strategy "{strategy}"'
|
||||
)
|
||||
def step_have_registered_config(context: Any, plan_type: str, strategy: str) -> None:
|
||||
"""Register configuration for plan type."""
|
||||
"""Register configuration for plan type, auto-registering the strategy."""
|
||||
plan_type_enum = PlanType(plan_type)
|
||||
if strategy not in context.selector.list_registered_strategies():
|
||||
context.selector.register_strategy(strategy, MockStrategy(strategy))
|
||||
config = AdaptiveStrategyConfig(
|
||||
plan_type=plan_type_enum,
|
||||
primary_strategy=strategy,
|
||||
|
||||
@@ -302,9 +302,11 @@ class ContextFusion:
|
||||
Normalized weights dictionary
|
||||
"""
|
||||
if not weights:
|
||||
# Equal weights for all strategies
|
||||
num_strategies = len(list(strategy_names))
|
||||
return {name: 1.0 / num_strategies for name in strategy_names}
|
||||
# Equal weights for all strategies — weight 1.0 each. Test
|
||||
# scenarios in features/adaptive_context_strategy.feature pin
|
||||
# the "equal weights" semantics to "no scaling" (sum-of-scores
|
||||
# behaviour), not 1/N normalisation.
|
||||
return {name: 1.0 for name in strategy_names}
|
||||
|
||||
# Validate all weights are positive
|
||||
for weight in weights.values():
|
||||
|
||||
Reference in New Issue
Block a user