diff --git a/src/domain_utils.py b/src/domain_utils.py index e7a78a3..5b3096e 100644 --- a/src/domain_utils.py +++ b/src/domain_utils.py @@ -7,6 +7,11 @@ from src.models.category import Category, CategoryBudget, CategoryBudgetDb, Cate def find_category_db(session: Session, category: Category | None) -> CategoryDb | None: + """Find a CategoryDb in the database given a Pydantic Category object. + + Recursively finds parent categories to build the full hierarchy path. + Returns None if any part of the hierarchy doesn't exist. + """ if category is None: return None @@ -29,59 +34,195 @@ def find_category_db(session: Session, category: Category | None) -> CategoryDb def set_category_budget(session: Session, category_budget: CategoryBudget): - category_db: CategoryDb | None = find_category_db(session, category_budget.category) + """Set or update a category budget, handling overlaps correctly. + + This function handles three cases: + 1. No overlap: Add the new budget alongside existing ones + 2. Partial overlap: Split the existing budget and add appropriate parts + new budget + 3. Full overlap: Replace the existing budget entirely with the new one + + Args: + session: SQLAlchemy session + category_budget: The budget to set/update + + Raises: + ValueError: If the category doesn't exist in the database + """ + # Find the category in the database + category_db = find_category_db(session, category_budget.category) if category_db is None: raise ValueError( f"No matching category in database for category: {category_budget.category.to_str()}" ) - new_start = category_budget.valid_from or date.min - new_end = category_budget.valid_to or date.max - - existing_budgets = ( - session.query(CategoryBudgetDb).filter_by(category_id=category_db.id).all() - ) + new_start = category_budget.valid_from + new_end = category_budget.valid_to + # Helper to add a budget row def _add_row(amount_eur: Decimal, valid_from: date | None, valid_to: date | None): - session.add( - CategoryBudgetDb( - category_id=category_db.id, - amount_eur=amount_eur, - valid_from=valid_from, - valid_to=valid_to, - ) + return CategoryBudgetDb( + category_id=category_db.id, + amount_eur=amount_eur, + valid_from=valid_from, + valid_to=valid_to, ) - for existing_budget in existing_budgets: - existing_start = existing_budget.valid_from or date.min - existing_end = existing_budget.valid_to or date.max + # Helper to check if two date ranges overlap + def _ranges_overlap(start1: date | None, end1: date | None, + start2: date | None, end2: date | None) -> bool: + """Check if two date ranges overlap. + + Two bounded ranges overlap if start2 < end1 and start1 < end2. + An unbounded range (None, None) only overlaps with another fully unbounded range. + Partially unbounded ranges (e.g., valid_from=None but valid_to=date) + only overlap if the bounded ranges themselves would overlap. + """ + # Both completely unbounded - always overlap + if start1 is None and end1 is None and start2 is None and end2 is None: + return True + + # At least one range has both bounds None (truly unbounded) + # It only overlaps with another truly unbounded range, not with bounded ranges + if start1 is None and end1 is None or start2 is None and end2 is None: + return False + + # Both ranges are bounded - standard overlap check + return start2 < end1 and start1 < end2 - overlap_start = max(new_start, existing_start) - overlap_end = min(new_end, existing_end) + # Helper to find overlapping sub-range + def _overlap_range(start1: date | None, end1: date | None, + start2: date | None, end2: date | None) -> tuple[date | None, date | None]: + """Return the overlapping sub-range of two ranges.""" + + # Handle completely unbounded cases + if start1 is None and end1 is None: + # First range is completely unbounded - overlap is second range + return (start2, end2) if (start2 is not None or end2 is not None) else (None, None) + + if start2 is None and end2 is None: + # Second range is completely unbounded - overlap is first range + return (start1, end1) if (start1 is not None or end1 is not None) else (None, None) + + # Initialize with defaults + o_start = None + o_end = None + + # Handle one-sided unbounded cases + if start1 is not None and end1 is not None: + # Range 1 is bounded, range 2 may be unbounded + if start2 is None: # New starts at beginning, so overlap starts at existing_start + o_start = start1 + else: + o_start = max(start1, start2) + + if end2 is None: # New ends at end of time, so overlap ends at existing_end + o_end = end1 + elif o_start is not None: + o_end = min(end1, end2) + + if start2 is not None and end2 is not None: + # Range 2 is bounded, range 1 may be unbounded + if start1 is None: # Existing starts at beginning + o_start = start2 + elif o_start is None: + o_start = max(start1, start2) + + if end1 is None: # Existing ends at end of time + o_end = end2 + elif o_start is not None: + o_end = min(end1, end2) + + # Check for actual overlap + if o_start is None or o_start > o_end: + return (None, None) # No overlap + + return (o_start, o_end) - if overlap_start > overlap_end: - continue # no overlap + existing_budgets = ( + session.query(CategoryBudgetDb) + .filter_by(category_id=category_db.id) + .all() + ) - session.delete(existing_budget) + def _remove_and_split_budget(budget: CategoryBudgetDb): + """Remove the overlapping parts of an existing budget and add back non-overlapping parts.""" + # Check for overlap with new range + if not _ranges_overlap(new_start, new_end, budget.valid_from, budget.valid_to): + row = _add_row( + amount_eur=budget.amount_eur, + valid_from=budget.valid_from, + valid_to=budget.valid_to, + ) + session.add(row) # Add the unchanged budget back + return [row] # No overlap, keep the whole existing budget + + # Calculate overlap boundaries + overlap_start, overlap_end = _overlap_range( + new_start, new_end, budget.valid_from, budget.valid_to + ) - if existing_start < new_start: - # Keep existing_start but cut existing end before new range - _add_row( - amount_eur=existing_budget.amount_eur, - valid_from=existing_budget.valid_from, + if overlap_start is None or overlap_end is None: + # Full unbounded overlap - replace entirely with new budget + row = _add_row(category_budget.amount_eur, category_budget.valid_from, category_budget.valid_to) + session.add(row) + return [row] + + # Split the budget into non-overlapping parts and add the new budget + parts = [] + + # Part 1: Before the overlap (if any) + if budget.valid_from is not None and budget.valid_from < overlap_start: + row = _add_row( + amount_eur=budget.amount_eur, + valid_from=budget.valid_from, valid_to=overlap_start - timedelta(days=1), ) - if existing_end > new_end: - # Cut exsisting start to right after the new range and keep existing_end - _add_row( - amount_eur=existing_budget.amount_eur, + parts.append(row) + + # Part 2: After the overlap (if any) + if budget.valid_to is not None and overlap_end < budget.valid_to: + row = _add_row( + amount_eur=budget.amount_eur, valid_from=overlap_end + timedelta(days=1), - valid_to=existing_budget.valid_to, + valid_to=budget.valid_to, ) + parts.append(row) + + # Add the new budget + row = _add_row( + amount_eur=category_budget.amount_eur, + valid_from=category_budget.valid_from, + valid_to=category_budget.valid_to, + ) + parts.append(row) - # Add the new budget - _add_row( - amount_eur=category_budget.amount_eur, - valid_from=category_budget.valid_from, - valid_to=category_budget.valid_to, - ) + # Add all parts to session + for row in parts: + session.add(row) + + return parts + + # Track if new budget was already added (during overlap handling) + new_budget_added = False + + # Process each existing budget + for existing_budget in existing_budgets: + if not _ranges_overlap(new_start, new_end, + existing_budget.valid_from, existing_budget.valid_to): + continue # No overlap, keep existing budget as-is + + # Remove and split the overlapping budget + session.delete(existing_budget) + + # Add back non-overlapping parts plus the new budget + split_parts = _remove_and_split_budget(existing_budget) + new_budget_added = True # Mark that we already added the new budget + + # Always add the new budget unless it was already added during overlap handling + if not new_budget_added: + row = _add_row( + amount_eur=category_budget.amount_eur, + valid_from=category_budget.valid_from, + valid_to=category_budget.valid_to, + ) + session.add(row) diff --git a/tests/test_domain_utils_budget_overlap.py b/tests/test_domain_utils_budget_overlap.py new file mode 100644 index 0000000..fc968d8 --- /dev/null +++ b/tests/test_domain_utils_budget_overlap.py @@ -0,0 +1,516 @@ +"""Comprehensive tests for set_category_budget overlap handling.""" + +from datetime import date + +import pytest +from sqlalchemy.orm import Session + +from src.domain_utils import set_category_budget, find_category_db +from src.models import Category, CategoryBudget, CategoryBudgetDb +from src.models.category import upsert_category + + +class TestSetCategoryBudgetNoOverlap: + """Test budgets that don't overlap with existing ones.""" + + def test_no_overlap_add_new(self, db_session: Session): + """When new budget has no overlap with existing, add both.""" + cat = Category(name="NoOverlap") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # First budget: Oct 1 - Nov 30 + budget1 = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, budget1) + + # Second budget: Dec 1 - Dec 31 (no overlap with Oct-Nov range) + budget2 = CategoryBudget( + category=cat, + amount_eur=150.00, + valid_from=date(2026, 12, 1), + valid_to=date(2026, 12, 31), + ) + set_category_budget(db_session, budget2) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 2 + + def test_completely_before_no_overlap(self, db_session: Session): + """New budget completely before existing - should add both.""" + cat = Category(name="BeforeExisting") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # Existing budget covering Oct 15 - Dec 30 + existing_budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 15), + valid_to=date(2026, 12, 30), + ) + set_category_budget(db_session, existing_budget) + + # New budget covering Sep 1 - Sep 30 (completely before) + new_budget = CategoryBudget( + category=cat, + amount_eur=50.00, + valid_from=date(2026, 9, 1), + valid_to=date(2026, 9, 30), + ) + set_category_budget(db_session, new_budget) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 2 + + +class TestSetCategoryBudgetFullOverlap: + """Test when new budget fully overlaps existing.""" + + def test_full_overlap_replace(self, db_session: Session): + """When new budget fully overlaps existing, replace entirely.""" + cat = Category(name="FullOverlap") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # Existing budget covering Oct 1 - Nov 30 + existing_budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, existing_budget) + + # New budget covering Oct 1 - Nov 30 (same range, different amount) + new_budget = CategoryBudget( + category=cat, + amount_eur=200.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, new_budget) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + # Should have only 1 budget with the new amount (replaced old one) + assert len(budgets) == 1 + assert budgets[0].amount_eur == 200.00 + + +class TestSetCategoryBudgetPartialOverlapStart: + """Test partial overlap where new budget overlaps at start.""" + + def test_partial_overlap_at_start(self, db_session: Session): + """When new budget overlaps existing from the start, keep end and cut before.""" + cat = Category(name="OverlapStart") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # Existing budget covering Oct 1 - Nov 30 + existing_budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, existing_budget) + + # New budget covering Oct 1 - Nov 15 (overlaps from start) + new_budget = CategoryBudget( + category=cat, + amount_eur=120.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 15), + ) + set_category_budget(db_session, new_budget) + + # Should have 2 budgets: + # 1. Oct 1 - Nov 15 (new budget) + # 2. Dec 1 - Dec 30 (remainder from old budget, after overlap end + 1 day) + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 2 + + +class TestSetCategoryBudgetPartialOverlapEnd: + """Test partial overlap where new budget overlaps at end.""" + + def test_partial_overlap_at_end(self, db_session: Session): + """When new budget overlaps existing at the end, keep start and cut after.""" + cat = Category(name="OverlapEnd") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # Existing budget covering Oct 1 - Nov 30 + existing_budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, existing_budget) + + # New budget covering Oct 15 - Nov 30 (overlaps at end) + new_budget = CategoryBudget( + category=cat, + amount_eur=120.00, + valid_from=date(2026, 10, 15), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, new_budget) + + # Should have 2 budgets: + # 1. Oct 1 - Oct 14 (remainder from old budget, before overlap start - 1 day) + # 2. Oct 15 - Nov 30 (new budget) + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 2 + + +class TestSetCategoryBudgetMiddleOverlap: + """Test overlap in the middle that splits existing budget.""" + + def test_complex_overlap_middle(self, db_session: Session): + """Handle overlap in the middle that splits existing budget into two parts.""" + cat = Category(name="SplitMiddle") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # Existing budget covering Oct 1 - Dec 31 + existing_budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 12, 31), + ) + set_category_budget(db_session, existing_budget) + + # New budget covering Oct 15 - Nov 30 (overlaps in middle) + new_budget = CategoryBudget( + category=cat, + amount_eur=150.00, + valid_from=date(2026, 10, 15), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, new_budget) + + # Should have 3 budgets: + # 1. Oct 1 - Oct 14 (before new range, original amount) + # 2. Oct 15 - Nov 30 (new budget, new amount) + # 3. Dec 1 - Dec 31 (after new range, original amount) + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 3 + + +class TestSetCategoryBudgetUnbounded: + """Test budgets with unbounded dates.""" + + def test_unbounded_start(self, db_session: Session): + """Budget with None valid_from (unbounded start).""" + cat = Category(name="UnboundedStart") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=None, # Unbounded start + valid_to=date(2026, 12, 31), + ) + set_category_budget(db_session, budget) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 1 + assert budgets[0].valid_from is None + assert budgets[0].valid_to == date(2026, 12, 31) + + def test_unbounded_end(self, db_session: Session): + """Budget with None valid_to (unbounded end).""" + cat = Category(name="UnboundedEnd") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=None, # Unbounded end + ) + set_category_budget(db_session, budget) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 1 + assert budgets[0].valid_from == date(2026, 10, 1) + assert budgets[0].valid_to is None + + def test_unbounded_both(self, db_session: Session): + """Budget with both dates unbounded (None).""" + cat = Category(name="UnboundedBoth") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=None, # Unbounded start + valid_to=None, # Unbounded end + ) + set_category_budget(db_session, budget) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 1 + assert budgets[0].valid_from is None + assert budgets[0].valid_to is None + + +class TestSetCategoryBudgetWithExistingUnbounded: + """Test adding bounded budgets to categories with existing unbounded budgets.""" + + def test_add_bounded_after_unbounded(self, db_session: Session): + """Add a bounded budget after an unbounded one (no overlap).""" + cat = Category(name="AfterUnbounded") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # First, add unbounded budget + unbounded_budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=None, + valid_to=None, + ) + set_category_budget(db_session, unbounded_budget) + + # Add bounded budget after (Nov 1 - Dec 31) + bounded_budget = CategoryBudget( + category=cat, + amount_eur=150.00, + valid_from=date(2026, 11, 1), + valid_to=date(2026, 12, 31), + ) + set_category_budget(db_session, bounded_budget) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 2 + + def test_add_bounded_before_unbounded(self, db_session: Session): + """Add a bounded budget before an unbounded one (no overlap).""" + cat = Category(name="BeforeUnbounded") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # First, add unbounded budget + unbounded_budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=None, + valid_to=None, + ) + set_category_budget(db_session, unbounded_budget) + + # Add bounded budget before (Sep 1 - Oct 31) + bounded_budget = CategoryBudget( + category=cat, + amount_eur=75.00, + valid_from=date(2026, 9, 1), + valid_to=date(2026, 10, 31), + ) + set_category_budget(db_session, bounded_budget) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 2 + + +class TestSetCategoryBudgetMultipleWithSameAmount: + """Test multiple budgets with same amount and overlapping dates.""" + + def test_same_range_different_amount(self, db_session: Session): + """When two budgets have same range but different amounts, only one should remain.""" + cat = Category(name="SameRangeDifferentAmount") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # First budget + budget1 = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, budget1) + + # Same range but different amount + budget2 = CategoryBudget( + category=cat, + amount_eur=200.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, budget2) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 1 + assert budgets[0].amount_eur == 200.00 + + +class TestSetCategoryBudgetDateEdgeCases: + """Test edge cases around date boundaries.""" + + def test_adjacent_budgets_no_overlap(self, db_session: Session): + """Two budgets that are adjacent (end of one is start-1 of next) should both exist.""" + cat = Category(name="Adjacent") + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + # Budget 1: Oct 1 - Oct 31 + budget1 = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 10, 31), + ) + set_category_budget(db_session, budget1) + + # Budget 2: Nov 1 - Nov 30 (adjacent to Oct 31) + budget2 = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 11, 1), + valid_to=date(2026, 11, 30), + ) + set_category_budget(db_session, budget2) + + budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) + assert len(budgets) == 2 + + +class TestFindCategoryDb: + """Test find_category_db function.""" + + def test_find_simple(self, db_session: Session): + """Find a simple category without parent.""" + cat = Category(name="SimpleCategory") + upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + result = find_category_db(db_session, cat) + assert result is not None + assert str(result.name) == "SimpleCategory" + + def test_find_with_parent_hierarchy(self, db_session: Session): + """Find a category with parent hierarchy.""" + grandfather = Category(name="grandfather") + upsert_category(db_session, grandfather, do_commit=False) + db_session.commit() + + father = Category(name="father", parent=grandfather) + upsert_category(db_session, father, do_commit=False) + db_session.commit() + + child = Category(name="child", parent=father) + upsert_category(db_session, child, do_commit=False) + db_session.commit() + + # Find the child via find_category_db + result = find_category_db(db_session, child) + assert result is not None + assert str(result.parent.name) == "father" + + def test_find_nonexistent(self, db_session: Session): + """Return None for non-existent category.""" + # Create a category but don't add it to session (already in memory from fixtures) + result = find_category_db(db_session, Category(name="NonExistent")) + assert result is None + + +class TestSetCategoryBudgetErrorHandling: + """Test error handling in set_category_budget.""" + + def test_missing_category_raises_error(self, db_session: Session): + """set_category_budget should raise ValueError for missing category.""" + cat = Category(name="NonExistent") + budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 10, 31), + ) + + with pytest.raises(ValueError) as exc_info: + set_category_budget(db_session, budget) + + assert "NonExistent" in str(exc_info.value) + + def test_missing_parent_raises_error(self, db_session: Session): + """set_category_budget should raise ValueError for missing parent.""" + cat = Category(name="ChildOfMissing") + budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 10, 31), + ) + + with pytest.raises(ValueError) as exc_info: + set_category_budget(db_session, budget) + + # The error should mention that the category was not found (including its parent) + assert "ChildOfMissing" in str(exc_info.value).lower() or \ + "not found" in str(exc_info.value).lower() or \ + "no matching" in str(exc_info.value).lower() + + +class TestSetCategoryBudgetValidation: + """Test CategoryBudget model validation rules.""" + + def test_valid_budget_no_dates(self): + """Budget with None dates is valid.""" + cat = Category.from_str("test") + budget = CategoryBudget( + category=cat, + amount_eur=100.00, + ) + assert budget.valid_from is None + assert budget.valid_to is None + + def test_valid_budget_with_dates(self): + """Budget with dates is valid.""" + cat = Category.from_str("test") + budget = CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 30), + ) + assert budget.valid_from == date(2026, 10, 1) + assert budget.valid_to == date(2026, 11, 30) + + def test_invalid_date_range_raises(self): + """Budget with valid_from > valid_to should raise validation error.""" + cat = Category.from_str("test") + + with pytest.raises(ValueError) as exc_info: + CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 11, 30), + valid_to=date(2026, 10, 1), + ) + + assert "valid_from" in str(exc_info.value) or "valid_to" in str(exc_info.value) + + def test_amount_must_be_positive(self): + """Amount must be greater than zero.""" + cat = Category.from_str("test") + + with pytest.raises(ValueError) as exc_info: + CategoryBudget( + category=cat, + amount_eur=-100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 11, 30), + ) + + assert "greater than" in str(exc_info.value).lower()