forked from HAL9000/cleveragents-core
freemo
051ee7c290
Added 52 new .feature files and corresponding _steps.py files targeting previously uncovered code paths in the following areas: - TUI layer: app, commands, persona (state/schema/registry), widgets, input (shell_exec, reference_parser) - Application services: plan lifecycle/service/executor, session, project, repo indexing, correction, checkpoint, actor, llm_actors, strategy coordinator, resource file watcher, service retry wiring - CLI commands: session, resource, repl, plan, db, automation_profile - Domain models: retry_policy, resource_type, cost_budget, docker_compose_analyzer, detail_level, _sql_string_aware, _postgresql_helpers - Core: circuit_breaker, retry_service_patterns - Infrastructure: repositories, transaction_sandbox, strategy_registry, plugins/loader, container - Config: settings - Agents: plan_generation, context_analysis, auto_debug - A2A: facade All new tests follow the Behave/Gherkin BDD standard. Resolved step definition collisions with unique prefixes. Fixed Alembic fileConfig logger disabling issue (disable_existing_loggers=False). ISSUES CLOSED: #1068
471 lines
16 KiB
Python
471 lines
16 KiB
Python
"""Step definitions for repo_indexing_utils_coverage.feature.
|
|
|
|
These steps target specific uncovered lines in repo_indexing_utils.py:
|
|
- Line 118: detect_language Dockerfile special-case return
|
|
- Line 175: _glob_match with '**' or '/' patterns (PurePosixPath.match)
|
|
- Lines 266, 268, 269: directory filtering in walk_and_index
|
|
- Line 274: hidden file skip
|
|
- Line 281: symlink skip
|
|
- Lines 285-286: ValueError from relative_to
|
|
- Lines 292-296: path > 1024 chars skip
|
|
- Lines 306-307: stat() OSError skip
|
|
- Line 312: non-regular file (FIFO) skip
|
|
- Lines 338-342: max_file_count cutoff
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.application.services.repo_indexing_utils import (
|
|
detect_language,
|
|
matches_policy,
|
|
walk_and_index,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
@given("the repo indexing utils module is imported")
|
|
def step_module_imported(context):
|
|
"""Ensure the module is importable."""
|
|
assert detect_language is not None
|
|
assert matches_policy is not None
|
|
assert walk_and_index is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# detect_language: Dockerfile (line 118)
|
|
# ---------------------------------------------------------------------------
|
|
@when('I detect the language for path "{path}"')
|
|
def step_detect_language(context, path):
|
|
context.detected_language = detect_language(path)
|
|
|
|
|
|
@then('the detected language should be "{expected}"')
|
|
def step_verify_language(context, expected):
|
|
assert context.detected_language == expected, (
|
|
f"Expected '{expected}', got '{context.detected_language}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# matches_policy: recursive / path globs (line 175)
|
|
# ---------------------------------------------------------------------------
|
|
@when('I check matches_policy for "{rel_path}" with include glob "{pattern}"')
|
|
def step_check_matches_policy(context, rel_path, pattern):
|
|
context.policy_result = matches_policy(
|
|
rel_path,
|
|
include_globs=(pattern,),
|
|
exclude_globs=(),
|
|
)
|
|
|
|
|
|
@then("the path should match the policy")
|
|
def step_path_matches(context):
|
|
assert context.policy_result is True, "Expected path to match policy"
|
|
|
|
|
|
@then("the path should not match the policy")
|
|
def step_path_not_matches(context):
|
|
assert context.policy_result is False, "Expected path NOT to match policy"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# walk_and_index: hidden / special directories filtered (lines 266-269)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a temporary directory with hidden and special subdirectories")
|
|
def step_create_dir_with_hidden_and_special(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.tmpdir = tmpdir
|
|
|
|
# Hidden directory — should be skipped
|
|
os.makedirs(os.path.join(tmpdir, ".git"))
|
|
with open(os.path.join(tmpdir, ".git", "config"), "w") as f:
|
|
f.write("gitconfig")
|
|
|
|
# __pycache__ — should be skipped
|
|
os.makedirs(os.path.join(tmpdir, "__pycache__"))
|
|
with open(os.path.join(tmpdir, "__pycache__", "mod.pyc"), "w") as f:
|
|
f.write("bytecode")
|
|
|
|
# node_modules — should be skipped
|
|
os.makedirs(os.path.join(tmpdir, "node_modules"))
|
|
with open(os.path.join(tmpdir, "node_modules", "pkg.js"), "w") as f:
|
|
f.write("js")
|
|
|
|
# venv — should be skipped
|
|
os.makedirs(os.path.join(tmpdir, "venv"))
|
|
with open(os.path.join(tmpdir, "venv", "activate.py"), "w") as f:
|
|
f.write("activate")
|
|
|
|
# Visible directory with a real file — should be indexed
|
|
os.makedirs(os.path.join(tmpdir, "src"))
|
|
with open(os.path.join(tmpdir, "src", "main.py"), "w") as f:
|
|
f.write("print('hello')")
|
|
|
|
def cleanup():
|
|
import shutil
|
|
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I run walk_and_index on the temporary directory")
|
|
def step_walk_and_index_tmpdir(context):
|
|
context.walk_results = walk_and_index(
|
|
Path(context.tmpdir),
|
|
include_globs=(),
|
|
exclude_globs=(),
|
|
max_file_size=None,
|
|
max_total_size=None,
|
|
)
|
|
|
|
|
|
@then("only files from visible non-special directories should be indexed")
|
|
def step_verify_only_visible_dirs(context):
|
|
paths = [r.path for r in context.walk_results]
|
|
# The only file should be src/main.py
|
|
assert len(paths) == 1, f"Expected 1 file, got {len(paths)}: {paths}"
|
|
assert paths[0] == os.path.join("src", "main.py"), (
|
|
f"Expected 'src/main.py', got '{paths[0]}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# walk_and_index: hidden files skipped (line 274)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a temporary directory with hidden files")
|
|
def step_create_dir_with_hidden_files(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.tmpdir_hidden = tmpdir
|
|
|
|
# Hidden file — should be skipped
|
|
with open(os.path.join(tmpdir, ".hidden_config"), "w") as f:
|
|
f.write("secret")
|
|
|
|
# Normal file — should be indexed
|
|
with open(os.path.join(tmpdir, "visible.py"), "w") as f:
|
|
f.write("x = 1")
|
|
|
|
def cleanup():
|
|
import shutil
|
|
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I run walk_and_index on the directory with hidden files")
|
|
def step_walk_hidden_files(context):
|
|
context.walk_results = walk_and_index(
|
|
Path(context.tmpdir_hidden),
|
|
include_globs=(),
|
|
exclude_globs=(),
|
|
max_file_size=None,
|
|
max_total_size=None,
|
|
)
|
|
|
|
|
|
@then("hidden files should not appear in the results")
|
|
def step_verify_no_hidden_files(context):
|
|
paths = [r.path for r in context.walk_results]
|
|
hidden = [p for p in paths if os.path.basename(p).startswith(".")]
|
|
assert len(hidden) == 0, f"Hidden files found: {hidden}"
|
|
assert "visible.py" in paths, f"Expected visible.py in {paths}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# walk_and_index: symlinks skipped (line 281)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a temporary directory with a symlink to a file")
|
|
def step_create_dir_with_symlink(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.tmpdir_symlink = tmpdir
|
|
|
|
# Real file
|
|
real_file = os.path.join(tmpdir, "real.py")
|
|
with open(real_file, "w") as f:
|
|
f.write("real content")
|
|
|
|
# Symlink pointing to the real file
|
|
link_path = os.path.join(tmpdir, "link_to_real.py")
|
|
os.symlink(real_file, link_path)
|
|
|
|
def cleanup():
|
|
import shutil
|
|
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I run walk_and_index on the directory with symlinks")
|
|
def step_walk_symlinks(context):
|
|
context.walk_results = walk_and_index(
|
|
Path(context.tmpdir_symlink),
|
|
include_globs=(),
|
|
exclude_globs=(),
|
|
max_file_size=None,
|
|
max_total_size=None,
|
|
)
|
|
|
|
|
|
@then("symlinked files should not appear in the results")
|
|
def step_verify_no_symlinks(context):
|
|
paths = [r.path for r in context.walk_results]
|
|
assert "link_to_real.py" not in paths, (
|
|
f"Symlink should be excluded but found in: {paths}"
|
|
)
|
|
assert "real.py" in paths, f"Expected real.py in {paths}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# walk_and_index: ValueError from relative_to (lines 285-286)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a temporary directory with a file that will cause a relative_to ValueError")
|
|
def step_create_dir_for_relative_to_error(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.tmpdir_relto = tmpdir
|
|
|
|
with open(os.path.join(tmpdir, "normal.py"), "w") as f:
|
|
f.write("x = 1")
|
|
|
|
def cleanup():
|
|
import shutil
|
|
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I run walk_and_index with mocked relative_to failure")
|
|
def step_walk_with_relative_to_mock(context):
|
|
"""Patch Path.relative_to so the first call raises ValueError,
|
|
but subsequent calls (or the resolve() call) work normally."""
|
|
original_relative_to = Path.relative_to
|
|
call_count = {"n": 0}
|
|
|
|
def patched_relative_to(self, other):
|
|
call_count["n"] += 1
|
|
# Fail for the first real file encountered to exercise lines 285-286
|
|
if call_count["n"] == 1:
|
|
raise ValueError("mocked relative_to failure")
|
|
return original_relative_to(self, other)
|
|
|
|
with patch.object(Path, "relative_to", patched_relative_to):
|
|
context.walk_results = walk_and_index(
|
|
Path(context.tmpdir_relto),
|
|
include_globs=(),
|
|
exclude_globs=(),
|
|
max_file_size=None,
|
|
max_total_size=None,
|
|
)
|
|
|
|
|
|
@then("the file causing ValueError should be skipped gracefully")
|
|
def step_verify_relative_to_skip(context):
|
|
# The function should complete without raising; the problematic file
|
|
# is simply skipped. Result may be empty (the only file was skipped).
|
|
assert isinstance(context.walk_results, list)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# walk_and_index: path > 1024 chars (lines 292-296)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a temporary directory with a deeply nested file exceeding 1024 char path")
|
|
def step_create_deeply_nested(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.tmpdir_deep = tmpdir
|
|
|
|
# Build a path whose relative portion > 1024 chars.
|
|
# Use 60-char directory names nested ~20 levels deep → ~1200 chars.
|
|
nested = tmpdir
|
|
for i in range(20):
|
|
segment = f"d{'x' * 58}{i:01d}" # 60 chars each
|
|
nested = os.path.join(nested, segment)
|
|
os.makedirs(nested, exist_ok=True)
|
|
|
|
deep_file = os.path.join(nested, "deep_file.py")
|
|
with open(deep_file, "w") as f:
|
|
f.write("deep = True")
|
|
|
|
# Also add a normal file at root level
|
|
with open(os.path.join(tmpdir, "shallow.py"), "w") as f:
|
|
f.write("shallow = True")
|
|
|
|
context.deep_file_rel = str(Path(deep_file).relative_to(Path(tmpdir).resolve()))
|
|
|
|
def cleanup():
|
|
import shutil
|
|
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I run walk_and_index on the deeply nested directory")
|
|
def step_walk_deeply_nested(context):
|
|
context.walk_results = walk_and_index(
|
|
Path(context.tmpdir_deep),
|
|
include_globs=(),
|
|
exclude_globs=(),
|
|
max_file_size=None,
|
|
max_total_size=None,
|
|
)
|
|
|
|
|
|
@then("the deeply nested file should be skipped")
|
|
def step_verify_deep_file_skipped(context):
|
|
paths = [r.path for r in context.walk_results]
|
|
# The deep file path exceeds 1024 chars so should be excluded
|
|
long_paths = [p for p in paths if len(p) > 1024]
|
|
assert len(long_paths) == 0, f"Long paths should be excluded: {long_paths}"
|
|
# shallow.py should still be there
|
|
assert "shallow.py" in paths, f"Expected shallow.py in {paths}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# walk_and_index: stat() OSError (lines 306-307)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a temporary directory with a file that will cause a stat OSError")
|
|
def step_create_dir_for_stat_error(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.tmpdir_stat = tmpdir
|
|
|
|
with open(os.path.join(tmpdir, "goodfile.py"), "w") as f:
|
|
f.write("good = True")
|
|
|
|
with open(os.path.join(tmpdir, "badfile.py"), "w") as f:
|
|
f.write("bad = True")
|
|
|
|
def cleanup():
|
|
import shutil
|
|
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I run walk_and_index with mocked stat failure")
|
|
def step_walk_with_stat_mock(context):
|
|
"""Patch Path.stat so it raises OSError for badfile.py.
|
|
|
|
Only intercept calls with follow_symlinks=True (the default),
|
|
so that is_symlink() → lstat() → stat(follow_symlinks=False)
|
|
still works normally.
|
|
"""
|
|
original_stat = Path.stat
|
|
|
|
def patched_stat(self, *args, **kwargs):
|
|
follow = kwargs.get("follow_symlinks", True)
|
|
if self.name == "badfile.py" and follow:
|
|
raise OSError("mocked stat failure")
|
|
return original_stat(self, *args, **kwargs)
|
|
|
|
with patch.object(Path, "stat", patched_stat):
|
|
context.walk_results = walk_and_index(
|
|
Path(context.tmpdir_stat),
|
|
include_globs=(),
|
|
exclude_globs=(),
|
|
max_file_size=None,
|
|
max_total_size=None,
|
|
)
|
|
|
|
|
|
@then("the file causing stat OSError should be skipped gracefully")
|
|
def step_verify_stat_error_skip(context):
|
|
paths = [r.path for r in context.walk_results]
|
|
assert "badfile.py" not in paths, (
|
|
f"badfile.py should be skipped due to OSError, got: {paths}"
|
|
)
|
|
assert "goodfile.py" in paths, f"Expected goodfile.py in {paths}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# walk_and_index: non-regular file / FIFO (line 312)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a temporary directory with a FIFO named pipe")
|
|
def step_create_dir_with_fifo(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.tmpdir_fifo = tmpdir
|
|
|
|
# Regular file
|
|
with open(os.path.join(tmpdir, "regular.py"), "w") as f:
|
|
f.write("regular = True")
|
|
|
|
# FIFO (named pipe) — should be skipped by S_ISREG check
|
|
fifo_path = os.path.join(tmpdir, "myfifo")
|
|
os.mkfifo(fifo_path)
|
|
|
|
def cleanup():
|
|
import shutil
|
|
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I run walk_and_index on the directory with the FIFO")
|
|
def step_walk_fifo(context):
|
|
context.walk_results = walk_and_index(
|
|
Path(context.tmpdir_fifo),
|
|
include_globs=(),
|
|
exclude_globs=(),
|
|
max_file_size=None,
|
|
max_total_size=None,
|
|
)
|
|
|
|
|
|
@then("the FIFO should not appear in the results")
|
|
def step_verify_no_fifo(context):
|
|
paths = [r.path for r in context.walk_results]
|
|
assert "myfifo" not in paths, f"FIFO should be excluded, got: {paths}"
|
|
assert "regular.py" in paths, f"Expected regular.py in {paths}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# walk_and_index: max_file_count cutoff (lines 338-342)
|
|
# ---------------------------------------------------------------------------
|
|
@given("a temporary directory with five regular files")
|
|
def step_create_dir_with_five_files(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.tmpdir_count = tmpdir
|
|
|
|
for i in range(5):
|
|
with open(os.path.join(tmpdir, f"file_{i:02d}.py"), "w") as f:
|
|
f.write(f"x = {i}")
|
|
|
|
def cleanup():
|
|
import shutil
|
|
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I run walk_and_index with max_file_count of {count:d}")
|
|
def step_walk_with_max_count(context, count):
|
|
context.walk_results = walk_and_index(
|
|
Path(context.tmpdir_count),
|
|
include_globs=(),
|
|
exclude_globs=(),
|
|
max_file_size=None,
|
|
max_total_size=None,
|
|
max_file_count=count,
|
|
)
|
|
|
|
|
|
@then("only {count:d} file records should be returned")
|
|
def step_verify_file_count(context, count):
|
|
actual = len(context.walk_results)
|
|
assert actual == count, (
|
|
f"Expected {count} records, got {actual}: "
|
|
f"{[r.path for r in context.walk_results]}"
|
|
)
|