khyari hamza
cea3bad758
- Use model_copy(update=...) in test helpers so Pydantic ge= validators are enforced during tests, not just env vars (NEW-1) - Promote _extract_plan_id_from_sandbox to public staticmethod and reuse in CLI active-plan detection to eliminate duplicate plan-ID parsing logic (NEW-2) - Cache _get_sandbox_dirs result for service instance lifetime and have CLI active-plan detection use the cached listing so /tmp is iterated only once per invocation (NEW-3)
294 lines
9.8 KiB
Python
294 lines
9.8 KiB
Python
"""Step definitions for CONC3 cleanup CLI commands and edge cases.
|
|
|
|
Exercises the Typer CLI commands (scan, purge, status) and
|
|
service-level edge cases for coverage.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
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 Settings instance with optional field overrides.
|
|
|
|
Uses ``model_copy(update=...)`` so that Pydantic validators
|
|
(including ``ge=`` bounds on retention fields) are enforced.
|
|
"""
|
|
Settings._instance = None
|
|
base = Settings()
|
|
if overrides:
|
|
return base.model_copy(update=overrides)
|
|
return base
|
|
|
|
|
|
def _create_sandbox_dir(
|
|
prefix: str = "ca-sandbox-testplan-",
|
|
age_hours: float = 72,
|
|
) -> Path:
|
|
"""Create a fake sandbox temp directory with a given age."""
|
|
d = Path(tempfile.mkdtemp(prefix=prefix))
|
|
old_time = time.time() - (age_hours * 3600)
|
|
os.utime(d, (old_time, old_time))
|
|
return d
|
|
|
|
|
|
# ── CLI command tests ─────────────────────────────────────────────
|
|
|
|
|
|
@given("a cleanup CLI runner")
|
|
def step_cli_runner(context: Context) -> None:
|
|
from typer.testing import CliRunner
|
|
|
|
context.cli_runner = CliRunner()
|
|
|
|
|
|
@given("a stale sandbox for CLI testing")
|
|
def step_stale_sandbox_cli(context: Context) -> None:
|
|
import shutil
|
|
|
|
context.cli_sandbox_dir = _create_sandbox_dir(age_hours=72)
|
|
|
|
if not hasattr(context, "_temp_paths"):
|
|
context._temp_paths = []
|
|
context._temp_paths.append(context.cli_sandbox_dir)
|
|
|
|
def _cleanup() -> None:
|
|
for p in getattr(context, "_temp_paths", []):
|
|
if p.exists():
|
|
if p.is_dir():
|
|
shutil.rmtree(p, ignore_errors=True)
|
|
else:
|
|
p.unlink(missing_ok=True)
|
|
|
|
if _cleanup not in context._cleanup_handlers:
|
|
context._cleanup_handlers.append(_cleanup)
|
|
|
|
|
|
@when("I invoke the cleanup scan command")
|
|
def step_invoke_scan(context: Context) -> None:
|
|
from cleveragents.cli.commands.cleanup import app as cleanup_app
|
|
|
|
context.cli_result = context.cli_runner.invoke(cleanup_app, ["scan"])
|
|
|
|
|
|
@when("I invoke the cleanup purge command with dry-run")
|
|
def step_invoke_purge_dry(context: Context) -> None:
|
|
from cleveragents.cli.commands.cleanup import app as cleanup_app
|
|
|
|
context.cli_result = context.cli_runner.invoke(cleanup_app, ["purge", "--dry-run"])
|
|
|
|
|
|
@when("I invoke the cleanup purge command with yes flag")
|
|
def step_invoke_purge_yes(context: Context) -> None:
|
|
from cleveragents.cli.commands.cleanup import app as cleanup_app
|
|
|
|
context.cli_result = context.cli_runner.invoke(cleanup_app, ["purge", "--yes"])
|
|
|
|
|
|
@when("I invoke the cleanup purge command with all and yes")
|
|
def step_invoke_purge_all_yes(context: Context) -> None:
|
|
from cleveragents.cli.commands.cleanup import app as cleanup_app
|
|
|
|
context.cli_result = context.cli_runner.invoke(
|
|
cleanup_app, ["purge", "--all", "--yes"]
|
|
)
|
|
|
|
|
|
@when("I invoke the cleanup status command")
|
|
def step_invoke_status(context: Context) -> None:
|
|
from cleveragents.cli.commands.cleanup import app as cleanup_app
|
|
|
|
context.cli_result = context.cli_runner.invoke(cleanup_app, ["status"])
|
|
|
|
|
|
@when("I invoke the cleanup purge command without confirmation")
|
|
def step_invoke_purge_no_confirm(context: Context) -> None:
|
|
from cleveragents.cli.commands.cleanup import app as cleanup_app
|
|
|
|
context.cli_result = context.cli_runner.invoke(cleanup_app, ["purge"], input="n\n")
|
|
|
|
|
|
@then("the CLI exit code should be {code:d}")
|
|
def step_cli_exit_code(context: Context, code: int) -> None:
|
|
assert context.cli_result.exit_code == code, (
|
|
f"Expected exit code {code}, got {context.cli_result.exit_code}. "
|
|
f"Output: {context.cli_result.output}"
|
|
)
|
|
|
|
|
|
@then('the cleanup CLI output should contain "{text}"')
|
|
def step_cleanup_cli_output_contains(context: Context, text: str) -> None:
|
|
assert text in context.cli_result.output, (
|
|
f"Expected '{text}' in output, got: {context.cli_result.output}"
|
|
)
|
|
|
|
|
|
# ── CleanupReport and CleanupService edge case tests ─────────────
|
|
|
|
|
|
@when("I create a cleanup report with known values")
|
|
def step_create_report(context: Context) -> None:
|
|
report = CleanupReport(dry_run=True)
|
|
report.sandboxes.scanned = 5
|
|
report.sandboxes.removed = 2
|
|
report.checkpoints.scanned = 100
|
|
context.report_dict = report.as_dict()
|
|
|
|
|
|
@then('the report dict should contain key "{key}"')
|
|
def step_report_dict_key(context: Context, key: str) -> None:
|
|
assert key in context.report_dict, (
|
|
f"Expected key '{key}' in report dict, got: {list(context.report_dict.keys())}"
|
|
)
|
|
|
|
|
|
@when("I create a cleanup service with invalid settings")
|
|
def step_invalid_settings(context: Context) -> None:
|
|
try:
|
|
CleanupService(settings="not-a-settings") # type: ignore[arg-type]
|
|
context.type_error_raised = False
|
|
except TypeError:
|
|
context.type_error_raised = True
|
|
|
|
|
|
@then("a TypeError should be raised")
|
|
def step_type_error(context: Context) -> None:
|
|
assert context.type_error_raised, "Expected TypeError to be raised"
|
|
|
|
|
|
@given("a cleanup service for age description testing")
|
|
def step_age_service(context: Context) -> None:
|
|
context.settings = _make_settings()
|
|
context.service = CleanupService(
|
|
settings=context.settings,
|
|
active_plan_ids=frozenset(),
|
|
)
|
|
|
|
|
|
@when("I check the age description for a {hours:d} hour old file")
|
|
def step_age_hours(context: Context, hours: int) -> None:
|
|
tmp = Path(tempfile.mktemp(prefix="ca-age-test-"))
|
|
tmp.write_text("test")
|
|
old_time = time.time() - (hours * 3600)
|
|
os.utime(tmp, (old_time, old_time))
|
|
context.age_desc = CleanupService._age_description(tmp)
|
|
tmp.unlink()
|
|
|
|
|
|
@when("I check the age description for a {minutes:d} minute old file")
|
|
def step_age_minutes(context: Context, minutes: int) -> None:
|
|
tmp = Path(tempfile.mktemp(prefix="ca-age-test-"))
|
|
tmp.write_text("test")
|
|
old_time = time.time() - (minutes * 60)
|
|
os.utime(tmp, (old_time, old_time))
|
|
context.age_desc = CleanupService._age_description(tmp)
|
|
tmp.unlink()
|
|
|
|
|
|
@when("I check the age description for a {days:d} day old file")
|
|
def step_age_days(context: Context, days: int) -> None:
|
|
tmp = Path(tempfile.mktemp(prefix="ca-age-test-"))
|
|
tmp.write_text("test")
|
|
old_time = time.time() - (days * 86400)
|
|
os.utime(tmp, (old_time, old_time))
|
|
context.age_desc = CleanupService._age_description(tmp)
|
|
tmp.unlink()
|
|
|
|
|
|
@then('the age description should contain "{text}"')
|
|
def step_age_contains(context: Context, text: str) -> None:
|
|
assert text in context.age_desc, (
|
|
f"Expected '{text}' in age description, got: {context.age_desc}"
|
|
)
|
|
|
|
|
|
# ── Full scan/purge integration via service ──────────────────────
|
|
|
|
|
|
@when("I run a full scan")
|
|
def step_full_scan(context: Context) -> None:
|
|
context.report = context.service.scan()
|
|
|
|
|
|
@then("the report should contain log stale items")
|
|
def step_log_stale_items(context: Context) -> None:
|
|
log_items = [i for i in context.report.stale_items if i.resource_type == "log"]
|
|
assert len(log_items) >= 1, f"Expected >= 1 log stale item, found {len(log_items)}"
|
|
|
|
|
|
@then("the report should contain backup stale items")
|
|
def step_backup_stale_items(context: Context) -> None:
|
|
backup_items = [
|
|
i for i in context.report.stale_items if i.resource_type == "backup"
|
|
]
|
|
assert len(backup_items) >= 1, (
|
|
f"Expected >= 1 backup stale item, found {len(backup_items)}"
|
|
)
|
|
|
|
|
|
@when("I run purge all")
|
|
def step_run_purge_all(context: Context) -> None:
|
|
context.report = context.service.purge(purge_all=True)
|
|
|
|
|
|
@then("the backup file should be removed")
|
|
def step_backup_removed(context: Context) -> None:
|
|
assert not context.backup_file.exists(), (
|
|
f"Backup file {context.backup_file} still exists"
|
|
)
|
|
|
|
|
|
@when("I run a full checkpoint scan")
|
|
def step_full_checkpoint_scan(context: Context) -> None:
|
|
context.service._get_checkpoint_base_dir = lambda: context.checkpoint_dir.parent
|
|
plan_dir = context.checkpoint_dir.parent / "test-plan"
|
|
if not plan_dir.exists():
|
|
context.checkpoint_dir.rename(plan_dir)
|
|
context.checkpoint_dir = plan_dir
|
|
context.service._get_checkpoint_base_dir = lambda: plan_dir.parent
|
|
context.report = context.service.scan()
|
|
|
|
|
|
@then("the report should contain checkpoint stale items")
|
|
def step_checkpoint_stale_items(context: Context) -> None:
|
|
cp_items = [
|
|
i for i in context.report.stale_items if i.resource_type == "checkpoint"
|
|
]
|
|
assert len(cp_items) >= 1, (
|
|
f"Expected >= 1 checkpoint stale item, found {len(cp_items)}"
|
|
)
|
|
|
|
|
|
@when("I run purge for checkpoints")
|
|
def step_purge_checkpoints_full(context: Context) -> None:
|
|
plan_dir = context.checkpoint_dir
|
|
context.service._get_checkpoint_base_dir = lambda: plan_dir.parent
|
|
target = plan_dir.parent / "test-plan-purge"
|
|
if not target.exists():
|
|
plan_dir.rename(target)
|
|
context.checkpoint_dir = target
|
|
context.service._get_checkpoint_base_dir = lambda: target.parent
|
|
context.report = context.service.purge()
|
|
|
|
|
|
@then("the report should show checkpoints removed")
|
|
def step_checkpoints_removed_report(context: Context) -> None:
|
|
assert context.report.checkpoints.removed >= 1, (
|
|
f"Expected >= 1 checkpoint removed, got {context.report.checkpoints.removed}"
|
|
)
|