fischer-agentkit/tests/unit/core/test_spec_manager.py

205 lines
6.2 KiB
Python

"""Tests for SpecManager — Spec 文档管理器"""
from __future__ import annotations
import time
from pathlib import Path
import pytest
from agentkit.core.spec_manager import Spec, SpecManager, SpecStep
@pytest.fixture
def specs_dir(tmp_path: Path) -> str:
"""Provide a temporary directory for spec files."""
return str(tmp_path / "specs")
@pytest.fixture
def mgr(specs_dir: str) -> SpecManager:
"""Create a SpecManager with a temporary directory."""
return SpecManager(specs_dir=specs_dir)
def make_spec(spec_id: str = "test-spec", goal: str = "test goal") -> Spec:
"""Create a test Spec."""
return Spec(
spec_id=spec_id,
goal=goal,
steps=[
SpecStep(step_id="s1", name="Step 1", description="First step"),
SpecStep(step_id="s2", name="Step 2", description="Second step", dependencies=["s1"]),
],
)
class TestSpecManagerCreateAndGet:
"""Test create and get a spec."""
def test_create_and_get(self, mgr: SpecManager):
spec = make_spec()
path = mgr.create(spec)
assert path.exists()
loaded = mgr.get(spec.spec_id)
assert loaded is not None
assert loaded.spec_id == spec.spec_id
assert loaded.goal == spec.goal
assert len(loaded.steps) == 2
assert loaded.steps[0].step_id == "s1"
assert loaded.steps[1].dependencies == ["s1"]
def test_create_writes_yaml_file(self, mgr: SpecManager, specs_dir: str):
spec = make_spec(spec_id="yaml-test")
mgr.create(spec)
yaml_path = Path(specs_dir) / "yaml-test.yaml"
assert yaml_path.exists()
def test_get_returns_from_cache(self, mgr: SpecManager):
spec = make_spec(spec_id="cached")
mgr.create(spec)
# Second get should hit cache
loaded = mgr.get("cached")
assert loaded is not None
assert loaded.spec_id == "cached"
class TestSpecManagerUpdate:
"""Test update spec fields."""
def test_update_goal(self, mgr: SpecManager):
spec = make_spec()
mgr.create(spec)
updated = mgr.update(spec.spec_id, goal="new goal")
assert updated is not None
assert updated.goal == "new goal"
def test_update_steps(self, mgr: SpecManager):
spec = make_spec()
mgr.create(spec)
new_steps = [
{"step_id": "s1", "name": "Step 1 Updated", "description": "Updated first step"},
]
updated = mgr.update(spec.spec_id, steps=new_steps)
assert updated is not None
assert len(updated.steps) == 1
assert updated.steps[0].name == "Step 1 Updated"
def test_update_metadata(self, mgr: SpecManager):
spec = make_spec()
mgr.create(spec)
updated = mgr.update(spec.spec_id, metadata={"key": "value"})
assert updated is not None
assert updated.metadata == {"key": "value"}
def test_update_nonexistent_returns_none(self, mgr: SpecManager):
result = mgr.update("nonexistent", goal="x")
assert result is None
class TestSpecManagerConfirm:
"""Test confirm sets status and confirmed_at."""
def test_confirm_sets_status_and_timestamp(self, mgr: SpecManager):
spec = make_spec()
mgr.create(spec)
assert spec.status == "draft"
assert spec.confirmed_at is None
confirmed = mgr.confirm(spec.spec_id)
assert confirmed is not None
assert confirmed.status == "confirmed"
assert confirmed.confirmed_at is not None
def test_confirm_marks_pending_steps_as_confirmed(self, mgr: SpecManager):
spec = make_spec()
mgr.create(spec)
confirmed = mgr.confirm(spec.spec_id)
assert confirmed is not None
for step in confirmed.steps:
assert step.status == "confirmed"
def test_confirm_nonexistent_returns_none(self, mgr: SpecManager):
result = mgr.confirm("nonexistent")
assert result is None
class TestSpecManagerList:
"""Test list_specs returns specs sorted by created_at desc."""
def test_list_specs_sorted_by_created_at_desc(self, mgr: SpecManager):
spec_a = make_spec(spec_id="spec-a", goal="A")
mgr.create(spec_a)
# Small delay to ensure different timestamps
time.sleep(0.01)
spec_b = make_spec(spec_id="spec-b", goal="B")
mgr.create(spec_b)
specs = mgr.list_specs()
assert len(specs) == 2
# Most recent first
assert specs[0].spec_id == "spec-b"
assert specs[1].spec_id == "spec-a"
def test_list_specs_filter_by_status(self, mgr: SpecManager):
spec_a = make_spec(spec_id="draft-spec")
mgr.create(spec_a)
spec_b = make_spec(spec_id="confirmed-spec")
mgr.create(spec_b)
mgr.confirm(spec_b.spec_id)
draft_specs = mgr.list_specs(status="draft")
assert len(draft_specs) == 1
assert draft_specs[0].spec_id == "draft-spec"
confirmed_specs = mgr.list_specs(status="confirmed")
assert len(confirmed_specs) == 1
assert confirmed_specs[0].spec_id == "confirmed-spec"
def test_list_specs_empty(self, mgr: SpecManager):
specs = mgr.list_specs()
assert specs == []
class TestSpecManagerDelete:
"""Test delete removes the spec."""
def test_delete_removes_spec(self, mgr: SpecManager):
spec = make_spec()
mgr.create(spec)
assert mgr.get(spec.spec_id) is not None
result = mgr.delete(spec.spec_id)
assert result is True
assert mgr.get(spec.spec_id) is None
def test_delete_removes_yaml_file(self, mgr: SpecManager, specs_dir: str):
spec = make_spec(spec_id="delete-me")
mgr.create(spec)
yaml_path = Path(specs_dir) / "delete-me.yaml"
assert yaml_path.exists()
mgr.delete("delete-me")
assert not yaml_path.exists()
def test_delete_nonexistent_returns_false(self, mgr: SpecManager):
result = mgr.delete("nonexistent")
assert result is False
class TestSpecManagerGetNonExistent:
"""Test get non-existent spec returns None."""
def test_get_nonexistent_returns_none(self, mgr: SpecManager):
result = mgr.get("does-not-exist")
assert result is None