forked from HAL9000/cleveragents-core
freemo
31472b5413
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate. ISSUES CLOSED: #1232
370 lines
13 KiB
Python
370 lines
13 KiB
Python
"""Step definitions for config_service_coverage_r3.feature.
|
|
|
|
Targets uncovered lines in config_service.py:
|
|
- Line 1195: project_root property
|
|
- Line 1201: project_config_path returns None
|
|
- Line 1208: local_config_path returns None
|
|
- Lines 1284-1285, 1289: write_scoped_config ValueError (no project root)
|
|
- Lines 1297-1298: write_scoped_config reads existing target file
|
|
- Lines 1336-1337: set_value redacts sensitive keys
|
|
- Lines 1351-1352: set_value catches event_bus emit failure
|
|
- Lines 1507-1508, 1510-1511: resolve verbose with local config winning
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import tomllib
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import tomlkit
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.application.services.config_service import (
|
|
ConfigLevel,
|
|
ConfigScope,
|
|
ConfigService,
|
|
)
|
|
from cleveragents.shared.redaction import REDACTED
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("cfscov3 a fresh temporary directory")
|
|
def step_cfscov3_fresh_tmpdir(context: Any) -> None:
|
|
context.cfscov3_tmpdir = Path(tempfile.mkdtemp())
|
|
context.cfscov3_config_dir = context.cfscov3_tmpdir / "cfg"
|
|
context.cfscov3_config_dir.mkdir(parents=True, exist_ok=True)
|
|
context.cfscov3_config_path = context.cfscov3_config_dir / "config.toml"
|
|
context.cfscov3_svc: ConfigService | None = None
|
|
context.cfscov3_error: Exception | None = None
|
|
context.cfscov3_result: Any = None
|
|
context.cfscov3_event_bus: MagicMock | None = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("cfscov3 a ConfigService with an explicit project root")
|
|
def step_cfscov3_svc_with_project_root(context: Any) -> None:
|
|
project_root = context.cfscov3_tmpdir / "myproject"
|
|
project_root.mkdir(parents=True, exist_ok=True)
|
|
context.cfscov3_project_root = project_root
|
|
context.cfscov3_svc = ConfigService(
|
|
config_dir=context.cfscov3_config_dir,
|
|
config_path=context.cfscov3_config_path,
|
|
project_root=project_root,
|
|
)
|
|
|
|
|
|
@given("cfscov3 a ConfigService with no project root")
|
|
def step_cfscov3_svc_no_project_root(context: Any) -> None:
|
|
context.cfscov3_svc = ConfigService(
|
|
config_dir=context.cfscov3_config_dir,
|
|
config_path=context.cfscov3_config_path,
|
|
project_root=None,
|
|
)
|
|
|
|
|
|
@given("cfscov3 a ConfigService with a project root containing an existing config.toml")
|
|
def step_cfscov3_svc_with_existing_project_config(context: Any) -> None:
|
|
project_root = context.cfscov3_tmpdir / "proj_existing"
|
|
project_root.mkdir(parents=True, exist_ok=True)
|
|
context.cfscov3_project_root = project_root
|
|
|
|
# Write an initial config.toml with an existing key
|
|
target = project_root / "config.toml"
|
|
doc = tomlkit.document()
|
|
doc["existing_key"] = "existing_value"
|
|
with open(target, "w") as fh:
|
|
tomlkit.dump(doc, fh)
|
|
|
|
context.cfscov3_svc = ConfigService(
|
|
config_dir=context.cfscov3_config_dir,
|
|
config_path=context.cfscov3_config_path,
|
|
project_root=project_root,
|
|
)
|
|
|
|
|
|
@given(
|
|
"cfscov3 a ConfigService with a project root containing an existing config.local.toml"
|
|
)
|
|
def step_cfscov3_svc_with_existing_local_config(context: Any) -> None:
|
|
project_root = context.cfscov3_tmpdir / "proj_existing_local"
|
|
project_root.mkdir(parents=True, exist_ok=True)
|
|
context.cfscov3_project_root = project_root
|
|
|
|
# Write an initial config.local.toml with an existing key
|
|
target = project_root / "config.local.toml"
|
|
doc = tomlkit.document()
|
|
doc["local_existing"] = "local_value"
|
|
with open(target, "w") as fh:
|
|
tomlkit.dump(doc, fh)
|
|
|
|
context.cfscov3_svc = ConfigService(
|
|
config_dir=context.cfscov3_config_dir,
|
|
config_path=context.cfscov3_config_path,
|
|
project_root=project_root,
|
|
)
|
|
|
|
|
|
@given("cfscov3 a ConfigService with a mock event bus")
|
|
def step_cfscov3_svc_with_mock_event_bus(context: Any) -> None:
|
|
bus = MagicMock()
|
|
context.cfscov3_event_bus = bus
|
|
context.cfscov3_svc = ConfigService(
|
|
config_dir=context.cfscov3_config_dir,
|
|
config_path=context.cfscov3_config_path,
|
|
event_bus=bus,
|
|
project_root=None,
|
|
)
|
|
|
|
|
|
@given("cfscov3 a ConfigService with a failing event bus")
|
|
def step_cfscov3_svc_with_failing_event_bus(context: Any) -> None:
|
|
bus = MagicMock()
|
|
bus.emit.side_effect = RuntimeError("emit exploded")
|
|
context.cfscov3_event_bus = bus
|
|
context.cfscov3_svc = ConfigService(
|
|
config_dir=context.cfscov3_config_dir,
|
|
config_path=context.cfscov3_config_path,
|
|
event_bus=bus,
|
|
project_root=None,
|
|
)
|
|
|
|
|
|
@given(
|
|
'cfscov3 a ConfigService with a local config containing "core.log.level" set to "TRACE"'
|
|
)
|
|
def step_cfscov3_svc_with_local_config_log_level(context: Any) -> None:
|
|
project_root = context.cfscov3_tmpdir / "proj_local_resolve"
|
|
project_root.mkdir(parents=True, exist_ok=True)
|
|
context.cfscov3_project_root = project_root
|
|
|
|
# Write config.local.toml with the key
|
|
local_cfg = project_root / "config.local.toml"
|
|
doc = tomlkit.document()
|
|
doc["core.log.level"] = "TRACE"
|
|
with open(local_cfg, "w") as fh:
|
|
tomlkit.dump(doc, fh)
|
|
|
|
context.cfscov3_svc = ConfigService(
|
|
config_dir=context.cfscov3_config_dir,
|
|
config_path=context.cfscov3_config_path,
|
|
project_root=project_root,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("cfscov3 I call write_scoped_config with PROJECT scope")
|
|
def step_cfscov3_write_scoped_project(context: Any) -> None:
|
|
try:
|
|
context.cfscov3_svc.write_scoped_config({"key": "val"}, ConfigScope.PROJECT)
|
|
except (ValueError, TypeError) as exc:
|
|
context.cfscov3_error = exc
|
|
|
|
|
|
@when("cfscov3 I call write_scoped_config with LOCAL scope")
|
|
def step_cfscov3_write_scoped_local(context: Any) -> None:
|
|
try:
|
|
context.cfscov3_svc.write_scoped_config({"key": "val"}, ConfigScope.LOCAL)
|
|
except (ValueError, TypeError) as exc:
|
|
context.cfscov3_error = exc
|
|
|
|
|
|
@when("cfscov3 I call write_scoped_config with PROJECT scope adding a new key")
|
|
def step_cfscov3_write_scoped_project_add_key(context: Any) -> None:
|
|
try:
|
|
context.cfscov3_svc.write_scoped_config(
|
|
{"new_key": "new_value"}, ConfigScope.PROJECT
|
|
)
|
|
except (ValueError, TypeError) as exc:
|
|
context.cfscov3_error = exc
|
|
|
|
|
|
@when("cfscov3 I call write_scoped_config with LOCAL scope adding a new key")
|
|
def step_cfscov3_write_scoped_local_add_key(context: Any) -> None:
|
|
try:
|
|
context.cfscov3_svc.write_scoped_config(
|
|
{"new_key": "new_value"}, ConfigScope.LOCAL
|
|
)
|
|
except (ValueError, TypeError) as exc:
|
|
context.cfscov3_error = exc
|
|
|
|
|
|
@when('cfscov3 I call set_value with sensitive key "{key}" and value "{value}"')
|
|
def step_cfscov3_set_value_sensitive(context: Any, key: str, value: str) -> None:
|
|
try:
|
|
context.cfscov3_svc.set_value(key, value)
|
|
except Exception as exc:
|
|
context.cfscov3_error = exc
|
|
|
|
|
|
@when('cfscov3 I call set_value with key "{key}" and value "{value}"')
|
|
def step_cfscov3_set_value(context: Any, key: str, value: str) -> None:
|
|
try:
|
|
context.cfscov3_svc.set_value(key, value)
|
|
except Exception as exc:
|
|
context.cfscov3_error = exc
|
|
|
|
|
|
@when('cfscov3 I resolve "{key}" with verbose True')
|
|
def step_cfscov3_resolve_verbose(context: Any, key: str) -> None:
|
|
# Ensure no env var interferes with resolution
|
|
entry = ConfigService.get_entry(key)
|
|
env_var_name = entry.env_var if entry else ""
|
|
patcher = patch.dict("os.environ", {}, clear=False)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
# Remove the specific env var if present to ensure local config wins
|
|
import os
|
|
|
|
saved = os.environ.pop(env_var_name, None)
|
|
if saved is not None:
|
|
context.add_cleanup(lambda: os.environ.__setitem__(env_var_name, saved))
|
|
|
|
context.cfscov3_result = context.cfscov3_svc.resolve(key, verbose=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("cfscov3 project_root should return the configured path")
|
|
def step_cfscov3_project_root_configured(context: Any) -> None:
|
|
actual = context.cfscov3_svc.project_root
|
|
expected = context.cfscov3_project_root
|
|
assert actual == expected, f"Expected project_root={expected}, got {actual}"
|
|
|
|
|
|
@then("cfscov3 project_root should return None")
|
|
def step_cfscov3_project_root_none(context: Any) -> None:
|
|
actual = context.cfscov3_svc.project_root
|
|
assert actual is None, f"Expected project_root=None, got {actual}"
|
|
|
|
|
|
@then("cfscov3 project_config_path should return None")
|
|
def step_cfscov3_project_config_path_none(context: Any) -> None:
|
|
actual = context.cfscov3_svc.project_config_path
|
|
assert actual is None, f"Expected project_config_path=None, got {actual}"
|
|
|
|
|
|
@then("cfscov3 local_config_path should return None")
|
|
def step_cfscov3_local_config_path_none(context: Any) -> None:
|
|
actual = context.cfscov3_svc.local_config_path
|
|
assert actual is None, f"Expected local_config_path=None, got {actual}"
|
|
|
|
|
|
@then('cfscov3 a ValueError mentioning "no project root" should be stored')
|
|
def step_cfscov3_valueerror_no_project_root(context: Any) -> None:
|
|
assert context.cfscov3_error is not None, (
|
|
"Expected a ValueError but none was raised"
|
|
)
|
|
assert isinstance(context.cfscov3_error, ValueError), (
|
|
f"Expected ValueError, got {type(context.cfscov3_error).__name__}"
|
|
)
|
|
msg = str(context.cfscov3_error).lower()
|
|
assert "no project root" in msg, (
|
|
f"Expected 'no project root' in error message, got: {context.cfscov3_error}"
|
|
)
|
|
|
|
|
|
@then("cfscov3 the project config.toml should contain both the old and new keys")
|
|
def step_cfscov3_verify_merged_project_config(context: Any) -> None:
|
|
assert context.cfscov3_error is None, (
|
|
f"write_scoped_config raised: {context.cfscov3_error}"
|
|
)
|
|
target = context.cfscov3_project_root / "config.toml"
|
|
assert target.exists(), "config.toml should exist"
|
|
with open(target, "rb") as fh:
|
|
data = tomllib.load(fh)
|
|
assert "existing_key" in data, (
|
|
f"Expected 'existing_key' in config, got keys: {list(data.keys())}"
|
|
)
|
|
assert data["existing_key"] == "existing_value"
|
|
assert "new_key" in data, (
|
|
f"Expected 'new_key' in config, got keys: {list(data.keys())}"
|
|
)
|
|
assert data["new_key"] == "new_value"
|
|
|
|
|
|
@then("cfscov3 the local config.local.toml should contain both the old and new keys")
|
|
def step_cfscov3_verify_merged_local_config(context: Any) -> None:
|
|
assert context.cfscov3_error is None, (
|
|
f"write_scoped_config raised: {context.cfscov3_error}"
|
|
)
|
|
target = context.cfscov3_project_root / "config.local.toml"
|
|
assert target.exists(), "config.local.toml should exist"
|
|
with open(target, "rb") as fh:
|
|
data = tomllib.load(fh)
|
|
assert "local_existing" in data, (
|
|
f"Expected 'local_existing' in config, got keys: {list(data.keys())}"
|
|
)
|
|
assert data["local_existing"] == "local_value"
|
|
assert "new_key" in data, (
|
|
f"Expected 'new_key' in config, got keys: {list(data.keys())}"
|
|
)
|
|
assert data["new_key"] == "new_value"
|
|
|
|
|
|
@then("cfscov3 the event bus should have received redacted values")
|
|
def step_cfscov3_verify_redacted_event(context: Any) -> None:
|
|
assert context.cfscov3_error is None, f"set_value raised: {context.cfscov3_error}"
|
|
bus = context.cfscov3_event_bus
|
|
assert bus.emit.called, "Expected event_bus.emit to be called"
|
|
call_args = bus.emit.call_args
|
|
event = call_args[0][0] # First positional argument
|
|
details = event.details
|
|
assert details["old_value"] == REDACTED, (
|
|
f"Expected old_value to be REDACTED, got: {details['old_value']}"
|
|
)
|
|
assert details["new_value"] == REDACTED, (
|
|
f"Expected new_value to be REDACTED, got: {details['new_value']}"
|
|
)
|
|
|
|
|
|
@then("cfscov3 no error should have been raised")
|
|
def step_cfscov3_no_error(context: Any) -> None:
|
|
assert context.cfscov3_error is None, (
|
|
f"Expected no error, but got: {context.cfscov3_error}"
|
|
)
|
|
|
|
|
|
@then('cfscov3 the resolved value should be "{expected}" from LOCAL source')
|
|
def step_cfscov3_resolved_local(context: Any, expected: str) -> None:
|
|
result = context.cfscov3_result
|
|
assert result.value == expected, (
|
|
f"Expected resolved value '{expected}', got '{result.value}'"
|
|
)
|
|
assert result.source == ConfigLevel.LOCAL, (
|
|
f"Expected source LOCAL, got {result.source}"
|
|
)
|
|
|
|
|
|
@then(
|
|
'cfscov3 the chain should contain a local entry with value "{expected}" and a path'
|
|
)
|
|
def step_cfscov3_chain_local_entry(context: Any, expected: str) -> None:
|
|
chain = context.cfscov3_result.chain
|
|
local_entries = [e for e in chain if e["source"] == ConfigLevel.LOCAL.value]
|
|
assert len(local_entries) == 1, (
|
|
f"Expected 1 LOCAL chain entry, got {len(local_entries)}: {chain}"
|
|
)
|
|
entry = local_entries[0]
|
|
assert entry["value"] == expected, (
|
|
f"Expected LOCAL chain value '{expected}', got '{entry['value']}'"
|
|
)
|
|
assert "path" in entry, "Expected 'path' key in LOCAL chain entry"
|
|
assert entry["path"], "Expected non-empty path in LOCAL chain entry"
|