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