From d4ebab96d693b9024cf09d9a8f720eadea5fc687 Mon Sep 17 00:00:00 2001 From: Mathias Scheider Date: Sun, 18 Oct 2026 09:40:31 +0200 Subject: [PATCH] Fix category budget overlaps and document the uv run test workflow. --- AGENTS.md | 14 +- pyproject.toml | 3 - src/domain_utils.py | 119 +++++----------- tests/factories.py | 4 +- tests/test_domain_utils_budget_overlap.py | 162 ++++++++++++++++++++++ 5 files changed, 207 insertions(+), 95 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 084d27b..606590c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -7,15 +7,17 @@ uv sync # Run migrations -alembic upgrade head +uv run alembic upgrade head # Run all tests -pytest +uv run pytest # Run a single test file -pytest tests/test_model_b4wexport.py -v +uv run pytest tests/test_model_b4wexport.py -v ``` +Use `uv run` for project commands so they use the project-managed Python environment and dependencies. + ## Project Structure - **`src/`**: Application code @@ -53,10 +55,10 @@ with get_session() as session: ```bash # Tests automatically run migrations via conftest fixture -pytest tests/ -v --tb=short +uv run pytest tests/ -v --tb=short # To test without migrations (in-memory SQLite), use DATABASE_URL="sqlite://" -export DATABASE_URL="sqlite://" pytest tests/test_db.py +DATABASE_URL="sqlite://" uv run pytest tests/test_db.py ``` ### Environment Variables @@ -129,7 +131,7 @@ item = B4wExportFactory.build( # build() for partial, create() for full ## Testing Tips -- Run single test: `pytest tests/test_model_b4wexport.py -v` +- Run single test: `uv run pytest tests/test_model_b4wexport.py -v` - View migrations: `ls alembic/versions/` - Test data location: `/scripts/dummy_data/` diff --git a/pyproject.toml b/pyproject.toml index 75c632a..641b12a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,9 +57,6 @@ dev = [ "ruff>=0.16.7", ] -[tool.pytest.ini_options] -addopts = "--cov=src --cov-report=term-missing" - [tool.coverage.run] branch = true source = ["src"] diff --git a/src/domain_utils.py b/src/domain_utils.py index 599a1f0..55e1794 100644 --- a/src/domain_utils.py +++ b/src/domain_utils.py @@ -75,53 +75,28 @@ def set_category_budget(session: Session, category_budget: CategoryBudget): start1: date | None, end1: date | None, start2: date | None, end2: date | None ) -> bool: """Check if two date ranges overlap.""" - # Both completely unbounded - always overlap - if start1 is None and end1 is None and start2 is None and end2 is None: + range1_unbounded = start1 is None and end1 is None + range2_unbounded = start2 is None and end2 is None + + # A fully unbounded budget acts as a default alongside bounded budgets. + if range1_unbounded and range2_unbounded: 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: + if range1_unbounded or range2_unbounded: return False - # Both ranges are bounded - standard overlap check - if end1 is None or end2 is None: # Defensive; mypy needs guards - return False - # At this point, start1/start2/type narrowing ensures they're not None either - _start1, _start2 = start1, start2 # type: ignore[misc] - return _start2 < end1 and _start1 < end2 # type: ignore[operator] + # None on one side of a partially bounded range means infinity. Date + # endpoints are inclusive, so equal endpoints also overlap. + starts_before_end1 = end1 is None or start2 is None or start2 <= end1 + starts_before_end2 = end2 is None or start1 is None or start1 <= end2 + return starts_before_end1 and starts_before_end2 # 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 first - if start1 is None and end1 is None: - return ( - (start2, end2) - if any(x is not None for x in (start2, end2)) - else (None, None) - ) - if start2 is None and end2 is None: - return ( - (start1, end1) - if any(x is not None for x in (start1, end1)) - else (None, None) - ) - - # Both ranges are bounded or partially unbounded (after handling fully unbounded) o_start = max((s for s in (start1, start2) if s is not None), default=None) o_end = min((e for e in (end1, end2) if e is not None), default=None) - - # If either was filtered out, there's no overlap - if o_start is None or o_end is None: - return (None, None) - - if o_start > o_end: - return (None, None) - return (o_start, o_end) existing_budgets = ( @@ -130,54 +105,33 @@ def set_category_budget(session: Session, category_budget: CategoryBudget): 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(new_amount, new_start, new_end) - 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( - budget.amount_eur, budget.valid_from, overlap_start - ONE_DAY + # Keep the old budget's portion before the new range, if any. + if overlap_start is not None and ( + budget.valid_from is None or budget.valid_from < overlap_start + ): + session.add( + _add_row( + budget.amount_eur, + budget.valid_from, + overlap_start - ONE_DAY, + ) ) - 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(budget.amount_eur, overlap_end + ONE_DAY, budget.valid_to) - parts.append(row) - - # Add the new budget - row = _add_row(new_amount, new_start, new_end) - 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 + # Keep the old budget's portion after the new range, if any. + if overlap_end is not None and ( + budget.valid_to is None or overlap_end < budget.valid_to + ): + session.add( + _add_row( + budget.amount_eur, + overlap_end + ONE_DAY, + budget.valid_to, + ) + ) # Process each existing budget for existing_budget in existing_budgets: @@ -188,12 +142,7 @@ def set_category_budget(session: Session, category_budget: CategoryBudget): # Remove and split the overlapping budget session.delete(existing_budget) - - # Add back non-overlapping parts plus the new budget _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(new_amount, new_start, new_end) - session.add(row) + # Add the requested budget once, even if it replaced multiple existing rows. + session.add(_add_row(new_amount, new_start, new_end)) diff --git a/tests/factories.py b/tests/factories.py index be16434..dd4e1f6 100644 --- a/tests/factories.py +++ b/tests/factories.py @@ -16,7 +16,9 @@ CATEGORY_PATHS = tuple( for line in ( Path(__file__).resolve().parents[1] / "scripts/initial_load/categories/categories.txt" - ).read_text(encoding="utf-8").splitlines() + ) + .read_text(encoding="utf-8") + .splitlines() if line.strip() ) diff --git a/tests/test_domain_utils_budget_overlap.py b/tests/test_domain_utils_budget_overlap.py index 3164887..546fbea 100644 --- a/tests/test_domain_utils_budget_overlap.py +++ b/tests/test_domain_utils_budget_overlap.py @@ -414,6 +414,168 @@ class TestSetCategoryBudgetDateEdgeCases: ) assert len(budgets) == 2 + def test_shared_endpoint_is_replaced(self, db_session: Session): + """Inclusive ranges that share a date must not overlap after replacement.""" + cat = CategoryFactory.build() + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + set_category_budget( + db_session, + CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 10, 15), + ), + ) + set_category_budget( + db_session, + CategoryBudget( + category=cat, + amount_eur=200.00, + valid_from=date(2026, 10, 15), + valid_to=date(2026, 10, 31), + ), + ) + + budgets = ( + db_session.query(CategoryBudgetDb) + .filter_by(category_id=cat_db.id) + .order_by(CategoryBudgetDb.valid_from) + .all() + ) + assert [(b.amount_eur, b.valid_from, b.valid_to) for b in budgets] == [ + (100.00, date(2026, 10, 1), date(2026, 10, 14)), + (200.00, date(2026, 10, 15), date(2026, 10, 31)), + ] + + +class TestSetCategoryBudgetRegressionCases: + def test_new_range_overlapping_multiple_budgets_is_added_once( + self, db_session: Session + ): + cat = CategoryFactory.build() + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + for start, end, amount in ( + (date(2026, 10, 1), date(2026, 10, 31), 100.00), + (date(2026, 12, 1), date(2026, 12, 31), 200.00), + ): + set_category_budget( + db_session, + CategoryBudget( + category=cat, + amount_eur=amount, + valid_from=start, + valid_to=end, + ), + ) + + set_category_budget( + db_session, + CategoryBudget( + category=cat, + amount_eur=300.00, + valid_from=date(2026, 10, 1), + valid_to=date(2026, 12, 31), + ), + ) + + budgets = ( + db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id).all() + ) + assert len(budgets) == 1 + assert budgets[0].amount_eur == 300.00 + assert budgets[0].valid_from == date(2026, 10, 1) + assert budgets[0].valid_to == date(2026, 12, 31) + + @pytest.mark.parametrize( + ("existing_start", "existing_end", "new_start", "new_end", "expected"), + [ + ( + date(2026, 10, 1), + date(2026, 12, 31), + None, + date(2026, 11, 15), + [ + (100.00, date(2026, 11, 16), date(2026, 12, 31)), + (200.00, None, date(2026, 11, 15)), + ], + ), + ( + date(2026, 10, 1), + date(2026, 12, 31), + date(2026, 11, 15), + None, + [ + (100.00, date(2026, 10, 1), date(2026, 11, 14)), + (200.00, date(2026, 11, 15), None), + ], + ), + ( + None, + date(2026, 12, 31), + date(2026, 10, 1), + date(2026, 10, 31), + [ + (100.00, None, date(2026, 9, 30)), + (200.00, date(2026, 10, 1), date(2026, 10, 31)), + (100.00, date(2026, 11, 1), date(2026, 12, 31)), + ], + ), + ( + date(2026, 10, 1), + None, + date(2026, 11, 1), + date(2026, 11, 30), + [ + (100.00, date(2026, 10, 1), date(2026, 10, 31)), + (200.00, date(2026, 11, 1), date(2026, 11, 30)), + (100.00, date(2026, 12, 1), None), + ], + ), + ], + ) + def test_partially_unbounded_ranges_are_split_correctly( + self, + db_session: Session, + existing_start: date | None, + existing_end: date | None, + new_start: date | None, + new_end: date | None, + expected: list[tuple[float, date | None, date | None]], + ): + cat = CategoryFactory.build() + cat_db = upsert_category(db_session, cat, do_commit=False) + db_session.commit() + + set_category_budget( + db_session, + CategoryBudget( + category=cat, + amount_eur=100.00, + valid_from=existing_start, + valid_to=existing_end, + ), + ) + set_category_budget( + db_session, + CategoryBudget( + category=cat, + amount_eur=200.00, + valid_from=new_start, + valid_to=new_end, + ), + ) + + budgets = ( + db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id).all() + ) + actual = {(float(b.amount_eur), b.valid_from, b.valid_to) for b in budgets} + assert actual == set(expected) + class TestFindCategoryDb: """Test find_category_db function."""