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 duplicates.get_canonical("https://example.com/dup") == "https://example.com/canonical"
195
196 # Remove duplicate
197 result = store.remove_duplicate("https://example.com/dup")
198 assert result is True
199
200 # Check duplicate was removed
201 duplicates = store.get_duplicates()
202 assert len(duplicates.duplicates) == 0
203 assert not duplicates.is_duplicate("https://example.com/dup")
204
205 def test_search_entries(self, temp_dir):
206 """Test searching entries."""
207 store = GitStore(temp_dir / "test_repo")
208
209 # Add user
210 store.add_user("testuser")
211
212 # Add entries with different content
213 entries_data = [
214 ("Test Python Programming", "Learning Python basics"),
215 ("JavaScript Tutorial", "Advanced JavaScript concepts"),
216 ("Python Web Development", "Building web apps with Python"),
217 ]
218
219 for title, summary in entries_data:
220 entry = AtomEntry(
221 id=f"https://example.com/entry/{title.lower().replace(' ', '-')}",
222 title=title,
223 link=HttpUrl(f"https://example.com/entry/{title.lower().replace(' ', '-')}"),
224 updated=datetime.now(),
225 summary=summary,
226 )
227 store.store_entry("testuser", entry)
228
229 # Search for Python entries
230 results = store.search_entries("Python")
231 assert len(results) == 2
232
233 # Search for specific user
234 results = store.search_entries("Python", username="testuser")
235 assert len(results) == 2
236
237 # Search with limit
238 results = store.search_entries("Python", limit=1)
239 assert len(results) == 1
240
241 # Search for non-existent term
242 results = store.search_entries("NonExistent")
243 assert len(results) == 0
244
245 def test_get_stats(self, temp_dir):
246 """Test getting repository statistics."""
247 store = GitStore(temp_dir / "test_repo")
248
249 # Get initial stats
250 stats = store.get_stats()
251 assert stats["total_users"] == 0
252 assert stats["total_entries"] == 0
253 assert stats["total_duplicates"] == 0
254
255 # Add user and entries
256 store.add_user("testuser")
257 for i in range(3):
258 entry = AtomEntry(
259 id=f"https://example.com/entry/{i}",
260 title=f"Test Entry {i}",
261 link=HttpUrl(f"https://example.com/entry/{i}"),
262 updated=datetime.now(),
263 )
264 store.store_entry("testuser", entry)
265
266 # Add duplicate
267 store.add_duplicate("https://example.com/dup", "https://example.com/canonical")
268
269 # Get updated stats
270 stats = store.get_stats()
271 assert stats["total_users"] == 1
272 assert stats["total_entries"] == 3
273 assert stats["total_duplicates"] == 1
274 assert "last_updated" in stats
275 assert "repository_size" in stats