WIN-JHFT4D3SIVT\caoxiaozhu
11160ec4d2
- M.2: ForgettingCurve, MemoryDecay, MemoryReinforcement (selective forgetting) - M.3: DailyDigestGenerator, ReminderScheduler, ProactiveInformer (proactive reminders) - M.4: MemoryExtractor with LLM-based memory extraction from conversations - M.5: MemoryRecallInjector with token budget control for prompt injection - All phases include comprehensive unit tests (109 tests passing) - Updated checklist.md to mark all tasks complete
221 lines
7.3 KiB
Python
221 lines
7.3 KiB
Python
"""
|
|
Tests for MemoryDecay (M.2)
|
|
|
|
Tests: evaluate(), archive_memory(), deprioritize_memory(), restore_from_archive().
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import UTC, datetime, timedelta
|
|
from unittest.mock import MagicMock
|
|
|
|
from app.services.memory.memory_decay import MemoryDecay
|
|
|
|
|
|
def create_mock_memory(
|
|
last_accessed_at=None,
|
|
importance_level: str = "medium",
|
|
decay_score: float = 1.0,
|
|
is_archived: bool = False,
|
|
archive_at=None,
|
|
):
|
|
"""Create a mock UserMemory for testing."""
|
|
memory = MagicMock()
|
|
memory.last_accessed_at = last_accessed_at
|
|
memory.importance_level = importance_level
|
|
memory.decay_score = decay_score
|
|
memory.is_archived = is_archived
|
|
memory.archive_at = archive_at
|
|
return memory
|
|
|
|
|
|
class TestMemoryDecayEvaluate:
|
|
"""Test evaluate() method."""
|
|
|
|
def test_evaluate_fresh_memory_keeps_active(self):
|
|
"""Fresh memory should be kept active."""
|
|
decay = MemoryDecay()
|
|
recent = datetime.now(UTC) - timedelta(hours=1)
|
|
memory = create_mock_memory(last_accessed_at=recent)
|
|
|
|
result = decay.evaluate(memory)
|
|
|
|
assert result["action"] == "keep_active"
|
|
assert result["should_archive"] is False
|
|
assert result["should_deprioritize"] is False
|
|
assert result["decay_score"] > 0.5
|
|
|
|
def test_evaluate_old_low_importance_archives(self):
|
|
"""Old low-importance memory should be archived."""
|
|
decay = MemoryDecay()
|
|
old = datetime.now(UTC) - timedelta(days=100)
|
|
memory = create_mock_memory(last_accessed_at=old, importance_level="low")
|
|
|
|
result = decay.evaluate(memory)
|
|
|
|
assert result["action"] == "archive"
|
|
assert result["should_archive"] is True
|
|
assert result["should_deprioritize"] is True
|
|
assert result["decay_score"] < 0.2
|
|
|
|
def test_evaluate_old_high_importance_deprioritizes(self):
|
|
"""Old high-importance memory may be deprioritized but not archived."""
|
|
decay = MemoryDecay()
|
|
# ~45 days for high: exp(-45/90) ≈ 0.6, still > 0.5
|
|
old = datetime.now(UTC) - timedelta(days=45)
|
|
memory = create_mock_memory(last_accessed_at=old, importance_level="high")
|
|
|
|
result = decay.evaluate(memory)
|
|
|
|
assert result["should_archive"] is False
|
|
assert result["should_deprioritize"] is False
|
|
assert 0.5 < result["decay_score"] < 0.7
|
|
|
|
def test_evaluate_boundary_deprioritize(self):
|
|
"""Memory at ~42 days medium importance should be deprioritized but not archived."""
|
|
decay = MemoryDecay()
|
|
# ~42 days for medium: exp(-42/30) ≈ 0.25 < 0.5, > 0.2
|
|
old = datetime.now(UTC) - timedelta(days=42)
|
|
memory = create_mock_memory(last_accessed_at=old, importance_level="medium")
|
|
|
|
result = decay.evaluate(memory)
|
|
|
|
assert result["action"] == "deprioritize"
|
|
assert result["should_deprioritize"] is True
|
|
assert result["should_archive"] is False
|
|
|
|
def test_evaluate_returns_all_keys(self):
|
|
"""evaluate() returns decay_score, should_archive, should_deprioritize, action."""
|
|
decay = MemoryDecay()
|
|
memory = create_mock_memory(last_accessed_at=datetime.now(UTC))
|
|
|
|
result = decay.evaluate(memory)
|
|
|
|
assert "decay_score" in result
|
|
assert "should_archive" in result
|
|
assert "should_deprioritize" in result
|
|
assert "action" in result
|
|
assert result["action"] in ("keep_active", "deprioritize", "archive")
|
|
|
|
|
|
class TestMemoryDecayArchiveMemory:
|
|
"""Test archive_memory() method."""
|
|
|
|
def test_archive_sets_is_archived_true(self):
|
|
"""archive_memory() sets is_archived = True."""
|
|
decay = MemoryDecay()
|
|
memory = create_mock_memory(is_archived=False)
|
|
|
|
result = decay.archive_memory(memory)
|
|
|
|
assert result.is_archived is True
|
|
|
|
def test_archive_sets_low_decay_score(self):
|
|
"""archive_memory() resets decay_score to 0.1."""
|
|
decay = MemoryDecay()
|
|
memory = create_mock_memory(decay_score=0.8)
|
|
|
|
result = decay.archive_memory(memory)
|
|
|
|
assert result.decay_score == 0.1
|
|
|
|
def test_archive_sets_archive_at_timestamp(self):
|
|
"""archive_memory() sets archive_at to current time."""
|
|
decay = MemoryDecay()
|
|
memory = create_mock_memory(archive_at=None)
|
|
|
|
before = datetime.now(UTC)
|
|
result = decay.archive_memory(memory)
|
|
after = datetime.now(UTC)
|
|
|
|
assert result.archive_at is not None
|
|
assert before <= result.archive_at <= after
|
|
|
|
def test_archive_preserves_other_fields(self):
|
|
"""archive_memory() does not modify other fields."""
|
|
decay = MemoryDecay()
|
|
memory = create_mock_memory(
|
|
last_accessed_at=datetime.now(UTC),
|
|
importance_level="high",
|
|
decay_score=0.5,
|
|
)
|
|
|
|
result = decay.archive_memory(memory)
|
|
|
|
assert result.last_accessed_at == memory.last_accessed_at
|
|
assert result.importance_level == "high"
|
|
|
|
|
|
class TestMemoryDecayDeprioritizeMemory:
|
|
"""Test deprioritize_memory() method."""
|
|
|
|
def test_deprioritize_updates_decay_score(self):
|
|
"""deprioritize_memory() recalculates decay_score."""
|
|
decay = MemoryDecay()
|
|
# Old memory will have low decay score
|
|
old = datetime.now(UTC) - timedelta(days=60)
|
|
memory = create_mock_memory(
|
|
last_accessed_at=old, importance_level="medium", decay_score=0.9
|
|
)
|
|
|
|
result = decay.deprioritize_memory(memory)
|
|
|
|
assert result.decay_score < 0.5 # Should be recalculated low
|
|
|
|
def test_deprioritize_does_not_archive(self):
|
|
"""deprioritize_memory() does NOT set is_archived."""
|
|
decay = MemoryDecay()
|
|
memory = create_mock_memory(is_archived=False)
|
|
|
|
result = decay.deprioritize_memory(memory)
|
|
|
|
assert result.is_archived is False
|
|
|
|
|
|
class TestMemoryDecayRestoreFromArchive:
|
|
"""Test restore_from_archive() method."""
|
|
|
|
def test_restore_clears_is_archived(self):
|
|
"""restore_from_archive() sets is_archived = False."""
|
|
decay = MemoryDecay()
|
|
memory = create_mock_memory(is_archived=True)
|
|
|
|
result = decay.restore_from_archive(memory)
|
|
|
|
assert result.is_archived is False
|
|
|
|
def test_restore_sets_decay_score_high(self):
|
|
"""restore_from_archive() sets decay_score to 0.8."""
|
|
decay = MemoryDecay()
|
|
memory = create_mock_memory(decay_score=0.1)
|
|
|
|
result = decay.restore_from_archive(memory)
|
|
|
|
assert result.decay_score == 0.8
|
|
|
|
def test_restore_updates_last_accessed(self):
|
|
"""restore_from_archive() updates last_accessed_at to now."""
|
|
decay = MemoryDecay()
|
|
old_time = datetime.now(UTC) - timedelta(days=30)
|
|
memory = create_mock_memory(
|
|
last_accessed_at=old_time, is_archived=True, archive_at=old_time
|
|
)
|
|
|
|
before = datetime.now(UTC)
|
|
result = decay.restore_from_archive(memory)
|
|
after = datetime.now(UTC)
|
|
|
|
assert before <= result.last_accessed_at <= after
|
|
|
|
def test_restore_clears_archive_at(self):
|
|
"""restore_from_archive() sets archive_at to None."""
|
|
decay = MemoryDecay()
|
|
memory = create_mock_memory(is_archived=True, archive_at=datetime.now(UTC))
|
|
|
|
result = decay.restore_from_archive(memory)
|
|
|
|
assert result.archive_at is None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|