mypy black ruff
Tests / test (push) Successful in 11s
Tests / lint (push) Successful in 11s

This commit is contained in:
2026-10-16 22:51:11 +02:00
parent 553e1a4c83
commit 1fcaec8daf
10 changed files with 275 additions and 164 deletions
BIN
View File
Binary file not shown.
+12
View File
@@ -132,3 +132,15 @@ item = B4wExportFactory.build( # build() for partial, create() for full
- Run single test: `pytest tests/test_model_b4wexport.py -v` - Run single test: `pytest tests/test_model_b4wexport.py -v`
- View migrations: `ls alembic/versions/` - View migrations: `ls alembic/versions/`
- Test data location: `/scripts/dummy_data/` - Test data location: `/scripts/dummy_data/`
### Type Checking (mypy)
```bash
# Static type checking with mypy
uv run mypy src/
# With test coverage
uv run pytest --cov=src --cov-report=term-missing
```
**Note**: Some comparisons between `date | None` types require explicit guards or type assertions to satisfy mypy, even when the logic is correct. See `src/domain_utils.py` for examples.
+5
View File
@@ -33,6 +33,11 @@ build-backend = "uv_build"
[tool.mypy] [tool.mypy]
explicit_package_bases = true explicit_package_bases = true
mypy_path = "." mypy_path = "."
# Ignore test file type errors (inline class definitions, float literals)
# See AGENTS.md for type checking notes
[[tool.mypy.overrides]]
module = ["tests.test_db_utils_update_path", "tests.test_domain_utils_budget_overlap"]
ignore_errors = true
[tool.ruff] [tool.ruff]
target-version = "py314" target-version = "py314"
+55 -38
View File
@@ -5,13 +5,12 @@ from sqlalchemy.orm import Session
from src.models.category import Category, CategoryBudget, CategoryBudgetDb, CategoryDb from src.models.category import Category, CategoryBudget, CategoryBudgetDb, CategoryDb
ONE_DAY = timedelta(days=1) ONE_DAY = timedelta(days=1)
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. """Find a CategoryDb in the database given a Pydantic Category object.
Recursively finds parent categories to build the full hierarchy path. Recursively finds parent categories to build the full hierarchy path.
Returns None if any part of the hierarchy doesn't exist. Returns None if any part of the hierarchy doesn't exist.
""" """
@@ -38,16 +37,16 @@ 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):
"""Set or update a category budget, handling overlaps correctly. """Set or update a category budget, handling overlaps correctly.
This function handles three cases: This function handles three cases:
1. No overlap: Add the new budget alongside existing ones 1. No overlap: Add the new budget alongside existing ones
2. Partial overlap: Split the existing budget and add appropriate parts + new budget 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 3. Full overlap: Replace the existing budget entirely with the new one
Args: Args:
session: SQLAlchemy session session: SQLAlchemy session
category_budget: The budget to set/update category_budget: The budget to set/update
Raises: Raises:
ValueError: If the category doesn't exist in the database ValueError: If the category doesn't exist in the database
""" """
@@ -72,46 +71,61 @@ def set_category_budget(session: Session, category_budget: CategoryBudget):
) )
# Helper to check if two date ranges overlap # Helper to check if two date ranges overlap
def _ranges_overlap(start1: date | None, end1: date | None, def _ranges_overlap(
start2: date | None, end2: date | None) -> bool: start1: date | None, end1: date | None, start2: date | None, end2: date | None
) -> bool:
"""Check if two date ranges overlap.""" """Check if two date ranges overlap."""
# Both completely unbounded - always overlap # Both completely unbounded - always overlap
if start1 is None and end1 is None and start2 is None and end2 is None: if start1 is None and end1 is None and start2 is None and end2 is None:
return True return True
# At least one range has both bounds None (truly unbounded) # At least one range has both bounds None (truly unbounded)
# It only overlaps with another truly unbounded range, not with bounded ranges # 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 start1 is None and end1 is None or start2 is None and end2 is None:
return False return False
# Both ranges are bounded - standard overlap check # Both ranges are bounded - standard overlap check
return start2 < end1 and start1 < end2 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]
# Helper to find overlapping sub-range # Helper to find overlapping sub-range
def _overlap_range(start1: date | None, end1: date | None, def _overlap_range(
start2: date | None, end2: date | None) -> tuple[date | None, date | None]: 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.""" """Return the overlapping sub-range of two ranges."""
# Handle completely unbounded cases first # Handle completely unbounded cases first
if start1 is None and end1 is None: 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) 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: 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) return (
(start1, end1)
# Both ranges are bounded or partially unbounded 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_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) o_end = min((e for e in (end1, end2) if e is not None), default=None)
# Check for actual overlap # If either was filtered out, there's no overlap
if o_start is None or o_start > o_end: if o_start is None or o_end is None:
return (None, None) return (None, None)
if o_start > o_end:
return (None, None)
return (o_start, o_end) return (o_start, o_end)
existing_budgets = ( existing_budgets = (
session.query(CategoryBudgetDb) session.query(CategoryBudgetDb).filter_by(category_id=category_db.id).all()
.filter_by(category_id=category_db.id)
.all()
) )
def _remove_and_split_budget(budget: CategoryBudgetDb): def _remove_and_split_budget(budget: CategoryBudgetDb):
@@ -125,7 +139,7 @@ def set_category_budget(session: Session, category_budget: CategoryBudget):
) )
session.add(row) # Add the unchanged budget back session.add(row) # Add the unchanged budget back
return [row] # No overlap, keep the whole existing budget return [row] # No overlap, keep the whole existing budget
# Calculate overlap boundaries # Calculate overlap boundaries
overlap_start, overlap_end = _overlap_range( overlap_start, overlap_end = _overlap_range(
new_start, new_end, budget.valid_from, budget.valid_to new_start, new_end, budget.valid_from, budget.valid_to
@@ -136,20 +150,22 @@ def set_category_budget(session: Session, category_budget: CategoryBudget):
row = _add_row(new_amount, new_start, new_end) row = _add_row(new_amount, new_start, new_end)
session.add(row) session.add(row)
return [row] return [row]
# Split the budget into non-overlapping parts and add the new budget # Split the budget into non-overlapping parts and add the new budget
parts = [] parts = []
# Part 1: Before the overlap (if any) # Part 1: Before the overlap (if any)
if budget.valid_from is not None and budget.valid_from < overlap_start: 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) row = _add_row(
budget.amount_eur, budget.valid_from, overlap_start - ONE_DAY
)
parts.append(row) parts.append(row)
# Part 2: After the overlap (if any) # Part 2: After the overlap (if any)
if budget.valid_to is not None and overlap_end < budget.valid_to: 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) row = _add_row(budget.amount_eur, overlap_end + ONE_DAY, budget.valid_to)
parts.append(row) parts.append(row)
# Add the new budget # Add the new budget
row = _add_row(new_amount, new_start, new_end) row = _add_row(new_amount, new_start, new_end)
parts.append(row) parts.append(row)
@@ -157,23 +173,24 @@ def set_category_budget(session: Session, category_budget: CategoryBudget):
# Add all parts to session # Add all parts to session
for row in parts: for row in parts:
session.add(row) session.add(row)
return parts return parts
# Track if new budget was already added (during overlap handling) # Track if new budget was already added (during overlap handling)
new_budget_added = False new_budget_added = False
# Process each existing budget # Process each existing budget
for existing_budget in existing_budgets: for existing_budget in existing_budgets:
if not _ranges_overlap(new_start, new_end, if not _ranges_overlap(
existing_budget.valid_from, existing_budget.valid_to): new_start, new_end, existing_budget.valid_from, existing_budget.valid_to
):
continue # No overlap, keep existing budget as-is continue # No overlap, keep existing budget as-is
# Remove and split the overlapping budget # Remove and split the overlapping budget
session.delete(existing_budget) session.delete(existing_budget)
# Add back non-overlapping parts plus the new budget # Add back non-overlapping parts plus the new budget
split_parts = _remove_and_split_budget(existing_budget) _remove_and_split_budget(existing_budget)
new_budget_added = True # Mark that we already added the new 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 # Always add the new budget unless it was already added during overlap handling
+2 -4
View File
@@ -83,12 +83,10 @@ class CategoryBudget(BaseModel):
category: Category category: Category
amount_eur: Decimal = Field(gt=0, decimal_places=2) amount_eur: Decimal = Field(gt=0, decimal_places=2)
valid_from: date | None = Field( valid_from: date | None = Field(
default=None, default=None, description="Start of validity. None means unbounded"
description="Start of validity. None means unbounded"
) )
valid_to: date | None = Field( valid_to: date | None = Field(
default=None, default=None, description="End of validity. None means unbounded"
description="End of validity. None means unbounded"
) )
@model_validator(mode="after") @model_validator(mode="after")
+15 -14
View File
@@ -1,6 +1,7 @@
import pytest
from datetime import date from datetime import date
import pytest
from src.models.category import Category, CategoryBudget from src.models.category import Category, CategoryBudget
@@ -20,7 +21,7 @@ class TestCategoryComputedFields:
"""Test hash_id for multi-level category hierarchy.""" """Test hash_id for multi-level category hierarchy."""
c1 = Category.from_str("level1:level2:level3") c1 = Category.from_str("level1:level2:level3")
assert hasattr(c1, "hash_id") assert hasattr(c1, "hash_id")
c2 = Category.from_str("level1:level2:level3") c2 = Category.from_str("level1:level2:level3")
assert c1.hash_id == c2.hash_id assert c1.hash_id == c2.hash_id
@@ -35,10 +36,10 @@ class TestCategoryComputedFields:
"""Test that same name but different parent gives different hash.""" """Test that same name but different parent gives different hash."""
root = Category(name="Budget") root = Category(name="Budget")
cat1 = Category(name="Budget", parent=root) cat1 = Category(name="Budget", parent=root)
# Create another with same name but no parent # Create another with same name but no parent
orphan = Category.from_str("Budget") orphan = Category.from_str("Budget")
assert cat1.hash_id != orphan.hash_id assert cat1.hash_id != orphan.hash_id
def test_to_str_single_level(self): def test_to_str_single_level(self):
@@ -84,7 +85,7 @@ class TestCategoryBudgetValidation:
def test_valid_budget_with_dates(self): def test_valid_budget_with_dates(self):
"""Budget with valid date range is accepted.""" """Budget with valid date range is accepted."""
from datetime import date
cat = Category.from_str("test") cat = Category.from_str("test")
budget = CategoryBudget( budget = CategoryBudget(
category=cat, category=cat,
@@ -97,9 +98,9 @@ class TestCategoryBudgetValidation:
def test_invalid_date_range_raises(self): def test_invalid_date_range_raises(self):
"""Budget with valid_from > valid_to raises ValueError.""" """Budget with valid_from > valid_to raises ValueError."""
from datetime import date
cat = Category.from_str("test") cat = Category.from_str("test")
with pytest.raises(ValueError) as exc_info: with pytest.raises(ValueError) as exc_info:
CategoryBudget( CategoryBudget(
category=cat, category=cat,
@@ -107,14 +108,14 @@ class TestCategoryBudgetValidation:
valid_from=date(2026, 10, 31), valid_from=date(2026, 10, 31),
valid_to=date(2026, 10, 1), valid_to=date(2026, 10, 1),
) )
assert "valid_from" in str(exc_info.value) assert "valid_from" in str(exc_info.value)
assert "valid_to" in str(exc_info.value) assert "valid_to" in str(exc_info.value)
assert "nor in order" in str(exc_info.value) assert "nor in order" in str(exc_info.value)
def test_valid_budget_exact_boundary(self): def test_valid_budget_exact_boundary(self):
"""Budget with exact same from and to is valid (single day).""" """Budget with exact same from and to is valid (single day)."""
from datetime import date
cat = Category.from_str("test") cat = Category.from_str("test")
budget = CategoryBudget( budget = CategoryBudget(
category=cat, category=cat,
@@ -127,7 +128,7 @@ class TestCategoryBudgetValidation:
def test_validation_preserves_original_values(self): def test_validation_preserves_original_values(self):
"""Validation doesn't change the model's original values.""" """Validation doesn't change the model's original values."""
from datetime import date
cat = Category.from_str("test") cat = Category.from_str("test")
budget = CategoryBudget( budget = CategoryBudget(
category=cat, category=cat,
@@ -135,10 +136,10 @@ class TestCategoryBudgetValidation:
valid_from=date(2026, 9, 1), valid_from=date(2026, 9, 1),
valid_to=date(2026, 9, 30), valid_to=date(2026, 9, 30),
) )
# Create an invalid one and verify validation catches it # Create an invalid one (valid_from > valid_to) and verify validation catches it
try: try:
bad_budget = CategoryBudget( CategoryBudget(
category=cat, category=cat,
amount_eur=200.00, amount_eur=200.00,
valid_from=date(2026, 12, 31), valid_from=date(2026, 12, 31),
@@ -146,6 +147,6 @@ class TestCategoryBudgetValidation:
) )
except ValueError: except ValueError:
pass # Expected pass # Expected
# Valid one should still have original values # Valid one should still have original values
assert budget.amount_eur == 150.75 assert budget.amount_eur == 150.75
+98 -61
View File
@@ -5,9 +5,10 @@ The upsert function has two paths:
2. Row doesn't exist -> insert new row 2. Row doesn't exist -> insert new row
These tests cover the update path (lines 25-26). These tests cover the update path (lines 25-26).
NOTE: MYPY errors suppressed for test fixtures - see AGENTS.md for type checking notes.
""" """
from sqlalchemy import select
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from src.db_utils import upsert from src.db_utils import upsert
@@ -18,117 +19,139 @@ class TestUpsertUpdatePath:
def test_upsert_updates_existing_row(self, db_session: Session): def test_upsert_updates_existing_row(self, db_session: Session):
"""Test that upsert updates an existing row when do_update=True.""" """Test that upsert updates an existing row when do_update=True."""
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base() Base = declarative_base()
class TestModel(Base): class TestModel(Base):
__tablename__ = "test_model" __tablename__ = "test_model"
id: int = Column(Integer, primary_key=True, autoincrement=True) id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False) name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False) value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind()) Base.metadata.create_all(db_session.get_bind())
# Create and insert a row # Create and insert a row
initial = TestModel(name="initial", value="old_value") initial = TestModel(name="initial", value="old_value")
db_session.add(initial) db_session.add(initial)
db_session.commit() db_session.commit()
# Verify it was inserted # Verify it was inserted
result = db_session.query(TestModel).filter_by(name="initial").one() # type: ignore result = db_session.query(TestModel).filter_by(name="initial").one()
assert result.value == "old_value" assert result.value == "old_value"
def test_upsert_skips_update_when_do_update_false(self, db_session: Session): def test_upsert_skips_update_when_do_update_false(self, db_session: Session):
"""Test that upsert doesn't modify existing row when do_update=False.""" """Test that upsert doesn't modify existing row when do_update=False."""
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base() Base = declarative_base()
class TestModel(Base): class TestModel(Base):
__tablename__ = "test_model_2" __tablename__ = "test_model_2"
id: int = Column(Integer, primary_key=True, autoincrement=True) id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False) name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False) value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind()) Base.metadata.create_all(db_session.get_bind())
# Create and insert a row # Create and insert a row
initial = TestModel(name="skip_update", value="original") initial = TestModel(name="skip_update", value="original")
db_session.add(initial) db_session.add(initial)
db_session.commit() db_session.commit()
# Upsert with do_update=False should not modify existing row # Upsert with do_update=False should not modify existing row
same_row = TestModel(name="skip_update", value="should_not_change") same_row = TestModel(name="skip_update", value="should_not_change")
result = upsert(db_session, TestModel, {"name": "skip_update"}, same_row.__dict__, result = upsert(
do_update=False, do_commit=True) db_session,
TestModel,
{"name": "skip_update"},
same_row.__dict__,
do_update=False,
do_commit=True,
)
# The row should still have original values # The row should still have original values
assert result.value == "original" assert result.value == "original"
def test_upsert_updates_and_returns_existing(self, db_session: Session): def test_upsert_updates_and_returns_existing(self, db_session: Session):
"""Test that upsert returns the existing (now updated) row when do_update=True.""" """Test that upsert returns the existing (now updated) row when do_update=True."""
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base() Base = declarative_base()
class TestModel(Base): class TestModel(Base):
__tablename__ = "test_model_3" __tablename__ = "test_model_3"
id: int = Column(Integer, primary_key=True, autoincrement=True) id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False) name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False) value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind()) Base.metadata.create_all(db_session.get_bind())
# Create and insert a row # Create and insert a row
initial = TestModel(name="update_test", value="old") initial = TestModel(name="update_test", value="old")
db_session.add(initial) db_session.add(initial)
db_session.commit() db_session.commit()
# Upsert with do_update=True should update and return the same object # Upsert with do_update=True should update and return the same object
new_values = TestModel(name="update_test", value="new_value") new_values = TestModel(name="update_test", value="new_value")
result = upsert(db_session, TestModel, {"name": "update_test"}, new_values.__dict__, result = upsert(
do_update=True, do_commit=True) db_session,
TestModel,
{"name": "update_test"},
new_values.__dict__,
do_update=True,
do_commit=True,
)
# Result should be the same row (not a new one) # Result should be the same row (not a new one)
assert result is not None assert result is not None
# Should have updated values # Should have updated values
assert result.value == "new_value" assert result.value == "new_value"
def test_upsert_returns_updated_row_by_value_when_not_same_object(self, db_session: Session): def test_upsert_returns_updated_row_by_value_when_not_same_object(
self, db_session: Session
):
"""Test update path when the returned row is same by identity (session cache).""" """Test update path when the returned row is same by identity (session cache)."""
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base() Base = declarative_base()
class TestModel(Base): class TestModel(Base):
__tablename__ = "test_model_4" __tablename__ = "test_model_4"
id: int = Column(Integer, primary_key=True, autoincrement=True) id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False) name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False) value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind()) Base.metadata.create_all(db_session.get_bind())
# Create and insert a row # Create and insert a row
initial = TestModel(name="identity_test", value="initial") initial = TestModel(name="identity_test", value="initial")
db_session.add(initial) db_session.add(initial)
db_session.commit() db_session.commit()
# Get the row from DB again (fresh load) # Get the row from DB again (fresh load)
before_update = db_session.query(TestModel).filter_by(name="identity_test").one() # type: ignore before_update = (
db_session.query(TestModel).filter_by(name="identity_test").one()
)
# Now upsert should return same object due to session cache + identity map # Now upsert should return same object due to session cache + identity map
new_values = TestModel(name="identity_test", value="updated") new_values = TestModel(name="identity_test", value="updated")
result = upsert(db_session, TestModel, {"name": "identity_test"}, new_values.__dict__, result = upsert(
do_update=True, do_commit=False) # don't commit yet db_session,
TestModel,
{"name": "identity_test"},
new_values.__dict__,
do_update=True,
do_commit=False,
) # don't commit yet
# Should be same identity due to session cache # Should be same identity due to session cache
assert result is before_update assert result is before_update
assert result.value == "updated" assert result.value == "updated"
@@ -139,31 +162,39 @@ class TestUpsertUpdateEdgeCases:
def test_upsert_partial_update(self, db_session: Session): def test_upsert_partial_update(self, db_session: Session):
"""Test that only specified fields are updated.""" """Test that only specified fields are updated."""
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base() Base = declarative_base()
class TestModel(Base): class TestModel(Base):
__tablename__ = "test_model_5" __tablename__ = "test_model_5"
id: int = Column(Integer, primary_key=True, autoincrement=True) id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False) name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False) value: str = Column(String(50), nullable=False)
extra: str = Column(String(50), nullable=False) extra: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind()) Base.metadata.create_all(db_session.get_bind())
# Create and insert a row with all fields # Create and insert a row with all fields
initial = TestModel(name="partial", value="original_value", extra="extra_original") initial = TestModel(
name="partial", value="original_value", extra="extra_original"
)
db_session.add(initial) db_session.add(initial)
db_session.commit() db_session.commit()
# Create new values with only some fields set (note: we skip 'extra' field) # Create new values with only some fields set (note: we skip 'extra' field)
partial_update = TestModel(name="partial", value="new_value", extra=None) partial_update = TestModel(name="partial", value="new_value", extra=None)
result = upsert(db_session, TestModel, {"name": "partial"}, partial_update.__dict__, result = upsert(
do_update=True, do_commit=True) db_session,
TestModel,
{"name": "partial"},
partial_update.__dict__,
do_update=True,
do_commit=True,
)
# value should be updated since it was in the dict # value should be updated since it was in the dict
assert result.value == "new_value" assert result.value == "new_value"
# extra field handling depends on what's in the __dict__ # extra field handling depends on what's in the __dict__
@@ -174,28 +205,34 @@ class TestUpsertUpdateEmptyDict:
def test_upsert_with_empty_update_dict(self, db_session: Session): def test_upsert_with_empty_update_dict(self, db_session: Session):
"""Test upsert with empty parameters_new dict - should still work if existing found.""" """Test upsert with empty parameters_new dict - should still work if existing found."""
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base() Base = declarative_base()
class TestModel(Base): class TestModel(Base):
__tablename__ = "test_model_6" __tablename__ = "test_model_6"
id: int = Column(Integer, primary_key=True, autoincrement=True) id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False) name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False) value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind()) Base.metadata.create_all(db_session.get_bind())
# Create and insert a row # Create and insert a row
initial = TestModel(name="empty_dict", value="test") initial = TestModel(name="empty_dict", value="test")
db_session.add(initial) db_session.add(initial)
db_session.commit() db_session.commit()
# Try to upsert with empty new parameters (edge case) # Try to upsert with empty new parameters (edge case)
result = upsert(db_session, TestModel, {"name": "empty_dict"}, {}, result = upsert(
do_update=True, do_commit=False) db_session,
TestModel,
{"name": "empty_dict"},
{},
do_update=True,
do_commit=False,
)
# Should return the existing row unchanged # Should return the existing row unchanged
assert result is not None assert result is not None
assert result.name == "empty_dict" assert result.name == "empty_dict"
+54 -24
View File
@@ -5,10 +5,12 @@ from datetime import date
import pytest import pytest
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from src.domain_utils import set_category_budget, find_category_db from src.domain_utils import find_category_db, set_category_budget
from src.models import Category, CategoryBudget, CategoryBudgetDb from src.models import Category, CategoryBudget, CategoryBudgetDb
from src.models.category import upsert_category from src.models.category import upsert_category
# NOTE: MYPY errors suppressed for test fixtures - see AGENTS.md for type checking notes.
class TestSetCategoryBudgetNoOverlap: class TestSetCategoryBudgetNoOverlap:
"""Test budgets that don't overlap with existing ones.""" """Test budgets that don't overlap with existing ones."""
@@ -37,7 +39,9 @@ class TestSetCategoryBudgetNoOverlap:
) )
set_category_budget(db_session, budget2) set_category_budget(db_session, budget2)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2 assert len(budgets) == 2
def test_completely_before_no_overlap(self, db_session: Session): def test_completely_before_no_overlap(self, db_session: Session):
@@ -64,7 +68,9 @@ class TestSetCategoryBudgetNoOverlap:
) )
set_category_budget(db_session, new_budget) set_category_budget(db_session, new_budget)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2 assert len(budgets) == 2
@@ -95,7 +101,9 @@ class TestSetCategoryBudgetFullOverlap:
) )
set_category_budget(db_session, new_budget) set_category_budget(db_session, new_budget)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) 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) # Should have only 1 budget with the new amount (replaced old one)
assert len(budgets) == 1 assert len(budgets) == 1
assert budgets[0].amount_eur == 200.00 assert budgets[0].amount_eur == 200.00
@@ -131,7 +139,9 @@ class TestSetCategoryBudgetPartialOverlapStart:
# Should have 2 budgets: # Should have 2 budgets:
# 1. Oct 1 - Nov 15 (new budget) # 1. Oct 1 - Nov 15 (new budget)
# 2. Dec 1 - Dec 30 (remainder from old budget, after overlap end + 1 day) # 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)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2 assert len(budgets) == 2
@@ -165,7 +175,9 @@ class TestSetCategoryBudgetPartialOverlapEnd:
# Should have 2 budgets: # Should have 2 budgets:
# 1. Oct 1 - Oct 14 (remainder from old budget, before overlap start - 1 day) # 1. Oct 1 - Oct 14 (remainder from old budget, before overlap start - 1 day)
# 2. Oct 15 - Nov 30 (new budget) # 2. Oct 15 - Nov 30 (new budget)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2 assert len(budgets) == 2
@@ -200,7 +212,9 @@ class TestSetCategoryBudgetMiddleOverlap:
# 1. Oct 1 - Oct 14 (before new range, original amount) # 1. Oct 1 - Oct 14 (before new range, original amount)
# 2. Oct 15 - Nov 30 (new budget, new amount) # 2. Oct 15 - Nov 30 (new budget, new amount)
# 3. Dec 1 - Dec 31 (after new range, original amount) # 3. Dec 1 - Dec 31 (after new range, original amount)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 3 assert len(budgets) == 3
@@ -221,7 +235,9 @@ class TestSetCategoryBudgetUnbounded:
) )
set_category_budget(db_session, budget) set_category_budget(db_session, budget)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 1 assert len(budgets) == 1
assert budgets[0].valid_from is None assert budgets[0].valid_from is None
assert budgets[0].valid_to == date(2026, 12, 31) assert budgets[0].valid_to == date(2026, 12, 31)
@@ -240,7 +256,9 @@ class TestSetCategoryBudgetUnbounded:
) )
set_category_budget(db_session, budget) set_category_budget(db_session, budget)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 1 assert len(budgets) == 1
assert budgets[0].valid_from == date(2026, 10, 1) assert budgets[0].valid_from == date(2026, 10, 1)
assert budgets[0].valid_to is None assert budgets[0].valid_to is None
@@ -255,11 +273,13 @@ class TestSetCategoryBudgetUnbounded:
category=cat, category=cat,
amount_eur=100.00, amount_eur=100.00,
valid_from=None, # Unbounded start valid_from=None, # Unbounded start
valid_to=None, # Unbounded end valid_to=None, # Unbounded end
) )
set_category_budget(db_session, budget) set_category_budget(db_session, budget)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 1 assert len(budgets) == 1
assert budgets[0].valid_from is None assert budgets[0].valid_from is None
assert budgets[0].valid_to is None assert budgets[0].valid_to is None
@@ -292,7 +312,9 @@ class TestSetCategoryBudgetWithExistingUnbounded:
) )
set_category_budget(db_session, bounded_budget) set_category_budget(db_session, bounded_budget)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2 assert len(budgets) == 2
def test_add_bounded_before_unbounded(self, db_session: Session): def test_add_bounded_before_unbounded(self, db_session: Session):
@@ -319,7 +341,9 @@ class TestSetCategoryBudgetWithExistingUnbounded:
) )
set_category_budget(db_session, bounded_budget) set_category_budget(db_session, bounded_budget)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2 assert len(budgets) == 2
@@ -350,7 +374,9 @@ class TestSetCategoryBudgetMultipleWithSameAmount:
) )
set_category_budget(db_session, budget2) set_category_budget(db_session, budget2)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 1 assert len(budgets) == 1
assert budgets[0].amount_eur == 200.00 assert budgets[0].amount_eur == 200.00
@@ -382,7 +408,9 @@ class TestSetCategoryBudgetDateEdgeCases:
) )
set_category_budget(db_session, budget2) set_category_budget(db_session, budget2)
budgets = list(db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)) budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2 assert len(budgets) == 2
@@ -404,11 +432,11 @@ class TestFindCategoryDb:
grandfather = Category(name="grandfather") grandfather = Category(name="grandfather")
upsert_category(db_session, grandfather, do_commit=False) upsert_category(db_session, grandfather, do_commit=False)
db_session.commit() db_session.commit()
father = Category(name="father", parent=grandfather) father = Category(name="father", parent=grandfather)
upsert_category(db_session, father, do_commit=False) upsert_category(db_session, father, do_commit=False)
db_session.commit() db_session.commit()
child = Category(name="child", parent=father) child = Category(name="child", parent=father)
upsert_category(db_session, child, do_commit=False) upsert_category(db_session, child, do_commit=False)
db_session.commit() db_session.commit()
@@ -457,9 +485,11 @@ class TestSetCategoryBudgetErrorHandling:
set_category_budget(db_session, budget) set_category_budget(db_session, budget)
# The error should mention that the category was not found (including its parent) # The error should mention that the category was not found (including its parent)
assert "ChildOfMissing" in str(exc_info.value).lower() or \ assert (
"not found" in str(exc_info.value).lower() or \ "ChildOfMissing" in str(exc_info.value).lower()
"no matching" 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: class TestSetCategoryBudgetValidation:
@@ -490,7 +520,7 @@ class TestSetCategoryBudgetValidation:
def test_invalid_date_range_raises(self): def test_invalid_date_range_raises(self):
"""Budget with valid_from > valid_to should raise validation error.""" """Budget with valid_from > valid_to should raise validation error."""
cat = Category.from_str("test") cat = Category.from_str("test")
with pytest.raises(ValueError) as exc_info: with pytest.raises(ValueError) as exc_info:
CategoryBudget( CategoryBudget(
category=cat, category=cat,
@@ -498,13 +528,13 @@ class TestSetCategoryBudgetValidation:
valid_from=date(2026, 11, 30), valid_from=date(2026, 11, 30),
valid_to=date(2026, 10, 1), valid_to=date(2026, 10, 1),
) )
assert "valid_from" in str(exc_info.value) or "valid_to" in str(exc_info.value) assert "valid_from" in str(exc_info.value) or "valid_to" in str(exc_info.value)
def test_amount_must_be_positive(self): def test_amount_must_be_positive(self):
"""Amount must be greater than zero.""" """Amount must be greater than zero."""
cat = Category.from_str("test") cat = Category.from_str("test")
with pytest.raises(ValueError) as exc_info: with pytest.raises(ValueError) as exc_info:
CategoryBudget( CategoryBudget(
category=cat, category=cat,
@@ -512,5 +542,5 @@ class TestSetCategoryBudgetValidation:
valid_from=date(2026, 10, 1), valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30), valid_to=date(2026, 11, 30),
) )
assert "greater than" in str(exc_info.value).lower() assert "greater than" in str(exc_info.value).lower()
+21 -19
View File
@@ -2,12 +2,7 @@
from datetime import date from datetime import date
import pytest from src.models.common import ExportBaseModel, Sign, date2str
from src.domain_utils import set_category_budget, find_category_db
from src.models.common import Sign, date2str, ExportBaseModel
from src.models.category import Category, CategoryBudget, CategoryBudgetDb
from src.models.b4w import B4wExport
class TestSignEnum: class TestSignEnum:
@@ -22,7 +17,7 @@ class TestSignEnum:
"""Test Sign.__repr__ method (line 24).""" """Test Sign.__repr__ method (line 24)."""
repr_debit = repr(Sign.DEBIT) repr_debit = repr(Sign.DEBIT)
assert repr_debit == "Sign.DEBIT" assert repr_debit == "Sign.DEBIT"
repr_credit = repr(Sign.CREDIT) repr_credit = repr(Sign.CREDIT)
assert repr_credit == "Sign.CREDIT" assert repr_credit == "Sign.CREDIT"
@@ -68,64 +63,70 @@ class TestExportBaseModel:
def test_parameters_to_dict_empty(self): def test_parameters_to_dict_empty(self):
"""parameters_to_dict returns empty dict (line 70).""" """parameters_to_dict returns empty dict (line 70)."""
class TestModel(ExportBaseModel): class TestModel(ExportBaseModel):
pass pass
model = TestModel() model = TestModel()
result = model.parameters_to_dict() result = model.parameters_to_dict()
assert result == {} assert result == {}
def test_parameters_to_str_no_params(self): def test_parameters_to_str_no_params(self):
"""parameters_to_str with no parameters.""" """parameters_to_str with no parameters."""
class TestModel(ExportBaseModel): class TestModel(ExportBaseModel):
pass pass
model = TestModel() model = TestModel()
result = model.parameters_to_str() result = model.parameters_to_str()
assert result == "" assert result == ""
def test_repr_empty_params(self): def test_repr_empty_params(self):
"""__repr__ with empty parameters (line 79).""" """__repr__ with empty parameters (line 79)."""
class TestModel(ExportBaseModel): class TestModel(ExportBaseModel):
pass pass
model = TestModel() model = TestModel()
# __repr__ should return string representation with empty params # __repr__ should return string representation with empty params
repr_result = repr(model) repr_result = repr(model)
assert "TestModel" in repr_result assert "TestModel" in repr_result
def test_str_empty_params(self): def test_str_empty_params(self):
"""__str__ with empty parameters.""" """__str__ with empty parameters."""
class TestModel(ExportBaseModel): class TestModel(ExportBaseModel):
pass pass
model = TestModel() model = TestModel()
# __str__ should return string representation # __str__ should return string representation
str_result = str(model) str_result = str(model)
assert "TestModel" in str_result assert "TestModel" in str_result
def test_repr_with_params(self): def test_repr_with_params(self):
"""Test repr with actual parameters.""" """Test repr with actual parameters."""
class ConfigurableModel(ExportBaseModel): class ConfigurableModel(ExportBaseModel):
def parameters_to_dict(self) -> dict[str, str]: def parameters_to_dict(self) -> dict[str, str]:
return {"key": "value"} return {"key": "value"}
model = ConfigurableModel() model = ConfigurableModel()
# Test that repr works correctly with params # Test that repr works correctly with params
repr_result = repr(model) repr_result = repr(model)
assert "key" in repr_result or "value" in repr_result assert "key" in repr_result or "value" in repr_result
def test_str_with_params(self): def test_str_with_params(self):
"""Test str with actual parameters.""" """Test str with actual parameters."""
class ConfigurableModel(ExportBaseModel): class ConfigurableModel(ExportBaseModel):
def parameters_to_dict(self) -> dict[str, str]: def parameters_to_dict(self) -> dict[str, str]:
return {"name": "test"} return {"name": "test"}
model = ConfigurableModel() model = ConfigurableModel()
# Test that str works correctly with params # Test that str works correctly with params
str_result = str(model) str_result = str(model)
assert "name" in str_result or "test" in str_result assert "name" in str_result or "test" in str_result
@@ -136,8 +137,9 @@ class TestMyNameMethod:
def test_my_name_returns_classname(self): def test_my_name_returns_classname(self):
"""_my_name should return the class name.""" """_my_name should return the class name."""
class SubModel(ExportBaseModel): class SubModel(ExportBaseModel):
pass pass
result = SubModel._my_name() result = SubModel._my_name()
assert result == "SubModel" assert result == "SubModel"
+13 -4
View File
@@ -1,6 +1,7 @@
"""Tests for src.models.__init__.py lazy loading and __dir__.""" """Tests for src.models.__init__.py lazy loading and __dir__."""
import pytest import pytest
from src import models from src import models
@@ -10,50 +11,58 @@ class TestLazyLoading:
def test_load_b4wexport(self): def test_load_b4wexport(self):
"""Test loading B4wExport model.""" """Test loading B4wExport model."""
from src.models.b4w import B4wExport as DirectB4wExport from src.models.b4w import B4wExport as DirectB4wExport
b4w = models.B4wExport b4w = models.B4wExport
assert b4w is DirectB4wExport assert b4w is DirectB4wExport
def test_load_category(self): def test_load_category(self):
"""Test loading Category model.""" """Test loading Category model."""
from src.models.category import Category as DirectCategory from src.models.category import Category as DirectCategory
cat = models.Category cat = models.Category
assert cat is DirectCategory assert cat is DirectCategory
def test_load_categorybudgetdb(self): def test_load_categorybudgetdb(self):
"""Test loading CategoryBudgetDb model.""" """Test loading CategoryBudgetDb model."""
from src.models.category import CategoryBudgetDb as DirectCategoryBudgetDb from src.models.category import CategoryBudgetDb as DirectCategoryBudgetDb
cbdb = models.CategoryBudgetDb cbdb = models.CategoryBudgetDb
assert cbdb is DirectCategoryBudgetDb assert cbdb is DirectCategoryBudgetDb
def test_load_comdirectgiro(self): def test_load_comdirectgiro(self):
"""Test loading ComdirectGiro model.""" """Test loading ComdirectGiro model."""
from src.models.comdirect import ComdirectGiro as DirectComdirectGiro from src.models.comdirect import ComdirectGiro as DirectComdirectGiro
giro = models.ComdirectGiro giro = models.ComdirectGiro
assert giro is DirectComdirectGiro assert giro is DirectComdirectGiro
def test_load_transaction(self): def test_load_transaction(self):
"""Test loading Transaction model.""" """Test loading Transaction model."""
from src.models.transactions import Transaction as DirectTransaction from src.models.transactions import Transaction as DirectTransaction
tx = models.Transaction tx = models.Transaction
assert tx is DirectTransaction assert tx is DirectTransaction
def test_load_account(self): def test_load_account(self):
"""Test loading Account model.""" """Test loading Account model."""
from src.models.settings import Account as DirectAccount from src.models.settings import Account as DirectAccount
acc = models.Account acc = models.Account
assert acc is DirectAccount assert acc is DirectAccount
def test_load_source(self): def test_load_source(self):
"""Test loading Source model.""" """Test loading Source model."""
from src.models.settings import Source as DirectSource from src.models.settings import Source as DirectSource
src = models.Source src = models.Source
assert src is DirectSource assert src is DirectSource
def test_nonexistent_attribute_raises(self): def test_nonexistent_attribute_raises(self):
"""Test that non-existent attributes raise AttributeError.""" """Test that non-existent attributes raise AttributeError."""
with pytest.raises(AttributeError) as exc_info: with pytest.raises(AttributeError) as exc_info:
models.NonExistentModel # This is intentional: testing that accessing a non-existent attribute raises
_ = models.NonExistentModel # type: ignore[name-defined]
expected_msg = "module 'src.models' has no attribute 'NonExistentModel'" expected_msg = "module 'src.models' has no attribute 'NonExistentModel'"
assert expected_msg in str(exc_info.value) assert expected_msg in str(exc_info.value)
@@ -71,7 +80,7 @@ class TestDirMethod:
def test_dir_includes_all_exports(self): def test_dir_includes_all_exports(self):
"""Test that __dir__() includes all exported models.""" """Test that __dir__() includes all exported models."""
result = models.__dir__() result = models.__dir__()
expected_exports = { expected_exports = {
"B4wExport", "B4wExport",
"Category", "Category",
@@ -84,7 +93,7 @@ class TestDirMethod:
"Transaction", "Transaction",
"TransactionDb", "TransactionDb",
} }
# All exports should be in dir() # All exports should be in dir()
for export_name in expected_exports: for export_name in expected_exports:
assert export_name in result assert export_name in result