some fixes
Tests / test (push) Successful in 16s
Tests / lint (push) Failing after 12s

This commit is contained in:
2026-10-16 22:15:19 +02:00
parent 2c7f8f4a67
commit 98a5e9332f
2 changed files with 695 additions and 38 deletions
+181 -40
View File
@@ -7,6 +7,11 @@ from src.models.category import Category, CategoryBudget, CategoryBudgetDb, Cate
def find_category_db(session: Session, category: Category | None) -> CategoryDb | None: 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: if category is None:
return 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): 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: if category_db is None:
raise ValueError( raise ValueError(
f"No matching category in database for category: {category_budget.category.to_str()}" f"No matching category in database for category: {category_budget.category.to_str()}"
) )
new_start = category_budget.valid_from or date.min new_start = category_budget.valid_from
new_end = category_budget.valid_to or date.max new_end = category_budget.valid_to
existing_budgets = (
session.query(CategoryBudgetDb).filter_by(category_id=category_db.id).all()
)
# Helper to add a budget row
def _add_row(amount_eur: Decimal, valid_from: date | None, valid_to: date | None): def _add_row(amount_eur: Decimal, valid_from: date | None, valid_to: date | None):
session.add( return CategoryBudgetDb(
CategoryBudgetDb( category_id=category_db.id,
category_id=category_db.id, amount_eur=amount_eur,
amount_eur=amount_eur, valid_from=valid_from,
valid_from=valid_from, valid_to=valid_to,
valid_to=valid_to,
)
) )
for existing_budget in existing_budgets: # Helper to check if two date ranges overlap
existing_start = existing_budget.valid_from or date.min def _ranges_overlap(start1: date | None, end1: date | None,
existing_end = existing_budget.valid_to or date.max start2: date | None, end2: date | None) -> bool:
"""Check if two date ranges overlap.
overlap_start = max(new_start, existing_start) Two bounded ranges overlap if start2 < end1 and start1 < end2.
overlap_end = min(new_end, existing_end) 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
if overlap_start > overlap_end: # At least one range has both bounds None (truly unbounded)
continue # no overlap # 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
session.delete(existing_budget) # Both ranges are bounded - standard overlap check
return start2 < end1 and start1 < end2
if existing_start < new_start: # Helper to find overlapping sub-range
# Keep existing_start but cut existing end before new range def _overlap_range(start1: date | None, end1: date | None,
_add_row( start2: date | None, end2: date | None) -> tuple[date | None, date | None]:
amount_eur=existing_budget.amount_eur, """Return the overlapping sub-range of two ranges."""
valid_from=existing_budget.valid_from,
# 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)
existing_budgets = (
session.query(CategoryBudgetDb)
.filter_by(category_id=category_db.id)
.all()
)
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 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), valid_to=overlap_start - timedelta(days=1),
) )
if existing_end > new_end: parts.append(row)
# Cut exsisting start to right after the new range and keep existing_end
_add_row(
amount_eur=existing_budget.amount_eur,
valid_from=overlap_end + timedelta(days=1),
valid_to=existing_budget.valid_to,
)
# Add the new budget # Part 2: After the overlap (if any)
_add_row( if budget.valid_to is not None and overlap_end < budget.valid_to:
amount_eur=category_budget.amount_eur, row = _add_row(
valid_from=category_budget.valid_from, amount_eur=budget.amount_eur,
valid_to=category_budget.valid_to, valid_from=overlap_end + timedelta(days=1),
) 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 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)
+516
View File
@@ -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()