k1nq
63d065011b
- Updated timestamps and AST hashes in manifest.json for various backend and frontend files to reflect recent changes. - Added new files related to MCC categories and optional hold reasons in the manifest. - Modified the README in the infra directory to correct the linting tool from 'mypy' to 'ty' for type checking in the backend project.
100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from alembic.config import Config
|
|
from sqlalchemy import create_engine, inspect
|
|
from sqlalchemy.orm import Session
|
|
|
|
from alembic import command
|
|
from app.categorization.models import CategoryMcc
|
|
from app.config import get_settings
|
|
from app.identity.models import User
|
|
from app.ledger.models import Account, Category
|
|
from app.spaces.models import FinancialSpace, Membership
|
|
|
|
|
|
def test_upgrade_backfills_one_pair_without_claiming_same_named_rows(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
) -> None:
|
|
database_url = f"sqlite:///{tmp_path / 'migration.db'}"
|
|
monkeypatch.setenv("DATABASE_URL", database_url)
|
|
get_settings.cache_clear()
|
|
config = Config(str(Path(__file__).parents[1] / "alembic.ini"))
|
|
config.set_main_option("script_location", str(Path(__file__).parents[1] / "alembic"))
|
|
|
|
try:
|
|
command.upgrade(config, "0002")
|
|
engine = create_engine(database_url)
|
|
with Session(engine) as db:
|
|
user = User(email="migration@example.com", password_hash="unused")
|
|
space = FinancialSpace(name="Legacy")
|
|
db.add_all([user, space])
|
|
db.flush()
|
|
db.add(Membership(space_id=space.id, user_id=user.id, role="owner"))
|
|
db.add(
|
|
Account(
|
|
space_id=space.id,
|
|
name="Savings",
|
|
kind="debit",
|
|
opened_on=date(2026, 1, 1),
|
|
)
|
|
)
|
|
db.add(
|
|
Category(
|
|
space_id=space.id,
|
|
name_override="Savings",
|
|
kind="monthly",
|
|
)
|
|
)
|
|
db.commit()
|
|
space_id = space.id
|
|
|
|
command.upgrade(config, "head")
|
|
command.upgrade(config, "head")
|
|
|
|
with Session(engine) as db:
|
|
accounts = db.query(Account).filter_by(space_id=space_id).all()
|
|
categories = db.query(Category).filter_by(space_id=space_id).all()
|
|
system_account = [item for item in accounts if item.system_key == "savings"]
|
|
system_category = [item for item in categories if item.system_key == "savings"]
|
|
assert len(system_account) == 1
|
|
assert len(system_category) == 1
|
|
assert system_category[0].linked_account_id == system_account[0].id
|
|
assert any(item.name == "Savings" and item.system_key is None for item in accounts)
|
|
assert any(
|
|
item.name_override == "Savings" and item.system_key is None for item in categories
|
|
)
|
|
default_categories = {
|
|
item.system_key: item
|
|
for item in categories
|
|
if item.system_key not in {None, "savings"}
|
|
}
|
|
assert len(default_categories) == 21
|
|
assert "groceries" in default_categories and "others" in default_categories
|
|
assert (
|
|
db.query(CategoryMcc)
|
|
.filter_by(
|
|
space_id=space_id,
|
|
category_id=default_categories["groceries"].id,
|
|
mcc_code=5411,
|
|
)
|
|
.count()
|
|
== 1
|
|
)
|
|
|
|
command.downgrade(config, "0003")
|
|
schema = inspect(engine)
|
|
assert schema.has_table("mcc_codes")
|
|
assert schema.has_table("bank_category_mappings")
|
|
assert not schema.has_table("category_mcc_codes")
|
|
with Session(engine) as db:
|
|
assert (
|
|
db.query(Category)
|
|
.filter(Category.space_id == space_id, Category.system_key == "groceries")
|
|
.count()
|
|
== 0
|
|
)
|
|
finally:
|
|
get_settings.cache_clear()
|