Manage Atom feeds in a persistent git repository
anil.recoil.org
1"""Tests for Git store functionality."""
2
3import json
4from datetime import datetime
5
6from pydantic import HttpUrl
7
8from thicket.core.git_store import GitStore
9from thicket.models import AtomEntry, DuplicateMap, UserMetadata
10
11
12class TestGitStore:
13 """Test the GitStore class."""
14
15 def test_init_new_repo(self, temp_dir):
16 """Test initializing a new Git repository."""
17 repo_path = temp_dir / "test_repo"
18 store = GitStore(repo_path)
19
20 assert store.repo_path == repo_path
21 assert store.repo is not None
22 assert repo_path.exists()
23 assert (repo_path / ".git").exists()
24 assert (repo_path / "index.json").exists()
25 assert (repo_path / "duplicates.json").exists()
26
27 def test_init_existing_repo(self, temp_dir):
28 """Test initializing with existing repository."""
29 repo_path = temp_dir / "test_repo"
30
31 # Create first store
32 store1 = GitStore(repo_path)
33 store1.add_user("testuser", display_name="Test User")
34
35 # Create second store pointing to same repo
36 store2 = GitStore(repo_path)
37 user = store2.get_user("testuser")
38
39 assert user is not None
40 assert user.username == "testuser"
41 assert user.display_name == "Test User"
42
43 def test_add_user(self, temp_dir):
44 """Test adding a user to the Git store."""
45 store = GitStore(temp_dir / "test_repo")
46
47 user = store.add_user(
48 username="testuser",
49 display_name="Test User",
50 email="test@example.com",
51 homepage="https://example.com",
52 icon="https://example.com/icon.png",
53 feeds=["https://example.com/feed.xml"],
54 )
55
56 assert isinstance(user, UserMetadata)
57 assert user.username == "testuser"
58 assert user.display_name == "Test User"
59 assert user.email == "test@example.com"
60 assert user.homepage == "https://example.com"
61 assert user.icon == "https://example.com/icon.png"
62 assert user.feeds == ["https://example.com/feed.xml"]
63 assert user.directory == "testuser"
64
65 # Check that user directory was created
66 user_dir = store.repo_path / "testuser"
67 assert user_dir.exists()
68
69 # Check user exists in index
70 stored_user = store.get_user("testuser")
71 assert stored_user is not None
72 assert stored_user.username == "testuser"
73 assert stored_user.display_name == "Test User"
74
75 def test_get_user(self, temp_dir):
76 """Test getting user metadata."""
77 store = GitStore(temp_dir / "test_repo")
78
79 # Add user
80 store.add_user("testuser", display_name="Test User")
81
82 # Get user
83 user = store.get_user("testuser")
84 assert user is not None
85 assert user.username == "testuser"
86 assert user.display_name == "Test User"
87
88 # Try to get non-existent user
89 non_user = store.get_user("nonexistent")
90 assert non_user is None
91
92 def test_store_entry(self, temp_dir):
93 """Test storing an entry."""
94 store = GitStore(temp_dir / "test_repo")
95
96 # Add user first
97 store.add_user("testuser")
98
99 # Create test entry
100 entry = AtomEntry(
101 id="https://example.com/entry/1",
102 title="Test Entry",
103 link=HttpUrl("https://example.com/entry/1"),
104 updated=datetime.now(),
105 summary="Test entry summary",
106 content="<p>Test content</p>",
107 )
108
109 # Store entry
110 result = store.store_entry("testuser", entry)
111 assert result is True
112
113 # Check that entry file was created
114 user_dir = store.repo_path / "testuser"
115 entry_files = list(user_dir.glob("*.json"))
116 entry_files = [f for f in entry_files if f.name != "metadata.json"]
117 assert len(entry_files) == 1
118
119 # Check entry content
120 with open(entry_files[0]) as f:
121 stored_entry = json.load(f)
122 assert stored_entry["title"] == "Test Entry"
123 assert stored_entry["id"] == "https://example.com/entry/1"
124
125 def test_get_entry(self, temp_dir):
126 """Test retrieving an entry."""
127 store = GitStore(temp_dir / "test_repo")
128
129 # Add user and entry
130 store.add_user("testuser")
131 entry = AtomEntry(
132 id="https://example.com/entry/1",
133 title="Test Entry",
134 link=HttpUrl("https://example.com/entry/1"),
135 updated=datetime.now(),
136 )
137 store.store_entry("testuser", entry)
138
139 # Get entry
140 retrieved = store.get_entry("testuser", "https://example.com/entry/1")
141 assert retrieved is not None
142 assert retrieved.title == "Test Entry"
143 assert retrieved.id == "https://example.com/entry/1"
144
145 # Try to get non-existent entry
146 non_entry = store.get_entry("testuser", "https://example.com/nonexistent")
147 assert non_entry is None
148
149 def test_list_entries(self, temp_dir):
150 """Test listing entries for a user."""
151 store = GitStore(temp_dir / "test_repo")
152
153 # Add user
154 store.add_user("testuser")
155
156 # Add multiple entries
157 for i in range(3):
158 entry = AtomEntry(
159 id=f"https://example.com/entry/{i}",
160 title=f"Test Entry {i}",
161 link=HttpUrl(f"https://example.com/entry/{i}"),
162 updated=datetime.now(),
163 )
164 store.store_entry("testuser", entry)
165
166 # List all entries
167 entries = store.list_entries("testuser")
168 assert len(entries) == 3
169
170 # List with limit
171 limited = store.list_entries("testuser", limit=2)
172 assert len(limited) == 2
173
174 # List for non-existent user
175 none_entries = store.list_entries("nonexistent")
176 assert len(none_entries) == 0
177
178 def test_duplicates(self, temp_dir):
179 """Test duplicate management."""
180 store = GitStore(temp_dir / "test_repo")
181
182 # Get initial duplicates (should be empty)
183 duplicates = store.get_duplicates()
184 assert isinstance(duplicates, DuplicateMap)
185 assert len(duplicates.duplicates) == 0
186
187 # Add duplicate
188 store.add_duplicate("https://example.com/dup", "https://example.com/canonical")
189
190 # Check duplicate was added
191 duplicates = store.get_duplicates()
192 assert len(duplicates.duplicates) == 1
193 assert duplicates.is_duplicate("https://example.com/dup")
194 assert (
195 duplicates.get_canonical("https://example.com/dup")
196 == "https://example.com/canonical"
197 )
198
199 # Remove duplicate
200 result = store.remove_duplicate("https://example.com/dup")
201 assert result is True
202
203 # Check duplicate was removed
204 duplicates = store.get_duplicates()
205 assert len(duplicates.duplicates) == 0
206 assert not duplicates.is_duplicate("https://example.com/dup")
207
208 def test_search_entries(self, temp_dir):
209 """Test searching entries."""
210 store = GitStore(temp_dir / "test_repo")
211
212 # Add user
213 store.add_user("testuser")
214
215 # Add entries with different content
216 entries_data = [
217 ("Test Python Programming", "Learning Python basics"),
218 ("JavaScript Tutorial", "Advanced JavaScript concepts"),
219 ("Python Web Development", "Building web apps with Python"),
220 ]
221
222 for title, summary in entries_data:
223 entry = AtomEntry(
224 id=f"https://example.com/entry/{title.lower().replace(' ', '-')}",
225 title=title,
226 link=HttpUrl(
227 f"https://example.com/entry/{title.lower().replace(' ', '-')}"
228 ),
229 updated=datetime.now(),
230 summary=summary,
231 )
232 store.store_entry("testuser", entry)
233
234 # Search for Python entries
235 results = store.search_entries("Python")
236 assert len(results) == 2
237
238 # Search for specific user
239 results = store.search_entries("Python", username="testuser")
240 assert len(results) == 2
241
242 # Search with limit
243 results = store.search_entries("Python", limit=1)
244 assert len(results) == 1
245
246 # Search for non-existent term
247 results = store.search_entries("NonExistent")
248 assert len(results) == 0
249
250 def test_get_stats(self, temp_dir):
251 """Test getting repository statistics."""
252 store = GitStore(temp_dir / "test_repo")
253
254 # Get initial stats
255 stats = store.get_stats()
256 assert stats["total_users"] == 0
257 assert stats["total_entries"] == 0
258 assert stats["total_duplicates"] == 0
259
260 # Add user and entries
261 store.add_user("testuser")
262 for i in range(3):
263 entry = AtomEntry(
264 id=f"https://example.com/entry/{i}",
265 title=f"Test Entry {i}",
266 link=HttpUrl(f"https://example.com/entry/{i}"),
267 updated=datetime.now(),
268 )
269 store.store_entry("testuser", entry)
270
271 # Add duplicate
272 store.add_duplicate("https://example.com/dup", "https://example.com/canonical")
273
274 # Get updated stats
275 stats = store.get_stats()
276 assert stats["total_users"] == 1
277 assert stats["total_entries"] == 3
278 assert stats["total_duplicates"] == 1
279 assert "last_updated" in stats
280 assert "repository_size" in stats