Files
placeholder/features/steps/git_checkout_handler_coverage_r3_steps.py
freemo 31472b5413 test(coverage): add Behave scenarios for 39 under-covered modules
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate.

ISSUES CLOSED: #1232
2026-03-31 21:47:12 +00:00

629 lines
21 KiB
Python

"""Step definitions for git_checkout_handler_coverage_r3.feature.
All steps use the ``gcov`` prefix to avoid Behave AmbiguousStep errors.
Exercises uncovered lines in git_checkout.py (GitCheckoutHandler).
"""
from __future__ import annotations
import os
import subprocess
import tempfile
from unittest.mock import MagicMock, patch
from behave import given, then, when
from cleveragents.domain.models.core.resource import PhysVirt, Resource
from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
_SUBPROCESS_RUN = "subprocess.run"
_HANDLER_MODULE = "cleveragents.resource.handlers.git_checkout"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_resource(location: str | None = None) -> Resource:
"""Create a minimal Resource with a git-checkout type."""
return Resource(
resource_id="01JQDVHN5X5QJKBMZ3AP8Y4G7K",
name="test-repo",
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
description="Test git checkout resource",
location=location,
)
def _init_git_repo(files: dict[str, bytes] | None = None) -> str:
"""Create a temporary git repo with an initial commit and optional files."""
repo = tempfile.mkdtemp(prefix="gcov-repo-")
subprocess.run(["git", "init"], cwd=repo, capture_output=True, check=True)
subprocess.run(
["git", "config", "user.email", "test@gcov.dev"],
cwd=repo,
capture_output=True,
check=True,
)
subprocess.run(
["git", "config", "user.name", "GCov Test"],
cwd=repo,
capture_output=True,
check=True,
)
subprocess.run(
["git", "config", "commit.gpgSign", "false"],
cwd=repo,
capture_output=True,
check=True,
)
# Always create a README so there's at least one tracked file
readme = os.path.join(repo, "README.md")
with open(readme, "w") as f:
f.write("# gcov test\n")
if files:
for name, data in files.items():
fpath = os.path.join(repo, name)
os.makedirs(os.path.dirname(fpath), exist_ok=True)
with open(fpath, "wb") as f:
f.write(data)
subprocess.run(["git", "add", "."], cwd=repo, capture_output=True, check=True)
subprocess.run(
["git", "commit", "-m", "init"],
cwd=repo,
capture_output=True,
check=True,
)
return repo
# ---------------------------------------------------------------------------
# GIVEN steps
# ---------------------------------------------------------------------------
@given("gcov a handler with a git repo containing tracked files")
def step_gcov_handler_with_tracked(ctx):
repo = _init_git_repo({"src/app.py": b"print('hi')\n"})
ctx.gcov_handler = GitCheckoutHandler()
ctx.gcov_resource = _make_resource(location=repo)
ctx.gcov_repo = repo
ctx.gcov_error = None
ctx.gcov_result = None
@given("gcov a handler with a git repo location")
def step_gcov_handler_plain(ctx):
repo = _init_git_repo()
ctx.gcov_handler = GitCheckoutHandler()
ctx.gcov_resource = _make_resource(location=repo)
ctx.gcov_repo = repo
ctx.gcov_error = None
ctx.gcov_result = None
@given('gcov a handler with a git repo and a real file "{filename}"')
def step_gcov_handler_with_file(ctx, filename):
content = b"hello world\n"
repo = _init_git_repo({filename: content})
ctx.gcov_handler = GitCheckoutHandler()
ctx.gcov_resource = _make_resource(location=repo)
ctx.gcov_repo = repo
ctx.gcov_file_content = content
ctx.gcov_error = None
ctx.gcov_result = None
@given('gcov a handler with a git repo and a real binary file "{filename}"')
def step_gcov_handler_with_binary_file(ctx, filename):
# Non-UTF-8 binary content
content = bytes(range(256))
repo = _init_git_repo({filename: content})
ctx.gcov_handler = GitCheckoutHandler()
ctx.gcov_resource = _make_resource(location=repo)
ctx.gcov_repo = repo
ctx.gcov_file_content = content
ctx.gcov_error = None
ctx.gcov_result = None
@given('gcov a handler with a git repo containing subdirectory "{dirname}"')
def step_gcov_handler_with_subdir(ctx, dirname):
files = {f"{dirname}/file.txt": b"data\n"}
repo = _init_git_repo(files)
ctx.gcov_handler = GitCheckoutHandler()
ctx.gcov_resource = _make_resource(location=repo)
ctx.gcov_repo = repo
ctx.gcov_error = None
ctx.gcov_result = None
@given('gcov git show returns binary data for path "{path}"')
def step_gcov_mock_git_show_binary(ctx, path):
"""Mock subprocess.run so that git show returns binary (non-UTF-8) data."""
real_run = subprocess.run
binary_data = bytes(range(128, 256)) # non-UTF-8
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "show" in cmd:
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = binary_data
mock_result.stderr = b""
return mock_result
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
# We also need the file to exist for _safe_resolve
fpath = os.path.join(ctx.gcov_repo, path)
os.makedirs(os.path.dirname(fpath), exist_ok=True)
with open(fpath, "wb") as f:
f.write(binary_data)
@given("gcov subprocess run is mocked to raise TimeoutExpired for git show")
def step_gcov_mock_timeout_git_show(ctx):
real_run = subprocess.run
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "show" in cmd:
raise subprocess.TimeoutExpired(cmd="git show", timeout=30)
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
@given("gcov subprocess run is mocked to raise OSError for git show")
def step_gcov_mock_oserror_git_show(ctx):
real_run = subprocess.run
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "show" in cmd:
raise OSError("mocked OS error")
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
@given("gcov subprocess run is mocked to fail git show with nonzero rc")
def step_gcov_mock_git_show_fail(ctx):
real_run = subprocess.run
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "show" in cmd:
mock_result = MagicMock()
mock_result.returncode = 128
mock_result.stdout = b""
mock_result.stderr = b"fatal: not a git repo"
return mock_result
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
@given("gcov subprocess run is mocked so git ls-tree returns nonzero")
def step_gcov_mock_lstree_fail(ctx):
real_run = subprocess.run
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "ls-tree" in cmd:
mock_result = MagicMock()
mock_result.returncode = 128
mock_result.stdout = ""
mock_result.stderr = "fatal: not a tree"
return mock_result
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
@given("gcov subprocess run is mocked for diff with no changes")
def step_gcov_mock_diff_no_changes(ctx):
real_run = subprocess.run
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "diff" in cmd and "--no-index" in cmd:
mock_result = MagicMock()
mock_result.returncode = 0 # 0 = no diff
mock_result.stdout = (
b"" if isinstance(cmd, list) and "--shortstat" not in cmd else ""
)
mock_result.stderr = b""
# Handle both text and binary mode
if kwargs.get("text"):
mock_result.stdout = ""
mock_result.stderr = ""
return mock_result
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
@given("gcov subprocess run is mocked so git ls-tree -d returns nonzero")
def step_gcov_mock_lstree_d_fail(ctx):
real_run = subprocess.run
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "ls-tree" in cmd and "-d" in cmd:
mock_result = MagicMock()
mock_result.returncode = 128
mock_result.stdout = ""
mock_result.stderr = "fatal: bad tree"
return mock_result
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
@given("gcov subprocess run is mocked so git ls-tree -d returns blank lines")
def step_gcov_mock_lstree_d_blank_lines(ctx):
real_run = subprocess.run
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "ls-tree" in cmd and "-d" in cmd:
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = "\nrealdir\n\n \n"
mock_result.stderr = ""
return mock_result
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
@given("gcov a handler and a resource with location")
def step_gcov_handler_and_resource(ctx):
repo = _init_git_repo()
ctx.gcov_handler = GitCheckoutHandler()
ctx.gcov_resource = _make_resource(location=repo)
ctx.gcov_repo = repo
ctx.gcov_error = None
ctx.gcov_result = None
@given("gcov a sandbox manager that returns None for get_sandbox")
def step_gcov_sandbox_mgr_none(ctx):
mgr = MagicMock()
mgr.get_sandbox.return_value = None
ctx.gcov_sandbox_manager = mgr
@given("gcov a sandbox manager that returns a sandbox with context")
def step_gcov_sandbox_mgr_with_context(ctx):
sandbox_ctx = MagicMock()
sandbox_ctx.sandbox_path = ctx.gcov_repo
sandbox = MagicMock()
sandbox.context = sandbox_ctx
mgr = MagicMock()
mgr.get_sandbox.return_value = sandbox
ctx.gcov_sandbox_manager = mgr
@given("gcov subprocess run is mocked so git tag returns nonzero")
def step_gcov_mock_git_tag_fail(ctx):
real_run = subprocess.run
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "tag" in cmd:
mock_result = MagicMock()
mock_result.returncode = 1
mock_result.stdout = ""
mock_result.stderr = "fatal: tag already exists"
return mock_result
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
@given("gcov subprocess run is mocked so git reset returns nonzero")
def step_gcov_mock_git_reset_fail(ctx):
real_run = subprocess.run
def side_effect(cmd, *args, **kwargs):
if isinstance(cmd, list) and "reset" in cmd:
mock_result = MagicMock()
mock_result.returncode = 1
mock_result.stdout = ""
mock_result.stderr = "fatal: ambiguous argument"
return mock_result
return real_run(cmd, *args, **kwargs)
patcher = patch(_SUBPROCESS_RUN, side_effect=side_effect)
patcher.start()
ctx.add_cleanup(patcher.stop)
# ---------------------------------------------------------------------------
# WHEN steps
# ---------------------------------------------------------------------------
@when("gcov read is called with empty path")
def step_gcov_read_empty_path(ctx):
try:
ctx.gcov_result = ctx.gcov_handler.read(resource=ctx.gcov_resource, path="")
except Exception as exc:
ctx.gcov_error = exc
@when('gcov read is called with path "{path}"')
def step_gcov_read_path(ctx, path):
try:
ctx.gcov_result = ctx.gcov_handler.read(resource=ctx.gcov_resource, path=path)
except Exception as exc:
ctx.gcov_error = exc
@when('gcov read is called with path "{path}" expecting error')
def step_gcov_read_path_error(ctx, path):
try:
ctx.gcov_result = ctx.gcov_handler.read(resource=ctx.gcov_resource, path=path)
except Exception as exc:
ctx.gcov_error = exc
@when("gcov delete is called with empty path expecting error")
def step_gcov_delete_empty_path(ctx):
try:
ctx.gcov_result = ctx.gcov_handler.delete(resource=ctx.gcov_resource, path="")
except Exception as exc:
ctx.gcov_error = exc
@when('gcov delete is called with path "{path}" expecting error')
def step_gcov_delete_path_error(ctx, path):
try:
ctx.gcov_result = ctx.gcov_handler.delete(resource=ctx.gcov_resource, path=path)
except Exception as exc:
ctx.gcov_error = exc
@when('gcov delete is called with path "{path}"')
def step_gcov_delete_path(ctx, path):
try:
ctx.gcov_result = ctx.gcov_handler.delete(resource=ctx.gcov_resource, path=path)
except Exception as exc:
ctx.gcov_error = exc
@when("gcov list_children is called")
def step_gcov_list_children(ctx):
try:
ctx.gcov_result = ctx.gcov_handler.list_children(resource=ctx.gcov_resource)
except Exception as exc:
ctx.gcov_error = exc
@when("gcov diff is called with the same location as other")
def step_gcov_diff_same(ctx):
try:
ctx.gcov_result = ctx.gcov_handler.diff(
resource=ctx.gcov_resource,
other_location=ctx.gcov_resource.location,
)
except Exception as exc:
ctx.gcov_error = exc
@when("gcov discover_children is called")
def step_gcov_discover_children(ctx):
try:
ctx.gcov_result = ctx.gcov_handler.discover_children(resource=ctx.gcov_resource)
except Exception as exc:
ctx.gcov_error = exc
@when("gcov create_checkpoint is called expecting error")
def step_gcov_create_checkpoint_error(ctx):
try:
ctx.gcov_result = ctx.gcov_handler.create_checkpoint(
resource=ctx.gcov_resource,
plan_id="plan-gcov-001",
sandbox_manager=ctx.gcov_sandbox_manager,
phase="execution",
)
except Exception as exc:
ctx.gcov_error = exc
@when("gcov rollback_to is called expecting error")
def step_gcov_rollback_error(ctx):
try:
ctx.gcov_result = ctx.gcov_handler.rollback_to(
resource=ctx.gcov_resource,
plan_id="plan-gcov-001",
checkpoint_id="checkpoint-gcov-20260101T000000",
sandbox_manager=ctx.gcov_sandbox_manager,
)
except Exception as exc:
ctx.gcov_error = exc
@when("gcov rollback_to is called")
def step_gcov_rollback(ctx):
try:
ctx.gcov_result = ctx.gcov_handler.rollback_to(
resource=ctx.gcov_resource,
plan_id="plan-gcov-001",
checkpoint_id="checkpoint-gcov-20260101T000000",
sandbox_manager=ctx.gcov_sandbox_manager,
)
except Exception as exc:
ctx.gcov_error = exc
# ---------------------------------------------------------------------------
# THEN steps
# ---------------------------------------------------------------------------
@then("gcov the result should be a Content with newline-joined file listing")
def step_gcov_check_content_listing(ctx):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert ctx.gcov_result is not None, "Expected a Content result"
text = ctx.gcov_result.data.decode("utf-8")
assert ctx.gcov_result.encoding == "utf-8"
# The repo has README.md and src/app.py tracked; listing should have entries
assert len(text.strip()) > 0, "Expected non-empty file listing"
@then("gcov the result encoding should be None")
def step_gcov_encoding_none(ctx):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert ctx.gcov_result is not None, "Expected a Content result"
assert ctx.gcov_result.encoding is None, (
f"Expected encoding=None, got {ctx.gcov_result.encoding!r}"
)
@then('gcov the result data should match the file contents of "{filename}"')
def step_gcov_data_matches_file(ctx, filename):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert ctx.gcov_result is not None, "Expected a Content result"
assert ctx.gcov_result.data == ctx.gcov_file_content, (
f"Data mismatch: expected {ctx.gcov_file_content!r}, got {ctx.gcov_result.data!r}"
)
@then("gcov a FileNotFoundError should be stored")
def step_gcov_fnf_error(ctx):
assert ctx.gcov_error is not None, "Expected an error but none occurred"
assert isinstance(ctx.gcov_error, FileNotFoundError), (
f"Expected FileNotFoundError, got {type(ctx.gcov_error).__name__}: {ctx.gcov_error}"
)
@then('gcov the result encoding should be "{encoding}"')
def step_gcov_encoding_value(ctx, encoding):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert ctx.gcov_result is not None, "Expected a result"
assert ctx.gcov_result.encoding == encoding, (
f"Expected encoding={encoding!r}, got {ctx.gcov_result.encoding!r}"
)
@then("gcov the result should have a content_hash")
def step_gcov_has_hash(ctx):
assert ctx.gcov_result is not None, "Expected a result"
assert ctx.gcov_result.content_hash is not None, "Expected content_hash to be set"
assert len(ctx.gcov_result.content_hash) == 64, (
f"Expected SHA-256 hex digest (64 chars), got {len(ctx.gcov_result.content_hash)}"
)
@then("gcov a PermissionError should be stored")
def step_gcov_perm_error(ctx):
assert ctx.gcov_error is not None, "Expected an error but none occurred"
assert isinstance(ctx.gcov_error, PermissionError), (
f"Expected PermissionError, got {type(ctx.gcov_error).__name__}: {ctx.gcov_error}"
)
@then("gcov the delete result should indicate success")
def step_gcov_delete_success(ctx):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert ctx.gcov_result is not None, "Expected a DeleteResult"
assert ctx.gcov_result.success is True, "Expected success=True"
@then('gcov the directory "{dirname}" should no longer exist')
def step_gcov_dir_gone(ctx, dirname):
target = os.path.join(ctx.gcov_repo, dirname)
assert not os.path.exists(target), f"Directory {target} still exists"
@then("gcov the result should be a sorted list of filesystem entries")
def step_gcov_sorted_fs_list(ctx):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert isinstance(ctx.gcov_result, list), (
f"Expected list, got {type(ctx.gcov_result).__name__}"
)
# The fallback os.listdir should return at least README.md
assert len(ctx.gcov_result) > 0, "Expected non-empty listing"
assert ctx.gcov_result == sorted(ctx.gcov_result), "Expected sorted order"
@then("gcov the diff result has_changes should be False")
def step_gcov_diff_no_changes(ctx):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert ctx.gcov_result is not None, "Expected a DiffResult"
assert ctx.gcov_result.has_changes is False, (
f"Expected has_changes=False, got {ctx.gcov_result.has_changes}"
)
@then("gcov the diff result unified_diff should be empty")
def step_gcov_diff_empty(ctx):
assert ctx.gcov_result.unified_diff == "", (
f"Expected empty unified_diff, got {ctx.gcov_result.unified_diff!r}"
)
@then("gcov the result should be an empty list")
def step_gcov_empty_list(ctx):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert ctx.gcov_result == [], f"Expected empty list, got {ctx.gcov_result!r}"
@then("gcov the result should contain only non-empty directory resources")
def step_gcov_non_empty_dir_resources(ctx):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert isinstance(ctx.gcov_result, list), (
f"Expected list, got {type(ctx.gcov_result).__name__}"
)
# "realdir" should be present; empty / whitespace-only names should be skipped
names = [r.name for r in ctx.gcov_result]
assert "" not in names, "Empty dirname should have been skipped"
assert "realdir" in names, f"Expected 'realdir' in result names, got {names}"
@then('gcov a RuntimeError should be stored with message "{msg}"')
def step_gcov_runtime_error_msg(ctx, msg):
assert ctx.gcov_error is not None, "Expected an error but none occurred"
assert isinstance(ctx.gcov_error, RuntimeError), (
f"Expected RuntimeError, got {type(ctx.gcov_error).__name__}: {ctx.gcov_error}"
)
assert msg in str(ctx.gcov_error), (
f"Expected '{msg}' in error message, got: {ctx.gcov_error}"
)
@then("gcov the rollback result success should be False")
def step_gcov_rollback_fail(ctx):
assert ctx.gcov_error is None, f"Unexpected error: {ctx.gcov_error}"
assert ctx.gcov_result is not None, "Expected a RollbackResult"
assert ctx.gcov_result.success is False, (
f"Expected success=False, got {ctx.gcov_result.success}"
)
@then('gcov the rollback result message should contain "{msg}"')
def step_gcov_rollback_msg(ctx, msg):
assert msg in ctx.gcov_result.message, (
f"Expected '{msg}' in message, got: {ctx.gcov_result.message!r}"
)