| """Real storage/cache behavior; runnable with stdlib unittest and no API keys.""" |
| import asyncio |
| import json |
| from pathlib import Path |
| import tempfile |
| import unittest |
| from unittest.mock import patch |
|
|
| from src.services.cache_service import CacheService |
| from src.services.metrics_service import MetricsService |
|
|
|
|
| class RuntimeRegressionTests(unittest.TestCase): |
| def setUp(self): |
| self.tmp = tempfile.TemporaryDirectory() |
| self.addCleanup(self.tmp.cleanup) |
| self.path = Path(self.tmp.name) / "metrics.json" |
| self.metrics = MetricsService(self.path) |
|
|
| def test_failed_write_retains_counts_for_retry(self): |
| self.metrics.incr("agent_requests", 3) |
| with patch("src.services.metrics_service.os.replace", side_effect=OSError("disk unavailable")): |
| asyncio.run(self.metrics.flush()) |
| self.assertEqual(self.metrics.snapshot()["agent_requests"], 3) |
| asyncio.run(self.metrics.flush()) |
| self.assertEqual(json.loads(self.path.read_text())["agent_requests"], 3) |
| asyncio.run(self.metrics.flush()) |
| self.assertEqual(self.metrics.snapshot()["agent_requests"], 3) |
|
|
| def test_invalid_json_shapes_recover_and_flush(self): |
| for raw in ('[]', 'null', '{"agent_requests": {}, "intents": 1, "errors":{"x":"bad"}}'): |
| self.path.write_text(raw) |
| self.metrics.incr("agent_requests") |
| self.metrics.incr_bucket("intents", "hours") |
| asyncio.run(self.metrics.flush()) |
| state = self.metrics.snapshot() |
| self.assertEqual(state["agent_requests"], 1) |
| self.assertEqual(state["intents"], {"hours": 1}) |
|
|
| def test_missing_file_defaults_do_not_share_nested_counters(self): |
| self.path.unlink() |
| self.metrics.incr_bucket("intents", "hours") |
| self.assertEqual(self.metrics.snapshot()["intents"], {"hours": 1}) |
| self.assertEqual(self.metrics.snapshot()["intents"], {"hours": 1}) |
|
|
| def test_zero_recent_queries_returns_empty(self): |
| self.metrics.log_query("library hours") |
| self.assertEqual(self.metrics.recent_queries(0), []) |
|
|
| def test_cache_enforces_bound_immediately(self): |
| cache = CacheService(max_size=2) |
| for i in range(10): |
| cache.set(str(i), i) |
| self.assertLessEqual(len(cache._cache), 2) |
|
|
| def test_cache_expires_values(self): |
| cache = CacheService() |
| cache.set("expired", "value", ttl=-1) |
| self.assertIsNone(cache.get("expired")) |
| cache.set("live", "value", ttl=30) |
| self.assertEqual(cache.get("live"), "value") |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
