This is page 2 of 2. Use http://codebase.md/chrisguidry/you-need-an-mcp?lines=true&page={x} to view the full context.
# Directory Structure
```
├── .envrc
├── .gitignore
├── .pre-commit-config.yaml
├── .python-version
├── CLAUDE.md
├── DESIGN.md
├── models.py
├── plans
│ └── caching.md
├── pyproject.toml
├── README.md
├── repository.py
├── server.py
├── tests
│ ├── assertions.py
│ ├── conftest.py
│ ├── test_accounts.py
│ ├── test_assertions.py
│ ├── test_budget_months.py
│ ├── test_categories.py
│ ├── test_payees.py
│ ├── test_repository.py
│ ├── test_scheduled_transactions.py
│ ├── test_transactions.py
│ ├── test_updates.py
│ └── test_utilities.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/tests/test_transactions.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for transaction-related functionality in YNAB MCP Server.
3 | """
4 |
5 | from datetime import date
6 | from unittest.mock import MagicMock
7 |
8 | import pytest
9 | import ynab
10 | from assertions import extract_response_data
11 | from conftest import create_ynab_transaction
12 | from fastmcp.client import Client, FastMCPTransport
13 | from fastmcp.exceptions import ToolError
14 |
15 |
16 | async def test_list_transactions_basic(
17 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
18 | ) -> None:
19 | """Test basic transaction listing without filters."""
20 |
21 | txn1 = create_ynab_transaction(
22 | id="txn-1",
23 | transaction_date=date(2024, 1, 15),
24 | amount=-50_000, # -$50.00 outflow
25 | memo="Grocery shopping",
26 | flag_color=ynab.TransactionFlagColor.RED,
27 | account_name="Checking",
28 | payee_id="payee-1",
29 | payee_name="Whole Foods",
30 | category_id="cat-1",
31 | category_name="Groceries",
32 | )
33 |
34 | txn2 = create_ynab_transaction(
35 | id="txn-2",
36 | transaction_date=date(2024, 1, 20),
37 | amount=-75_000, # -$75.00 outflow
38 | memo="Dinner",
39 | cleared=ynab.TransactionClearedStatus.UNCLEARED,
40 | account_name="Checking",
41 | payee_id="payee-2",
42 | payee_name="Restaurant XYZ",
43 | category_id="cat-2",
44 | category_name="Dining Out",
45 | )
46 |
47 | # Add a deleted transaction that should be filtered out
48 | txn_deleted = create_ynab_transaction(
49 | id="txn-deleted",
50 | transaction_date=date(2024, 1, 10),
51 | amount=-25_000,
52 | memo="Deleted transaction",
53 | account_name="Checking",
54 | payee_id="payee-3",
55 | payee_name="Store ABC",
56 | category_id="cat-1",
57 | category_name="Groceries",
58 | deleted=True, # Should be excluded
59 | )
60 |
61 | # Mock repository to return transactions
62 | mock_repository.get_transactions.return_value = [txn2, txn1, txn_deleted]
63 |
64 | result = await mcp_client.call_tool("list_transactions", {})
65 |
66 | response_data = extract_response_data(result)
67 |
68 | # Should have 2 transactions (deleted one excluded)
69 | assert len(response_data["transactions"]) == 2
70 |
71 | # Should be sorted by date descending
72 | assert response_data["transactions"][0]["id"] == "txn-2"
73 | assert response_data["transactions"][0]["date"] == "2024-01-20"
74 | assert response_data["transactions"][0]["amount"] == "-75"
75 | assert response_data["transactions"][0]["payee_name"] == "Restaurant XYZ"
76 | assert response_data["transactions"][0]["category_name"] == "Dining Out"
77 |
78 | assert response_data["transactions"][1]["id"] == "txn-1"
79 | assert response_data["transactions"][1]["date"] == "2024-01-15"
80 | assert response_data["transactions"][1]["amount"] == "-50"
81 | assert response_data["transactions"][1]["flag"] == "Red"
82 |
83 | # Check pagination
84 | assert response_data["pagination"]["total_count"] == 2
85 | assert response_data["pagination"]["has_more"] is False
86 |
87 |
88 | async def test_list_transactions_with_account_filter(
89 | mock_repository: MagicMock,
90 | mcp_client: Client[FastMCPTransport],
91 | ) -> None:
92 | """Test transaction listing filtered by account."""
93 | # Create transaction
94 | txn = create_ynab_transaction(
95 | id="txn-acc-1",
96 | transaction_date=date(2024, 2, 1),
97 | amount=-30_000,
98 | memo="Account filtered",
99 | account_id="acc-checking",
100 | account_name="Main Checking",
101 | payee_id="payee-1",
102 | payee_name="Store",
103 | category_id="cat-1",
104 | category_name="Shopping",
105 | )
106 |
107 | # Mock repository to return filtered transactions
108 | mock_repository.get_transactions_by_filters.return_value = [txn]
109 |
110 | result = await mcp_client.call_tool(
111 | "list_transactions", {"account_id": "acc-checking"}
112 | )
113 |
114 | response_data = extract_response_data(result)
115 | assert len(response_data["transactions"]) == 1
116 | assert response_data["transactions"][0]["account_id"] == "acc-checking"
117 |
118 | # Verify correct repository method was called
119 | mock_repository.get_transactions_by_filters.assert_called_once_with(
120 | account_id="acc-checking",
121 | category_id=None,
122 | payee_id=None,
123 | since_date=None,
124 | )
125 |
126 |
127 | async def test_list_transactions_with_amount_filters(
128 | mock_repository: MagicMock,
129 | mcp_client: Client[FastMCPTransport],
130 | ) -> None:
131 | """Test transaction listing with amount range filters."""
132 | # Create transactions with different amounts
133 | txn_small = create_ynab_transaction(
134 | id="txn-small",
135 | transaction_date=date(2024, 3, 1),
136 | amount=-25_000, # -$25
137 | memo="Small purchase",
138 | payee_id="payee-1",
139 | payee_name="Coffee Shop",
140 | category_id="cat-1",
141 | category_name="Dining Out",
142 | )
143 |
144 | txn_medium = create_ynab_transaction(
145 | id="txn-medium",
146 | transaction_date=date(2024, 3, 2),
147 | amount=-60_000, # -$60
148 | memo="Medium purchase",
149 | payee_id="payee-2",
150 | payee_name="Restaurant",
151 | category_id="cat-1",
152 | category_name="Dining Out",
153 | )
154 |
155 | txn_large = create_ynab_transaction(
156 | id="txn-large",
157 | transaction_date=date(2024, 3, 3),
158 | amount=-120_000, # -$120
159 | memo="Large purchase",
160 | payee_id="payee-3",
161 | payee_name="Electronics Store",
162 | category_id="cat-2",
163 | category_name="Shopping",
164 | )
165 |
166 | # Mock repository to return all transactions for filtering
167 | mock_repository.get_transactions.return_value = [txn_small, txn_medium, txn_large]
168 |
169 | # Test with min_amount filter (transactions >= -$50)
170 | result = await mcp_client.call_tool(
171 | "list_transactions",
172 | {
173 | "min_amount": -50.0 # -$50
174 | },
175 | )
176 |
177 | response_data = extract_response_data(result)
178 | assert response_data is not None
179 | # Should only include small transaction (-$25 > -$50)
180 | assert len(response_data["transactions"]) == 1
181 | assert response_data["transactions"][0]["id"] == "txn-small"
182 |
183 | # Test with max_amount filter (transactions <= -$100)
184 | result = await mcp_client.call_tool(
185 | "list_transactions",
186 | {
187 | "max_amount": -100.0 # -$100
188 | },
189 | )
190 |
191 | response_data = extract_response_data(result)
192 | assert response_data is not None
193 | # Should only include large transaction (-$120 < -$100)
194 | assert len(response_data["transactions"]) == 1
195 | assert response_data["transactions"][0]["id"] == "txn-large"
196 |
197 | # Test with both min and max filters
198 | result = await mcp_client.call_tool(
199 | "list_transactions",
200 | {
201 | "min_amount": -80.0, # >= -$80
202 | "max_amount": -40.0, # <= -$40
203 | },
204 | )
205 |
206 | response_data = extract_response_data(result)
207 | assert response_data is not None
208 | # Should only include medium transaction (-$60)
209 | assert len(response_data["transactions"]) == 1
210 | assert response_data["transactions"][0]["id"] == "txn-medium"
211 |
212 |
213 | async def test_list_transactions_with_subtransactions(
214 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
215 | ) -> None:
216 | """Test transaction listing with split transactions (subtransactions)."""
217 | sub1 = ynab.SubTransaction(
218 | id="sub-1",
219 | transaction_id="txn-split",
220 | amount=-30_000, # -$30
221 | memo="Groceries portion",
222 | payee_id=None,
223 | payee_name=None,
224 | category_id="cat-groceries",
225 | category_name="Groceries",
226 | transfer_account_id=None,
227 | transfer_transaction_id=None,
228 | deleted=False,
229 | )
230 |
231 | sub2 = ynab.SubTransaction(
232 | id="sub-2",
233 | transaction_id="txn-split",
234 | amount=-20_000, # -$20
235 | memo="Household items",
236 | payee_id=None,
237 | payee_name=None,
238 | category_id="cat-household",
239 | category_name="Household",
240 | transfer_account_id=None,
241 | transfer_transaction_id=None,
242 | deleted=False,
243 | )
244 |
245 | # Deleted subtransaction should be filtered out
246 | sub_deleted = ynab.SubTransaction(
247 | id="sub-deleted",
248 | transaction_id="txn-split",
249 | amount=-10_000,
250 | memo="Deleted sub",
251 | payee_id=None,
252 | payee_name=None,
253 | category_id="cat-other",
254 | category_name="Other",
255 | transfer_account_id=None,
256 | transfer_transaction_id=None,
257 | deleted=True,
258 | )
259 |
260 | # Create split transaction
261 | txn_split = create_ynab_transaction(
262 | id="txn-split",
263 | transaction_date=date(2024, 4, 1),
264 | amount=-50_000, # -$50 total
265 | memo="Split transaction at Target",
266 | payee_id="payee-target",
267 | payee_name="Target",
268 | category_id=None, # Split transactions don't have a single category
269 | category_name=None,
270 | subtransactions=[sub1, sub2, sub_deleted],
271 | )
272 |
273 | # Mock repository to return split transaction
274 | mock_repository.get_transactions.return_value = [txn_split]
275 |
276 | result = await mcp_client.call_tool("list_transactions", {})
277 |
278 | response_data = extract_response_data(result)
279 | assert response_data is not None
280 | assert len(response_data["transactions"]) == 1
281 |
282 | txn = response_data["transactions"][0]
283 | assert txn["id"] == "txn-split"
284 | assert txn["amount"] == "-50"
285 |
286 | # Should have 2 subtransactions (deleted one excluded)
287 | assert len(txn["subtransactions"]) == 2
288 | assert txn["subtransactions"][0]["id"] == "sub-1"
289 | assert txn["subtransactions"][0]["amount"] == "-30"
290 | assert txn["subtransactions"][0]["category_name"] == "Groceries"
291 | assert txn["subtransactions"][1]["id"] == "sub-2"
292 | assert txn["subtransactions"][1]["amount"] == "-20"
293 | assert txn["subtransactions"][1]["category_name"] == "Household"
294 |
295 |
296 | async def test_list_transactions_pagination(
297 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
298 | ) -> None:
299 | """Test transaction listing with pagination."""
300 | # Create many transactions to test pagination
301 | transactions = []
302 | for i in range(5):
303 | txn = create_ynab_transaction(
304 | id=f"txn-{i}",
305 | transaction_date=date(2024, 1, i + 1),
306 | amount=-10_000 * (i + 1),
307 | memo=f"Transaction {i}",
308 | payee_id=f"payee-{i}",
309 | payee_name=f"Store {i}",
310 | category_id="cat-1",
311 | category_name="Shopping",
312 | )
313 | transactions.append(txn)
314 |
315 | # Mock repository to return all transactions
316 | mock_repository.get_transactions.return_value = transactions
317 |
318 | # Test first page
319 | result = await mcp_client.call_tool("list_transactions", {"limit": 2, "offset": 0})
320 |
321 | response_data = extract_response_data(result)
322 | assert response_data is not None
323 | assert len(response_data["transactions"]) == 2
324 | assert response_data["pagination"]["total_count"] == 5
325 | assert response_data["pagination"]["has_more"] is True
326 |
327 | # Transactions should be sorted by date descending
328 | assert response_data["transactions"][0]["id"] == "txn-4"
329 | assert response_data["transactions"][1]["id"] == "txn-3"
330 |
331 | # Test second page
332 | result = await mcp_client.call_tool("list_transactions", {"limit": 2, "offset": 2})
333 |
334 | response_data = extract_response_data(result)
335 | assert response_data is not None
336 | assert len(response_data["transactions"]) == 2
337 | assert response_data["transactions"][0]["id"] == "txn-2"
338 | assert response_data["transactions"][1]["id"] == "txn-1"
339 |
340 |
341 | async def test_list_transactions_with_category_filter(
342 | mock_repository: MagicMock,
343 | mcp_client: Client[FastMCPTransport],
344 | ) -> None:
345 | """Test transaction listing filtered by category."""
346 |
347 | # Create transaction
348 | txn = create_ynab_transaction(
349 | id="txn-cat-1",
350 | transaction_date=date(2024, 2, 1),
351 | amount=-40_000,
352 | memo="Category filtered",
353 | payee_id="payee-1",
354 | payee_name="Store",
355 | category_id="cat-dining",
356 | category_name="Dining Out",
357 | )
358 |
359 | # Mock repository to return filtered transactions
360 | mock_repository.get_transactions_by_filters.return_value = [txn]
361 |
362 | result = await mcp_client.call_tool(
363 | "list_transactions", {"category_id": "cat-dining"}
364 | )
365 |
366 | response_data = extract_response_data(result)
367 | assert len(response_data["transactions"]) == 1
368 | assert response_data["transactions"][0]["category_id"] == "cat-dining"
369 |
370 | # Verify correct repository method was called
371 | mock_repository.get_transactions_by_filters.assert_called_once_with(
372 | account_id=None,
373 | category_id="cat-dining",
374 | payee_id=None,
375 | since_date=None,
376 | )
377 |
378 |
379 | async def test_list_transactions_with_payee_filter(
380 | mock_repository: MagicMock,
381 | mcp_client: Client[FastMCPTransport],
382 | ) -> None:
383 | """Test transaction listing filtered by payee."""
384 | # Create transaction
385 | txn = create_ynab_transaction(
386 | id="txn-payee-1",
387 | transaction_date=date(2024, 3, 1),
388 | amount=-80_000,
389 | memo="Payee filtered",
390 | payee_id="payee-amazon",
391 | payee_name="Amazon",
392 | category_id="cat-shopping",
393 | category_name="Shopping",
394 | )
395 |
396 | # Mock repository to return filtered transactions
397 | mock_repository.get_transactions_by_filters.return_value = [txn]
398 |
399 | result = await mcp_client.call_tool(
400 | "list_transactions", {"payee_id": "payee-amazon"}
401 | )
402 |
403 | response_data = extract_response_data(result)
404 | assert len(response_data["transactions"]) == 1
405 | assert response_data["transactions"][0]["payee_id"] == "payee-amazon"
406 |
407 | # Verify correct repository method was called
408 | mock_repository.get_transactions_by_filters.assert_called_once_with(
409 | account_id=None,
410 | category_id=None,
411 | payee_id="payee-amazon",
412 | since_date=None,
413 | )
414 |
415 |
416 | async def test_split_transaction_payee_inheritance(
417 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
418 | ) -> None:
419 | """Test that subtransactions inherit parent payee when their payee is null."""
420 | # Create subtransactions where payee is null (simulating API response issue)
421 | sub1 = ynab.SubTransaction(
422 | id="sub-1",
423 | transaction_id="txn-split",
424 | amount=-30_000,
425 | memo="Groceries portion",
426 | payee_id=None, # Null payee_id
427 | payee_name=None, # Null payee_name (the bug we're fixing)
428 | category_id="cat-groceries",
429 | category_name="Groceries",
430 | transfer_account_id=None,
431 | transfer_transaction_id=None,
432 | deleted=False,
433 | )
434 |
435 | sub2 = ynab.SubTransaction(
436 | id="sub-2",
437 | transaction_id="txn-split",
438 | amount=-20_000,
439 | memo="Household items",
440 | payee_id=None, # Null payee_id
441 | payee_name=None, # Null payee_name (the bug we're fixing)
442 | category_id="cat-household",
443 | category_name="Household",
444 | transfer_account_id=None,
445 | transfer_transaction_id=None,
446 | deleted=False,
447 | )
448 |
449 | # Create parent transaction with valid payee (what user sees in YNAB interface)
450 | txn_split = create_ynab_transaction(
451 | id="txn-split",
452 | transaction_date=date(2024, 8, 11),
453 | amount=-50_000,
454 | memo="Split transaction at Walmart",
455 | payee_id="payee-walmart",
456 | payee_name="Walmart", # Parent has payee name
457 | category_id=None, # Split transactions don't have single category
458 | category_name=None,
459 | subtransactions=[sub1, sub2],
460 | )
461 |
462 | # Mock repository to return split transaction
463 | mock_repository.get_transactions.return_value = [txn_split]
464 |
465 | result = await mcp_client.call_tool("list_transactions", {})
466 |
467 | response_data = extract_response_data(result)
468 | assert response_data is not None
469 | assert len(response_data["transactions"]) == 1
470 |
471 | txn = response_data["transactions"][0]
472 | assert txn["id"] == "txn-split"
473 | assert txn["payee_name"] == "Walmart" # Parent should have payee name
474 | assert txn["payee_id"] == "payee-walmart"
475 |
476 | # Both subtransactions should inherit parent payee info
477 | assert len(txn["subtransactions"]) == 2
478 |
479 | assert txn["subtransactions"][0]["id"] == "sub-1"
480 | assert txn["subtransactions"][0]["payee_name"] == "Walmart" # Inherited!
481 | assert txn["subtransactions"][0]["payee_id"] == "payee-walmart" # Inherited!
482 |
483 | assert txn["subtransactions"][1]["id"] == "sub-2"
484 | assert txn["subtransactions"][1]["payee_name"] == "Walmart" # Inherited!
485 | assert txn["subtransactions"][1]["payee_id"] == "payee-walmart" # Inherited!
486 |
487 |
488 | async def test_hybrid_transaction_subtransaction_payee_resolution(
489 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
490 | ) -> None:
491 | """Test HybridTransaction subtransactions that need parent payee resolution."""
492 | # Create a HybridTransaction subtransaction (like from filtered API)
493 | from datetime import date
494 |
495 | from ynab.models.hybrid_transaction import HybridTransaction
496 |
497 | # This simulates what we get from get_transactions_by_filters()
498 | hybrid_subtxn = HybridTransaction(
499 | id="28a0ce46-a33b-4c3b-bcfc-633a05d9f9ec",
500 | date=date(2025, 8, 11),
501 | amount=-239660, # $239.66 in milliunits
502 | memo=None,
503 | cleared=ynab.TransactionClearedStatus.RECONCILED,
504 | approved=True,
505 | flag_color=None,
506 | flag_name=None,
507 | account_id="914dcb14-13da-49d2-86de-ba241c48f047",
508 | account_name="American Express",
509 | payee_id=None, # Missing payee info (the bug)
510 | payee_name=None, # Missing payee info (the bug)
511 | category_id="cd7c0b0e-7895-4f9f-aa1e-b6e0a22020cd",
512 | category_name="Groceries",
513 | transfer_account_id=None,
514 | transfer_transaction_id=None,
515 | matched_transaction_id=None,
516 | import_id="YNAB:-339660:2025-08-11:1",
517 | import_payee_name=None,
518 | import_payee_name_original=None,
519 | debt_transaction_type=None,
520 | deleted=False,
521 | type="subtransaction", # This is key!
522 | parent_transaction_id="5db47639-1867-41df-a807-23cc23b0ffe9",
523 | )
524 |
525 | # Create the parent transaction that has the payee info
526 | parent_txn = create_ynab_transaction(
527 | id="5db47639-1867-41df-a807-23cc23b0ffe9",
528 | transaction_date=date(2025, 8, 11),
529 | amount=-339660, # Total amount
530 | memo=None,
531 | payee_id="payee-walmart",
532 | payee_name="Walmart", # Parent has the payee name
533 | category_id=None,
534 | category_name="Split",
535 | )
536 |
537 | # Mock repository to return both transactions
538 | mock_repository.get_transactions_by_filters.return_value = [hybrid_subtxn]
539 | mock_repository.get_transaction_by_id.return_value = parent_txn # For parent lookup
540 |
541 | result = await mcp_client.call_tool(
542 | "list_transactions", {"category_id": "cd7c0b0e-7895-4f9f-aa1e-b6e0a22020cd"}
543 | )
544 |
545 | response_data = extract_response_data(result)
546 | assert response_data is not None
547 | assert len(response_data["transactions"]) == 1
548 |
549 | txn = response_data["transactions"][0]
550 | assert txn["id"] == "28a0ce46-a33b-4c3b-bcfc-633a05d9f9ec"
551 | assert txn["amount"] == "-239.66"
552 |
553 | # The key test: should have resolved parent payee info
554 | assert txn["payee_name"] == "Walmart" # Resolved from parent!
555 | assert txn["payee_id"] == "payee-walmart" # Resolved from parent!
556 | assert (
557 | txn["parent_transaction_id"] == "5db47639-1867-41df-a807-23cc23b0ffe9"
558 | ) # Should surface parent ID
559 |
560 |
561 | async def test_hybrid_transaction_with_missing_parent(
562 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
563 | ) -> None:
564 | """Test HybridTransaction subtransaction when parent is not found."""
565 | from datetime import date
566 |
567 | from ynab.models.hybrid_transaction import HybridTransaction
568 |
569 | # Create a HybridTransaction subtransaction with non-existent parent
570 | hybrid_subtxn = HybridTransaction(
571 | id="orphan-subtxn",
572 | date=date(2025, 8, 11),
573 | amount=-50000,
574 | memo=None,
575 | cleared=ynab.TransactionClearedStatus.CLEARED,
576 | approved=True,
577 | flag_color=None,
578 | flag_name=None,
579 | account_id="acc-1",
580 | account_name="Test Account",
581 | payee_id=None,
582 | payee_name=None,
583 | category_id="cat-1",
584 | category_name="Test Category",
585 | transfer_account_id=None,
586 | transfer_transaction_id=None,
587 | matched_transaction_id=None,
588 | import_id=None,
589 | import_payee_name=None,
590 | import_payee_name_original=None,
591 | debt_transaction_type=None,
592 | deleted=False,
593 | type="subtransaction",
594 | parent_transaction_id="nonexistent-parent-id", # Parent doesn't exist
595 | )
596 |
597 | # Mock repository - parent transaction not found
598 | mock_repository.get_transactions_by_filters.return_value = [hybrid_subtxn]
599 | # Mock get_transaction_by_id to raise an exception (transaction not found)
600 | mock_repository.get_transaction_by_id.side_effect = Exception(
601 | "Transaction not found"
602 | )
603 |
604 | # Should raise an exception when parent lookup fails
605 | with pytest.raises(ToolError, match="Transaction not found"):
606 | await mcp_client.call_tool("list_transactions", {"category_id": "cat-1"})
607 |
608 |
609 | async def test_hybrid_transaction_parent_resolver_exception(
610 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
611 | ) -> None:
612 | """Test HybridTransaction when parent resolver throws exception."""
613 | from datetime import date
614 |
615 | from ynab.models.hybrid_transaction import HybridTransaction
616 |
617 | hybrid_subtxn = HybridTransaction(
618 | id="exception-subtxn",
619 | date=date(2025, 8, 11),
620 | amount=-75000,
621 | memo=None,
622 | cleared=ynab.TransactionClearedStatus.CLEARED,
623 | approved=True,
624 | flag_color=None,
625 | flag_name=None,
626 | account_id="acc-1",
627 | account_name="Test Account",
628 | payee_id=None,
629 | payee_name=None,
630 | category_id="cat-1",
631 | category_name="Test Category",
632 | transfer_account_id=None,
633 | transfer_transaction_id=None,
634 | matched_transaction_id=None,
635 | import_id=None,
636 | import_payee_name=None,
637 | import_payee_name_original=None,
638 | debt_transaction_type=None,
639 | deleted=False,
640 | type="subtransaction",
641 | parent_transaction_id="exception-parent-id",
642 | )
643 |
644 | # Mock repository to return the subtransaction
645 | mock_repository.get_transactions_by_filters.return_value = [hybrid_subtxn]
646 | # Make get_transaction_by_id raise an exception to test exception handling
647 | mock_repository.get_transaction_by_id.side_effect = Exception("Database error")
648 |
649 | # Should raise an exception when parent lookup fails
650 | with pytest.raises(ToolError, match="Database error"):
651 | await mcp_client.call_tool("list_transactions", {"category_id": "cat-1"})
652 |
653 |
654 | async def test_hybrid_transaction_parent_with_null_payee(
655 | mock_repository: MagicMock, mcp_client: Client[FastMCPTransport]
656 | ) -> None:
657 | """Test HybridTransaction when parent transaction also has null payee."""
658 | from datetime import date
659 |
660 | from ynab.models.hybrid_transaction import HybridTransaction
661 |
662 | hybrid_subtxn = HybridTransaction(
663 | id="null-payee-subtxn",
664 | date=date(2025, 8, 11),
665 | amount=-80000,
666 | memo=None,
667 | cleared=ynab.TransactionClearedStatus.CLEARED,
668 | approved=True,
669 | flag_color=None,
670 | flag_name=None,
671 | account_id="acc-1",
672 | account_name="Test Account",
673 | payee_id=None,
674 | payee_name=None,
675 | category_id="cat-1",
676 | category_name="Test Category",
677 | transfer_account_id=None,
678 | transfer_transaction_id=None,
679 | matched_transaction_id=None,
680 | import_id=None,
681 | import_payee_name=None,
682 | import_payee_name_original=None,
683 | debt_transaction_type=None,
684 | deleted=False,
685 | type="subtransaction",
686 | parent_transaction_id="null-payee-parent-id",
687 | )
688 |
689 | # Create parent transaction that also has null payee info
690 | parent_txn = create_ynab_transaction(
691 | id="null-payee-parent-id",
692 | transaction_date=date(2025, 8, 11),
693 | amount=-80000,
694 | memo=None,
695 | payee_id=None, # Parent also has null payee
696 | payee_name=None, # Parent also has null payee
697 | category_id=None,
698 | category_name="Split",
699 | )
700 |
701 | # Mock repository to return both
702 | mock_repository.get_transactions_by_filters.return_value = [hybrid_subtxn]
703 | mock_repository.get_transaction_by_id.return_value = (
704 | parent_txn # Parent found but has null payee
705 | )
706 |
707 | result = await mcp_client.call_tool("list_transactions", {"category_id": "cat-1"})
708 |
709 | response_data = extract_response_data(result)
710 | assert response_data is not None
711 | assert len(response_data["transactions"]) == 1
712 |
713 | txn = response_data["transactions"][0]
714 | assert txn["id"] == "null-payee-subtxn"
715 |
716 | # Should remain null when parent also has null payee
717 | assert txn["payee_name"] is None
718 | assert txn["payee_id"] is None
719 | assert txn["parent_transaction_id"] == "null-payee-parent-id"
720 |
```
--------------------------------------------------------------------------------
/server.py:
--------------------------------------------------------------------------------
```python
1 | import logging
2 | import os
3 | from collections.abc import Sequence
4 | from datetime import date, datetime
5 | from decimal import Decimal
6 | from typing import Literal
7 |
8 | import ynab
9 | from fastmcp import FastMCP
10 |
11 | from models import (
12 | Account,
13 | AccountsResponse,
14 | BudgetMonth,
15 | CategoriesResponse,
16 | Category,
17 | CategoryGroup,
18 | PaginationInfo,
19 | Payee,
20 | PayeesResponse,
21 | ScheduledTransaction,
22 | ScheduledTransactionsResponse,
23 | Transaction,
24 | TransactionsResponse,
25 | milliunits_to_currency,
26 | )
27 | from repository import YNABRepository
28 |
29 | # Configure logging
30 | logging.basicConfig(
31 | level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
32 | )
33 | logger = logging.getLogger(__name__)
34 |
35 | mcp = FastMCP[None](
36 | name="YNAB",
37 | instructions="""
38 | Gives you access to a user's YNAB budget, including accounts, categories, and
39 | transactions. If a user is ever asking about budgeting, their personal finances,
40 | banking, saving, or investing, their YNAB budget is very relevant to them.
41 | When the user asks about budget categories and "how much is left", they are
42 | talking about the current month.
43 |
44 | Budget categories are grouped into category groups, which are important groupings
45 | to the user and should be displayed in a hierarchical manner. Categories will have
46 | the category_group_name and category_group_id available.
47 |
48 | The server operates on a single budget configured via the YNAB_BUDGET environment
49 | variable. All tools work with this budget automatically.
50 | """,
51 | )
52 |
53 |
54 | # Load configuration at module import - fail fast if not configured
55 | BUDGET_ID = os.environ["YNAB_BUDGET"]
56 | ACCESS_TOKEN = os.environ["YNAB_ACCESS_TOKEN"]
57 | ynab_api_configuration = ynab.Configuration(access_token=ACCESS_TOKEN)
58 |
59 | # Initialize repository at module level
60 | _repository = YNABRepository(budget_id=BUDGET_ID, access_token=ACCESS_TOKEN)
61 |
62 |
63 | def _paginate_items[T](
64 | items: list[T], limit: int, offset: int
65 | ) -> tuple[list[T], PaginationInfo]:
66 | """Apply pagination to a list of items and return the page with pagination info."""
67 | total_count = len(items)
68 | start_index = offset
69 | end_index = min(offset + limit, total_count)
70 | items_page = items[start_index:end_index]
71 |
72 | has_more = end_index < total_count
73 |
74 | pagination = PaginationInfo(
75 | total_count=total_count,
76 | limit=limit,
77 | offset=offset,
78 | has_more=has_more,
79 | )
80 |
81 | return items_page, pagination
82 |
83 |
84 | def _filter_active_items[T](
85 | items: list[T],
86 | *,
87 | exclude_deleted: bool = True,
88 | exclude_hidden: bool = False,
89 | exclude_closed: bool = False,
90 | ) -> list[T]:
91 | """Filter items to exclude deleted/hidden/closed based on flags."""
92 | filtered = []
93 | for item in items:
94 | if exclude_deleted and getattr(item, "deleted", False):
95 | continue
96 | if exclude_hidden and getattr(item, "hidden", False):
97 | continue
98 | if exclude_closed and getattr(item, "closed", False):
99 | continue
100 | filtered.append(item)
101 | return filtered
102 |
103 |
104 | def _build_category_group_map(
105 | category_groups: list[ynab.CategoryGroupWithCategories],
106 | ) -> dict[str, str]:
107 | """Build a mapping of category_id to category_group_name."""
108 | mapping = {}
109 | for category_group in category_groups:
110 | for category in category_group.categories:
111 | mapping[category.id] = category_group.name
112 | return mapping
113 |
114 |
115 | def convert_month_to_date(
116 | month: date | Literal["current", "last", "next"],
117 | ) -> date:
118 | """Convert month parameter to appropriate date object for YNAB API.
119 |
120 | Args:
121 | month: Month in ISO format (date object), or "current", "last", "next" literals
122 |
123 | Returns:
124 | date object representing the first day of the specified month:
125 | - "current": first day of current month
126 | - "last": first day of previous month
127 | - "next": first day of next month
128 | - date object unchanged if already a date
129 | """
130 | if isinstance(month, date):
131 | return month
132 |
133 | today = datetime.now().date()
134 | year, month_num = today.year, today.month
135 |
136 | match month:
137 | case "current":
138 | return date(year, month_num, 1)
139 | case "last":
140 | return (
141 | date(year - 1, 12, 1)
142 | if month_num == 1
143 | else date(year, month_num - 1, 1)
144 | )
145 | case "next":
146 | return (
147 | date(year + 1, 1, 1)
148 | if month_num == 12
149 | else date(year, month_num + 1, 1)
150 | )
151 | case _:
152 | raise ValueError(f"Invalid month value: {month}")
153 |
154 |
155 | @mcp.tool()
156 | def list_accounts(
157 | limit: int = 100,
158 | offset: int = 0,
159 | ) -> AccountsResponse:
160 | """List accounts with pagination.
161 |
162 | Only returns open/active accounts. Closed accounts are excluded automatically.
163 |
164 | Args:
165 | limit: Maximum number of accounts to return per page (default: 100)
166 | offset: Number of accounts to skip for pagination (default: 0)
167 |
168 | Returns:
169 | AccountsResponse with accounts list and pagination information
170 | """
171 | # Get accounts from repository (handles sync automatically if needed)
172 | accounts = _repository.get_accounts()
173 |
174 | # Apply existing filtering and pagination logic
175 | active_accounts = _filter_active_items(accounts, exclude_closed=True)
176 | all_accounts = [Account.from_ynab(account) for account in active_accounts]
177 |
178 | accounts_page, pagination = _paginate_items(all_accounts, limit, offset)
179 |
180 | return AccountsResponse(accounts=accounts_page, pagination=pagination)
181 |
182 |
183 | @mcp.tool()
184 | def list_categories(
185 | limit: int = 50,
186 | offset: int = 0,
187 | ) -> CategoriesResponse:
188 | """List categories with pagination.
189 |
190 | Only returns active/visible categories. Hidden and deleted categories are excluded
191 | automatically.
192 |
193 | Args:
194 | limit: Maximum number of categories to return per page (default: 50)
195 | offset: Number of categories to skip for pagination (default: 0)
196 |
197 | Returns:
198 | CategoriesResponse with categories list and pagination information
199 | """
200 | category_groups = _repository.get_category_groups()
201 |
202 | all_categories = []
203 | for category_group in category_groups:
204 | active_categories = _filter_active_items(
205 | category_group.categories, exclude_hidden=True
206 | )
207 | for category in active_categories:
208 | all_categories.append(
209 | Category.from_ynab(category, category_group.name).model_dump()
210 | )
211 |
212 | categories_page, pagination = _paginate_items(all_categories, limit, offset)
213 |
214 | # Convert dict categories back to Category objects
215 | category_objects = [Category(**cat_dict) for cat_dict in categories_page]
216 |
217 | return CategoriesResponse(categories=category_objects, pagination=pagination)
218 |
219 |
220 | @mcp.tool()
221 | def list_category_groups() -> list[CategoryGroup]:
222 | """List category groups (lighter weight than full categories).
223 |
224 | Returns:
225 | List of category groups
226 | """
227 | category_groups = _repository.get_category_groups()
228 | active_groups = _filter_active_items(category_groups)
229 | groups = [
230 | CategoryGroup.from_ynab(category_group) for category_group in active_groups
231 | ]
232 |
233 | return groups
234 |
235 |
236 | @mcp.tool()
237 | def get_budget_month(
238 | month: date | Literal["current", "last", "next"] = "current",
239 | limit: int = 50,
240 | offset: int = 0,
241 | ) -> BudgetMonth:
242 | """Get budget data for a specific month including category budgets, activity, and
243 | balances with pagination.
244 |
245 | Only returns active/visible categories. Hidden and deleted categories are excluded
246 | automatically.
247 |
248 | Args:
249 | month: Specifies which budget month to retrieve:
250 | • "current": Current calendar month
251 | • "last": Previous calendar month
252 | • "next": Next calendar month
253 | • date object: Specific month (uses first day of month)
254 | Examples: "current", date(2024, 3, 1) for March 2024 (default: "current")
255 | limit: Maximum number of categories to return per page (default: 50)
256 | offset: Number of categories to skip for pagination (default: 0)
257 |
258 | Returns:
259 | BudgetMonth with month info, categories, and pagination
260 | """
261 | converted_month = convert_month_to_date(month)
262 | month_data = _repository.get_budget_month(converted_month)
263 |
264 | # Map category IDs to group names
265 | category_groups = _repository.get_category_groups()
266 | category_group_map = _build_category_group_map(category_groups)
267 | all_categories = []
268 |
269 | active_categories = _filter_active_items(month_data.categories, exclude_hidden=True)
270 | for category in active_categories:
271 | group_name = category_group_map.get(category.id)
272 | all_categories.append(Category.from_ynab(category, group_name))
273 |
274 | categories_page, pagination = _paginate_items(all_categories, limit, offset)
275 |
276 | return BudgetMonth(
277 | month=month_data.month,
278 | note=month_data.note,
279 | income=milliunits_to_currency(month_data.income),
280 | budgeted=milliunits_to_currency(month_data.budgeted),
281 | activity=milliunits_to_currency(month_data.activity),
282 | to_be_budgeted=milliunits_to_currency(month_data.to_be_budgeted),
283 | age_of_money=month_data.age_of_money,
284 | categories=categories_page,
285 | pagination=pagination,
286 | )
287 |
288 |
289 | @mcp.tool()
290 | def get_month_category_by_id(
291 | category_id: str,
292 | month: date | Literal["current", "last", "next"] = "current",
293 | ) -> Category:
294 | """Get a specific category's data for a specific month.
295 |
296 | Args:
297 | category_id: Unique identifier for the category (required)
298 | month: Specifies which budget month to retrieve:
299 | • "current": Current calendar month
300 | • "last": Previous calendar month
301 | • "next": Next calendar month
302 | • date object: Specific month (uses first day of month)
303 | Examples: "current", date(2024, 3, 1) for March 2024 (default: "current")
304 |
305 | Returns:
306 | Category with budget data for the specified month
307 | """
308 | converted_month = convert_month_to_date(month)
309 | category = _repository.get_month_category_by_id(converted_month, category_id)
310 |
311 | # Fetch category groups to get group name
312 | category_groups = _repository.get_category_groups()
313 | category_group_map = _build_category_group_map(category_groups)
314 | group_name = category_group_map.get(category_id)
315 |
316 | return Category.from_ynab(category, group_name)
317 |
318 |
319 | @mcp.tool()
320 | def list_transactions(
321 | account_id: str | None = None,
322 | category_id: str | None = None,
323 | payee_id: str | None = None,
324 | since_date: date | None = None,
325 | min_amount: Decimal | None = None,
326 | max_amount: Decimal | None = None,
327 | limit: int = 25,
328 | offset: int = 0,
329 | ) -> TransactionsResponse:
330 | """List transactions with powerful filtering options for financial analysis.
331 |
332 | This tool supports various filters that can be combined:
333 | - Filter by account to see transactions for a specific account
334 | - Filter by category to analyze spending in a category (e.g., "Dining Out")
335 | - Filter by payee to see all transactions with a specific merchant (e.g., "Amazon")
336 | - Filter by date range using since_date
337 | - Filter by amount range using min_amount and/or max_amount
338 |
339 | Example queries this tool can answer:
340 | - "Show me all transactions over $50 in Dining Out this year"
341 | → Use: category_id="cat_dining_out_id", min_amount=50.00,
342 | since_date=date(2024, 1, 1)
343 | - "How much have I spent at Amazon this month"
344 | → Use: payee_id="payee_amazon_id", since_date=date(2024, 12, 1)
345 | - "List recent transactions in my checking account"
346 | → Use: account_id="acc_checking_id"
347 |
348 | Args:
349 | account_id: Filter by specific account (optional)
350 | category_id: Filter by specific category (optional)
351 | payee_id: Filter by specific payee (optional)
352 | since_date: Only show transactions on or after this date. Accepts date objects
353 | in YYYY-MM-DD format (e.g., date(2024, 1, 1)) (optional)
354 | min_amount: Only show transactions with amount >= this value in currency units.
355 | Use negative values for outflows/expenses
356 | (e.g., -50.00 for $50+ expenses) (optional)
357 | max_amount: Only show transactions with amount <= this value in currency units.
358 | Use negative values for outflows/expenses
359 | (e.g., -10.00 for under $10 expenses) (optional)
360 | limit: Maximum number of transactions to return per page (default: 25)
361 | offset: Number of transactions to skip for pagination (default: 0)
362 |
363 | Returns:
364 | TransactionsResponse with filtered transactions and pagination info
365 | """
366 | # Use repository to get transactions with appropriate filtering
367 | transactions_data: Sequence[ynab.TransactionDetail | ynab.HybridTransaction]
368 | if account_id or category_id or payee_id or since_date:
369 | # Use filtered endpoint for specific filters
370 | transactions_data = _repository.get_transactions_by_filters(
371 | account_id=account_id,
372 | category_id=category_id,
373 | payee_id=payee_id,
374 | since_date=since_date,
375 | )
376 | else:
377 | # Use cached transactions for general queries
378 | transactions_data = _repository.get_transactions()
379 |
380 | active_transactions = _filter_active_items(list(transactions_data))
381 | all_transactions = []
382 | for txn in active_transactions:
383 | # Apply amount filters (check milliunits directly for efficiency)
384 | if (
385 | min_amount is not None
386 | and txn.amount is not None
387 | and txn.amount < (min_amount * 1000)
388 | ):
389 | continue
390 | if (
391 | max_amount is not None
392 | and txn.amount is not None
393 | and txn.amount > (max_amount * 1000)
394 | ):
395 | continue
396 |
397 | all_transactions.append(Transaction.from_ynab(txn, _repository))
398 |
399 | # Sort by date descending (most recent first)
400 | all_transactions.sort(key=lambda t: t.date, reverse=True)
401 |
402 | transactions_page, pagination = _paginate_items(all_transactions, limit, offset)
403 |
404 | return TransactionsResponse(transactions=transactions_page, pagination=pagination)
405 |
406 |
407 | @mcp.tool()
408 | def list_payees(
409 | limit: int = 50,
410 | offset: int = 0,
411 | ) -> PayeesResponse:
412 | """List payees for a specific budget with pagination.
413 |
414 | Payees are the entities you pay money to (merchants, people, companies, etc.).
415 | This tool helps you find payee IDs for filtering transactions or analyzing spending
416 | patterns. Only returns active payees. Deleted payees are excluded automatically.
417 |
418 | Example queries this tool can answer:
419 | - "List all my payees"
420 | - "Find the payee ID for Amazon"
421 | - "Show me all merchants I've paid"
422 |
423 | Args:
424 | limit: Maximum number of payees to return per page (default: 50)
425 | offset: Number of payees to skip for pagination (default: 0)
426 |
427 | Returns:
428 | PayeesResponse with payees list and pagination information
429 | """
430 | # Get payees from repository (syncs automatically if needed)
431 | payees = _repository.get_payees()
432 |
433 | active_payees = _filter_active_items(payees)
434 | all_payees = [Payee.from_ynab(payee) for payee in active_payees]
435 |
436 | # Sort by name for easier browsing
437 | all_payees.sort(key=lambda p: p.name.lower())
438 |
439 | payees_page, pagination = _paginate_items(all_payees, limit, offset)
440 |
441 | return PayeesResponse(payees=payees_page, pagination=pagination)
442 |
443 |
444 | @mcp.tool()
445 | def find_payee(
446 | name_search: str,
447 | limit: int = 10,
448 | ) -> PayeesResponse:
449 | """Find payees by searching for name substrings (case-insensitive).
450 |
451 | This tool is perfect for finding specific payees when you know part of their name.
452 | Much more efficient than paginating through all payees with list_payees.
453 | Only returns active payees. Deleted payees are excluded automatically.
454 |
455 | Example queries this tool can answer:
456 | - "Find Amazon payee ID" (use name_search="amazon")
457 | - "Show me all Starbucks locations" (use name_search="starbucks")
458 | - "Find payees with 'grocery' in the name" (use name_search="grocery")
459 |
460 | Args:
461 | name_search: Search term to match against payee names (case-insensitive
462 | substring match). Examples: "amazon", "starbucks", "grocery"
463 | limit: Maximum number of matching payees to return (default: 10)
464 |
465 | Returns:
466 | PayeesResponse with matching payees and pagination information
467 | """
468 | # Get payees from repository (syncs automatically if needed)
469 | payees = _repository.get_payees()
470 |
471 | active_payees = _filter_active_items(payees)
472 | search_term = name_search.lower().strip()
473 | matching_payees = [
474 | Payee.from_ynab(payee)
475 | for payee in active_payees
476 | if search_term in payee.name.lower()
477 | ]
478 |
479 | # Sort by name for easier browsing
480 | matching_payees.sort(key=lambda p: p.name.lower())
481 |
482 | # Apply limit (no offset since this is a search, not pagination)
483 | limited_payees = matching_payees[:limit]
484 |
485 | # Create pagination info showing search results
486 | total_count = len(matching_payees)
487 | has_more = len(matching_payees) > limit
488 |
489 | pagination = PaginationInfo(
490 | total_count=total_count,
491 | limit=limit,
492 | offset=0,
493 | has_more=has_more,
494 | )
495 |
496 | return PayeesResponse(payees=limited_payees, pagination=pagination)
497 |
498 |
499 | @mcp.tool()
500 | def list_scheduled_transactions(
501 | account_id: str | None = None,
502 | category_id: str | None = None,
503 | payee_id: str | None = None,
504 | frequency: str | None = None,
505 | upcoming_days: int | None = None,
506 | min_amount: Decimal | None = None,
507 | max_amount: Decimal | None = None,
508 | limit: int = 25,
509 | offset: int = 0,
510 | ) -> ScheduledTransactionsResponse:
511 | """List scheduled transactions with powerful filtering options for analysis.
512 |
513 | This tool supports various filters that can be combined:
514 | - Filter by account to see scheduled transactions for a specific account
515 | - Filter by category to analyze recurring spending (e.g., "Monthly Bills")
516 | - Filter by payee to see scheduled transactions (e.g., "Netflix")
517 | - Filter by frequency to find daily, weekly, monthly, etc. recurring transactions
518 | - Filter by upcoming_days to see what's scheduled in the next N days
519 | - Filter by amount range using min_amount and/or max_amount
520 |
521 | Example queries this tool can answer:
522 | - "Show me all monthly recurring expenses" (use frequency="monthly")
523 | - "What bills are due in the next 7 days?" (use upcoming_days=7)
524 | - "List all Netflix subscriptions" (use payee search first, then filter by payee_id)
525 | - "Show scheduled transactions over $100" (use min_amount=100)
526 |
527 | Args:
528 | account_id: Filter by specific account (optional)
529 | category_id: Filter by specific category (optional)
530 | payee_id: Filter by specific payee (optional)
531 | frequency: Filter by recurrence frequency. Valid values:
532 | • never, daily, weekly
533 | • everyOtherWeek, twiceAMonth, every4Weeks
534 | • monthly, everyOtherMonth, every3Months, every4Months
535 | • twiceAYear, yearly, everyOtherYear
536 | (optional)
537 | upcoming_days: Only show scheduled transactions with next occurrence
538 | within this many days (optional)
539 | min_amount: Only show scheduled transactions with amount >= this value
540 | in currency units. Use negative values for outflows/expenses
541 | (optional)
542 | max_amount: Only show scheduled transactions with amount <= this value
543 | in currency units. Use negative values for outflows/expenses
544 | (optional)
545 | limit: Maximum number of scheduled transactions to return per page (default: 25)
546 | offset: Number of scheduled transactions to skip for pagination (default: 0)
547 |
548 | Returns:
549 | ScheduledTransactionsResponse with filtered scheduled transactions and
550 | pagination info
551 | """
552 | scheduled_transactions_data = _repository.get_scheduled_transactions()
553 | active_scheduled_transactions = _filter_active_items(scheduled_transactions_data)
554 | all_scheduled_transactions = []
555 | for st in active_scheduled_transactions:
556 | # Apply filters
557 | if account_id and st.account_id != account_id:
558 | continue
559 | if category_id and st.category_id != category_id:
560 | continue
561 | if payee_id and st.payee_id != payee_id:
562 | continue
563 | if frequency and st.frequency != frequency:
564 | continue
565 |
566 | # Apply upcoming_days filter
567 | if upcoming_days is not None:
568 | days_until_next = (st.date_next - datetime.now().date()).days
569 | if days_until_next > upcoming_days:
570 | continue
571 |
572 | # Apply amount filters (check milliunits directly for efficiency)
573 | if min_amount is not None and st.amount < (min_amount * 1000):
574 | continue
575 | if max_amount is not None and st.amount > (max_amount * 1000):
576 | continue
577 |
578 | all_scheduled_transactions.append(ScheduledTransaction.from_ynab(st))
579 |
580 | # Sort by next date ascending (earliest scheduled first)
581 | all_scheduled_transactions.sort(key=lambda st: st.date_next)
582 |
583 | scheduled_transactions_page, pagination = _paginate_items(
584 | all_scheduled_transactions, limit, offset
585 | )
586 |
587 | return ScheduledTransactionsResponse(
588 | scheduled_transactions=scheduled_transactions_page, pagination=pagination
589 | )
590 |
591 |
592 | @mcp.tool()
593 | def update_category_budget(
594 | category_id: str,
595 | budgeted: Decimal,
596 | month: date | Literal["current", "last", "next"] = "current",
597 | ) -> Category:
598 | """Update the budgeted amount for a category in a specific month.
599 |
600 | This tool allows you to assign money to budget categories, which is essential
601 | for monthly budget maintenance and reallocation.
602 |
603 | IMPORTANT: For categories with NEED goals (refill up to X monthly), budget the
604 | full goal_target amount regardless of current balance. These goals expect the
605 | full target to be budgeted each month.
606 |
607 | Args:
608 | category_id: Unique identifier for the category to update (required)
609 | budgeted: Amount to budget for this category in currency units (required)
610 | month: Budget month to update:
611 | • "current": Current calendar month
612 | • "last": Previous calendar month
613 | • "next": Next calendar month
614 | • date object: Specific month (uses first day of month)
615 | (default: "current")
616 |
617 | Returns:
618 | Category with updated budget information
619 | """
620 | converted_month = convert_month_to_date(month)
621 |
622 | # Convert currency units to milliunits
623 | budgeted_milliunits = int(budgeted * 1000)
624 |
625 | # Use repository update method with cache invalidation
626 | updated_category = _repository.update_month_category(
627 | category_id, converted_month, budgeted_milliunits
628 | )
629 |
630 | # Get category group name for the response
631 | category_groups = _repository.get_category_groups()
632 | category_group_map = _build_category_group_map(category_groups)
633 | group_name = category_group_map.get(category_id)
634 |
635 | return Category.from_ynab(updated_category, group_name)
636 |
637 |
638 | @mcp.tool()
639 | def update_transaction(
640 | transaction_id: str,
641 | category_id: str | None = None,
642 | payee_id: str | None = None,
643 | memo: str | None = None,
644 | ) -> Transaction:
645 | """Update an existing transaction's details.
646 |
647 | This tool allows you to modify transaction properties, most commonly
648 | to assign the correct category to imported or uncategorized transactions.
649 |
650 | Args:
651 | transaction_id: Unique identifier for the transaction to update (required)
652 | category_id: Category ID to assign (optional)
653 | payee_id: Payee ID to assign (optional)
654 | memo: Transaction memo (optional)
655 |
656 | Returns:
657 | Transaction with updated information
658 | """
659 | # First, get the existing transaction to preserve its current values
660 | existing_txn = _repository.get_transaction_by_id(transaction_id)
661 |
662 | # Build the update data starting with existing transaction values
663 | update_data = {
664 | "account_id": existing_txn.account_id,
665 | "date": existing_txn.var_date, # ExistingTransaction uses 'date'
666 | "amount": existing_txn.amount,
667 | "payee_id": existing_txn.payee_id,
668 | "payee_name": existing_txn.payee_name,
669 | "category_id": existing_txn.category_id,
670 | "memo": existing_txn.memo,
671 | "cleared": existing_txn.cleared,
672 | "approved": existing_txn.approved,
673 | "flag_color": existing_txn.flag_color,
674 | "subtransactions": existing_txn.subtransactions,
675 | }
676 |
677 | # Apply only the fields we want to change
678 | if category_id is not None:
679 | update_data["category_id"] = category_id
680 | if payee_id is not None:
681 | update_data["payee_id"] = payee_id
682 | if memo is not None:
683 | update_data["memo"] = memo
684 |
685 | # Use repository update method with cache invalidation
686 | updated_transaction = _repository.update_transaction(transaction_id, update_data)
687 |
688 | return Transaction.from_ynab(updated_transaction, _repository)
689 |
```
--------------------------------------------------------------------------------
/tests/test_scheduled_transactions.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Tests for scheduled transaction functionality in YNAB MCP Server.
3 |
4 | Tests the list_scheduled_transactions tool with various filters and scenarios.
5 | """
6 |
7 | from datetime import date
8 | from unittest.mock import MagicMock
9 |
10 | import ynab
11 | from assertions import extract_response_data
12 | from fastmcp.client import Client, FastMCPTransport
13 |
14 |
15 | async def test_list_scheduled_transactions_basic(
16 | mock_repository: MagicMock,
17 | mcp_client: Client[FastMCPTransport],
18 | ) -> None:
19 | """Test basic scheduled transaction listing without filters."""
20 |
21 | st1 = ynab.ScheduledTransactionDetail(
22 | id="st-1",
23 | date_first=date(2024, 1, 1),
24 | date_next=date(2024, 2, 1),
25 | frequency="monthly",
26 | amount=-120_000, # -$120.00 outflow
27 | memo="Netflix subscription",
28 | flag_color=ynab.TransactionFlagColor.RED,
29 | flag_name="Entertainment",
30 | account_id="acc-1",
31 | account_name="Checking",
32 | payee_id="payee-1",
33 | payee_name="Netflix",
34 | category_id="cat-1",
35 | category_name="Entertainment",
36 | transfer_account_id=None,
37 | deleted=False,
38 | subtransactions=[],
39 | )
40 |
41 | st2 = ynab.ScheduledTransactionDetail(
42 | id="st-2",
43 | date_first=date(2024, 1, 15),
44 | date_next=date(2024, 1, 29),
45 | frequency="weekly",
46 | amount=-5_000, # -$5.00 outflow
47 | memo="Weekly coffee",
48 | flag_color=None,
49 | flag_name=None,
50 | account_id="acc-1",
51 | account_name="Checking",
52 | payee_id="payee-2",
53 | payee_name="Coffee Shop",
54 | category_id="cat-2",
55 | category_name="Dining Out",
56 | transfer_account_id=None,
57 | deleted=False,
58 | subtransactions=[],
59 | )
60 |
61 | # Add a deleted scheduled transaction that should be filtered out
62 | st_deleted = ynab.ScheduledTransactionDetail(
63 | id="st-deleted",
64 | date_first=date(2024, 1, 1),
65 | date_next=date(2024, 3, 1),
66 | frequency="monthly",
67 | amount=-50_000,
68 | memo="Deleted subscription",
69 | flag_color=None,
70 | flag_name=None,
71 | account_id="acc-1",
72 | account_name="Checking",
73 | payee_id="payee-3",
74 | payee_name="Old Service",
75 | category_id="cat-1",
76 | category_name="Entertainment",
77 | transfer_account_id=None,
78 | deleted=True, # Should be excluded
79 | subtransactions=[],
80 | )
81 |
82 | # Mock repository to return scheduled transactions
83 | mock_repository.get_scheduled_transactions.return_value = [
84 | st2,
85 | st1,
86 | st_deleted,
87 | ]
88 |
89 | result = await mcp_client.call_tool("list_scheduled_transactions", {})
90 |
91 | response_data = extract_response_data(result)
92 |
93 | # Should have 2 scheduled transactions (deleted one excluded)
94 | assert len(response_data["scheduled_transactions"]) == 2
95 |
96 | # Should be sorted by next date ascending (earliest scheduled first)
97 | assert response_data["scheduled_transactions"][0]["id"] == "st-2"
98 | assert response_data["scheduled_transactions"][0]["date_next"] == "2024-01-29"
99 | assert response_data["scheduled_transactions"][0]["frequency"] == "weekly"
100 | assert response_data["scheduled_transactions"][0]["amount"] == "-5"
101 | assert response_data["scheduled_transactions"][0]["payee_name"] == "Coffee Shop"
102 |
103 | assert response_data["scheduled_transactions"][1]["id"] == "st-1"
104 | assert response_data["scheduled_transactions"][1]["date_next"] == "2024-02-01"
105 | assert response_data["scheduled_transactions"][1]["frequency"] == "monthly"
106 | assert response_data["scheduled_transactions"][1]["amount"] == "-120"
107 | assert response_data["scheduled_transactions"][1]["flag"] == "Entertainment (Red)"
108 |
109 | # Check pagination
110 | assert response_data["pagination"]["total_count"] == 2
111 | assert response_data["pagination"]["has_more"] is False
112 |
113 |
114 | async def test_list_scheduled_transactions_with_frequency_filter(
115 | mock_repository: MagicMock,
116 | mcp_client: Client[FastMCPTransport],
117 | ) -> None:
118 | """Test scheduled transaction listing filtered by frequency."""
119 |
120 | st_monthly = ynab.ScheduledTransactionDetail(
121 | id="st-monthly",
122 | date_first=date(2024, 1, 1),
123 | date_next=date(2024, 2, 1),
124 | frequency="monthly",
125 | amount=-100_000,
126 | memo="Monthly bill",
127 | flag_color=None,
128 | flag_name=None,
129 | account_id="acc-1",
130 | account_name="Checking",
131 | payee_id="payee-1",
132 | payee_name="Electric Company",
133 | category_id="cat-1",
134 | category_name="Utilities",
135 | transfer_account_id=None,
136 | deleted=False,
137 | subtransactions=[],
138 | )
139 |
140 | st_weekly = ynab.ScheduledTransactionDetail(
141 | id="st-weekly",
142 | date_first=date(2024, 1, 8),
143 | date_next=date(2024, 1, 15),
144 | frequency="weekly",
145 | amount=-2_500,
146 | memo="Weekly groceries",
147 | flag_color=None,
148 | flag_name=None,
149 | account_id="acc-1",
150 | account_name="Checking",
151 | payee_id="payee-2",
152 | payee_name="Grocery Store",
153 | category_id="cat-2",
154 | category_name="Groceries",
155 | transfer_account_id=None,
156 | deleted=False,
157 | subtransactions=[],
158 | )
159 |
160 | # Mock repository to return scheduled transactions
161 | mock_repository.get_scheduled_transactions.return_value = [st_monthly, st_weekly]
162 |
163 | # Test filtering by monthly frequency
164 | result = await mcp_client.call_tool(
165 | "list_scheduled_transactions", {"frequency": "monthly"}
166 | )
167 |
168 | response_data = extract_response_data(result)
169 |
170 | # Should only have the monthly scheduled transaction
171 | assert len(response_data["scheduled_transactions"]) == 1
172 | assert response_data["scheduled_transactions"][0]["id"] == "st-monthly"
173 | assert response_data["scheduled_transactions"][0]["frequency"] == "monthly"
174 | assert (
175 | response_data["scheduled_transactions"][0]["payee_name"] == "Electric Company"
176 | )
177 |
178 | # Check pagination
179 | assert response_data["pagination"]["total_count"] == 1
180 | assert response_data["pagination"]["has_more"] is False
181 |
182 |
183 | async def test_list_scheduled_transactions_with_upcoming_days_filter(
184 | mock_repository: MagicMock,
185 | mcp_client: Client[FastMCPTransport],
186 | ) -> None:
187 | """Test scheduled transaction listing filtered by upcoming days."""
188 |
189 | # Scheduled for 5 days from now
190 | st_soon = ynab.ScheduledTransactionDetail(
191 | id="st-soon",
192 | date_first=date(2024, 1, 1),
193 | date_next=date(2024, 1, 20), # 5 days from "today" (2024-01-15)
194 | frequency="monthly",
195 | amount=-50_000,
196 | memo="Due soon",
197 | flag_color=None,
198 | flag_name=None,
199 | account_id="acc-1",
200 | account_name="Checking",
201 | payee_id="payee-1",
202 | payee_name="Due Soon Co",
203 | category_id="cat-1",
204 | category_name="Bills",
205 | transfer_account_id=None,
206 | deleted=False,
207 | subtransactions=[],
208 | )
209 |
210 | # Scheduled for 15 days from now
211 | st_later = ynab.ScheduledTransactionDetail(
212 | id="st-later",
213 | date_first=date(2024, 1, 1),
214 | date_next=date(2024, 1, 30), # 15 days from "today" (2024-01-15)
215 | frequency="monthly",
216 | amount=-75_000,
217 | memo="Due later",
218 | flag_color=None,
219 | flag_name=None,
220 | account_id="acc-1",
221 | account_name="Checking",
222 | payee_id="payee-2",
223 | payee_name="Due Later Co",
224 | category_id="cat-1",
225 | category_name="Bills",
226 | transfer_account_id=None,
227 | deleted=False,
228 | subtransactions=[],
229 | )
230 |
231 | # Mock repository to return scheduled transactions
232 | mock_repository.get_scheduled_transactions.return_value = [st_soon, st_later]
233 |
234 | # Mock datetime.now() to return a fixed date for testing
235 | from unittest.mock import patch
236 |
237 | import server
238 |
239 | with patch.object(server, "datetime") as mock_datetime:
240 | mock_datetime.now.return_value.date.return_value = date(2024, 1, 15)
241 |
242 | # Test filtering by upcoming 7 days
243 | result = await mcp_client.call_tool(
244 | "list_scheduled_transactions", {"upcoming_days": 7}
245 | )
246 |
247 | response_data = extract_response_data(result)
248 |
249 | # Should only have the transaction due within 7 days
250 | assert len(response_data["scheduled_transactions"]) == 1
251 | assert response_data["scheduled_transactions"][0]["id"] == "st-soon"
252 | assert response_data["scheduled_transactions"][0]["payee_name"] == "Due Soon Co"
253 |
254 |
255 | async def test_list_scheduled_transactions_with_amount_filter(
256 | mock_repository: MagicMock,
257 | mcp_client: Client[FastMCPTransport],
258 | ) -> None:
259 | """Test scheduled transaction listing filtered by amount range."""
260 |
261 | st_small = ynab.ScheduledTransactionDetail(
262 | id="st-small",
263 | date_first=date(2024, 1, 1),
264 | date_next=date(2024, 2, 1),
265 | frequency="monthly",
266 | amount=-1_000, # -$1.00
267 | memo="Small expense",
268 | flag_color=None,
269 | flag_name=None,
270 | account_id="acc-1",
271 | account_name="Checking",
272 | payee_id="payee-1",
273 | payee_name="Small Store",
274 | category_id="cat-1",
275 | category_name="Misc",
276 | transfer_account_id=None,
277 | deleted=False,
278 | subtransactions=[],
279 | )
280 |
281 | st_large = ynab.ScheduledTransactionDetail(
282 | id="st-large",
283 | date_first=date(2024, 1, 1),
284 | date_next=date(2024, 2, 1),
285 | frequency="monthly",
286 | amount=-500_000, # -$500.00
287 | memo="Large expense",
288 | flag_color=None,
289 | flag_name=None,
290 | account_id="acc-1",
291 | account_name="Checking",
292 | payee_id="payee-2",
293 | payee_name="Large Store",
294 | category_id="cat-1",
295 | category_name="Bills",
296 | transfer_account_id=None,
297 | deleted=False,
298 | subtransactions=[],
299 | )
300 |
301 | # Mock repository to return scheduled transactions
302 | mock_repository.get_scheduled_transactions.return_value = [st_small, st_large]
303 |
304 | # Test filtering by minimum amount (expenses <= -$10, i.e., larger expenses)
305 | result = await mcp_client.call_tool(
306 | "list_scheduled_transactions", {"max_amount": -10}
307 | )
308 |
309 | response_data = extract_response_data(result)
310 |
311 | # Should only have the large transaction (<= -$10)
312 | assert len(response_data["scheduled_transactions"]) == 1
313 | assert response_data["scheduled_transactions"][0]["id"] == "st-large"
314 | assert response_data["scheduled_transactions"][0]["amount"] == "-500"
315 |
316 |
317 | async def test_list_scheduled_transactions_with_account_filter(
318 | mock_repository: MagicMock,
319 | mcp_client: Client[FastMCPTransport],
320 | ) -> None:
321 | """Test scheduled transaction listing filtered by account."""
322 |
323 | st_checking = ynab.ScheduledTransactionDetail(
324 | id="st-checking",
325 | date_first=date(2024, 1, 1),
326 | date_next=date(2024, 2, 1),
327 | frequency="monthly",
328 | amount=-100_000,
329 | memo="Checking account expense",
330 | flag_color=None,
331 | flag_name=None,
332 | account_id="acc-checking",
333 | account_name="Checking",
334 | payee_id="payee-1",
335 | payee_name="Merchant A",
336 | category_id="cat-1",
337 | category_name="Bills",
338 | transfer_account_id=None,
339 | deleted=False,
340 | subtransactions=[],
341 | )
342 |
343 | st_savings = ynab.ScheduledTransactionDetail(
344 | id="st-savings",
345 | date_first=date(2024, 1, 1),
346 | date_next=date(2024, 2, 1),
347 | frequency="monthly",
348 | amount=-50_000,
349 | memo="Savings account expense",
350 | flag_color=None,
351 | flag_name=None,
352 | account_id="acc-savings",
353 | account_name="Savings",
354 | payee_id="payee-2",
355 | payee_name="Merchant B",
356 | category_id="cat-1",
357 | category_name="Bills",
358 | transfer_account_id=None,
359 | deleted=False,
360 | subtransactions=[],
361 | )
362 |
363 | # Mock repository to return scheduled transactions
364 | mock_repository.get_scheduled_transactions.return_value = [st_checking, st_savings]
365 |
366 | # Test filtering by checking account
367 | result = await mcp_client.call_tool(
368 | "list_scheduled_transactions", {"account_id": "acc-checking"}
369 | )
370 |
371 | response_data = extract_response_data(result)
372 |
373 | # Should only have the checking account scheduled transaction
374 | assert len(response_data["scheduled_transactions"]) == 1
375 | assert response_data["scheduled_transactions"][0]["id"] == "st-checking"
376 | assert response_data["scheduled_transactions"][0]["account_name"] == "Checking"
377 |
378 |
379 | async def test_list_scheduled_transactions_with_category_filter(
380 | mock_repository: MagicMock,
381 | mcp_client: Client[FastMCPTransport],
382 | ) -> None:
383 | """Test scheduled transaction listing filtered by category."""
384 |
385 | st_bills = ynab.ScheduledTransactionDetail(
386 | id="st-bills",
387 | date_first=date(2024, 1, 1),
388 | date_next=date(2024, 2, 1),
389 | frequency="monthly",
390 | amount=-100_000,
391 | memo="Monthly bill",
392 | flag_color=None,
393 | flag_name=None,
394 | account_id="acc-1",
395 | account_name="Checking",
396 | payee_id="payee-1",
397 | payee_name="Utility Co",
398 | category_id="cat-bills",
399 | category_name="Bills",
400 | transfer_account_id=None,
401 | deleted=False,
402 | subtransactions=[],
403 | )
404 |
405 | st_entertainment = ynab.ScheduledTransactionDetail(
406 | id="st-entertainment",
407 | date_first=date(2024, 1, 1),
408 | date_next=date(2024, 2, 1),
409 | frequency="monthly",
410 | amount=-1_500,
411 | memo="Entertainment subscription",
412 | flag_color=None,
413 | flag_name=None,
414 | account_id="acc-1",
415 | account_name="Checking",
416 | payee_id="payee-2",
417 | payee_name="Streaming Service",
418 | category_id="cat-entertainment",
419 | category_name="Entertainment",
420 | transfer_account_id=None,
421 | deleted=False,
422 | subtransactions=[],
423 | )
424 |
425 | # Mock repository to return scheduled transactions
426 | mock_repository.get_scheduled_transactions.return_value = [
427 | st_bills,
428 | st_entertainment,
429 | ]
430 |
431 | # Test filtering by bills category
432 | result = await mcp_client.call_tool(
433 | "list_scheduled_transactions", {"category_id": "cat-bills"}
434 | )
435 |
436 | response_data = extract_response_data(result)
437 |
438 | # Should only have the bills category scheduled transaction
439 | assert len(response_data["scheduled_transactions"]) == 1
440 | assert response_data["scheduled_transactions"][0]["id"] == "st-bills"
441 | assert response_data["scheduled_transactions"][0]["category_name"] == "Bills"
442 |
443 |
444 | async def test_list_scheduled_transactions_with_min_amount_filter(
445 | mock_repository: MagicMock,
446 | mcp_client: Client[FastMCPTransport],
447 | ) -> None:
448 | """Test scheduled transaction listing filtered by minimum amount."""
449 |
450 | st_small = ynab.ScheduledTransactionDetail(
451 | id="st-small",
452 | date_first=date(2024, 1, 1),
453 | date_next=date(2024, 2, 1),
454 | frequency="monthly",
455 | amount=-1_000, # -$1.00
456 | memo="Small expense",
457 | flag_color=None,
458 | flag_name=None,
459 | account_id="acc-1",
460 | account_name="Checking",
461 | payee_id="payee-1",
462 | payee_name="Small Store",
463 | category_id="cat-1",
464 | category_name="Misc",
465 | transfer_account_id=None,
466 | deleted=False,
467 | subtransactions=[],
468 | )
469 |
470 | st_large = ynab.ScheduledTransactionDetail(
471 | id="st-large",
472 | date_first=date(2024, 1, 1),
473 | date_next=date(2024, 2, 1),
474 | frequency="monthly",
475 | amount=-500_000, # -$500.00
476 | memo="Large expense",
477 | flag_color=None,
478 | flag_name=None,
479 | account_id="acc-1",
480 | account_name="Checking",
481 | payee_id="payee-2",
482 | payee_name="Large Store",
483 | category_id="cat-1",
484 | category_name="Bills",
485 | transfer_account_id=None,
486 | deleted=False,
487 | subtransactions=[],
488 | )
489 |
490 | # Mock repository to return scheduled transactions
491 | mock_repository.get_scheduled_transactions.return_value = [st_small, st_large]
492 |
493 | # Test filtering by minimum amount (only expenses >= -$5, excludes -$500)
494 | result = await mcp_client.call_tool(
495 | "list_scheduled_transactions", {"min_amount": -5}
496 | )
497 |
498 | response_data = extract_response_data(result)
499 |
500 | # Should only have the small transaction (>= -$5)
501 | assert len(response_data["scheduled_transactions"]) == 1
502 | assert response_data["scheduled_transactions"][0]["id"] == "st-small"
503 | assert response_data["scheduled_transactions"][0]["amount"] == "-1"
504 |
505 |
506 | async def test_list_scheduled_transactions_with_payee_filter(
507 | mock_repository: MagicMock,
508 | mcp_client: Client[FastMCPTransport],
509 | ) -> None:
510 | """Test scheduled transaction listing filtered by payee."""
511 |
512 | st_netflix = ynab.ScheduledTransactionDetail(
513 | id="st-netflix",
514 | date_first=date(2024, 1, 1),
515 | date_next=date(2024, 2, 1),
516 | frequency="monthly",
517 | amount=-1_500,
518 | memo="Netflix subscription",
519 | flag_color=None,
520 | flag_name=None,
521 | account_id="acc-1",
522 | account_name="Checking",
523 | payee_id="payee-netflix",
524 | payee_name="Netflix",
525 | category_id="cat-1",
526 | category_name="Entertainment",
527 | transfer_account_id=None,
528 | deleted=False,
529 | subtransactions=[],
530 | )
531 |
532 | st_spotify = ynab.ScheduledTransactionDetail(
533 | id="st-spotify",
534 | date_first=date(2024, 1, 1),
535 | date_next=date(2024, 2, 1),
536 | frequency="monthly",
537 | amount=-1_000,
538 | memo="Spotify subscription",
539 | flag_color=None,
540 | flag_name=None,
541 | account_id="acc-1",
542 | account_name="Checking",
543 | payee_id="payee-spotify",
544 | payee_name="Spotify",
545 | category_id="cat-1",
546 | category_name="Entertainment",
547 | transfer_account_id=None,
548 | deleted=False,
549 | subtransactions=[],
550 | )
551 |
552 | # Mock repository to return scheduled transactions
553 | mock_repository.get_scheduled_transactions.return_value = [st_netflix, st_spotify]
554 |
555 | # Test filtering by Netflix payee
556 | result = await mcp_client.call_tool(
557 | "list_scheduled_transactions", {"payee_id": "payee-netflix"}
558 | )
559 |
560 | response_data = extract_response_data(result)
561 |
562 | # Should only have the Netflix scheduled transaction
563 | assert len(response_data["scheduled_transactions"]) == 1
564 | assert response_data["scheduled_transactions"][0]["id"] == "st-netflix"
565 | assert response_data["scheduled_transactions"][0]["payee_name"] == "Netflix"
566 |
567 |
568 | async def test_list_scheduled_transactions_pagination(
569 | mock_repository: MagicMock,
570 | mcp_client: Client[FastMCPTransport],
571 | ) -> None:
572 | """Test scheduled transaction listing with pagination."""
573 |
574 | # Create multiple scheduled transactions
575 | scheduled_transactions = []
576 | for i in range(15):
577 | st = ynab.ScheduledTransactionDetail(
578 | id=f"st-{i}",
579 | date_first=date(2024, 1, 1),
580 | date_next=date(2024, 2, i + 1), # Different next dates for sorting
581 | frequency="monthly",
582 | amount=-10_000 * (i + 1),
583 | memo=f"Transaction {i}",
584 | flag_color=None,
585 | flag_name=None,
586 | account_id="acc-1",
587 | account_name="Checking",
588 | payee_id=f"payee-{i}",
589 | payee_name=f"Payee {i}",
590 | category_id="cat-1",
591 | category_name="Bills",
592 | transfer_account_id=None,
593 | deleted=False,
594 | subtransactions=[],
595 | )
596 | scheduled_transactions.append(st)
597 |
598 | # Mock repository to return scheduled transactions
599 | mock_repository.get_scheduled_transactions.return_value = scheduled_transactions
600 |
601 | # Test first page with limit
602 | result = await mcp_client.call_tool(
603 | "list_scheduled_transactions", {"limit": 5, "offset": 0}
604 | )
605 |
606 | response_data = extract_response_data(result)
607 |
608 | # Should have 5 scheduled transactions
609 | assert len(response_data["scheduled_transactions"]) == 5
610 | assert response_data["pagination"]["total_count"] == 15
611 | assert response_data["pagination"]["has_more"] is True
612 |
613 | # Test second page
614 | result = await mcp_client.call_tool(
615 | "list_scheduled_transactions", {"limit": 5, "offset": 5}
616 | )
617 |
618 | response_data = extract_response_data(result)
619 |
620 | # Should have next 5 scheduled transactions
621 | assert len(response_data["scheduled_transactions"]) == 5
622 | assert response_data["pagination"]["total_count"] == 15
623 | assert response_data["pagination"]["has_more"] is True
624 |
625 |
626 | async def test_list_scheduled_transactions_with_subtransactions(
627 | mock_repository: MagicMock,
628 | mcp_client: Client[FastMCPTransport],
629 | ) -> None:
630 | """Test scheduled transaction listing with split transactions (subtransactions)."""
631 |
632 | # Create scheduled subtransactions
633 | sub1 = ynab.ScheduledSubTransaction(
634 | id="sub-1",
635 | scheduled_transaction_id="st-split",
636 | amount=-30_000, # -$30.00 for groceries
637 | memo="Groceries portion",
638 | payee_id="payee-1",
639 | payee_name="Grocery Store",
640 | category_id="cat-groceries",
641 | category_name="Groceries",
642 | transfer_account_id=None,
643 | deleted=False,
644 | )
645 |
646 | sub2 = ynab.ScheduledSubTransaction(
647 | id="sub-2",
648 | scheduled_transaction_id="st-split",
649 | amount=-20_000, # -$20.00 for household
650 | memo="Household portion",
651 | payee_id="payee-1",
652 | payee_name="Grocery Store",
653 | category_id="cat-household",
654 | category_name="Household",
655 | transfer_account_id=None,
656 | deleted=False,
657 | )
658 |
659 | st_split = ynab.ScheduledTransactionDetail(
660 | id="st-split",
661 | date_first=date(2024, 1, 1),
662 | date_next=date(2024, 2, 1),
663 | frequency="monthly",
664 | amount=-50_000, # -$50.00 total (should equal sum of subtransactions)
665 | memo="Split transaction",
666 | flag_color=None,
667 | flag_name=None,
668 | account_id="acc-1",
669 | account_name="Checking",
670 | payee_id="payee-1",
671 | payee_name="Grocery Store",
672 | category_id=None, # Split transactions don't have a main category
673 | category_name=None,
674 | transfer_account_id=None,
675 | deleted=False,
676 | subtransactions=[sub1, sub2],
677 | )
678 |
679 | # Mock repository to return scheduled transactions
680 | mock_repository.get_scheduled_transactions.return_value = [st_split]
681 |
682 | result = await mcp_client.call_tool("list_scheduled_transactions", {})
683 |
684 | response_data = extract_response_data(result)
685 |
686 | # Should have 1 scheduled transaction with subtransactions
687 | assert len(response_data["scheduled_transactions"]) == 1
688 | st = response_data["scheduled_transactions"][0]
689 |
690 | assert st["id"] == "st-split"
691 | assert st["amount"] == "-50"
692 | assert st["memo"] == "Split transaction"
693 |
694 | # Check subtransactions
695 | assert len(st["subtransactions"]) == 2
696 |
697 | assert st["subtransactions"][0]["id"] == "sub-1"
698 | assert st["subtransactions"][0]["amount"] == "-30"
699 | assert st["subtransactions"][0]["category_name"] == "Groceries"
700 |
701 | assert st["subtransactions"][1]["id"] == "sub-2"
702 | assert st["subtransactions"][1]["amount"] == "-20"
703 | assert st["subtransactions"][1]["category_name"] == "Household"
704 |
705 |
706 | async def test_list_scheduled_transactions_with_deleted_subtransactions(
707 | mock_repository: MagicMock,
708 | mcp_client: Client[FastMCPTransport],
709 | ) -> None:
710 | """Test scheduled transaction listing excludes deleted subtransactions."""
711 |
712 | # Create active and deleted scheduled subtransactions
713 | sub_active = ynab.ScheduledSubTransaction(
714 | id="sub-active",
715 | scheduled_transaction_id="st-mixed",
716 | amount=-30_000, # -$30.00
717 | memo="Active subtransaction",
718 | payee_id="payee-1",
719 | payee_name="Store",
720 | category_id="cat-1",
721 | category_name="Active Category",
722 | transfer_account_id=None,
723 | deleted=False,
724 | )
725 |
726 | sub_deleted = ynab.ScheduledSubTransaction(
727 | id="sub-deleted",
728 | scheduled_transaction_id="st-mixed",
729 | amount=-20_000, # -$20.00
730 | memo="Deleted subtransaction",
731 | payee_id="payee-1",
732 | payee_name="Store",
733 | category_id="cat-2",
734 | category_name="Deleted Category",
735 | transfer_account_id=None,
736 | deleted=True, # Should be excluded
737 | )
738 |
739 | st_mixed = ynab.ScheduledTransactionDetail(
740 | id="st-mixed",
741 | date_first=date(2024, 1, 1),
742 | date_next=date(2024, 2, 1),
743 | frequency="monthly",
744 | amount=-50_000, # -$50.00 total
745 | memo="Mixed subtransactions",
746 | flag_color=None,
747 | flag_name=None,
748 | account_id="acc-1",
749 | account_name="Checking",
750 | payee_id="payee-1",
751 | payee_name="Store",
752 | category_id=None,
753 | category_name=None,
754 | transfer_account_id=None,
755 | deleted=False,
756 | subtransactions=[sub_active, sub_deleted],
757 | )
758 |
759 | # Mock repository to return scheduled transactions
760 | mock_repository.get_scheduled_transactions.return_value = [st_mixed]
761 |
762 | result = await mcp_client.call_tool("list_scheduled_transactions", {})
763 |
764 | response_data = extract_response_data(result)
765 |
766 | # Should have 1 scheduled transaction with only active subtransactions
767 | assert len(response_data["scheduled_transactions"]) == 1
768 | st = response_data["scheduled_transactions"][0]
769 |
770 | assert st["id"] == "st-mixed"
771 | assert st["amount"] == "-50"
772 |
773 | # Should only have the active subtransaction (deleted one excluded)
774 | assert len(st["subtransactions"]) == 1
775 | assert st["subtransactions"][0]["id"] == "sub-active"
776 | assert st["subtransactions"][0]["category_name"] == "Active Category"
777 |
```
--------------------------------------------------------------------------------
/tests/test_repository.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Test suite for YNABRepository differential sync functionality.
3 | """
4 |
5 | import logging
6 | import threading
7 | import time
8 | from datetime import date, datetime, timedelta
9 | from typing import Any
10 | from unittest.mock import MagicMock, patch
11 |
12 | import pytest
13 | import ynab
14 | from conftest import create_ynab_account, create_ynab_payee
15 | from ynab.exceptions import ConflictException
16 |
17 | from repository import YNABRepository
18 |
19 |
20 | @pytest.fixture
21 | def repository() -> YNABRepository:
22 | """Create a repository instance for testing."""
23 | repo = YNABRepository(budget_id="test-budget", access_token="test-token")
24 |
25 | # Disable background sync by default to prevent real API calls during tests
26 | repo._background_sync_enabled = False
27 |
28 | return repo
29 |
30 |
31 | def test_repository_initial_sync(repository: YNABRepository) -> None:
32 | """Test repository initial sync without server knowledge."""
33 | account1 = create_ynab_account(id="acc-1", name="Checking")
34 | account2 = create_ynab_account(id="acc-2", name="Savings")
35 |
36 | accounts_response = ynab.AccountsResponse(
37 | data=ynab.AccountsResponseData(
38 | accounts=[account1, account2], server_knowledge=100
39 | )
40 | )
41 |
42 | with patch("ynab.ApiClient") as mock_client_class:
43 | mock_client = MagicMock()
44 | mock_client_class.return_value.__enter__.return_value = mock_client
45 |
46 | mock_accounts_api = MagicMock()
47 | mock_accounts_api.get_accounts.return_value = accounts_response
48 |
49 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
50 | repository.sync_accounts()
51 |
52 | # Verify initial sync called without last_knowledge_of_server
53 | mock_accounts_api.get_accounts.assert_called_once_with("test-budget")
54 |
55 | # Verify data was stored
56 | accounts = repository.get_accounts()
57 | assert len(accounts) == 2
58 | assert accounts[0].id == "acc-1"
59 | assert accounts[1].id == "acc-2"
60 |
61 | # Verify server knowledge was stored
62 | assert repository._server_knowledge["accounts"] == 100
63 | assert repository.is_initialized()
64 |
65 |
66 | def test_repository_delta_sync(repository: YNABRepository) -> None:
67 | """Test repository delta sync with server knowledge."""
68 | # Set up initial state
69 | account1 = create_ynab_account(id="acc-1", name="Checking")
70 | repository._data["accounts"] = [account1]
71 | repository._server_knowledge["accounts"] = 100
72 | repository._last_sync = datetime.now()
73 |
74 | # Delta sync with updated account and new account
75 | updated_account1 = create_ynab_account(id="acc-1", name="Updated Checking")
76 | new_account = create_ynab_account(id="acc-2", name="New Savings")
77 |
78 | delta_response = ynab.AccountsResponse(
79 | data=ynab.AccountsResponseData(
80 | accounts=[updated_account1, new_account], server_knowledge=110
81 | )
82 | )
83 |
84 | with patch("ynab.ApiClient") as mock_client_class:
85 | mock_client = MagicMock()
86 | mock_client_class.return_value.__enter__.return_value = mock_client
87 |
88 | mock_accounts_api = MagicMock()
89 | mock_accounts_api.get_accounts.return_value = delta_response
90 |
91 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
92 | repository.sync_accounts()
93 |
94 | # Verify delta sync called with last_knowledge_of_server
95 | mock_accounts_api.get_accounts.assert_called_once_with(
96 | "test-budget", last_knowledge_of_server=100
97 | )
98 |
99 | # Verify deltas were applied
100 | accounts = repository.get_accounts()
101 | assert len(accounts) == 2
102 |
103 | # Find accounts by ID
104 | acc1 = next(acc for acc in accounts if acc.id == "acc-1")
105 | acc2 = next(acc for acc in accounts if acc.id == "acc-2")
106 |
107 | assert acc1.name == "Updated Checking" # Updated
108 | assert acc2.name == "New Savings" # Added
109 |
110 | # Verify server knowledge was updated
111 | assert repository._server_knowledge["accounts"] == 110
112 |
113 |
114 | def test_repository_handles_deleted_accounts(repository: YNABRepository) -> None:
115 | """Test repository handles deleted accounts in delta sync."""
116 | # Set up initial state with two accounts
117 | account1 = create_ynab_account(id="acc-1", name="Checking")
118 | account2 = create_ynab_account(id="acc-2", name="Savings")
119 | repository._data["accounts"] = [account1, account2]
120 | repository._server_knowledge["accounts"] = 100
121 | repository._last_sync = datetime.now()
122 |
123 | # Delta with one deleted account
124 | deleted_account = create_ynab_account(id="acc-2", name="Savings", deleted=True)
125 |
126 | delta_response = ynab.AccountsResponse(
127 | data=ynab.AccountsResponseData(accounts=[deleted_account], server_knowledge=110)
128 | )
129 |
130 | with patch("ynab.ApiClient") as mock_client_class:
131 | mock_client = MagicMock()
132 | mock_client_class.return_value.__enter__.return_value = mock_client
133 |
134 | mock_accounts_api = MagicMock()
135 | mock_accounts_api.get_accounts.return_value = delta_response
136 |
137 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
138 | repository.sync_accounts()
139 |
140 | # Verify deleted account was removed
141 | accounts = repository.get_accounts()
142 | assert len(accounts) == 1
143 | assert accounts[0].id == "acc-1" # Only checking account remains
144 |
145 |
146 | def test_repository_fallback_to_full_refresh_on_error(
147 | repository: YNABRepository,
148 | ) -> None:
149 | """Test repository falls back to full refresh when delta sync fails."""
150 | # Set up initial state
151 | repository._server_knowledge["accounts"] = 100
152 | repository._last_sync = datetime.now()
153 |
154 | account1 = create_ynab_account(id="acc-1", name="Checking")
155 | account2 = create_ynab_account(id="acc-2", name="Savings")
156 |
157 | full_response = ynab.AccountsResponse(
158 | data=ynab.AccountsResponseData(
159 | accounts=[account1, account2], server_knowledge=120
160 | )
161 | )
162 |
163 | with patch("ynab.ApiClient") as mock_client_class:
164 | mock_client = MagicMock()
165 | mock_client_class.return_value.__enter__.return_value = mock_client
166 |
167 | mock_accounts_api = MagicMock()
168 | # First call (delta) raises API exception
169 | # Second call (full refresh) succeeds
170 | mock_accounts_api.get_accounts.side_effect = [
171 | ynab.ApiException(status=500, reason="Server Error"),
172 | full_response,
173 | ]
174 |
175 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
176 | repository.sync_accounts()
177 |
178 | # Verify two calls were made
179 | assert mock_accounts_api.get_accounts.call_count == 2
180 |
181 | # First call with server knowledge (delta attempt)
182 | first_call = mock_accounts_api.get_accounts.call_args_list[0]
183 | assert first_call[0] == ("test-budget",)
184 | assert first_call[1] == {"last_knowledge_of_server": 100}
185 |
186 | # Second call without server knowledge (full refresh)
187 | second_call = mock_accounts_api.get_accounts.call_args_list[1]
188 | assert second_call[0] == ("test-budget",)
189 | assert "last_knowledge_of_server" not in second_call[1]
190 |
191 | # Verify data was stored from full refresh
192 | accounts = repository.get_accounts()
193 | assert len(accounts) == 2
194 | assert repository._server_knowledge["accounts"] == 120
195 |
196 |
197 | def test_repository_lazy_initialization(repository: YNABRepository) -> None:
198 | """Test repository initializes automatically when data is requested."""
199 | account1 = create_ynab_account(id="acc-1", name="Checking")
200 |
201 | accounts_response = ynab.AccountsResponse(
202 | data=ynab.AccountsResponseData(accounts=[account1], server_knowledge=100)
203 | )
204 |
205 | with patch("ynab.ApiClient") as mock_client_class:
206 | mock_client = MagicMock()
207 | mock_client_class.return_value.__enter__.return_value = mock_client
208 |
209 | mock_accounts_api = MagicMock()
210 | mock_accounts_api.get_accounts.return_value = accounts_response
211 |
212 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
213 | # Repository is not initialized initially
214 | assert not repository.is_initialized()
215 |
216 | # Calling get_accounts should trigger sync
217 | accounts = repository.get_accounts()
218 |
219 | # Verify sync was called
220 | mock_accounts_api.get_accounts.assert_called_once()
221 |
222 | # Verify data is available
223 | assert len(accounts) == 1
224 | assert accounts[0].id == "acc-1"
225 | assert repository.is_initialized()
226 |
227 |
228 | def test_repository_thread_safety(repository: YNABRepository) -> None:
229 | """Test repository operations are thread-safe."""
230 | # This test verifies the locking mechanism works
231 | account1 = create_ynab_account(id="acc-1", name="Checking")
232 |
233 | accounts_response = ynab.AccountsResponse(
234 | data=ynab.AccountsResponseData(accounts=[account1], server_knowledge=100)
235 | )
236 |
237 | with patch("ynab.ApiClient") as mock_client_class:
238 | mock_client = MagicMock()
239 | mock_client_class.return_value.__enter__.return_value = mock_client
240 |
241 | mock_accounts_api = MagicMock()
242 | mock_accounts_api.get_accounts.return_value = accounts_response
243 |
244 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
245 | # Multiple calls should be safe
246 | repository.sync_accounts()
247 | accounts1 = repository.get_accounts()
248 | accounts2 = repository.get_accounts()
249 | last_sync1 = repository.last_sync_time()
250 | last_sync2 = repository.last_sync_time()
251 |
252 | # All operations should complete successfully
253 | assert len(accounts1) == 1
254 | assert len(accounts2) == 1
255 | assert last_sync1 is not None
256 | assert last_sync2 is not None
257 | assert last_sync1 == last_sync2
258 |
259 |
260 | def test_repository_payees_initial_sync(repository: YNABRepository) -> None:
261 | """Test repository initial sync for payees without server knowledge."""
262 | payee1 = create_ynab_payee(id="payee-1", name="Amazon")
263 | payee2 = create_ynab_payee(id="payee-2", name="Starbucks")
264 |
265 | payees_response = ynab.PayeesResponse(
266 | data=ynab.PayeesResponseData(payees=[payee1, payee2], server_knowledge=100)
267 | )
268 |
269 | with patch("ynab.ApiClient") as mock_client_class:
270 | mock_client = MagicMock()
271 | mock_client_class.return_value.__enter__.return_value = mock_client
272 |
273 | mock_payees_api = MagicMock()
274 | mock_payees_api.get_payees.return_value = payees_response
275 |
276 | with patch("ynab.PayeesApi", return_value=mock_payees_api):
277 | repository.sync_payees()
278 |
279 | # Verify initial sync called without last_knowledge_of_server
280 | mock_payees_api.get_payees.assert_called_once_with("test-budget")
281 |
282 | # Verify data was stored
283 | payees = repository.get_payees()
284 | assert len(payees) == 2
285 | assert payees[0].id == "payee-1"
286 | assert payees[1].id == "payee-2"
287 |
288 | # Verify server knowledge was stored
289 | assert repository._server_knowledge["payees"] == 100
290 |
291 |
292 | def test_repository_payees_delta_sync(repository: YNABRepository) -> None:
293 | """Test repository delta sync for payees with server knowledge."""
294 | # Set up initial state
295 | payee1 = create_ynab_payee(id="payee-1", name="Amazon")
296 | repository._data["payees"] = [payee1]
297 | repository._server_knowledge["payees"] = 100
298 | repository._last_sync = datetime.now()
299 |
300 | # Delta sync with updated payee and new payee
301 | updated_payee1 = create_ynab_payee(id="payee-1", name="Amazon.com")
302 | new_payee = create_ynab_payee(id="payee-2", name="Target")
303 |
304 | delta_response = ynab.PayeesResponse(
305 | data=ynab.PayeesResponseData(
306 | payees=[updated_payee1, new_payee], server_knowledge=110
307 | )
308 | )
309 |
310 | with patch("ynab.ApiClient") as mock_client_class:
311 | mock_client = MagicMock()
312 | mock_client_class.return_value.__enter__.return_value = mock_client
313 |
314 | mock_payees_api = MagicMock()
315 | mock_payees_api.get_payees.return_value = delta_response
316 |
317 | with patch("ynab.PayeesApi", return_value=mock_payees_api):
318 | repository.sync_payees()
319 |
320 | # Verify delta sync called with last_knowledge_of_server
321 | mock_payees_api.get_payees.assert_called_once_with(
322 | "test-budget", last_knowledge_of_server=100
323 | )
324 |
325 | # Verify deltas were applied
326 | payees = repository.get_payees()
327 | assert len(payees) == 2
328 |
329 | # Find payees by ID
330 | p1 = next(p for p in payees if p.id == "payee-1")
331 | p2 = next(p for p in payees if p.id == "payee-2")
332 |
333 | assert p1.name == "Amazon.com" # Updated
334 | assert p2.name == "Target" # Added
335 |
336 | # Verify server knowledge was updated
337 | assert repository._server_knowledge["payees"] == 110
338 |
339 |
340 | def test_repository_payees_handles_deleted(repository: YNABRepository) -> None:
341 | """Test repository handles deleted payees in delta sync."""
342 | # Set up initial state with two payees
343 | payee1 = create_ynab_payee(id="payee-1", name="Amazon")
344 | payee2 = create_ynab_payee(id="payee-2", name="Old Store")
345 | repository._data["payees"] = [payee1, payee2]
346 | repository._server_knowledge["payees"] = 100
347 | repository._last_sync = datetime.now()
348 |
349 | # Delta with one deleted payee
350 | deleted_payee = create_ynab_payee(id="payee-2", name="Old Store", deleted=True)
351 |
352 | delta_response = ynab.PayeesResponse(
353 | data=ynab.PayeesResponseData(payees=[deleted_payee], server_knowledge=110)
354 | )
355 |
356 | with patch("ynab.ApiClient") as mock_client_class:
357 | mock_client = MagicMock()
358 | mock_client_class.return_value.__enter__.return_value = mock_client
359 |
360 | mock_payees_api = MagicMock()
361 | mock_payees_api.get_payees.return_value = delta_response
362 |
363 | with patch("ynab.PayeesApi", return_value=mock_payees_api):
364 | repository.sync_payees()
365 |
366 | # Verify deleted payee was removed
367 | payees = repository.get_payees()
368 | assert len(payees) == 1
369 | assert payees[0].id == "payee-1" # Only Amazon remains
370 |
371 |
372 | def test_repository_payees_lazy_initialization(repository: YNABRepository) -> None:
373 | """Test payees repository initializes automatically when data is requested."""
374 | payee1 = create_ynab_payee(id="payee-1", name="Amazon")
375 |
376 | payees_response = ynab.PayeesResponse(
377 | data=ynab.PayeesResponseData(payees=[payee1], server_knowledge=100)
378 | )
379 |
380 | with patch("ynab.ApiClient") as mock_client_class:
381 | mock_client = MagicMock()
382 | mock_client_class.return_value.__enter__.return_value = mock_client
383 |
384 | mock_payees_api = MagicMock()
385 | mock_payees_api.get_payees.return_value = payees_response
386 |
387 | with patch("ynab.PayeesApi", return_value=mock_payees_api):
388 | # Repository payees is not initialized initially
389 | assert "payees" not in repository._data
390 |
391 | # Calling get_payees should trigger sync
392 | payees = repository.get_payees()
393 |
394 | # Verify sync was called
395 | mock_payees_api.get_payees.assert_called_once()
396 |
397 | # Verify data is available
398 | assert len(payees) == 1
399 | assert payees[0].id == "payee-1"
400 |
401 |
402 | def create_ynab_category_group(
403 | *,
404 | id: str = "group-1",
405 | name: str = "Test Group",
406 | deleted: bool = False,
407 | **kwargs: Any,
408 | ) -> ynab.CategoryGroupWithCategories:
409 | """Create a YNAB CategoryGroupWithCategories for testing with sensible defaults."""
410 | categories = kwargs.get("categories", [])
411 | return ynab.CategoryGroupWithCategories(
412 | id=id,
413 | name=name,
414 | hidden=kwargs.get("hidden", False),
415 | deleted=deleted,
416 | categories=categories,
417 | )
418 |
419 |
420 | def test_repository_category_groups_initial_sync(repository: YNABRepository) -> None:
421 | """Test repository initial sync for category groups without server knowledge."""
422 | group1 = create_ynab_category_group(id="group-1", name="Monthly Bills")
423 | group2 = create_ynab_category_group(id="group-2", name="Everyday Expenses")
424 |
425 | categories_response = ynab.CategoriesResponse(
426 | data=ynab.CategoriesResponseData(
427 | category_groups=[group1, group2], server_knowledge=100
428 | )
429 | )
430 |
431 | with patch("ynab.ApiClient") as mock_client_class:
432 | mock_client = MagicMock()
433 | mock_client_class.return_value.__enter__.return_value = mock_client
434 |
435 | mock_categories_api = MagicMock()
436 | mock_categories_api.get_categories.return_value = categories_response
437 |
438 | with patch("ynab.CategoriesApi", return_value=mock_categories_api):
439 | repository.sync_category_groups()
440 |
441 | # Verify initial sync called without last_knowledge_of_server
442 | mock_categories_api.get_categories.assert_called_once_with("test-budget")
443 |
444 | # Verify data was stored
445 | category_groups = repository.get_category_groups()
446 | assert len(category_groups) == 2
447 | assert category_groups[0].id == "group-1"
448 | assert category_groups[1].id == "group-2"
449 |
450 | # Verify server knowledge was stored
451 | assert repository._server_knowledge["category_groups"] == 100
452 |
453 |
454 | def test_repository_category_groups_delta_sync(repository: YNABRepository) -> None:
455 | """Test repository delta sync for category groups with server knowledge."""
456 | # Set up initial state
457 | group1 = create_ynab_category_group(id="group-1", name="Monthly Bills")
458 | repository._data["category_groups"] = [group1]
459 | repository._server_knowledge["category_groups"] = 100
460 | repository._last_sync = datetime.now()
461 |
462 | # Delta sync with updated group and new group
463 | updated_group1 = create_ynab_category_group(id="group-1", name="Fixed Expenses")
464 | new_group = create_ynab_category_group(id="group-2", name="Variable Expenses")
465 |
466 | delta_response = ynab.CategoriesResponse(
467 | data=ynab.CategoriesResponseData(
468 | category_groups=[updated_group1, new_group], server_knowledge=110
469 | )
470 | )
471 |
472 | with patch("ynab.ApiClient") as mock_client_class:
473 | mock_client = MagicMock()
474 | mock_client_class.return_value.__enter__.return_value = mock_client
475 |
476 | mock_categories_api = MagicMock()
477 | mock_categories_api.get_categories.return_value = delta_response
478 |
479 | with patch("ynab.CategoriesApi", return_value=mock_categories_api):
480 | repository.sync_category_groups()
481 |
482 | # Verify delta sync called with last_knowledge_of_server
483 | mock_categories_api.get_categories.assert_called_once_with(
484 | "test-budget", last_knowledge_of_server=100
485 | )
486 |
487 | # Verify deltas were applied
488 | category_groups = repository.get_category_groups()
489 | assert len(category_groups) == 2
490 |
491 | # Find groups by ID
492 | g1 = next(g for g in category_groups if g.id == "group-1")
493 | g2 = next(g for g in category_groups if g.id == "group-2")
494 |
495 | assert g1.name == "Fixed Expenses" # Updated
496 | assert g2.name == "Variable Expenses" # Added
497 |
498 | # Verify server knowledge was updated
499 | assert repository._server_knowledge["category_groups"] == 110
500 |
501 |
502 | def test_repository_category_groups_handles_deleted(repository: YNABRepository) -> None:
503 | """Test repository handles deleted category groups in delta sync."""
504 | # Set up initial state with two groups
505 | group1 = create_ynab_category_group(id="group-1", name="Monthly Bills")
506 | group2 = create_ynab_category_group(id="group-2", name="Old Category")
507 | repository._data["category_groups"] = [group1, group2]
508 | repository._server_knowledge["category_groups"] = 100
509 | repository._last_sync = datetime.now()
510 |
511 | # Delta with one deleted group
512 | deleted_group = create_ynab_category_group(
513 | id="group-2", name="Old Category", deleted=True
514 | )
515 |
516 | delta_response = ynab.CategoriesResponse(
517 | data=ynab.CategoriesResponseData(
518 | category_groups=[deleted_group], server_knowledge=110
519 | )
520 | )
521 |
522 | with patch("ynab.ApiClient") as mock_client_class:
523 | mock_client = MagicMock()
524 | mock_client_class.return_value.__enter__.return_value = mock_client
525 |
526 | mock_categories_api = MagicMock()
527 | mock_categories_api.get_categories.return_value = delta_response
528 |
529 | with patch("ynab.CategoriesApi", return_value=mock_categories_api):
530 | repository.sync_category_groups()
531 |
532 | # Verify deleted group was removed
533 | category_groups = repository.get_category_groups()
534 | assert len(category_groups) == 1
535 | assert category_groups[0].id == "group-1" # Only Monthly Bills remains
536 |
537 |
538 | def test_repository_category_groups_lazy_initialization(
539 | repository: YNABRepository,
540 | ) -> None:
541 | """Test category groups repository initializes automatically when data requested."""
542 | group1 = create_ynab_category_group(id="group-1", name="Monthly Bills")
543 |
544 | categories_response = ynab.CategoriesResponse(
545 | data=ynab.CategoriesResponseData(category_groups=[group1], server_knowledge=100)
546 | )
547 |
548 | with patch("ynab.ApiClient") as mock_client_class:
549 | mock_client = MagicMock()
550 | mock_client_class.return_value.__enter__.return_value = mock_client
551 |
552 | mock_categories_api = MagicMock()
553 | mock_categories_api.get_categories.return_value = categories_response
554 |
555 | with patch("ynab.CategoriesApi", return_value=mock_categories_api):
556 | # Repository category groups is not initialized initially
557 | assert "category_groups" not in repository._data
558 |
559 | # Calling get_category_groups should trigger sync
560 | category_groups = repository.get_category_groups()
561 |
562 | # Verify sync was called
563 | mock_categories_api.get_categories.assert_called_once()
564 |
565 | # Verify data is available
566 | assert len(category_groups) == 1
567 | assert category_groups[0].id == "group-1"
568 |
569 |
570 | def create_ynab_transaction(
571 | *,
572 | id: str = "txn-1",
573 | account_id: str = "acc-1",
574 | amount: int = -50_000, # -$50.00
575 | memo: str | None = "Test Transaction",
576 | cleared: str = "cleared",
577 | approved: bool = True,
578 | deleted: bool = False,
579 | **kwargs: Any,
580 | ) -> ynab.TransactionDetail:
581 | """Create a YNAB TransactionDetail for testing with sensible defaults."""
582 | return ynab.TransactionDetail(
583 | id=id,
584 | date=kwargs.get("date", date.today()),
585 | amount=amount,
586 | memo=memo,
587 | cleared=ynab.TransactionClearedStatus(cleared),
588 | approved=approved,
589 | flag_color=kwargs.get("flag_color"),
590 | account_id=account_id,
591 | account_name=kwargs.get("account_name", "Test Account"),
592 | payee_id=kwargs.get("payee_id"),
593 | payee_name=kwargs.get("payee_name", "Test Payee"),
594 | category_id=kwargs.get("category_id"),
595 | category_name=kwargs.get("category_name"),
596 | transfer_account_id=kwargs.get("transfer_account_id"),
597 | transfer_transaction_id=kwargs.get("transfer_transaction_id"),
598 | matched_transaction_id=kwargs.get("matched_transaction_id"),
599 | import_id=kwargs.get("import_id"),
600 | import_payee_name=kwargs.get("import_payee_name"),
601 | import_payee_name_original=kwargs.get("import_payee_name_original"),
602 | debt_transaction_type=kwargs.get("debt_transaction_type"),
603 | subtransactions=kwargs.get("subtransactions", []),
604 | deleted=deleted,
605 | )
606 |
607 |
608 | def test_repository_transactions_initial_sync(repository: YNABRepository) -> None:
609 | """Test repository initial sync for transactions without server knowledge."""
610 | txn1 = create_ynab_transaction(id="txn-1", amount=-25_000, memo="Groceries")
611 | txn2 = create_ynab_transaction(id="txn-2", amount=-15_000, memo="Gas")
612 |
613 | transactions_response = ynab.TransactionsResponse(
614 | data=ynab.TransactionsResponseData(
615 | transactions=[txn1, txn2], server_knowledge=100
616 | )
617 | )
618 |
619 | with patch("ynab.ApiClient") as mock_client_class:
620 | mock_client = MagicMock()
621 | mock_client_class.return_value.__enter__.return_value = mock_client
622 |
623 | mock_transactions_api = MagicMock()
624 | mock_transactions_api.get_transactions.return_value = transactions_response
625 |
626 | with patch("ynab.TransactionsApi", return_value=mock_transactions_api):
627 | repository.sync_transactions()
628 |
629 | # Verify initial sync called without last_knowledge_of_server
630 | mock_transactions_api.get_transactions.assert_called_once_with("test-budget")
631 |
632 | # Verify data was stored
633 | transactions = repository.get_transactions()
634 | assert len(transactions) == 2
635 | assert transactions[0].id == "txn-1"
636 | assert transactions[1].id == "txn-2"
637 |
638 | # Verify server knowledge was stored
639 | assert repository._server_knowledge["transactions"] == 100
640 |
641 |
642 | def test_repository_transactions_delta_sync(repository: YNABRepository) -> None:
643 | """Test repository delta sync for transactions with server knowledge."""
644 | # Set up initial state
645 | txn1 = create_ynab_transaction(id="txn-1", amount=-25_000, memo="Groceries")
646 | repository._data["transactions"] = [txn1]
647 | repository._server_knowledge["transactions"] = 100
648 | repository._last_sync = datetime.now()
649 |
650 | # Delta sync with updated transaction and new transaction
651 | updated_txn1 = create_ynab_transaction(
652 | id="txn-1", amount=-25_000, memo="Groceries (Updated)"
653 | )
654 | new_txn = create_ynab_transaction(id="txn-2", amount=-15_000, memo="Gas")
655 |
656 | delta_response = ynab.TransactionsResponse(
657 | data=ynab.TransactionsResponseData(
658 | transactions=[updated_txn1, new_txn], server_knowledge=110
659 | )
660 | )
661 |
662 | with patch("ynab.ApiClient") as mock_client_class:
663 | mock_client = MagicMock()
664 | mock_client_class.return_value.__enter__.return_value = mock_client
665 |
666 | mock_transactions_api = MagicMock()
667 | mock_transactions_api.get_transactions.return_value = delta_response
668 |
669 | with patch("ynab.TransactionsApi", return_value=mock_transactions_api):
670 | repository.sync_transactions()
671 |
672 | # Verify delta sync called with last_knowledge_of_server
673 | mock_transactions_api.get_transactions.assert_called_once_with(
674 | "test-budget", last_knowledge_of_server=100
675 | )
676 |
677 | # Verify deltas were applied
678 | transactions = repository.get_transactions()
679 | assert len(transactions) == 2
680 |
681 | # Find transactions by ID
682 | t1 = next(t for t in transactions if t.id == "txn-1")
683 | t2 = next(t for t in transactions if t.id == "txn-2")
684 |
685 | assert t1.memo == "Groceries (Updated)" # Updated
686 | assert t2.memo == "Gas" # Added
687 |
688 | # Verify server knowledge was updated
689 | assert repository._server_knowledge["transactions"] == 110
690 |
691 |
692 | def test_repository_transactions_handles_deleted(repository: YNABRepository) -> None:
693 | """Test repository handles deleted transactions in delta sync."""
694 | # Set up initial state with two transactions
695 | txn1 = create_ynab_transaction(id="txn-1", amount=-25_000, memo="Groceries")
696 | txn2 = create_ynab_transaction(id="txn-2", amount=-15_000, memo="Gas")
697 | repository._data["transactions"] = [txn1, txn2]
698 | repository._server_knowledge["transactions"] = 100
699 | repository._last_sync = datetime.now()
700 |
701 | # Delta with one deleted transaction
702 | deleted_txn = create_ynab_transaction(
703 | id="txn-2", amount=-15_000, memo="Gas", deleted=True
704 | )
705 |
706 | delta_response = ynab.TransactionsResponse(
707 | data=ynab.TransactionsResponseData(
708 | transactions=[deleted_txn], server_knowledge=110
709 | )
710 | )
711 |
712 | with patch("ynab.ApiClient") as mock_client_class:
713 | mock_client = MagicMock()
714 | mock_client_class.return_value.__enter__.return_value = mock_client
715 |
716 | mock_transactions_api = MagicMock()
717 | mock_transactions_api.get_transactions.return_value = delta_response
718 |
719 | with patch("ynab.TransactionsApi", return_value=mock_transactions_api):
720 | repository.sync_transactions()
721 |
722 | # Verify deleted transaction was removed
723 | transactions = repository.get_transactions()
724 | assert len(transactions) == 1
725 | assert transactions[0].id == "txn-1" # Only groceries transaction remains
726 |
727 |
728 | def test_repository_transactions_lazy_initialization(
729 | repository: YNABRepository,
730 | ) -> None:
731 | """Test transactions repository initializes automatically when data requested."""
732 | txn1 = create_ynab_transaction(id="txn-1", amount=-25_000, memo="Groceries")
733 |
734 | transactions_response = ynab.TransactionsResponse(
735 | data=ynab.TransactionsResponseData(transactions=[txn1], server_knowledge=100)
736 | )
737 |
738 | with patch("ynab.ApiClient") as mock_client_class:
739 | mock_client = MagicMock()
740 | mock_client_class.return_value.__enter__.return_value = mock_client
741 |
742 | mock_transactions_api = MagicMock()
743 | mock_transactions_api.get_transactions.return_value = transactions_response
744 |
745 | with patch("ynab.TransactionsApi", return_value=mock_transactions_api):
746 | # Repository transactions is not initialized initially
747 | assert "transactions" not in repository._data
748 |
749 | # Calling get_transactions should trigger sync
750 | transactions = repository.get_transactions()
751 |
752 | # Verify sync was called
753 | mock_transactions_api.get_transactions.assert_called_once()
754 |
755 | # Verify data is available
756 | assert len(transactions) == 1
757 | assert transactions[0].id == "txn-1"
758 |
759 |
760 | # ===== EDGE CASE AND ERROR HANDLING TESTS =====
761 |
762 |
763 | def test_repository_needs_sync_functionality(repository: YNABRepository) -> None:
764 | """Test needs_sync() method behavior."""
765 | # Fresh repository should need sync
766 | assert repository.needs_sync()
767 |
768 | # After setting last_sync to now, should not need sync
769 | repository._last_sync = datetime.now()
770 | assert not repository.needs_sync()
771 |
772 | # After 6 minutes, should need sync (default threshold is 5 minutes)
773 | repository._last_sync = datetime.now() - timedelta(minutes=6)
774 | assert repository.needs_sync()
775 |
776 | # Custom threshold - should not need sync at 3 minutes with 5 minute threshold
777 | repository._last_sync = datetime.now() - timedelta(minutes=3)
778 | assert not repository.needs_sync(max_age_minutes=5)
779 |
780 | # Custom threshold - should need sync at 3 minutes with 2 minute threshold
781 | assert repository.needs_sync(max_age_minutes=2)
782 |
783 |
784 | def test_repository_conflict_exception_fallback(repository: YNABRepository) -> None:
785 | """Test that ConflictException triggers fallback to full sync."""
786 | repository._server_knowledge["accounts"] = 100
787 | repository._last_sync = datetime.now()
788 |
789 | account1 = create_ynab_account(id="acc-1", name="Checking")
790 | full_response = ynab.AccountsResponse(
791 | data=ynab.AccountsResponseData(accounts=[account1], server_knowledge=120)
792 | )
793 |
794 | with patch("ynab.ApiClient") as mock_client_class:
795 | mock_client = MagicMock()
796 | mock_client_class.return_value.__enter__.return_value = mock_client
797 |
798 | mock_accounts_api = MagicMock()
799 | # First call (delta) raises ConflictException (409)
800 | # Second call (full refresh) succeeds
801 | mock_accounts_api.get_accounts.side_effect = [
802 | ConflictException(status=409, reason="Conflict"),
803 | full_response,
804 | ]
805 |
806 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
807 | repository.sync_accounts()
808 |
809 | # Verify fallback behavior
810 | assert mock_accounts_api.get_accounts.call_count == 2
811 | accounts = repository.get_accounts()
812 | assert len(accounts) == 1
813 | assert repository._server_knowledge["accounts"] == 120
814 |
815 |
816 | def test_repository_rate_limit_retry_behavior(repository: YNABRepository) -> None:
817 | """Test that 429 rate limit triggers retry with exponential backoff."""
818 | account1 = create_ynab_account(id="acc-1", name="Checking")
819 | success_response = ynab.AccountsResponse(
820 | data=ynab.AccountsResponseData(accounts=[account1], server_knowledge=100)
821 | )
822 |
823 | with patch("ynab.ApiClient") as mock_client_class:
824 | mock_client = MagicMock()
825 | mock_client_class.return_value.__enter__.return_value = mock_client
826 |
827 | mock_accounts_api = MagicMock()
828 | # First call raises 429, second call succeeds
829 | mock_accounts_api.get_accounts.side_effect = [
830 | ynab.ApiException(status=429, reason="Too Many Requests"),
831 | success_response,
832 | ]
833 |
834 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
835 | with patch("time.sleep") as mock_sleep:
836 | repository.sync_accounts()
837 |
838 | # Verify retry behavior
839 | assert mock_accounts_api.get_accounts.call_count == 2
840 | mock_sleep.assert_called_once_with(1) # First retry waits 2^0 = 1 second
841 | accounts = repository.get_accounts()
842 | assert len(accounts) == 1
843 |
844 |
845 | def test_repository_rate_limit_max_retries_exceeded(repository: YNABRepository) -> None:
846 | """Test that repeated 429s eventually give up after max retries."""
847 | with patch("ynab.ApiClient") as mock_client_class:
848 | mock_client = MagicMock()
849 | mock_client_class.return_value.__enter__.return_value = mock_client
850 |
851 | mock_accounts_api = MagicMock()
852 | # Always return 429
853 | mock_accounts_api.get_accounts.side_effect = ynab.ApiException(
854 | status=429, reason="Too Many Requests"
855 | )
856 |
857 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
858 | with patch("time.sleep") as mock_sleep:
859 | with pytest.raises(ynab.ApiException) as exc_info:
860 | repository.sync_accounts()
861 |
862 | # Verify max retries behavior (3 attempts total)
863 | assert mock_accounts_api.get_accounts.call_count == 3
864 | assert exc_info.value.status == 429
865 | # Should have called sleep twice (after first and second attempts)
866 | assert mock_sleep.call_count == 2
867 |
868 |
869 | def test_repository_unexpected_exception_not_caught(repository: YNABRepository) -> None:
870 | """Test that unexpected exceptions are re-raised, not silently caught."""
871 | with patch("ynab.ApiClient") as mock_client_class:
872 | mock_client = MagicMock()
873 | mock_client_class.return_value.__enter__.return_value = mock_client
874 |
875 | mock_accounts_api = MagicMock()
876 | # Raise a non-API exception
877 | mock_accounts_api.get_accounts.side_effect = ValueError("Unexpected error")
878 |
879 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
880 | with pytest.raises(ValueError) as exc_info:
881 | repository.sync_accounts()
882 |
883 | assert str(exc_info.value) == "Unexpected error"
884 |
885 |
886 | def test_repository_background_sync_not_blocking(repository: YNABRepository) -> None:
887 | """Test that background sync doesn't block data access."""
888 | # Enable background sync for this test
889 | repository._background_sync_enabled = True
890 |
891 | # Set up stale data
892 | account1 = create_ynab_account(id="acc-1", name="Checking")
893 | repository._data["accounts"] = [account1]
894 | repository._last_sync = datetime.now() - timedelta(minutes=10) # Stale
895 |
896 | # Mock needs_sync to return True (stale)
897 | with patch.object(repository, "needs_sync", return_value=True):
898 | with patch.object(repository, "_trigger_background_sync") as mock_bg_sync:
899 | # Getting accounts should return existing data immediately
900 | accounts = repository.get_accounts()
901 |
902 | # Verify we got the stale data instantly
903 | assert len(accounts) == 1
904 | assert accounts[0].id == "acc-1"
905 |
906 | # Verify background sync was triggered
907 | mock_bg_sync.assert_called_once_with("accounts")
908 |
909 |
910 | def test_repository_background_sync_error_handling(repository: YNABRepository) -> None:
911 | """Test that background sync errors don't crash or affect data access."""
912 | account1 = create_ynab_account(id="acc-1", name="Checking")
913 | repository._data["accounts"] = [account1]
914 | repository._last_sync = datetime.now() - timedelta(minutes=10) # Stale
915 |
916 | # Should still return existing data without crashing
917 | accounts = repository.get_accounts()
918 | assert len(accounts) == 1
919 | assert accounts[0].id == "acc-1"
920 |
921 |
922 | def test_repository_concurrent_access_safety(repository: YNABRepository) -> None:
923 | """Test that concurrent access to repository data is thread-safe."""
924 |
925 | # Set up initial data using the lock to ensure thread safety
926 | account1 = create_ynab_account(id="acc-1", name="Checking")
927 | with repository._lock:
928 | repository._data["accounts"] = [account1]
929 | repository._last_sync = datetime.now()
930 |
931 | results = []
932 | errors = []
933 | results_lock = threading.Lock()
934 |
935 | # Track which thread should fail for error coverage
936 | should_fail = [True] # Use list to make it mutable in closure
937 |
938 | def access_data() -> None:
939 | try:
940 | # Make the first thread fail to test exception handling
941 | if should_fail[0]:
942 | should_fail[0] = False # Only fail once
943 | raise RuntimeError("Test error for coverage")
944 |
945 | accounts = repository.get_accounts()
946 | # Simulate some processing time
947 | time.sleep(0.01)
948 | sync_time = repository.last_sync_time()
949 |
950 | # Thread-safe result collection
951 | with results_lock:
952 | results.append((len(accounts), sync_time is not None))
953 | except Exception as e:
954 | with results_lock:
955 | errors.append(e)
956 |
957 | # Start multiple threads accessing data concurrently
958 | threads = []
959 | for _ in range(10):
960 | t = threading.Thread(target=access_data)
961 | threads.append(t)
962 | t.start()
963 |
964 | for t in threads:
965 | t.join()
966 |
967 | # Verify error was captured and other threads succeeded
968 | assert len(errors) == 1, f"Expected 1 error, got {len(errors)}: {errors}"
969 | assert isinstance(errors[0], RuntimeError), (
970 | f"Expected RuntimeError, got {type(errors[0])}"
971 | )
972 | assert len(results) == 9, (
973 | f"Expected 9 results, got {len(results)}: {results}"
974 | ) # 9 successful threads
975 |
976 | # Extract length and sync_time results from successful threads
977 | length_results = [r[0] for r in results]
978 | sync_time_results = [r[1] for r in results]
979 |
980 | assert all(r == 1 for r in length_results), (
981 | f"Length results not all 1: {length_results}"
982 | )
983 | assert all(r is True for r in sync_time_results), (
984 | f"Sync time results not all True: {sync_time_results}"
985 | )
986 |
987 |
988 | def test_repository_concurrent_access_error_handling(
989 | repository: YNABRepository,
990 | ) -> None:
991 | """Test that errors in concurrent access are properly captured."""
992 |
993 | # Set up initial data
994 | account1 = create_ynab_account(id="acc-1", name="Checking")
995 | with repository._lock:
996 | repository._data["accounts"] = [account1]
997 | repository._last_sync = datetime.now()
998 |
999 | errors = []
1000 | results_lock = threading.Lock()
1001 |
1002 | def access_data_with_error() -> None:
1003 | try:
1004 | # Intentionally cause an error by accessing invalid attribute
1005 | _ = repository.get_accounts()
1006 | raise ValueError("Test error for coverage")
1007 | except Exception as e:
1008 | with results_lock:
1009 | errors.append(e)
1010 |
1011 | # Start a thread that will cause an error
1012 | thread = threading.Thread(target=access_data_with_error)
1013 | thread.start()
1014 | thread.join()
1015 |
1016 | # Verify the error was captured
1017 | assert len(errors) == 1
1018 | assert isinstance(errors[0], ValueError)
1019 | assert str(errors[0]) == "Test error for coverage"
1020 |
1021 |
1022 | def test_repository_lazy_init_only_syncs_once(repository: YNABRepository) -> None:
1023 | """Test that lazy initialization only syncs once even with concurrent access."""
1024 | account1 = create_ynab_account(id="acc-1", name="Checking")
1025 | success_response = ynab.AccountsResponse(
1026 | data=ynab.AccountsResponseData(accounts=[account1], server_knowledge=100)
1027 | )
1028 |
1029 | with patch("ynab.ApiClient") as mock_client_class:
1030 | mock_client = MagicMock()
1031 | mock_client_class.return_value.__enter__.return_value = mock_client
1032 |
1033 | mock_accounts_api = MagicMock()
1034 | mock_accounts_api.get_accounts.return_value = success_response
1035 |
1036 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1037 | # Multiple calls to get_accounts should only sync once
1038 | accounts1 = repository.get_accounts()
1039 | accounts2 = repository.get_accounts()
1040 | accounts3 = repository.get_accounts()
1041 |
1042 | # Verify only one API call was made
1043 | mock_accounts_api.get_accounts.assert_called_once()
1044 |
1045 | # All results should be consistent
1046 | assert len(accounts1) == len(accounts2) == len(accounts3) == 1
1047 |
1048 |
1049 | def test_repository_handles_empty_api_responses(repository: YNABRepository) -> None:
1050 | """Test repository gracefully handles empty API responses."""
1051 | empty_response = ynab.AccountsResponse(
1052 | data=ynab.AccountsResponseData(accounts=[], server_knowledge=100)
1053 | )
1054 |
1055 | with patch("ynab.ApiClient") as mock_client_class:
1056 | mock_client = MagicMock()
1057 | mock_client_class.return_value.__enter__.return_value = mock_client
1058 |
1059 | mock_accounts_api = MagicMock()
1060 | mock_accounts_api.get_accounts.return_value = empty_response
1061 |
1062 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1063 | repository.sync_accounts()
1064 |
1065 | # Should handle empty response gracefully
1066 | accounts = repository.get_accounts()
1067 | assert len(accounts) == 0
1068 | assert repository._server_knowledge["accounts"] == 100
1069 | assert repository.is_initialized()
1070 |
1071 |
1072 | def test_repository_server_knowledge_progression(repository: YNABRepository) -> None:
1073 | """Test that server knowledge progresses correctly through multiple syncs."""
1074 | account1 = create_ynab_account(id="acc-1", name="Checking")
1075 |
1076 | # First sync
1077 | response1 = ynab.AccountsResponse(
1078 | data=ynab.AccountsResponseData(accounts=[account1], server_knowledge=100)
1079 | )
1080 |
1081 | # Second sync with higher knowledge
1082 | account2 = create_ynab_account(id="acc-2", name="Savings")
1083 | response2 = ynab.AccountsResponse(
1084 | data=ynab.AccountsResponseData(
1085 | accounts=[account1, account2], server_knowledge=110
1086 | )
1087 | )
1088 |
1089 | with patch("ynab.ApiClient") as mock_client_class:
1090 | mock_client = MagicMock()
1091 | mock_client_class.return_value.__enter__.return_value = mock_client
1092 |
1093 | mock_accounts_api = MagicMock()
1094 | mock_accounts_api.get_accounts.side_effect = [response1, response2]
1095 |
1096 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1097 | # First sync
1098 | repository.sync_accounts()
1099 | assert repository._server_knowledge["accounts"] == 100
1100 |
1101 | # Second sync should use previous knowledge
1102 | repository.sync_accounts()
1103 | assert repository._server_knowledge["accounts"] == 110
1104 |
1105 | # Verify second call used delta sync
1106 | calls = mock_accounts_api.get_accounts.call_args_list
1107 | assert len(calls) == 2
1108 | assert calls[0][1] == {} # First call without last_knowledge
1109 | assert calls[1][1] == {
1110 | "last_knowledge_of_server": 100
1111 | } # Second call with knowledge
1112 |
1113 |
1114 | def test_repository_mixed_entity_types_independent(repository: YNABRepository) -> None:
1115 | """Test that different entity types sync independently."""
1116 | # Set up different sync states for different entity types
1117 | account1 = create_ynab_account(id="acc-1", name="Checking")
1118 | payee1 = create_ynab_payee(id="payee-1", name="Amazon")
1119 |
1120 | accounts_response = ynab.AccountsResponse(
1121 | data=ynab.AccountsResponseData(accounts=[account1], server_knowledge=100)
1122 | )
1123 | payees_response = ynab.PayeesResponse(
1124 | data=ynab.PayeesResponseData(payees=[payee1], server_knowledge=200)
1125 | )
1126 |
1127 | with patch("ynab.ApiClient") as mock_client_class:
1128 | mock_client = MagicMock()
1129 | mock_client_class.return_value.__enter__.return_value = mock_client
1130 |
1131 | mock_accounts_api = MagicMock()
1132 | mock_accounts_api.get_accounts.return_value = accounts_response
1133 |
1134 | mock_payees_api = MagicMock()
1135 | mock_payees_api.get_payees.return_value = payees_response
1136 |
1137 | with (
1138 | patch("ynab.AccountsApi", return_value=mock_accounts_api),
1139 | patch("ynab.PayeesApi", return_value=mock_payees_api),
1140 | ):
1141 | # Sync different entity types
1142 | repository.sync_accounts()
1143 | repository.sync_payees()
1144 |
1145 | # Verify independent server knowledge tracking
1146 | assert repository._server_knowledge["accounts"] == 100
1147 | assert repository._server_knowledge["payees"] == 200
1148 |
1149 | # Verify data is separate
1150 | accounts = repository.get_accounts()
1151 | payees = repository.get_payees()
1152 | assert len(accounts) == 1
1153 | assert len(payees) == 1
1154 |
1155 |
1156 | def test_repository_background_sync_thread_safety(repository: YNABRepository) -> None:
1157 | """Test that background sync threading doesn't cause issues."""
1158 |
1159 | # Enable background sync for this test
1160 | repository._background_sync_enabled = True
1161 |
1162 | # Set up initial stale data
1163 | account1 = create_ynab_account(id="acc-1", name="Checking")
1164 | repository._data["accounts"] = [account1]
1165 | repository._last_sync = datetime.now() - timedelta(minutes=10) # Stale
1166 |
1167 | # Track thread creations
1168 | original_thread = threading.Thread
1169 | created_threads = []
1170 |
1171 | def track_thread_creation(*args: Any, **kwargs: Any) -> threading.Thread:
1172 | thread = original_thread(*args, **kwargs)
1173 | created_threads.append(thread)
1174 | return thread
1175 |
1176 | # Mock the actual sync to prevent real API calls
1177 | with patch.object(repository, "sync_accounts"):
1178 | # Mock needs_sync to return True for stale data
1179 | with patch.object(repository, "needs_sync", return_value=True):
1180 | with patch("threading.Thread", side_effect=track_thread_creation):
1181 | # Multiple rapid calls should trigger background sync threads
1182 | repository.get_accounts()
1183 | repository.get_accounts()
1184 | repository.get_accounts()
1185 |
1186 | # Give threads a moment to be created
1187 | time.sleep(0.1)
1188 |
1189 | # Should have created threads for background sync (up to 3, one per call)
1190 | assert len(created_threads) <= 3 # At most one per call
1191 |
1192 |
1193 | def test_repository_preserves_data_during_failed_sync(
1194 | repository: YNABRepository,
1195 | ) -> None:
1196 | """Test that existing data is preserved when sync fails."""
1197 | # Set up initial good data
1198 | account1 = create_ynab_account(id="acc-1", name="Checking")
1199 | repository._data["accounts"] = [account1]
1200 | repository._server_knowledge["accounts"] = 100
1201 | repository._last_sync = datetime.now()
1202 |
1203 | # Mock sync to fail
1204 | with patch("ynab.ApiClient") as mock_client_class:
1205 | mock_client = MagicMock()
1206 | mock_client_class.return_value.__enter__.return_value = mock_client
1207 |
1208 | mock_accounts_api = MagicMock()
1209 | mock_accounts_api.get_accounts.side_effect = ynab.ApiException(
1210 | status=500, reason="Server Error"
1211 | )
1212 |
1213 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1214 | with pytest.raises(ynab.ApiException):
1215 | repository.sync_accounts()
1216 |
1217 | # Original data should still be there
1218 | accounts = repository.get_accounts()
1219 | assert len(accounts) == 1
1220 | assert accounts[0].id == "acc-1"
1221 | assert repository._server_knowledge["accounts"] == 100
1222 |
1223 |
1224 | def test_repository_handles_malformed_api_responses(repository: YNABRepository) -> None:
1225 | """Test repository handles malformed or unexpected API response structures."""
1226 | # Mock a response that might have unexpected structure
1227 | malformed_response = MagicMock()
1228 | malformed_response.data.accounts = None # Unexpected None
1229 | malformed_response.data.server_knowledge = 100
1230 |
1231 | with patch("ynab.ApiClient") as mock_client_class:
1232 | mock_client = MagicMock()
1233 | mock_client_class.return_value.__enter__.return_value = mock_client
1234 |
1235 | mock_accounts_api = MagicMock()
1236 | mock_accounts_api.get_accounts.return_value = malformed_response
1237 |
1238 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1239 | # Should handle malformed response gracefully
1240 | with pytest.raises((AttributeError, TypeError)):
1241 | repository.sync_accounts()
1242 |
1243 |
1244 | def test_repository_sync_entity_atomic_updates(repository: YNABRepository) -> None:
1245 | """Test that _sync_entity updates are atomic."""
1246 | # Set up initial data
1247 | account1 = create_ynab_account(id="acc-1", name="Checking")
1248 | repository._data["accounts"] = [account1]
1249 | repository._server_knowledge["accounts"] = 100
1250 | repository._last_sync = datetime.now()
1251 |
1252 | # Mock successful API call but failing delta application
1253 | account2 = create_ynab_account(id="acc-2", name="Savings")
1254 | success_response = ynab.AccountsResponse(
1255 | data=ynab.AccountsResponseData(
1256 | accounts=[account1, account2], server_knowledge=110
1257 | )
1258 | )
1259 |
1260 | with patch("ynab.ApiClient") as mock_client_class:
1261 | mock_client = MagicMock()
1262 | mock_client_class.return_value.__enter__.return_value = mock_client
1263 |
1264 | mock_accounts_api = MagicMock()
1265 | mock_accounts_api.get_accounts.return_value = success_response
1266 |
1267 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1268 | # Mock _apply_deltas to fail
1269 | with patch.object(
1270 | repository, "_apply_deltas", side_effect=Exception("Delta failed")
1271 | ):
1272 | # Sync should fail
1273 | with pytest.raises(Exception, match="Delta failed"):
1274 | repository.sync_accounts()
1275 |
1276 | # Original data should be unchanged due to atomic failure
1277 | accounts = repository.get_accounts()
1278 | assert len(accounts) == 1
1279 | assert accounts[0].id == "acc-1"
1280 | assert repository._server_knowledge["accounts"] == 100 # Should not be updated
1281 |
1282 |
1283 | def test_repository_handles_very_large_server_knowledge_values(
1284 | repository: YNABRepository,
1285 | ) -> None:
1286 | """Test repository handles very large server knowledge values correctly."""
1287 | # Test with a very large server knowledge value
1288 | large_knowledge = 999_999_999_999
1289 |
1290 | account1 = create_ynab_account(id="acc-1", name="Checking")
1291 | response = ynab.AccountsResponse(
1292 | data=ynab.AccountsResponseData(
1293 | accounts=[account1], server_knowledge=large_knowledge
1294 | )
1295 | )
1296 |
1297 | with patch("ynab.ApiClient") as mock_client_class:
1298 | mock_client = MagicMock()
1299 | mock_client_class.return_value.__enter__.return_value = mock_client
1300 |
1301 | mock_accounts_api = MagicMock()
1302 | mock_accounts_api.get_accounts.return_value = response
1303 |
1304 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1305 | repository.sync_accounts()
1306 |
1307 | # Should handle large values correctly
1308 | assert repository._server_knowledge["accounts"] == large_knowledge
1309 |
1310 | # Should be able to use it in subsequent delta calls
1311 | with patch("ynab.ApiClient") as mock_client_class:
1312 | mock_client = MagicMock()
1313 | mock_client_class.return_value.__enter__.return_value = mock_client
1314 |
1315 | mock_accounts_api = MagicMock()
1316 | mock_accounts_api.get_accounts.return_value = response
1317 |
1318 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1319 | repository.sync_accounts()
1320 |
1321 | # Verify large knowledge was passed correctly
1322 | mock_accounts_api.get_accounts.assert_called_with(
1323 | "test-budget", last_knowledge_of_server=large_knowledge
1324 | )
1325 |
1326 |
1327 | def test_repository_background_sync_respects_staleness_threshold(
1328 | repository: YNABRepository,
1329 | ) -> None:
1330 | """Test that background sync only triggers when data is actually stale."""
1331 | # Set up fresh data (not stale)
1332 | account1 = create_ynab_account(id="acc-1", name="Checking")
1333 | repository._data["accounts"] = [account1]
1334 | repository._last_sync = datetime.now() - timedelta(minutes=2) # Fresh (< 5 minutes)
1335 |
1336 | with patch.object(repository, "_trigger_background_sync") as mock_bg_sync:
1337 | # Getting accounts should NOT trigger background sync
1338 | accounts = repository.get_accounts()
1339 |
1340 | # Verify we got the data
1341 | assert len(accounts) == 1
1342 | assert accounts[0].id == "acc-1"
1343 |
1344 | # Verify background sync was NOT triggered
1345 | mock_bg_sync.assert_not_called()
1346 |
1347 |
1348 | def test_repository_error_logging_behavior(repository: YNABRepository) -> None:
1349 | """Test that errors are properly logged with appropriate levels."""
1350 |
1351 | # Capture log messages
1352 | log_messages = []
1353 |
1354 | class TestLogHandler(logging.Handler):
1355 | def emit(self, record: Any) -> None:
1356 | log_messages.append((record.levelname, record.getMessage()))
1357 |
1358 | # Add test handler to repository logger
1359 | test_handler = TestLogHandler()
1360 | repository_logger = logging.getLogger("repository")
1361 | repository_logger.addHandler(test_handler)
1362 | repository_logger.setLevel(logging.DEBUG)
1363 |
1364 | try:
1365 | with patch("ynab.ApiClient") as mock_client_class:
1366 | mock_client = MagicMock()
1367 | mock_client_class.return_value.__enter__.return_value = mock_client
1368 |
1369 | mock_accounts_api = MagicMock()
1370 |
1371 | # Set up initial server knowledge to trigger delta sync path
1372 | repository._server_knowledge["accounts"] = 50
1373 |
1374 | # Test different error scenarios
1375 | # 1. ConflictException should log as INFO (expected)
1376 | mock_accounts_api.get_accounts.side_effect = [
1377 | ConflictException(status=409, reason="Conflict"),
1378 | ynab.AccountsResponse(
1379 | data=ynab.AccountsResponseData(accounts=[], server_knowledge=100)
1380 | ),
1381 | ]
1382 |
1383 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1384 | repository.sync_accounts()
1385 |
1386 | # 2. Generic ApiException should log as WARNING
1387 | # Reset server knowledge for next test
1388 | repository._server_knowledge["accounts"] = 60
1389 | mock_accounts_api.get_accounts.side_effect = [
1390 | ynab.ApiException(status=500, reason="Server Error"),
1391 | ynab.AccountsResponse(
1392 | data=ynab.AccountsResponseData(accounts=[], server_knowledge=100)
1393 | ),
1394 | ]
1395 |
1396 | with patch("ynab.AccountsApi", return_value=mock_accounts_api):
1397 | repository.sync_accounts()
1398 |
1399 | # Verify appropriate log levels were used
1400 | info_logs = [msg for level, msg in log_messages if level == "INFO"]
1401 | warning_logs = [msg for level, msg in log_messages if level == "WARNING"]
1402 |
1403 | assert any("conflict" in msg.lower() for msg in info_logs)
1404 | assert any("api error" in msg.lower() for msg in warning_logs)
1405 |
1406 | finally:
1407 | # Clean up
1408 | repository_logger.removeHandler(test_handler)
1409 |
```