-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathtest_summary_cache.py
More file actions
89 lines (74 loc) · 3.46 KB
/
Copy pathtest_summary_cache.py
File metadata and controls
89 lines (74 loc) · 3.46 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
"""Phase 3 — mtime-keyed summary disk cache."""
from __future__ import annotations
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if REPO_ROOT not in sys.path:
sys.path.insert(0, REPO_ROOT)
from services import summary_cache
from pathlib import Path
from services.summary_cache import (
fingerprint_workspace_storage,
get_cached_projects,
set_cached_projects,
)
class TestSummaryCache(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.cache_patch = patch.object(summary_cache, "CACHE_DIR", self.tmp.name)
self.cache_patch.start()
summary_cache.PROJECTS_CACHE_FILE = Path(self.tmp.name) / "projects.json"
def tearDown(self):
self.cache_patch.stop()
self.tmp.cleanup()
def test_cache_hit_when_fingerprint_unchanged(self):
fp = {"version": 1, "workspace_path": "/ws", "global_db_mtime_ns": 100}
projects = [{"id": "a", "name": "A", "conversationCount": 1, "lastModified": "x"}]
warnings: list = []
set_cached_projects(fp, projects, warnings)
hit = get_cached_projects(fp)
self.assertIsNotNone(hit)
assert hit is not None
self.assertEqual(hit[0], projects)
def test_cache_miss_when_fingerprint_changes(self):
fp1 = {"version": 1, "workspace_path": "/ws", "global_db_mtime_ns": 100}
fp2 = {**fp1, "global_db_mtime_ns": 101}
set_cached_projects(fp1, [{"id": "a"}], [])
self.assertIsNone(get_cached_projects(fp2))
def test_nocache_env(self):
with patch.dict(os.environ, {"CURSOR_CHAT_BROWSER_NOCACHE": "1"}):
self.assertTrue(summary_cache.nocache_enabled())
def test_fingerprint_includes_workspace_files(self):
with tempfile.TemporaryDirectory() as ws:
entry_dir = os.path.join(ws, "entry1")
os.makedirs(entry_dir)
db = os.path.join(entry_dir, "state.vscdb")
with open(db, "wb") as f:
f.write(b"x")
entries = [{"name": "entry1", "workspaceJsonPath": os.path.join(entry_dir, "workspace.json")}]
fp = fingerprint_workspace_storage(ws, entries, global_db_path=None, rules=[])
self.assertTrue(fp["workspace_files"])
def test_workspace_files_fingerprint_round_trip(self):
"""JSON cache round-trip must match freshly computed workspace_files fingerprints."""
with tempfile.TemporaryDirectory() as ws:
entry_dir = os.path.join(ws, "entry1")
os.makedirs(entry_dir)
db = os.path.join(entry_dir, "state.vscdb")
with open(db, "wb") as f:
f.write(b"x")
entries = [{"name": "entry1", "workspaceJsonPath": os.path.join(entry_dir, "workspace.json")}]
fp = fingerprint_workspace_storage(ws, entries, global_db_path=None, rules=[])
projects = [{"id": "a", "name": "A", "conversationCount": 1, "lastModified": "x"}]
warnings: list = []
set_cached_projects(fp, projects, warnings)
fp2 = fingerprint_workspace_storage(ws, entries, global_db_path=None, rules=[])
hit = get_cached_projects(fp2)
self.assertIsNotNone(hit, msg="cache miss after JSON round-trip of workspace_files")
assert hit is not None
self.assertEqual(hit[0], projects)
if __name__ == "__main__":
unittest.main()