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
526 lines
20 KiB
Python
526 lines
20 KiB
Python
"""Step definitions for fs_directory_coverage.feature.
|
|
|
|
Exercises uncovered lines in fs_directory.py:
|
|
L133 — delete() with empty path raises PermissionError
|
|
L143 — delete() removes subdirectory via shutil.rmtree
|
|
L197 — diff() with file missing from resource dir (a_lines = [])
|
|
L204 — diff() with file missing from other dir (b_lines = [])
|
|
L287-289 — create_checkpoint() no sandbox RuntimeError
|
|
L329-331 — rollback_to() no sandbox RuntimeError
|
|
L336-339 — rollback_to() missing/expired checkpoint
|
|
L347,349 — rollback_to() .git skip + rmtree in sandbox clear
|
|
L357,360 — rollback_to() .git skip + copytree in snapshot restore
|
|
L374-381 — discard_checkpoints()
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.domain.models.core.resource import (
|
|
PhysVirt,
|
|
Resource,
|
|
ResourceCapabilities,
|
|
)
|
|
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
|
|
|
|
__all__: list[str] = []
|
|
|
|
_ULID_COUNTER = 900_000
|
|
|
|
|
|
def _next_ulid() -> str:
|
|
"""Generate a valid 26-char Crockford Base32 ID for tests."""
|
|
global _ULID_COUNTER
|
|
_ULID_COUNTER += 1
|
|
cb32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
|
|
n = _ULID_COUNTER
|
|
chars: list[str] = []
|
|
for _ in range(26):
|
|
chars.append(cb32[n % 32])
|
|
n //= 32
|
|
return "".join(reversed(chars))
|
|
|
|
|
|
def _make_resource(location: str) -> Resource:
|
|
return Resource(
|
|
resource_id=_next_ulid(),
|
|
resource_type_name="fs-directory",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=location,
|
|
capabilities=ResourceCapabilities(
|
|
readable=True, writable=True, sandboxable=True
|
|
),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Delete with empty path — L133
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('fsdcov a temp directory with a file "{fname}"')
|
|
def step_fsdcov_given_temp_dir_file(context: Context, fname: str) -> None:
|
|
d = tempfile.mkdtemp(prefix="fsdcov_del_")
|
|
context.add_cleanup(lambda: shutil.rmtree(d, ignore_errors=True))
|
|
Path(d, fname).write_text("content", encoding="utf-8")
|
|
context.fsdcov_handler = FsDirectoryHandler()
|
|
context.fsdcov_resource = _make_resource(d)
|
|
context.fsdcov_tmpdir = d
|
|
|
|
|
|
@when("fsdcov I delete with an empty path")
|
|
def step_fsdcov_delete_empty_path(context: Context) -> None:
|
|
handler: FsDirectoryHandler = context.fsdcov_handler
|
|
try:
|
|
handler.delete(resource=context.fsdcov_resource, path="")
|
|
context.fsdcov_error = None
|
|
except Exception as exc:
|
|
context.fsdcov_error = exc
|
|
|
|
|
|
@then('fsdcov a PermissionError is raised with message "{fragment}"')
|
|
def step_fsdcov_permission_error(context: Context, fragment: str) -> None:
|
|
assert context.fsdcov_error is not None, "Expected an error but none was raised"
|
|
assert isinstance(context.fsdcov_error, PermissionError), (
|
|
f"Expected PermissionError, got {type(context.fsdcov_error).__name__}"
|
|
)
|
|
assert fragment in str(context.fsdcov_error), (
|
|
f"Expected '{fragment}' in '{context.fsdcov_error}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Delete subdirectory — L143
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('fsdcov a temp directory with a subdirectory "{dirname}" containing "{fname}"')
|
|
def step_fsdcov_given_temp_dir_subdir(
|
|
context: Context, dirname: str, fname: str
|
|
) -> None:
|
|
d = tempfile.mkdtemp(prefix="fsdcov_delsub_")
|
|
context.add_cleanup(lambda: shutil.rmtree(d, ignore_errors=True))
|
|
sub = Path(d, dirname)
|
|
sub.mkdir(parents=True, exist_ok=True)
|
|
(sub / fname).write_text("inner content", encoding="utf-8")
|
|
context.fsdcov_handler = FsDirectoryHandler()
|
|
context.fsdcov_resource = _make_resource(d)
|
|
context.fsdcov_tmpdir = d
|
|
|
|
|
|
@when('fsdcov I delete subdirectory "{dirname}"')
|
|
def step_fsdcov_delete_subdir(context: Context, dirname: str) -> None:
|
|
handler: FsDirectoryHandler = context.fsdcov_handler
|
|
try:
|
|
context.fsdcov_delete_result = handler.delete(
|
|
resource=context.fsdcov_resource, path=dirname
|
|
)
|
|
context.fsdcov_error = None
|
|
except Exception as exc:
|
|
context.fsdcov_error = exc
|
|
context.fsdcov_delete_result = None
|
|
|
|
|
|
@then("fsdcov the delete result is successful")
|
|
def step_fsdcov_delete_success(context: Context) -> None:
|
|
assert context.fsdcov_error is None, f"Unexpected error: {context.fsdcov_error}"
|
|
assert context.fsdcov_delete_result is not None
|
|
assert context.fsdcov_delete_result.success is True
|
|
|
|
|
|
@then('fsdcov the subdirectory "{dirname}" no longer exists')
|
|
def step_fsdcov_subdir_gone(context: Context, dirname: str) -> None:
|
|
target = Path(context.fsdcov_tmpdir) / dirname
|
|
assert not target.exists(), (
|
|
f"Expected '{dirname}' to be removed, but it still exists"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Diff with files on only one side — L197, L204
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('fsdcov a temp directory "{label}" with no files')
|
|
def step_fsdcov_given_empty_dir(context: Context, label: str) -> None:
|
|
d = tempfile.mkdtemp(prefix=f"fsdcov_{label}_")
|
|
context.add_cleanup(lambda: shutil.rmtree(d, ignore_errors=True))
|
|
if not hasattr(context, "fsdcov_diff_dirs"):
|
|
context.fsdcov_diff_dirs = {}
|
|
context.fsdcov_diff_dirs[label] = d
|
|
|
|
|
|
@given('fsdcov a temp directory "{label}" with file "{fname}" containing "{content}"')
|
|
def step_fsdcov_given_dir_with_file(
|
|
context: Context, label: str, fname: str, content: str
|
|
) -> None:
|
|
d = tempfile.mkdtemp(prefix=f"fsdcov_{label}_")
|
|
context.add_cleanup(lambda: shutil.rmtree(d, ignore_errors=True))
|
|
Path(d, fname).write_text(content, encoding="utf-8")
|
|
if not hasattr(context, "fsdcov_diff_dirs"):
|
|
context.fsdcov_diff_dirs = {}
|
|
context.fsdcov_diff_dirs[label] = d
|
|
|
|
|
|
@when("fsdcov I diff dir_a against dir_b")
|
|
def step_fsdcov_diff(context: Context) -> None:
|
|
handler = FsDirectoryHandler()
|
|
dir_a = context.fsdcov_diff_dirs["dir_a"]
|
|
dir_b = context.fsdcov_diff_dirs["dir_b"]
|
|
resource = _make_resource(dir_a)
|
|
context.fsdcov_diff_result = handler.diff(resource=resource, other_location=dir_b)
|
|
|
|
|
|
@then("fsdcov the diff shows changes with insertions")
|
|
def step_fsdcov_diff_insertions(context: Context) -> None:
|
|
r = context.fsdcov_diff_result
|
|
assert r.has_changes is True, "Expected diff to have changes"
|
|
assert r.insertions > 0, f"Expected insertions > 0, got {r.insertions}"
|
|
|
|
|
|
@then("fsdcov the diff shows changes with deletions")
|
|
def step_fsdcov_diff_deletions(context: Context) -> None:
|
|
r = context.fsdcov_diff_result
|
|
assert r.has_changes is True, "Expected diff to have changes"
|
|
assert r.deletions > 0, f"Expected deletions > 0, got {r.deletions}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_checkpoint with no sandbox — L287-289
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("fsdcov a handler and a resource with location")
|
|
def step_fsdcov_handler_and_resource(context: Context) -> None:
|
|
d = tempfile.mkdtemp(prefix="fsdcov_ckpt_")
|
|
context.add_cleanup(lambda: shutil.rmtree(d, ignore_errors=True))
|
|
context.fsdcov_handler = FsDirectoryHandler()
|
|
context.fsdcov_resource = _make_resource(d)
|
|
context.fsdcov_tmpdir = d
|
|
|
|
|
|
@given("fsdcov a sandbox manager that returns None for get_sandbox")
|
|
def step_fsdcov_sandbox_mgr_none(context: Context) -> None:
|
|
mgr = MagicMock()
|
|
mgr.get_sandbox.return_value = None
|
|
context.fsdcov_sandbox_manager = mgr
|
|
|
|
|
|
@given("fsdcov a sandbox manager that returns sandbox with no context")
|
|
def step_fsdcov_sandbox_mgr_no_context(context: Context) -> None:
|
|
mgr = MagicMock()
|
|
sandbox_mock = MagicMock()
|
|
sandbox_mock.context = None
|
|
mgr.get_sandbox.return_value = sandbox_mock
|
|
context.fsdcov_sandbox_manager = mgr
|
|
|
|
|
|
@when("fsdcov I call create_checkpoint")
|
|
def step_fsdcov_call_create_checkpoint(context: Context) -> None:
|
|
handler: FsDirectoryHandler = context.fsdcov_handler
|
|
try:
|
|
handler.create_checkpoint(
|
|
resource=context.fsdcov_resource,
|
|
plan_id="test-plan-1",
|
|
sandbox_manager=context.fsdcov_sandbox_manager,
|
|
)
|
|
context.fsdcov_error = None
|
|
except Exception as exc:
|
|
context.fsdcov_error = exc
|
|
|
|
|
|
@then('fsdcov a RuntimeError is raised with message "{fragment}"')
|
|
def step_fsdcov_runtime_error(context: Context, fragment: str) -> None:
|
|
assert context.fsdcov_error is not None, "Expected an error but none was raised"
|
|
assert isinstance(context.fsdcov_error, RuntimeError), (
|
|
f"Expected RuntimeError, got {type(context.fsdcov_error).__name__}"
|
|
)
|
|
assert fragment in str(context.fsdcov_error), (
|
|
f"Expected '{fragment}' in '{context.fsdcov_error}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# rollback_to with no sandbox — L329-331
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('fsdcov I call rollback_to with checkpoint "{ckpt_id}"')
|
|
def step_fsdcov_rollback_to(context: Context, ckpt_id: str) -> None:
|
|
handler: FsDirectoryHandler = context.fsdcov_handler
|
|
try:
|
|
context.fsdcov_rollback_result = handler.rollback_to(
|
|
resource=context.fsdcov_resource,
|
|
plan_id="test-plan-1",
|
|
checkpoint_id=ckpt_id,
|
|
sandbox_manager=context.fsdcov_sandbox_manager,
|
|
)
|
|
context.fsdcov_error = None
|
|
except Exception as exc:
|
|
context.fsdcov_error = exc
|
|
context.fsdcov_rollback_result = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# rollback_to with missing checkpoint — L336-339
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("fsdcov a sandbox manager with a valid sandbox")
|
|
def step_fsdcov_sandbox_mgr_valid(context: Context) -> None:
|
|
sandbox_path = tempfile.mkdtemp(prefix="fsdcov_sb_")
|
|
context.add_cleanup(lambda: shutil.rmtree(sandbox_path, ignore_errors=True))
|
|
|
|
mgr = MagicMock()
|
|
sandbox_mock = MagicMock()
|
|
ctx_mock = MagicMock()
|
|
ctx_mock.sandbox_path = sandbox_path
|
|
sandbox_mock.context = ctx_mock
|
|
mgr.get_sandbox.return_value = sandbox_mock
|
|
|
|
context.fsdcov_sandbox_manager = mgr
|
|
context.fsdcov_sandbox_path = sandbox_path
|
|
|
|
|
|
@then('fsdcov the rollback result is unsuccessful with message "{fragment}"')
|
|
def step_fsdcov_rollback_unsuccessful(context: Context, fragment: str) -> None:
|
|
assert context.fsdcov_error is None, f"Unexpected exception: {context.fsdcov_error}"
|
|
r = context.fsdcov_rollback_result
|
|
assert r is not None, "Expected a rollback result"
|
|
assert r.success is False, f"Expected success=False, got {r.success}"
|
|
assert fragment in r.message, f"Expected '{fragment}' in '{r.message}'"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# rollback_to with expired checkpoint path — L336-339
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("fsdcov a handler with a checkpoint pointing to a deleted path")
|
|
def step_fsdcov_handler_expired_ckpt(context: Context) -> None:
|
|
d = tempfile.mkdtemp(prefix="fsdcov_expired_")
|
|
context.add_cleanup(lambda: shutil.rmtree(d, ignore_errors=True))
|
|
context.fsdcov_handler = FsDirectoryHandler()
|
|
context.fsdcov_resource = _make_resource(d)
|
|
|
|
# Create a temp dir, record it as a checkpoint, then delete it
|
|
expired_dir = tempfile.mkdtemp(prefix="fsdcov_snap_expired_")
|
|
context.fsdcov_expired_ckpt_id = "expired-ckpt-xyz"
|
|
context.fsdcov_handler._checkpoints[context.fsdcov_expired_ckpt_id] = expired_dir
|
|
shutil.rmtree(expired_dir)
|
|
|
|
|
|
@when("fsdcov I call rollback_to with the expired checkpoint")
|
|
def step_fsdcov_rollback_expired(context: Context) -> None:
|
|
handler: FsDirectoryHandler = context.fsdcov_handler
|
|
try:
|
|
context.fsdcov_rollback_result = handler.rollback_to(
|
|
resource=context.fsdcov_resource,
|
|
plan_id="test-plan-1",
|
|
checkpoint_id=context.fsdcov_expired_ckpt_id,
|
|
sandbox_manager=context.fsdcov_sandbox_manager,
|
|
)
|
|
context.fsdcov_error = None
|
|
except Exception as exc:
|
|
context.fsdcov_error = exc
|
|
context.fsdcov_rollback_result = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# rollback_to with .git skip and directory handling — L347,349,357,360
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("fsdcov a handler with a real checkpoint snapshot")
|
|
def step_fsdcov_handler_real_checkpoint(context: Context) -> None:
|
|
# Create snapshot directory with: .git/, a subdir, and a file
|
|
snap_parent = tempfile.mkdtemp(prefix="fsdcov_snap_")
|
|
context.add_cleanup(lambda: shutil.rmtree(snap_parent, ignore_errors=True))
|
|
snap_path = Path(snap_parent) / "snapshot"
|
|
snap_path.mkdir()
|
|
|
|
# .git directory in snapshot (should be skipped during restore)
|
|
(snap_path / ".git").mkdir()
|
|
(snap_path / ".git" / "HEAD").write_text("ref: refs/heads/main\n")
|
|
|
|
# A subdirectory to restore (exercises copytree — L360)
|
|
(snap_path / "src").mkdir()
|
|
(snap_path / "src" / "main.py").write_text("print('hello')\n")
|
|
|
|
# A regular file to restore (exercises copy2)
|
|
(snap_path / "README.md").write_text("# Project\n")
|
|
|
|
context.fsdcov_handler = FsDirectoryHandler()
|
|
|
|
# Register the snapshot as a checkpoint on the handler
|
|
context.fsdcov_real_ckpt_id = "real-ckpt-001"
|
|
context.fsdcov_handler._checkpoints[context.fsdcov_real_ckpt_id] = str(snap_path)
|
|
|
|
# Create a resource location
|
|
res_dir = tempfile.mkdtemp(prefix="fsdcov_resdir_")
|
|
context.add_cleanup(lambda: shutil.rmtree(res_dir, ignore_errors=True))
|
|
context.fsdcov_resource = _make_resource(res_dir)
|
|
|
|
|
|
@given("fsdcov a sandbox with .git dir and other content")
|
|
def step_fsdcov_sandbox_with_git(context: Context) -> None:
|
|
sandbox_path = tempfile.mkdtemp(prefix="fsdcov_sb_git_")
|
|
context.add_cleanup(lambda: shutil.rmtree(sandbox_path, ignore_errors=True))
|
|
|
|
sb = Path(sandbox_path)
|
|
# .git directory — must be preserved (L347)
|
|
(sb / ".git").mkdir()
|
|
(sb / ".git" / "config").write_text("[core]\n")
|
|
|
|
# A subdirectory — must be removed by rmtree (L349)
|
|
(sb / "old_dir").mkdir()
|
|
(sb / "old_dir" / "old.py").write_text("old\n")
|
|
|
|
# A regular file — must be removed by unlink (L351)
|
|
(sb / "stale.txt").write_text("stale\n")
|
|
|
|
mgr = MagicMock()
|
|
sandbox_mock = MagicMock()
|
|
ctx_mock = MagicMock()
|
|
ctx_mock.sandbox_path = sandbox_path
|
|
sandbox_mock.context = ctx_mock
|
|
mgr.get_sandbox.return_value = sandbox_mock
|
|
|
|
context.fsdcov_sandbox_manager = mgr
|
|
context.fsdcov_sandbox_path = sandbox_path
|
|
|
|
|
|
@when("fsdcov I call rollback_to with the real checkpoint")
|
|
def step_fsdcov_rollback_real(context: Context) -> None:
|
|
handler: FsDirectoryHandler = context.fsdcov_handler
|
|
try:
|
|
context.fsdcov_rollback_result = handler.rollback_to(
|
|
resource=context.fsdcov_resource,
|
|
plan_id="test-plan-1",
|
|
checkpoint_id=context.fsdcov_real_ckpt_id,
|
|
sandbox_manager=context.fsdcov_sandbox_manager,
|
|
)
|
|
context.fsdcov_error = None
|
|
except Exception as exc:
|
|
context.fsdcov_error = exc
|
|
context.fsdcov_rollback_result = None
|
|
|
|
|
|
@then("fsdcov the rollback result is successful")
|
|
def step_fsdcov_rollback_success(context: Context) -> None:
|
|
assert context.fsdcov_error is None, f"Unexpected error: {context.fsdcov_error}"
|
|
r = context.fsdcov_rollback_result
|
|
assert r is not None, "Expected a rollback result"
|
|
assert r.success is True, f"Expected success=True, got {r.success}"
|
|
|
|
|
|
@then("fsdcov the .git directory is preserved in the sandbox")
|
|
def step_fsdcov_git_preserved(context: Context) -> None:
|
|
sb = Path(context.fsdcov_sandbox_path)
|
|
assert (sb / ".git").exists(), ".git directory was removed but should be preserved"
|
|
assert (sb / ".git" / "config").exists(), (
|
|
".git/config was removed but should be preserved"
|
|
)
|
|
|
|
|
|
@then("fsdcov the snapshot content is restored in the sandbox")
|
|
def step_fsdcov_snapshot_restored(context: Context) -> None:
|
|
sb = Path(context.fsdcov_sandbox_path)
|
|
# Subdirectory from snapshot should be restored (L360)
|
|
assert (sb / "src").is_dir(), "Snapshot 'src/' directory was not restored"
|
|
assert (sb / "src" / "main.py").exists(), "Snapshot 'src/main.py' was not restored"
|
|
assert (sb / "src" / "main.py").read_text() == "print('hello')\n"
|
|
# File from snapshot should be restored
|
|
assert (sb / "README.md").exists(), "Snapshot 'README.md' was not restored"
|
|
assert (sb / "README.md").read_text() == "# Project\n"
|
|
# Old content should be gone
|
|
assert not (sb / "old_dir").exists(), "Old sandbox content was not cleared"
|
|
assert not (sb / "stale.txt").exists(), "Old sandbox file was not cleared"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# discard_checkpoints — L374-381
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('fsdcov a handler with two checkpoints for plan "{plan_id}"')
|
|
def step_fsdcov_two_checkpoints(context: Context, plan_id: str) -> None:
|
|
context.fsdcov_handler = FsDirectoryHandler()
|
|
context.fsdcov_plan_id = plan_id
|
|
|
|
# Create two temp snapshot dirs and register them
|
|
for i in range(2):
|
|
snap_dir = tempfile.mkdtemp(prefix=f"fsdcov_disc_{i}_")
|
|
snap_path = Path(snap_dir) / "snapshot"
|
|
snap_path.mkdir()
|
|
(snap_path / f"file{i}.txt").write_text(f"data{i}")
|
|
ckpt_id = f"checkpoint-{plan_id}-2024010{i}"
|
|
context.fsdcov_handler._checkpoints[ckpt_id] = str(snap_path)
|
|
|
|
# Track parent dirs for cleanup verification
|
|
context.fsdcov_discard_snap_parents = [
|
|
str(Path(v).parent) for v in context.fsdcov_handler._checkpoints.values()
|
|
]
|
|
for p in context.fsdcov_discard_snap_parents:
|
|
context.add_cleanup(lambda pp=p: shutil.rmtree(pp, ignore_errors=True))
|
|
|
|
|
|
@when('fsdcov I call discard_checkpoints with plan_id "{plan_id}"')
|
|
def step_fsdcov_discard_checkpoints(context: Context, plan_id: str) -> None:
|
|
context.fsdcov_discard_count = context.fsdcov_handler.discard_checkpoints(
|
|
plan_id=plan_id
|
|
)
|
|
|
|
|
|
@then("fsdcov discard_checkpoints returns {count:d}")
|
|
def step_fsdcov_discard_returns(context: Context, count: int) -> None:
|
|
assert context.fsdcov_discard_count == count, (
|
|
f"Expected {count}, got {context.fsdcov_discard_count}"
|
|
)
|
|
|
|
|
|
@then('fsdcov no checkpoints remain for plan "{plan_id}"')
|
|
def step_fsdcov_no_checkpoints_for_plan(context: Context, plan_id: str) -> None:
|
|
remaining = [cid for cid in context.fsdcov_handler._checkpoints if plan_id in cid]
|
|
assert remaining == [], (
|
|
f"Expected no checkpoints for '{plan_id}', found {remaining}"
|
|
)
|
|
|
|
|
|
@given("fsdcov a handler with checkpoints for different plans")
|
|
def step_fsdcov_multi_plan_checkpoints(context: Context) -> None:
|
|
context.fsdcov_handler = FsDirectoryHandler()
|
|
|
|
for plan_name in ("plan-x", "plan-y"):
|
|
snap_dir = tempfile.mkdtemp(prefix=f"fsdcov_mp_{plan_name}_")
|
|
snap_path = Path(snap_dir) / "snapshot"
|
|
snap_path.mkdir()
|
|
(snap_path / "f.txt").write_text("data")
|
|
ckpt_id = f"checkpoint-{plan_name}-20240101"
|
|
context.fsdcov_handler._checkpoints[ckpt_id] = str(snap_path)
|
|
context.add_cleanup(lambda d=snap_dir: shutil.rmtree(d, ignore_errors=True))
|
|
|
|
|
|
@when("fsdcov I call discard_checkpoints with empty plan_id")
|
|
def step_fsdcov_discard_all(context: Context) -> None:
|
|
context.fsdcov_discard_count = context.fsdcov_handler.discard_checkpoints(
|
|
plan_id=""
|
|
)
|
|
|
|
|
|
@then("fsdcov all checkpoints are removed")
|
|
def step_fsdcov_all_removed(context: Context) -> None:
|
|
assert len(context.fsdcov_handler._checkpoints) == 0, (
|
|
f"Expected 0 checkpoints, found {len(context.fsdcov_handler._checkpoints)}"
|
|
)
|
|
assert context.fsdcov_discard_count >= 2, (
|
|
f"Expected at least 2 removed, got {context.fsdcov_discard_count}"
|
|
)
|