"""
Comprehensive tests for timestamp-based memory recall functionality
"""
import pytest
import asyncio
import time
from datetime import datetime, timedelta, date
import sys
import os
# Add the src directory to the path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
from mcp_memory_service.models.memory import Memory
from mcp_memory_service.storage.chroma import ChromaMemoryStorage
from mcp_memory_service.utils.hashing import generate_content_hash
from mcp_memory_service.utils.time_parser import extract_time_expression, parse_time_expression
class TestTimestampRecall:
"""Test class for timestamp-based memory recall"""
@pytest.fixture
async def storage(self, tmp_path):
"""Create a temporary storage instance for testing"""
storage_path = str(tmp_path / "test_chroma_db")
storage = ChromaMemoryStorage(storage_path, preload_model=True)
yield storage
# Cleanup happens automatically with tmp_path
@pytest.fixture
async def populated_storage(self, storage):
"""Create a storage instance with test memories"""
# Create memories at specific timestamps
now = time.time()
test_data = [
# Today's memories
{"content": "Morning coffee", "offset": 0, "tags": ["today", "morning"]},
{"content": "Lunch meeting", "offset": -4 * 3600, "tags": ["today", "afternoon"]},
{"content": "Evening workout", "offset": -8 * 3600, "tags": ["today", "evening"]},
# Yesterday's memories
{"content": "Yesterday morning", "offset": -24 * 3600, "tags": ["yesterday", "morning"]},
{"content": "Yesterday lunch", "offset": -28 * 3600, "tags": ["yesterday", "afternoon"]},
# Last week's memories
{"content": "Last Monday meeting", "offset": -7 * 24 * 3600, "tags": ["lastweek", "monday"]},
{"content": "Last Friday party", "offset": -3 * 24 * 3600, "tags": ["lastweek", "friday"]},
# Last month's memories
{"content": "Monthly review", "offset": -30 * 24 * 3600, "tags": ["lastmonth", "review"]},
# Specific date memories (for precise testing)
{"content": "New Year 2025", "timestamp": datetime(2025, 1, 1, 12, 0, 0).timestamp(), "tags": ["holiday"]},
{"content": "Valentine's Day", "timestamp": datetime(2025, 2, 14, 18, 0, 0).timestamp(), "tags": ["holiday"]},
]
# Store all memories
for data in test_data:
if "timestamp" in data:
timestamp = data["timestamp"]
else:
timestamp = now + data["offset"]
memory = Memory(
content=data["content"],
content_hash=generate_content_hash(data["content"]),
tags=data["tags"],
created_at=timestamp,
updated_at=timestamp
)
await storage.store(memory)
return storage
@pytest.mark.asyncio
async def test_timestamp_precision(self, storage):
"""Test that timestamps are stored with full float precision"""
# Create memories with sub-second precision
base_time = time.time()
memories = []
for i in range(5):
timestamp = base_time + (i * 0.1) # 0.1 second intervals
memory = Memory(
content=f"Memory {i}",
content_hash=generate_content_hash(f"Memory {i}"),
created_at=timestamp,
updated_at=timestamp
)
success, _ = await storage.store(memory)
assert success
memories.append((timestamp, f"Memory {i}"))
# Verify each memory has unique timestamp
all_data = storage.collection.get(include=["metadatas", "documents"])
stored_timestamps = [m.get("timestamp") for m in all_data["metadatas"]]
# All timestamps should be unique
assert len(set(stored_timestamps)) == len(stored_timestamps)
# All timestamps should be floats
for ts in stored_timestamps:
assert isinstance(ts, float)
@pytest.mark.asyncio
async def test_natural_language_time_parsing(self):
"""Test parsing of various natural language time expressions"""
test_cases = [
("yesterday", True),
("last week", True),
("2 days ago", True),
("last month", True),
("this morning", True),
("last summer", True),
("christmas", True),
("first quarter of 2024", True),
("random text", False),
]
for query, should_parse in test_cases:
start_ts, end_ts = parse_time_expression(query)
if should_parse:
assert start_ts is not None or end_ts is not None, f"Failed to parse: {query}"
else:
assert start_ts is None and end_ts is None, f"Incorrectly parsed: {query}"
@pytest.mark.asyncio
async def test_recall_yesterday(self, populated_storage):
"""Test recalling memories from yesterday"""
results = await populated_storage.recall(
query="yesterday",
n_results=10
)
# We should get yesterday's memories
assert len(results) == 2
contents = [r.memory.content for r in results]
assert "Yesterday morning" in contents
assert "Yesterday lunch" in contents
@pytest.mark.asyncio
async def test_recall_last_week(self, populated_storage):
"""Test recalling memories from last week"""
start_ts, end_ts = parse_time_expression("last week")
results = await populated_storage.recall(
query=None,
n_results=10,
start_timestamp=start_ts,
end_timestamp=end_ts
)
# Should get last week's memories
contents = [r.memory.content for r in results]
assert any("Last" in c and "week" in ' '.join(r.memory.tags) for c, r in zip(contents, results))
@pytest.mark.asyncio
async def test_recall_with_semantic_and_time(self, populated_storage):
"""Test combined semantic and time-based recall"""
# Extract time expression and search for "meeting" in last 10 days
results = await populated_storage.recall(
query="meeting",
n_results=10,
start_timestamp=time.time() - (10 * 24 * 3600),
end_timestamp=time.time()
)
# Should find meetings within time range
contents = [r.memory.content for r in results]
assert any("meeting" in c.lower() for c in contents)
@pytest.mark.asyncio
async def test_recall_specific_date_range(self, populated_storage):
"""Test recall with specific date range"""
# Query for January 2025
start_ts = datetime(2025, 1, 1).timestamp()
end_ts = datetime(2025, 1, 31, 23, 59, 59).timestamp()
results = await populated_storage.recall(
query=None,
n_results=10,
start_timestamp=start_ts,
end_timestamp=end_ts
)
# Should find New Year memory
contents = [r.memory.content for r in results]
assert "New Year 2025" in contents
@pytest.mark.asyncio
async def test_recall_time_of_day(self, populated_storage):
"""Test recall with time of day expressions"""
# Get today's date at morning time
today = date.today()
morning_start = datetime.combine(today, datetime.min.time().replace(hour=5))
morning_end = datetime.combine(today, datetime.min.time().replace(hour=11, minute=59, second=59))
results = await populated_storage.recall(
query=None,
n_results=10,
start_timestamp=morning_start.timestamp(),
end_timestamp=morning_end.timestamp()
)
# Should find morning memories from today
contents = [r.memory.content for r in results]
assert any("morning" in c.lower() or "morning" in ' '.join(r.memory.tags)
for c, r in zip(contents, results))
@pytest.mark.asyncio
async def test_edge_cases(self, storage):
"""Test edge cases for timestamp handling"""
# Test with None timestamps
results = await storage.recall(query="test", n_results=5)
assert isinstance(results, list)
# Test with very old timestamp
old_time = datetime(1970, 1, 1).timestamp()
memory = Memory(
content="Very old memory",
content_hash=generate_content_hash("Very old memory"),
created_at=old_time
)
success, _ = await storage.store(memory)
assert success
# Test with future timestamp
future_time = datetime(2030, 1, 1).timestamp()
memory = Memory(
content="Future memory",
content_hash=generate_content_hash("Future memory"),
created_at=future_time
)
success, _ = await storage.store(memory)
assert success
@pytest.mark.asyncio
async def test_timestamp_normalization(self):
"""Test the normalize_timestamp function"""
# Test with datetime object
dt = datetime.now()
normalized = ChromaMemoryStorage.normalize_timestamp(dt)
assert isinstance(normalized, float)
assert normalized == time.mktime(dt.timetuple())
# Test with float
ts = time.time()
normalized = ChromaMemoryStorage.normalize_timestamp(ts)
assert normalized == ts
# Test with int
ts_int = int(time.time())
normalized = ChromaMemoryStorage.normalize_timestamp(ts_int)
assert normalized == float(ts_int)
if __name__ == "__main__":
# Run with pytest
pytest.main([__file__, "-v"])