Coverage for tests / unit / ai / test_cache.py: 100%
103 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-03 18:53 +0000
1"""Tests for the AI suggestion cache."""
3from __future__ import annotations
5import json
6import time
7from pathlib import Path
9from assertpy import assert_that
11from lintro.ai.cache import (
12 CACHE_DIR,
13 _cache_key,
14 _evict_lru,
15 cache_suggestion,
16 get_cached_suggestion,
17)
18from lintro.ai.models import AIFixSuggestion
21def _make_suggestion(**kwargs: object) -> AIFixSuggestion:
22 """Create a minimal AIFixSuggestion for tests."""
23 defaults = {
24 "file": "test.py",
25 "line": 1,
26 "code": "E001",
27 "original_code": "x",
28 "suggested_code": "y",
29 }
30 defaults.update(kwargs)
31 return AIFixSuggestion(**defaults) # type: ignore[arg-type]
34# -- _cache_key --------------------------------------------------------------
37def test_cache_key_is_deterministic() -> None:
38 """Same inputs always produce the same key."""
39 key1 = _cache_key("content", "E001", 10, "msg")
40 key2 = _cache_key("content", "E001", 10, "msg")
41 assert_that(key1).is_equal_to(key2)
44def test_cache_key_differs_for_different_inputs() -> None:
45 """Different inputs produce different keys."""
46 key1 = _cache_key("content", "E001", 10, "msg")
47 key2 = _cache_key("content", "E002", 10, "msg")
48 assert_that(key1).is_not_equal_to(key2)
51def test_cache_key_is_16_hex_chars() -> None:
52 """Cache key is a 16-character hex string."""
53 key = _cache_key("content", "E001", 10, "msg")
54 assert_that(key).matches(r"^[0-9a-f]{16}$")
57# -- get_cached_suggestion ---------------------------------------------------
60def test_cache_miss_returns_none(tmp_path: Path) -> None:
61 """A cache miss returns None."""
62 result = get_cached_suggestion(tmp_path, "content", "E001", 10, "msg")
63 assert_that(result).is_none()
66def test_cache_hit_returns_data(tmp_path: Path) -> None:
67 """A cache hit returns the stored suggestion as AIFixSuggestion."""
68 root = tmp_path
69 suggestion = _make_suggestion()
70 cache_suggestion(root, "content", "E001", 10, "msg", suggestion)
72 result = get_cached_suggestion(root, "content", "E001", 10, "msg")
73 assert_that(result).is_not_none()
74 assert_that(result).is_instance_of(AIFixSuggestion)
75 assert_that(result.file).is_equal_to("test.py")
76 assert_that(result.original_code).is_equal_to("x")
77 assert_that(result.suggested_code).is_equal_to("y")
80def test_expired_cache_returns_none_and_deletes_file(tmp_path: Path) -> None:
81 """An expired cache entry returns None and removes the file."""
82 root = tmp_path
83 suggestion = {"file": "test.py", "line": 1, "code": "E001"}
85 # Write a cache entry with a timestamp far in the past
86 key = _cache_key("content", "E001", 10, "msg")
87 cache_dir = root / CACHE_DIR
88 cache_dir.mkdir(parents=True, exist_ok=True)
89 cache_file = cache_dir / f"{key}.json"
90 cache_file.write_text(
91 json.dumps({"timestamp": time.time() - 7200, "suggestion": suggestion}),
92 )
94 result = get_cached_suggestion(root, "content", "E001", 10, "msg", ttl=3600)
95 assert_that(result).is_none()
96 assert_that(cache_file.exists()).is_false()
99# -- cache_suggestion --------------------------------------------------------
102def test_cache_stores_to_correct_path(tmp_path: Path) -> None:
103 """cache_suggestion writes a JSON file under CACHE_DIR."""
104 root = tmp_path
105 suggestion = _make_suggestion()
106 cache_suggestion(root, "content", "E001", 10, "msg", suggestion)
108 key = _cache_key("content", "E001", 10, "msg")
109 cache_file = root / CACHE_DIR / f"{key}.json"
110 assert_that(cache_file.exists()).is_true()
112 data = json.loads(cache_file.read_text())
113 assert_that(data).contains_key("timestamp", "suggestion")
114 assert_that(data["suggestion"]["file"]).is_equal_to("test.py")
117# -- LRU eviction -------------------------------------------------------------
120def test_evict_lru_removes_oldest_entries(tmp_path: Path) -> None:
121 """_evict_lru removes the least recently modified entries."""
122 import os
124 cache_dir = tmp_path / "cache"
125 cache_dir.mkdir()
127 # Create 5 files with staggered mtime values
128 for i in range(5):
129 f = cache_dir / f"entry_{i}.json"
130 f.write_text(json.dumps({"i": i}))
131 # Set mtime so entry_0 is oldest, entry_4 is newest
132 os.utime(f, (1000 + i, 1000 + i))
134 _evict_lru(cache_dir, max_entries=3)
136 remaining = sorted(p.name for p in cache_dir.glob("*.json"))
137 assert_that(remaining).is_equal_to(["entry_2.json", "entry_3.json", "entry_4.json"])
140def test_evict_lru_no_op_when_under_limit(tmp_path: Path) -> None:
141 """_evict_lru does nothing when entry count is within the limit."""
142 cache_dir = tmp_path / "cache"
143 cache_dir.mkdir()
145 for i in range(3):
146 (cache_dir / f"entry_{i}.json").write_text("{}")
148 _evict_lru(cache_dir, max_entries=5)
150 remaining = list(cache_dir.glob("*.json"))
151 assert_that(remaining).is_length(3)
154def test_cache_suggestion_evicts_when_over_max(tmp_path: Path) -> None:
155 """cache_suggestion triggers LRU eviction when max_entries is exceeded."""
156 import os
158 root = tmp_path
159 max_entries = 3
161 # Pre-populate cache with entries that have known mtime ordering
162 cache_dir = root / CACHE_DIR
163 cache_dir.mkdir(parents=True, exist_ok=True)
164 for i in range(3):
165 f = cache_dir / f"old_entry_{i}.json"
166 f.write_text(json.dumps({"timestamp": time.time(), "suggestion": {"i": i}}))
167 # Make entries progressively older
168 os.utime(f, (1000 + i, 1000 + i))
170 # Adding one more entry should evict the oldest (old_entry_0)
171 suggestion = _make_suggestion(code="E999")
172 cache_suggestion(
173 root, "new", "E999", 1, "new msg", suggestion, max_entries=max_entries,
174 )
176 remaining = sorted(p.name for p in cache_dir.glob("*.json"))
177 # old_entry_0 (mtime=1000) should have been evicted
178 assert_that(remaining).does_not_contain("old_entry_0.json")
179 # Should have exactly max_entries files
180 assert_that(remaining).is_length(max_entries)
183def test_get_cached_suggestion_updates_mtime(tmp_path: Path) -> None:
184 """Accessing a cached entry touches the file so it becomes most recent."""
185 import os
187 root = tmp_path
188 suggestion = _make_suggestion()
189 cache_suggestion(root, "content", "E001", 10, "msg", suggestion)
191 key = _cache_key("content", "E001", 10, "msg")
192 cache_file = root / CACHE_DIR / f"{key}.json"
194 # Set mtime to the past
195 os.utime(cache_file, (1000, 1000))
196 old_mtime = cache_file.stat().st_mtime
198 # Access the entry
199 result = get_cached_suggestion(root, "content", "E001", 10, "msg")
200 assert_that(result).is_not_none()
202 new_mtime = cache_file.stat().st_mtime
203 assert_that(new_mtime).is_greater_than(old_mtime)