CommonAutoRearsh/tests/test_mutation_engine.py

108 lines
4.7 KiB
Python

from __future__ import annotations
import shutil
import tempfile
import unittest
from pathlib import Path
from engine.artifact_manager import ArtifactManager
from engine.models import (
ArtifactSpec,
BudgetSpec,
LoggingSpec,
MutationSpec,
MutatorSpec,
ObjectiveSpec,
PolicySpec,
RunnerSpec,
ScorerParseSpec,
ScorerSpec,
TaskSpec,
)
from engine.mutation_engine import MutationValidationError, validate_candidate_changes
def _make_task(task_root: Path, allowed_file_types: list[str], max_changed_lines: int) -> TaskSpec:
return TaskSpec(
id="mutation-test",
description="Mutation validation fixture.",
artifacts=ArtifactSpec(include=["fixtures/*"], exclude=[], max_files_per_iteration=10),
mutation=MutationSpec(
mode="direct_edit",
allowed_file_types=allowed_file_types,
max_changed_lines=max_changed_lines,
),
mutator=MutatorSpec(type="command", command="python -c \"print('mutate')\"", cwd="tasks/demo", timeout_seconds=30),
runner=RunnerSpec(command="python -c \"print('runner ok')\"", cwd="tasks/demo", timeout_seconds=30),
scorer=ScorerSpec(
type="command",
command="python -c \"print('{\\\"score\\\": 1.0, \\\"metrics\\\": {}}')\"",
timeout_seconds=30,
parse=ScorerParseSpec(format="json", score_field="score", metrics_field="metrics"),
),
objective=ObjectiveSpec(primary_metric="score", direction="maximize"),
constraints=[],
policy=PolicySpec(keep_if="better_primary", tie_breakers=[], on_failure="discard"),
budget=BudgetSpec(max_iterations=1, max_failures=1),
logging=LoggingSpec(results_file="work/results.jsonl", candidate_dir="work/candidates"),
root_dir=task_root,
)
class MutationEngineTest(unittest.TestCase):
def test_rejects_too_many_changed_lines_in_candidate_root(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
baseline_root = Path(tmp) / "baseline"
candidate_root = Path(tmp) / "candidate"
(baseline_root / "fixtures").mkdir(parents=True)
(baseline_root / "fixtures" / "note.md").write_text("line 1\nline 2\n", encoding="utf-8")
shutil.copytree(baseline_root, candidate_root)
baseline_task = _make_task(baseline_root, allowed_file_types=[".md"], max_changed_lines=1)
snapshot = ArtifactManager(baseline_task).snapshot()
(candidate_root / "fixtures" / "note.md").write_text("line 1\nline 2\nline 3\n", encoding="utf-8")
with self.assertRaises(MutationValidationError) as ctx:
validate_candidate_changes(baseline_task, snapshot, candidate_root)
self.assertIn("changed lines", str(ctx.exception))
def test_rejects_new_file_with_disallowed_extension_in_candidate_root(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
baseline_root = Path(tmp) / "baseline"
candidate_root = Path(tmp) / "candidate"
(baseline_root / "fixtures").mkdir(parents=True)
(baseline_root / "fixtures" / "note.md").write_text("line 1\n", encoding="utf-8")
shutil.copytree(baseline_root, candidate_root)
baseline_task = _make_task(baseline_root, allowed_file_types=[".md"], max_changed_lines=10)
snapshot = ArtifactManager(baseline_task).snapshot()
(candidate_root / "fixtures" / "extra.txt").write_text("new file\n", encoding="utf-8")
with self.assertRaises(MutationValidationError) as ctx:
validate_candidate_changes(baseline_task, snapshot, candidate_root)
self.assertIn("disallowed file type", str(ctx.exception))
def test_rejects_renamed_file_with_disallowed_extension(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
baseline_root = Path(tmp) / "baseline"
candidate_root = Path(tmp) / "candidate"
(baseline_root / "fixtures").mkdir(parents=True)
(baseline_root / "fixtures" / "note.md").write_text("line 1\n", encoding="utf-8")
shutil.copytree(baseline_root, candidate_root)
baseline_task = _make_task(baseline_root, allowed_file_types=[".md"], max_changed_lines=10)
snapshot = ArtifactManager(baseline_task).snapshot()
(candidate_root / "fixtures" / "note.md").unlink()
(candidate_root / "fixtures" / "note.txt").write_text("line 1\n", encoding="utf-8")
with self.assertRaises(MutationValidationError) as ctx:
validate_candidate_changes(baseline_task, snapshot, candidate_root)
self.assertIn("disallowed file type", str(ctx.exception))
if __name__ == "__main__":
unittest.main()