"""Tests for metrics collection system."""
import time
import pytest
from pathlib import Path
import tempfile
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from amicus.metrics import MetricsCollector
@pytest.fixture
def temp_db():
"""Create temporary database for testing"""
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "test_metrics.db"
yield db_path
def test_metrics_initialization(temp_db):
"""Test metrics collector initialization"""
collector = MetricsCollector(temp_db)
assert temp_db.exists()
stats = collector.get_stats()
assert stats['total_events'] == 0
assert collector.is_enabled()
def test_record_metric(temp_db):
"""Test recording a metric"""
collector = MetricsCollector(temp_db)
collector.record("test.metric", {
"value": 42,
"message": "test"
}, node_id="test-node")
stats = collector.get_stats()
assert stats['total_events'] == 1
def test_query_metrics(temp_db):
"""Test querying metrics"""
collector = MetricsCollector(temp_db)
# Record multiple metrics
for i in range(5):
collector.record("test.counter", {"count": i}, node_id="node-1")
time.sleep(0.01)
# Query all
results = collector.query()
assert len(results) == 5
# Query filtered by metric
results = collector.query(metric="test.counter")
assert len(results) == 5
# Query filtered by node
results = collector.query(node_id="node-1")
assert len(results) == 5
def test_aggregate_metrics(temp_db):
"""Test aggregation functions"""
collector = MetricsCollector(temp_db)
# Record test data
values = [10, 20, 30, 40, 50]
for v in values:
collector.record("test.values", {"amount": v})
# Test aggregations
avg = collector.aggregate("test.values", "amount", "avg")
assert avg == 30.0
total = collector.aggregate("test.values", "amount", "sum")
assert total == 150.0
minimum = collector.aggregate("test.values", "amount", "min")
assert minimum == 10.0
maximum = collector.aggregate("test.values", "amount", "max")
assert maximum == 50.0
count = collector.aggregate("test.values", "amount", "count")
assert count == 5.0
def test_time_range_query(temp_db):
"""Test time range filtering"""
collector = MetricsCollector(temp_db)
start_time = time.time()
# Record metrics over time
collector.record("test.early", {"value": 1})
time.sleep(0.05)
mid_time = time.time()
time.sleep(0.05)
collector.record("test.late", {"value": 2})
end_time = time.time()
# Query full range
all_results = collector.query(since=start_time, until=end_time)
assert len(all_results) == 2
# Query first half
early_results = collector.query(since=start_time, until=mid_time)
assert len(early_results) == 1
assert early_results[0]['metric'] == "test.early"
def test_cleanup_old_metrics(temp_db):
"""Test retention and cleanup"""
collector = MetricsCollector(temp_db)
# Record metrics with old timestamps
old_time = time.time() - (31 * 86400) # 31 days ago
with collector._get_conn() as conn:
conn.execute(
"INSERT INTO events (timestamp, metric, data) VALUES (?, ?, ?)",
(old_time, "test.old", "{}")
)
# Record current metric
collector.record("test.current", {"value": 1})
# Verify both exist
assert collector.get_stats()['total_events'] == 2
# Cleanup with 30-day retention
deleted = collector.cleanup(retention_days=30)
assert deleted == 1
# Verify old metric removed
assert collector.get_stats()['total_events'] == 1
def test_sanitization(temp_db):
"""Test data sanitization"""
collector = MetricsCollector(temp_db)
# Record metric with sensitive data
collector.record("test.sensitive", {
"username": "alice",
"api_key": "secret-key-123",
"value": 42
})
# Query and verify sanitization
results = collector.query(metric="test.sensitive")
assert len(results) == 1
data = results[0]['data']
assert data['username'] == "alice"
assert data['api_key'] == "<redacted>"
assert data['value'] == 42
def test_enable_disable(temp_db):
"""Test enabling/disabling metrics"""
collector = MetricsCollector(temp_db)
# Should be enabled by default
assert collector.is_enabled()
# Record metric
collector.record("test.enabled", {"value": 1})
assert collector.get_stats()['total_events'] == 1
# Disable
collector.set_enabled(False)
assert not collector.is_enabled()
# Try to record (should be ignored)
collector.record("test.disabled", {"value": 2})
assert collector.get_stats()['total_events'] == 1 # No increase
# Re-enable
collector.set_enabled(True)
collector.record("test.re-enabled", {"value": 3})
assert collector.get_stats()['total_events'] == 2
def test_stats(temp_db):
"""Test statistics reporting"""
collector = MetricsCollector(temp_db)
# Record various metrics
collector.record("metric.a", {"value": 1}, node_id="node-1")
collector.record("metric.b", {"value": 2}, node_id="node-1")
collector.record("metric.a", {"value": 3}, node_id="node-2")
stats = collector.get_stats()
assert stats['total_events'] == 3
assert stats['unique_metrics'] == 2
assert stats['unique_nodes'] == 2
assert stats['db_size_bytes'] > 0
if __name__ == '__main__':
pytest.main([__file__, '-v'])
