forked from HAL9000/cleveragents-core
428 lines
17 KiB
Python
428 lines
17 KiB
Python
"""Step definitions for CleanupService uncovered-lines coverage tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.application.services.cleanup_service import (
|
|
CleanupReport,
|
|
CleanupService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_settings(**overrides: object) -> Settings:
|
|
"""Create a real Settings instance with optional overrides.
|
|
|
|
We build a fresh ``Settings`` (which has sensible defaults for
|
|
every cleanup-related field), then patch individual attributes
|
|
for the specific test at hand.
|
|
"""
|
|
s = Settings()
|
|
for key, value in overrides.items():
|
|
object.__setattr__(s, key, value)
|
|
return s
|
|
|
|
|
|
# ── Given steps ──────────────────────────────────────────────────
|
|
|
|
|
|
@given("cleanup coverage has a CleanupService with default settings")
|
|
def step_cleanup_cov_service_default(context: Context) -> None:
|
|
context.cleanup_settings = _make_settings()
|
|
context.cleanup_service = CleanupService(context.cleanup_settings)
|
|
|
|
|
|
@given("cleanup coverage has a CleanupService with max {n:d} checkpoints")
|
|
def step_cleanup_cov_service_max_checkpoints(context: Context, n: int) -> None:
|
|
context.cleanup_settings = _make_settings(cleanup_checkpoint_max_per_plan=n)
|
|
context.cleanup_service = CleanupService(context.cleanup_settings)
|
|
|
|
|
|
@given("cleanup coverage has a CleanupService with max 1 checkpoint override")
|
|
def step_cleanup_cov_service_max_1_checkpoint(context: Context) -> None:
|
|
"""Create service with max_per_plan=1 (bypassing pydantic ge=2 validation)
|
|
so that 2 files > max_count but len(files) < 3, hitting L286-287."""
|
|
settings = _make_settings()
|
|
object.__setattr__(settings, "cleanup_checkpoint_max_per_plan", 1)
|
|
context.cleanup_settings = settings
|
|
context.cleanup_service = CleanupService(context.cleanup_settings)
|
|
|
|
|
|
@given('cleanup coverage has a temp directory with a subdirectory named "{name}"')
|
|
def step_cleanup_cov_temp_dir_with_subdir(context: Context, name: str) -> None:
|
|
context.cleanup_temp_dir = Path(tempfile.mkdtemp())
|
|
subdir = context.cleanup_temp_dir / name
|
|
subdir.mkdir()
|
|
# Make the subdirectory old so it would be "expired" by age
|
|
old_time = time.time() - 400 * 86400
|
|
os.utime(str(subdir), (old_time, old_time))
|
|
|
|
|
|
@given("cleanup coverage has a CleanupService with log dir containing expired files")
|
|
def step_cleanup_cov_service_with_expired_logs(context: Context) -> None:
|
|
tmp = Path(tempfile.mkdtemp())
|
|
context.cleanup_temp_dir = tmp
|
|
log_dir = tmp / "logs"
|
|
log_dir.mkdir()
|
|
# Create an expired log file
|
|
expired_file = log_dir / "old.log"
|
|
expired_file.write_text("old log data")
|
|
old_time = time.time() - 400 * 86400
|
|
os.utime(str(expired_file), (old_time, old_time))
|
|
context.cleanup_settings = _make_settings(
|
|
log_dir=log_dir,
|
|
cleanup_log_retention_days=1,
|
|
)
|
|
context.cleanup_service = CleanupService(context.cleanup_settings)
|
|
|
|
|
|
@given("cleanup coverage has a CleanupService with backup dir containing expired files")
|
|
def step_cleanup_cov_service_with_expired_backups(context: Context) -> None:
|
|
tmp = Path(tempfile.mkdtemp())
|
|
context.cleanup_temp_dir = tmp
|
|
data_dir = tmp / "data"
|
|
data_dir.mkdir()
|
|
backup_dir = data_dir / "backups"
|
|
backup_dir.mkdir()
|
|
# Create an expired backup file
|
|
expired_file = backup_dir / "old_backup.tar.gz"
|
|
expired_file.write_text("old backup data")
|
|
old_time = time.time() - 400 * 86400
|
|
os.utime(str(expired_file), (old_time, old_time))
|
|
context.cleanup_settings = _make_settings(
|
|
data_dir=data_dir,
|
|
cleanup_backup_retention_days=1,
|
|
)
|
|
context.cleanup_service = CleanupService(context.cleanup_settings)
|
|
|
|
|
|
# ── When steps ───────────────────────────────────────────────────
|
|
|
|
|
|
@when("cleanup coverage calls _get_sandbox_dirs with tmp not existing")
|
|
def step_cleanup_cov_get_sandbox_dirs_no_tmp(context: Context) -> None:
|
|
svc = context.cleanup_service
|
|
# Clear the cache so the method actually runs its logic
|
|
svc._sandbox_dirs_cache = None
|
|
with patch(
|
|
"cleveragents.application.services.cleanup_service.tempfile"
|
|
) as mock_tmp:
|
|
mock_tmp.gettempdir.return_value = "/nonexistent_tmp_path_xyz"
|
|
with patch(
|
|
"cleveragents.application.services.cleanup_service.Path"
|
|
) as mock_path_cls:
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = False
|
|
mock_path_cls.return_value = mock_path_instance
|
|
context.cleanup_result = svc._get_sandbox_dirs()
|
|
|
|
|
|
@when("cleanup coverage calls _get_sandbox_dirs with iterdir raising OSError")
|
|
def step_cleanup_cov_get_sandbox_dirs_iterdir_oserror(context: Context) -> None:
|
|
svc = context.cleanup_service
|
|
svc._sandbox_dirs_cache = None
|
|
with patch(
|
|
"cleveragents.application.services.cleanup_service.tempfile"
|
|
) as mock_tmp:
|
|
mock_tmp.gettempdir.return_value = "/some_tmp"
|
|
with patch(
|
|
"cleveragents.application.services.cleanup_service.Path"
|
|
) as mock_path_cls:
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.iterdir.side_effect = OSError("permission denied")
|
|
mock_path_cls.return_value = mock_path_instance
|
|
context.cleanup_result = svc._get_sandbox_dirs()
|
|
|
|
|
|
@when(
|
|
"cleanup coverage calls _get_sandbox_dirs with is_dir raising OSError on one entry"
|
|
)
|
|
def step_cleanup_cov_get_sandbox_dirs_isdir_oserror(context: Context) -> None:
|
|
svc = context.cleanup_service
|
|
svc._sandbox_dirs_cache = None
|
|
|
|
# Create two mock path entries: one that raises OSError on is_dir,
|
|
# one that works fine and matches the sandbox prefix
|
|
bad_entry = MagicMock()
|
|
bad_entry.is_dir.side_effect = OSError("broken")
|
|
bad_entry.name = "ca-sandbox-plan1-abc"
|
|
|
|
good_entry = MagicMock()
|
|
good_entry.is_dir.return_value = True
|
|
good_entry.name = "ca-sandbox-plan2-def"
|
|
|
|
with patch(
|
|
"cleveragents.application.services.cleanup_service.tempfile"
|
|
) as mock_tmp:
|
|
mock_tmp.gettempdir.return_value = "/some_tmp"
|
|
with patch(
|
|
"cleveragents.application.services.cleanup_service.Path"
|
|
) as mock_path_cls:
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.iterdir.return_value = [bad_entry, good_entry]
|
|
mock_path_cls.return_value = mock_path_instance
|
|
context.cleanup_result = svc._get_sandbox_dirs()
|
|
context.cleanup_good_entry = good_entry
|
|
|
|
|
|
@when('cleanup coverage extracts plan id from path "{dirname}"')
|
|
def step_cleanup_cov_extract_plan_id(context: Context, dirname: str) -> None:
|
|
context.cleanup_plan_id = CleanupService.extract_plan_id_from_sandbox(Path(dirname))
|
|
|
|
|
|
@when("cleanup coverage checks staleness of a path that raises OSError on stat")
|
|
def step_cleanup_cov_is_stale_oserror(context: Context) -> None:
|
|
mock_path = MagicMock()
|
|
mock_path.stat.side_effect = OSError("no such file")
|
|
context.cleanup_stale_result = context.cleanup_service._is_sandbox_stale(mock_path)
|
|
|
|
|
|
@when("cleanup coverage purges sandboxes and rmtree raises OSError")
|
|
def step_cleanup_cov_purge_sandboxes_rmtree_oserror(context: Context) -> None:
|
|
svc = context.cleanup_service
|
|
report = CleanupReport(dry_run=False)
|
|
|
|
# Create a fake stale sandbox dir
|
|
stale_dir = MagicMock()
|
|
stale_dir.name = "ca-sandbox-testplan-abc123"
|
|
stale_dir.stat.return_value = MagicMock(
|
|
st_mtime=time.time() - 999999,
|
|
)
|
|
|
|
svc._sandbox_dirs_cache = [stale_dir]
|
|
|
|
with patch(
|
|
"cleveragents.application.services.cleanup_service.shutil"
|
|
) as mock_shutil:
|
|
mock_shutil.rmtree.side_effect = OSError("permission denied")
|
|
svc._purge_sandboxes(report)
|
|
|
|
context.cleanup_report = report
|
|
|
|
|
|
@when("cleanup coverage scans checkpoints for a non-existent directory")
|
|
def step_cleanup_cov_scan_checkpoints_nodir(context: Context) -> None:
|
|
fake_dir = Path(tempfile.mkdtemp()) / "nonexistent"
|
|
context.cleanup_result = context.cleanup_service.scan_checkpoints_for_plan(fake_dir)
|
|
|
|
|
|
@when("cleanup coverage scans checkpoints for a directory with {n:d} files")
|
|
def step_cleanup_cov_scan_checkpoints_few_files(context: Context, n: int) -> None:
|
|
tmp = Path(tempfile.mkdtemp())
|
|
context.cleanup_temp_dir = tmp
|
|
for i in range(n):
|
|
(tmp / f"checkpoint_{i:04d}.json").write_text("{}")
|
|
context.cleanup_result = context.cleanup_service.scan_checkpoints_for_plan(tmp)
|
|
|
|
|
|
@when("cleanup coverage prunes checkpoints and unlink raises OSError")
|
|
def step_cleanup_cov_prune_checkpoints_unlink_oserror(context: Context) -> None:
|
|
tmp = Path(tempfile.mkdtemp())
|
|
context.cleanup_temp_dir = tmp
|
|
# Create 5 files to exceed max_per_plan=2 → middle 3 are excess,
|
|
# but keep first and last → prune from middle
|
|
for i in range(5):
|
|
(tmp / f"checkpoint_{i:04d}.json").write_text("{}")
|
|
|
|
# Patch Path.unlink on the excess files to raise OSError
|
|
def failing_unlink(self, *args, **kwargs):
|
|
raise OSError("permission denied")
|
|
|
|
with patch.object(Path, "unlink", failing_unlink):
|
|
context.cleanup_prune_result = (
|
|
context.cleanup_service.prune_checkpoints_for_plan(tmp)
|
|
)
|
|
|
|
|
|
@when("cleanup coverage scans sessions with a None updated_at entry")
|
|
def step_cleanup_cov_scan_sessions_none(context: Context) -> None:
|
|
sessions = [{"id": "s1", "updated_at": None}]
|
|
context.cleanup_result = context.cleanup_service.scan_inactive_sessions(sessions)
|
|
|
|
|
|
@when("cleanup coverage scans sessions with a string updated_at that is old")
|
|
def step_cleanup_cov_scan_sessions_string(context: Context) -> None:
|
|
old_date = (datetime.now(tz=UTC) - timedelta(days=365)).isoformat()
|
|
sessions = [{"id": "s1", "updated_at": old_date}]
|
|
context.cleanup_result = context.cleanup_service.scan_inactive_sessions(sessions)
|
|
|
|
|
|
@when("cleanup coverage scans sessions with a naive datetime updated_at that is old")
|
|
def step_cleanup_cov_scan_sessions_naive_dt(context: Context) -> None:
|
|
# Create a naive datetime (no tzinfo) that is old
|
|
old_date = datetime.now() - timedelta(days=365)
|
|
assert old_date.tzinfo is None, "should be naive"
|
|
sessions = [{"id": "s1", "updated_at": old_date}]
|
|
context.cleanup_result = context.cleanup_service.scan_inactive_sessions(sessions)
|
|
|
|
|
|
@when("cleanup coverage scans expired files in that directory")
|
|
def step_cleanup_cov_scan_expired_files_subdir(context: Context) -> None:
|
|
context.cleanup_result = context.cleanup_service.scan_expired_files(
|
|
context.cleanup_temp_dir,
|
|
retention_days=1,
|
|
pattern="*.log",
|
|
)
|
|
|
|
|
|
@when("cleanup coverage scans expired files where stat raises OSError")
|
|
def step_cleanup_cov_scan_expired_files_stat_oserror(context: Context) -> None:
|
|
# We need is_file() to return True, but then the explicit stat()
|
|
# call at L413 to raise OSError. Use a mock Path for the file
|
|
# and patch directory.glob to yield it.
|
|
tmp = Path(tempfile.mkdtemp())
|
|
context.cleanup_temp_dir = tmp
|
|
|
|
mock_file = MagicMock(spec=Path)
|
|
mock_file.is_file.return_value = True
|
|
mock_file.stat.side_effect = OSError("stat failed")
|
|
|
|
with patch.object(Path, "glob", return_value=[mock_file]):
|
|
context.cleanup_result = context.cleanup_service.scan_expired_files(
|
|
tmp,
|
|
retention_days=0,
|
|
pattern="*.log",
|
|
)
|
|
|
|
|
|
@when("cleanup coverage purges logs and unlink raises OSError")
|
|
def step_cleanup_cov_purge_logs_oserror(context: Context) -> None:
|
|
svc = context.cleanup_service
|
|
report = CleanupReport(dry_run=False)
|
|
|
|
def failing_unlink(self, *args, **kwargs):
|
|
raise OSError("permission denied")
|
|
|
|
with patch.object(Path, "unlink", failing_unlink):
|
|
svc._purge_logs(report)
|
|
|
|
context.cleanup_report = report
|
|
|
|
|
|
@when("cleanup coverage purges backups and unlink raises OSError")
|
|
def step_cleanup_cov_purge_backups_oserror(context: Context) -> None:
|
|
svc = context.cleanup_service
|
|
report = CleanupReport(dry_run=False)
|
|
|
|
def failing_unlink(self, *args, **kwargs):
|
|
raise OSError("permission denied")
|
|
|
|
with patch.object(Path, "unlink", failing_unlink):
|
|
svc._purge_backups(report)
|
|
|
|
context.cleanup_report = report
|
|
|
|
|
|
@when("cleanup coverage calls _age_description on a path where stat raises OSError")
|
|
def step_cleanup_cov_age_description_oserror(context: Context) -> None:
|
|
mock_path = MagicMock()
|
|
mock_path.stat.side_effect = OSError("no such file")
|
|
context.cleanup_age_desc = CleanupService._age_description(mock_path)
|
|
|
|
|
|
# ── Then steps ───────────────────────────────────────────────────
|
|
|
|
|
|
@then("cleanup coverage _get_sandbox_dirs result should be empty")
|
|
def step_cleanup_cov_sandbox_dirs_empty(context: Context) -> None:
|
|
assert context.cleanup_result == [], (
|
|
f"Expected empty list, got {context.cleanup_result}"
|
|
)
|
|
|
|
|
|
@then("cleanup coverage _get_sandbox_dirs result should contain only the safe entry")
|
|
def step_cleanup_cov_sandbox_dirs_one_entry(context: Context) -> None:
|
|
assert len(context.cleanup_result) == 1, (
|
|
f"Expected 1 entry, got {len(context.cleanup_result)}"
|
|
)
|
|
assert context.cleanup_result[0] is context.cleanup_good_entry
|
|
|
|
|
|
@then("cleanup coverage extracted plan id should be None")
|
|
def step_cleanup_cov_plan_id_none(context: Context) -> None:
|
|
assert context.cleanup_plan_id is None, (
|
|
f"Expected None, got {context.cleanup_plan_id!r}"
|
|
)
|
|
|
|
|
|
@then("cleanup coverage staleness result should be False")
|
|
def step_cleanup_cov_stale_false(context: Context) -> None:
|
|
assert context.cleanup_stale_result is False
|
|
|
|
|
|
@then("cleanup coverage purge sandbox report should show skipped count of {n:d}")
|
|
def step_cleanup_cov_purge_sandbox_skipped(context: Context, n: int) -> None:
|
|
assert context.cleanup_report.sandboxes.skipped == n, (
|
|
f"Expected skipped={n}, got {context.cleanup_report.sandboxes.skipped}"
|
|
)
|
|
|
|
|
|
@then("cleanup coverage checkpoint scan result should be empty")
|
|
def step_cleanup_cov_checkpoint_scan_empty(context: Context) -> None:
|
|
assert context.cleanup_result == [], (
|
|
f"Expected empty list, got {context.cleanup_result}"
|
|
)
|
|
|
|
|
|
@then("cleanup coverage prune result should be {n:d}")
|
|
def step_cleanup_cov_prune_result(context: Context, n: int) -> None:
|
|
assert context.cleanup_prune_result == n, (
|
|
f"Expected {n}, got {context.cleanup_prune_result}"
|
|
)
|
|
|
|
|
|
@then("cleanup coverage inactive sessions result should be empty")
|
|
def step_cleanup_cov_sessions_empty(context: Context) -> None:
|
|
assert context.cleanup_result == [], (
|
|
f"Expected empty list, got {context.cleanup_result}"
|
|
)
|
|
|
|
|
|
@then("cleanup coverage inactive sessions result should contain that session")
|
|
def step_cleanup_cov_sessions_has_session(context: Context) -> None:
|
|
assert len(context.cleanup_result) == 1, (
|
|
f"Expected 1 inactive session, got {len(context.cleanup_result)}"
|
|
)
|
|
assert context.cleanup_result[0]["id"] == "s1"
|
|
|
|
|
|
@then("cleanup coverage expired files result should be empty")
|
|
def step_cleanup_cov_expired_empty(context: Context) -> None:
|
|
assert context.cleanup_result == [], (
|
|
f"Expected empty list, got {context.cleanup_result}"
|
|
)
|
|
|
|
|
|
@then("cleanup coverage purge logs report should show skipped count of {n:d}")
|
|
def step_cleanup_cov_purge_logs_skipped(context: Context, n: int) -> None:
|
|
assert context.cleanup_report.logs.skipped == n, (
|
|
f"Expected logs.skipped={n}, got {context.cleanup_report.logs.skipped}"
|
|
)
|
|
|
|
|
|
@then("cleanup coverage purge backups report should show skipped count of {n:d}")
|
|
def step_cleanup_cov_purge_backups_skipped(context: Context, n: int) -> None:
|
|
assert context.cleanup_report.backups.skipped == n, (
|
|
f"Expected backups.skipped={n}, got {context.cleanup_report.backups.skipped}"
|
|
)
|
|
|
|
|
|
@then('cleanup coverage age description should be "{expected}"')
|
|
def step_cleanup_cov_age_desc(context: Context, expected: str) -> None:
|
|
assert context.cleanup_age_desc == expected, (
|
|
f"Expected {expected!r}, got {context.cleanup_age_desc!r}"
|
|
)
|