forked from HAL9000/cleveragents-core
a808c395f9
Add 53 new .feature files and corresponding step definition files targeting uncovered lines identified in build/coverage.xml. Fix AmbiguousStep conflicts in 7 pre-existing step files by disambiguating step text. New tests cover: ACP clients/facade, actor CLI/config, application container, ACMS service/strategies, async worker, automation profile CLI, autonomy guardrail, bridge, change model, config CLI/service, context service, cross-plan correction, database models, decision service, decomposition clustering/service, discovery handler, langchain chat provider, langgraph nodes, materializers, multi-project service, plan apply/CLI/lifecycle/model/ preflight/resume/service, PostgreSQL analyzer, project CLI/context CLI, provider registry, reactive application/route, repositories, resolver handler, resource registry service, resume model, retry patterns, sandbox protocol, server CLI, skill CLI/service, skills registry, subplan execution/service, system CLI, UKO loader, UoW, and YAML template engine. Closes #645
198 lines
7.3 KiB
Python
198 lines
7.3 KiB
Python
"""Step definitions for config_service_coverage_boost.feature.
|
|
|
|
Targets uncovered lines in config_service.py:
|
|
- Line 1267: resolve() verbose + cli_value (CLI_FLAG chain entry with value)
|
|
- Lines 1279-1283: resolve() verbose + env var (ENV_VAR chain entry with value)
|
|
- Lines 1404-1405: set_project_value() with empty project_name
|
|
- Line 1415: set_project_value() non-dict project section
|
|
- Line 1418: set_project_value() non-dict project overrides
|
|
- Line 1450: get_project_overrides() non-dict project section
|
|
- Line 1453: get_project_overrides() non-dict project data
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.application.services.config_service import (
|
|
ConfigLevel,
|
|
ConfigService,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a fresh temporary config directory for coverage boost tests")
|
|
def step_fresh_temp_dir(context: Any) -> None:
|
|
context.cb_tmpdir = Path(tempfile.mkdtemp())
|
|
context.cb_config_dir = context.cb_tmpdir / "cfg"
|
|
context.cb_config_path = context.cb_config_dir / "config.toml"
|
|
context.cb_svc = None
|
|
context.cb_result = None
|
|
context.cb_error = None
|
|
context.cb_env_patches: list[Any] = []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a coverage boost ConfigService with empty config")
|
|
def step_svc_empty(context: Any) -> None:
|
|
context.cb_config_dir.mkdir(parents=True, exist_ok=True)
|
|
context.cb_svc = ConfigService(
|
|
config_dir=context.cb_config_dir,
|
|
config_path=context.cb_config_path,
|
|
)
|
|
|
|
|
|
@given('the coverage boost env var "{name}" is injected as "{value}"')
|
|
def step_set_env_var(context: Any, name: str, value: str) -> None:
|
|
patcher = patch.dict(os.environ, {name: value})
|
|
patcher.start()
|
|
context.cb_env_patches.append(patcher)
|
|
|
|
|
|
@given("a coverage boost ConfigService with project section set to a string")
|
|
def step_svc_project_section_string(context: Any) -> None:
|
|
context.cb_config_dir.mkdir(parents=True, exist_ok=True)
|
|
# Write a TOML file where "project" is a plain string, not a table.
|
|
# tomlkit won't let us write project = "bad" then treat it as a table,
|
|
# so we write raw TOML manually.
|
|
context.cb_config_path.write_text('project = "not_a_dict"\n')
|
|
context.cb_svc = ConfigService(
|
|
config_dir=context.cb_config_dir,
|
|
config_path=context.cb_config_path,
|
|
)
|
|
|
|
|
|
@given(
|
|
'a coverage boost ConfigService with project overrides for "{proj}" set to a string'
|
|
)
|
|
def step_svc_project_overrides_string(context: Any, proj: str) -> None:
|
|
context.cb_config_dir.mkdir(parents=True, exist_ok=True)
|
|
# Write TOML where [project] exists but project.<proj> is a string.
|
|
raw = f'[project]\n{proj} = "not_a_dict"\n'
|
|
context.cb_config_path.write_text(raw)
|
|
context.cb_svc = ConfigService(
|
|
config_dir=context.cb_config_dir,
|
|
config_path=context.cb_config_path,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I resolve "{key}" with cli_value "{cli_val}" and verbose True')
|
|
def step_resolve_cli_verbose(context: Any, key: str, cli_val: str) -> None:
|
|
context.cb_result = context.cb_svc.resolve(key, cli_value=cli_val, verbose=True)
|
|
|
|
|
|
@when('I resolve "{key}" with verbose True and no cli_value')
|
|
def step_resolve_verbose_no_cli(context: Any, key: str) -> None:
|
|
context.cb_result = context.cb_svc.resolve(key, verbose=True)
|
|
|
|
|
|
@when(
|
|
'I call set_project_value with empty project name and key "{key}" value {value:d}'
|
|
)
|
|
def step_set_project_value_empty_name(context: Any, key: str, value: int) -> None:
|
|
try:
|
|
context.cb_svc.set_project_value("", key, value)
|
|
context.cb_error = None
|
|
except (ValueError, TypeError) as exc:
|
|
context.cb_error = exc
|
|
|
|
|
|
@when('I call set_project_value for project "{proj}" key "{key}" value {value:d}')
|
|
def step_set_project_value_normal(
|
|
context: Any, proj: str, key: str, value: int
|
|
) -> None:
|
|
try:
|
|
context.cb_svc.set_project_value(proj, key, value)
|
|
context.cb_error = None
|
|
except (ValueError, TypeError) as exc:
|
|
context.cb_error = exc
|
|
|
|
|
|
@when('I call get_project_overrides for project "{proj}"')
|
|
def step_get_project_overrides(context: Any, proj: str) -> None:
|
|
context.cb_result = context.cb_svc.get_project_overrides(proj)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the chain should contain a CLI_FLAG entry with value "{expected}"')
|
|
def step_verify_cli_flag_chain_entry(context: Any, expected: str) -> None:
|
|
chain = context.cb_result.chain
|
|
cli_entries = [e for e in chain if e["source"] == ConfigLevel.CLI_FLAG.value]
|
|
assert len(cli_entries) == 1, f"Expected 1 CLI_FLAG entry, got {len(cli_entries)}"
|
|
assert cli_entries[0]["value"] == expected, (
|
|
f"Expected CLI_FLAG value '{expected}', got '{cli_entries[0]['value']}'"
|
|
)
|
|
|
|
|
|
@then(
|
|
'the chain should contain an ENV_VAR entry with value "{expected}" '
|
|
'and env_name "{env_name}"'
|
|
)
|
|
def step_verify_env_var_chain_entry(context: Any, expected: str, env_name: str) -> None:
|
|
# Clean up env patches after capturing the resolve result
|
|
for p in context.cb_env_patches:
|
|
p.stop()
|
|
context.cb_env_patches.clear()
|
|
|
|
chain = context.cb_result.chain
|
|
env_entries = [e for e in chain if e["source"] == ConfigLevel.ENV_VAR.value]
|
|
assert len(env_entries) == 1, f"Expected 1 ENV_VAR entry, got {len(env_entries)}"
|
|
entry = env_entries[0]
|
|
assert entry["value"] == expected, (
|
|
f"Expected ENV_VAR value '{expected}', got '{entry['value']}'"
|
|
)
|
|
assert entry["env_name"] == env_name, (
|
|
f"Expected env_name '{env_name}', got '{entry['env_name']}'"
|
|
)
|
|
|
|
|
|
@then('a ValueError with message "{fragment}" should be raised')
|
|
def step_verify_valueerror_message(context: Any, fragment: str) -> None:
|
|
assert context.cb_error is not None, "Expected a ValueError but none was raised"
|
|
assert isinstance(context.cb_error, ValueError), (
|
|
f"Expected ValueError, got {type(context.cb_error).__name__}"
|
|
)
|
|
assert fragment in str(context.cb_error), (
|
|
f"Expected '{fragment}' in error message, got: {context.cb_error}"
|
|
)
|
|
|
|
|
|
@then("the call should succeed without error")
|
|
def step_verify_no_error(context: Any) -> None:
|
|
assert context.cb_error is None, f"Expected no error but got: {context.cb_error}"
|
|
|
|
|
|
@then('get_project_overrides for "{proj}" should contain key "{key}"')
|
|
def step_verify_overrides_contain_key(context: Any, proj: str, key: str) -> None:
|
|
overrides = context.cb_svc.get_project_overrides(proj)
|
|
assert key in overrides, (
|
|
f"Expected '{key}' in project overrides, got: {list(overrides.keys())}"
|
|
)
|
|
|
|
|
|
@then("the coverage boost result should be an empty dictionary")
|
|
def step_verify_empty_dict(context: Any) -> None:
|
|
assert context.cb_result == {}, f"Expected empty dict, got: {context.cb_result}"
|