diff --git a/AGENTS.md b/AGENTS.md index 9c3ff70..b7fc8fc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -23,6 +23,10 @@ Use four-space indentation, double-quoted strings, and Ruff's 88-character line Use pytest and favor focused unit tests with row dictionaries or temporary SQLite databases. Integration tests are opt-in: set `MONEYWIZ_TEST_DB_PATH` to an absolute path to a disposable MoneyWiz database, then run `.venv/bin/python -m pytest tests`. They skip when the variable is unset. Do not use a live personal database. CI runs unit tests, Ruff, and mypy; no minimum coverage threshold is configured. +When automatic schema resolution is added or migrated, preserve caller-supplied aliases and converters, resolve inherited effective specifications rather than own-class defaults, and cover explicit aliases, inherited fields, partial consumer customization, and record construction. Restrict automatic row-evidence resolution to consumers that require it. + +Treat record identity admission as part of recoverable row processing: preflight every ID and GID collision before changing either lookup map, then perform insertion inside the row-recovery boundary. Cover duplicate identities, unchanged first-record lookups, and later valid-row recovery. + ## Commit & Pull Request Guidelines Recent commits use short subjects such as `Fix schema profile validation and record construction`; capitalization and prefixes vary. Write a brief imperative subject that names the change. In pull requests, explain behavior changes, list the checks run, and link a relevant issue when one exists. Include a CLI example when changing user-visible shell behavior. diff --git a/README.md b/README.md index 3bf3a34..2eae49d 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,13 @@ profile.validate() ``` `SchemaProfileResolver(db_path, baseline=None).resolve()` enriches a profile -with database-specific tag-join table details. When no baseline is provided, it +with database-specific tag-join table details and investment column choices. +Explicit baseline alias overrides are preserved; automatic detection applies to +investment fields that retain the default aliases. +Holdings and buy/sell transactions are resolved separately using physical +columns and entity-specific row evidence. If populated investment entities have +no safe column choice, resolution raises `UnsupportedInvestmentSchemaError` +before records are loaded. When no baseline is provided, it uses `DEFAULT_SCHEMA_PROFILE`. The resolved profile can be passed to `MoneywizApi` or the lower-level `DatabaseAccessor`: @@ -101,6 +107,10 @@ a raw row, reports field-resolution failures together, and creates the model instance. Record classes receive mapped field values and do not depend on `SchemaProfile`. +Managers skip malformed rows and report them through `manager.load_errors`, +which returns a copy of the skipped records' IDs, model names, and error types. +Unsupported database schemas stop loading instead of being treated as row errors. + ## Tests Run unit tests without a database: diff --git a/src/moneywiz_api/__init__.py b/src/moneywiz_api/__init__.py index dff6c14..cb3f5f6 100644 --- a/src/moneywiz_api/__init__.py +++ b/src/moneywiz_api/__init__.py @@ -3,11 +3,13 @@ from moneywiz_api.schema.schema_profile_resolver import ( DEFAULT_SCHEMA_PROFILE, SchemaProfileResolver, + UnsupportedInvestmentSchemaError, ) __all__ = [ "MoneywizApi", "SchemaProfile", "SchemaProfileResolver", + "UnsupportedInvestmentSchemaError", "DEFAULT_SCHEMA_PROFILE", ] diff --git a/src/moneywiz_api/database_accessor.py b/src/moneywiz_api/database_accessor.py index e951d2b..811e801 100644 --- a/src/moneywiz_api/database_accessor.py +++ b/src/moneywiz_api/database_accessor.py @@ -2,7 +2,7 @@ from collections import defaultdict from decimal import Decimal from pathlib import Path -from typing import Any, Dict, List, Tuple +from typing import Any, Callable, Dict, List, Tuple, cast from moneywiz_api.schema.raw_data_handler import RawDataHandler as RDH from moneywiz_api.model.record import Record @@ -59,6 +59,13 @@ def ent_for(self, typename: str) -> ENT_ID: assert ent_id is not None, f"Unknown typename {typename}" return ent_id + def _table_exists(self, table_name: str) -> bool: + row = self._con.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", + (table_name,), + ).fetchone() + return row is not None + def query_objects(self, typenames: List[str]) -> List[Any]: cur = self._con.cursor() res = cur.execute( @@ -70,7 +77,12 @@ def query_objects(self, typenames: List[str]) -> List[Any]: ) return res.fetchall() - def get_record(self, pk_id: ID, constructor: type[Record] = Record): + def _construct_record(self, row, constructor: Callable): + if isinstance(constructor, type) and issubclass(constructor, Record): + return self.schema_profile.create_record(row, constructor) + return cast(Callable, constructor)(row) + + def get_record(self, pk_id: ID, constructor: Callable = Record): cur = self._con.cursor() res = cur.execute( """ @@ -80,9 +92,9 @@ def get_record(self, pk_id: ID, constructor: type[Record] = Record): [pk_id], ) - return self.schema_profile.create_record(res.fetchone(), constructor) + return self._construct_record(res.fetchone(), constructor) - def get_record_by_gid(self, gid: GID, constructor: type[Record] = Record): + def get_record_by_gid(self, gid: GID, constructor: Callable = Record): cur = self._con.cursor() res = cur.execute( """ @@ -92,10 +104,12 @@ def get_record_by_gid(self, gid: GID, constructor: type[Record] = Record): [gid], ) - return self.schema_profile.create_record(res.fetchone(), constructor) + return self._construct_record(res.fetchone(), constructor) def get_category_assignment(self) -> Dict[ID, List[Tuple[ID, Decimal]]]: transaction_map: Dict[ID, List[Tuple[ID, Decimal]]] = defaultdict(list) + if not self._table_exists("ZCATEGORYASSIGMENT"): + return transaction_map cur = self._con.cursor() res = cur.execute( """ @@ -111,6 +125,8 @@ def get_category_assignment(self) -> Dict[ID, List[Tuple[ID, Decimal]]]: def get_refund_maps(self) -> Dict[ID, ID]: refund_to_withdraw: Dict[ID, ID] = {} + if not self._table_exists("ZWITHDRAWREFUNDTRANSACTIONLINK"): + return refund_to_withdraw cur = self._con.cursor() res = cur.execute( """ diff --git a/src/moneywiz_api/managers/record_manager.py b/src/moneywiz_api/managers/record_manager.py index 99049cc..838a936 100644 --- a/src/moneywiz_api/managers/record_manager.py +++ b/src/moneywiz_api/managers/record_manager.py @@ -1,17 +1,21 @@ from abc import ABC, abstractmethod +import logging from typing import Dict, Generic, TypeVar from moneywiz_api.database_accessor import DatabaseAccessor from moneywiz_api.model.record import Record +from moneywiz_api.schema.schema_profile_resolver import UnsupportedInvestmentSchemaError from moneywiz_api.types import GID, ID T = TypeVar("T", bound=Record) +logger = logging.getLogger(__name__) class RecordManager(ABC, Generic[T]): def __init__(self) -> None: self._records: Dict[ID, T] = {} self._gid_to_id: Dict[GID, ID] = {} + self._load_errors: list[tuple[ID, str, str]] = [] @property @abstractmethod @@ -23,22 +27,36 @@ def load(self, db_accessor: DatabaseAccessor) -> None: for record in records: typename = db_accessor.typename_for(record["Z_ENT"]) - assert typename in self.ents, ( - f"Unknown typename {typename} for record {record}" - ) - - model_cls = self.ents[typename] - obj = db_accessor.schema_profile.create_record(record, model_cls) - obj.validate() - self.add(obj) + if typename in self.ents: + try: + obj = db_accessor.schema_profile.create_record( + record, self.ents[typename] + ) + obj.validate() + self.add(obj) + except UnsupportedInvestmentSchemaError: + raise + except (AssertionError, KeyError, RuntimeError, ValueError) as exc: + record_id = record.get("Z_PK") + detail = type(exc).__name__ + self._load_errors.append((record_id, typename, detail)) + logger.debug( + "Skipping unreadable %s record %s: %s", + typename, + record_id, + detail, + ) + continue def add(self, record: T) -> None: - self._records[record.id] = record + if record.id in self._records: + raise RuntimeError(f"Duplicate id {record.id}") if record.gid in self._gid_to_id: raise RuntimeError( f"Duplicate gid for {record}, existing record Id {self._gid_to_id[record.gid]}" ) + self._records[record.id] = record self._gid_to_id[record.gid] = record.id def get(self, record_id: ID) -> T | None: @@ -53,5 +71,10 @@ def get_by_gid(self, gid: GID) -> T | None: def records(self) -> Dict[ID, T]: return self._records + @property + def load_errors(self) -> list[tuple[ID, str, str]]: + """Return records skipped during best-effort read parsing.""" + return list(self._load_errors) + def __repr__(self): return "\n".join(f"{key}: {value}" for key, value in self.records().items()) diff --git a/src/moneywiz_api/model/transaction.py b/src/moneywiz_api/model/transaction.py index 2d70231..aa90e99 100644 --- a/src/moneywiz_api/model/transaction.py +++ b/src/moneywiz_api/model/transaction.py @@ -336,6 +336,8 @@ def __init__(self, row, field_values): super().__init__(row, field_values) # Fixes + if self.original_exchange_rate == 0: + raise ValueError("cannot read a transfer with a zero exchange rate") if ( self.recipient_amount in (None, Decimal(0)) and self.original_amount is not None @@ -344,6 +346,11 @@ def __init__(self, row, field_values): self.recipient_amount = -self.original_amount * self.original_exchange_rate if self.recipient_amount is not None: self.recipient_amount = abs(self.recipient_amount) + if self.original_amount == 0 and self.recipient_amount not in ( + None, + Decimal(0), + ): + self.original_amount = self.amount def validate(self) -> None: super().validate() diff --git a/src/moneywiz_api/schema/raw_data_handler.py b/src/moneywiz_api/schema/raw_data_handler.py index a8b3527..49ae790 100644 --- a/src/moneywiz_api/schema/raw_data_handler.py +++ b/src/moneywiz_api/schema/raw_data_handler.py @@ -30,10 +30,13 @@ def get_decimal(raw_value: Any) -> Decimal: @staticmethod def filter_row(row: Dict[str, Any]) -> Dict[str, Any]: copy = {k: v for k, v in row.items()} - del copy["ZMANUALHISTORICALPRICESPERSHARE"] - del copy["ZIMPORTLINKIDARRAY2"] - del copy["ZIMPORTLINKIDARRAY"] - del copy["ZBANKLOGOPRIMARYCOLOR"] + for key in ( + "ZMANUALHISTORICALPRICESPERSHARE", + "ZIMPORTLINKIDARRAY2", + "ZIMPORTLINKIDARRAY", + "ZBANKLOGOPRIMARYCOLOR", + ): + copy.pop(key, None) return { k: v for k, v in copy.items() diff --git a/src/moneywiz_api/schema/schema_profile_resolver.py b/src/moneywiz_api/schema/schema_profile_resolver.py index 7f5a8c7..9306478 100644 --- a/src/moneywiz_api/schema/schema_profile_resolver.py +++ b/src/moneywiz_api/schema/schema_profile_resolver.py @@ -4,7 +4,14 @@ import sqlite3 from pathlib import Path +from moneywiz_api.model.investment_holding import InvestmentHolding +from moneywiz_api.model.transaction import ( + InvestmentBuyTransaction, + InvestmentSellTransaction, +) + from moneywiz_api.schema.schema_fields import ( + FieldSpec, datetime_field, decimal_field, is_one_field, @@ -15,6 +22,10 @@ from moneywiz_api.utils import get_datetime +class UnsupportedInvestmentSchemaError(ValueError): + """Raised when investment columns cannot be selected safely for a store.""" + + class SchemaProfileResolver: """Build a profile from a database path and an optional baseline profile.""" @@ -23,19 +34,149 @@ def __init__(self, db_path: str | Path, baseline: SchemaProfile | None = None): self._baseline = DEFAULT_SCHEMA_PROFILE if baseline is None else baseline def resolve(self) -> SchemaProfile: - """Return the baseline enriched with tag-table details from the database.""" - if self._baseline.tag_table_info is not None: - return self._baseline - + """Return the baseline enriched with database-specific schema details.""" connection = sqlite3.connect(self._db_path, uri=True) connection.row_factory = sqlite3.Row try: - return self._baseline.with_tag_table_info( - self._get_tags_table_info(connection) + column_map = { + model_name: dict(fields) + for model_name, fields in self._baseline.column_map.items() + } + self._resolve_investment_fields(connection, column_map) + return SchemaProfile(column_map).with_tag_table_info( + self._baseline.tag_table_info or self._get_tags_table_info(connection) ) finally: connection.close() + def _resolve_investment_fields( + self, + connection: sqlite3.Connection, + column_map: dict[str | type, dict[str, FieldSpec]], + ) -> None: + tables = { + row["name"] + for row in connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ).fetchall() + } + if "ZSYNCOBJECT" not in tables or "Z_PRIMARYKEY" not in tables: + return + + entities: dict[str, int] = { + row["Z_NAME"]: row["Z_ENT"] + for row in connection.execute( + "SELECT Z_ENT, Z_NAME FROM Z_PRIMARYKEY WHERE Z_NAME IN (?, ?, ?)", + ( + "InvestmentHolding", + "InvestmentBuyTransaction", + "InvestmentSellTransaction", + ), + ).fetchall() + } + if not entities: + return + + active_entity_ids = { + row["Z_ENT"] + for row in connection.execute( + "SELECT DISTINCT Z_ENT FROM ZSYNCOBJECT WHERE Z_ENT IN (" + + ", ".join("?" for _ in entities) + + ")", + tuple(entities.values()), + ).fetchall() + } + active_entities = { + name: entity_id + for name, entity_id in entities.items() + if entity_id in active_entity_ids + } + if not active_entities: + return + + columns = { + row["name"] + for row in connection.execute('PRAGMA table_info("ZSYNCOBJECT")') + } + for models, field_name, aliases in ( + ( + (InvestmentHolding,), + "number_of_shares", + ("ZNUMBEROFSHARES", "ZNUMBEROFSHARES1"), + ), + ( + (InvestmentBuyTransaction, InvestmentSellTransaction), + "number_of_shares", + ("ZNUMBEROFSHARES", "ZNUMBEROFSHARES1"), + ), + ( + (InvestmentBuyTransaction, InvestmentSellTransaction), + "price_per_share", + ("ZPRICEPERSHARE", "ZPRICEPERSHARE1"), + ), + ): + specs = { + model: self._baseline.specs_for(model)[field_name] + for model in models + if model.__name__ in active_entities + } + automatic_specs = { + model: spec + for model, spec in specs.items() + if spec.aliases + == DEFAULT_SCHEMA_PROFILE.specs_for(model)[field_name].aliases + } + if not automatic_specs: + continue + column = self._column_for_entities( + connection, + columns, + aliases[0], + aliases[1], + tuple(active_entities[model.__name__] for model in automatic_specs), + ) + if column is None: + consumer = ( + "holding shares column" + if models[0] is InvestmentHolding + else "transaction columns" + ) + raise UnsupportedInvestmentSchemaError( + f"unsupported investment schema profile: ambiguous {consumer}" + ) + for model, spec in automatic_specs.items(): + column_map.setdefault(model.__name__, {})[field_name] = FieldSpec( + (column,), spec.converter + ) + + @staticmethod + def _column_for_entities( + connection: sqlite3.Connection, + columns: set[str], + unsuffixed: str, + suffixed: str, + entity_ids: tuple[int, ...], + ) -> str | None: + has_unsuffixed = unsuffixed in columns + has_suffixed = suffixed in columns + if has_unsuffixed != has_suffixed: + return unsuffixed if has_unsuffixed else suffixed + if not has_unsuffixed: + return None + + placeholders = ", ".join("?" for _ in entity_ids) + counts = connection.execute( + f'SELECT COUNT("{unsuffixed}"), COUNT("{suffixed}") ' + f'FROM "ZSYNCOBJECT" WHERE "Z_ENT" IN ({placeholders})', + entity_ids, + ).fetchone() + unsuffixed_count, suffixed_count = counts[0], counts[1] + if unsuffixed_count and not suffixed_count: + return unsuffixed + if suffixed_count and not unsuffixed_count: + return suffixed + return None + @staticmethod def _get_tags_table_info(connection: sqlite3.Connection) -> TagTableInfo: cursor = connection.cursor() diff --git a/tests/unit/test_database_accessor.py b/tests/unit/test_database_accessor.py index e1d12e0..be56d83 100644 --- a/tests/unit/test_database_accessor.py +++ b/tests/unit/test_database_accessor.py @@ -49,3 +49,16 @@ def test_get_tags_map_raises_when_tags_table_is_missing(tmp_path): with pytest.raises(sqlite3.OperationalError, match="no such table"): accessor.get_tags_map() + + +def test_get_record_preserves_callable_constructors(tmp_path): + db_path = tmp_path / "moneywiz.sqlite" + accessor = _create_accessor(db_path) + accessor._con.execute("CREATE TABLE ZSYNCOBJECT (Z_PK INTEGER, ZGID TEXT)") + accessor._con.execute("INSERT INTO ZSYNCOBJECT VALUES (1, 'record-gid')") + + def get_gid(row): + return row["ZGID"] + + assert accessor.get_record(1, get_gid) == "record-gid" + assert accessor.get_record_by_gid("record-gid", get_gid) == "record-gid" diff --git a/tests/unit/test_profiled_investment_values.py b/tests/unit/test_profiled_investment_values.py new file mode 100644 index 0000000..1583b9e --- /dev/null +++ b/tests/unit/test_profiled_investment_values.py @@ -0,0 +1,532 @@ +import sqlite3 +from decimal import Decimal +from typing import cast +from unittest.mock import Mock, patch + +import pytest + +from moneywiz_api import ( + DEFAULT_SCHEMA_PROFILE, + MoneywizApi, + SchemaProfile, + SchemaProfileResolver, +) +from moneywiz_api.database_accessor import DatabaseAccessor +from moneywiz_api.managers.investment_holding_manager import InvestmentHoldingManager +from moneywiz_api.managers.transaction_manager import TransactionManager +from moneywiz_api.model.investment_holding import InvestmentHolding +from moneywiz_api.model.transaction import ( + InvestmentBuyTransaction, + InvestmentSellTransaction, +) +from moneywiz_api.schema.schema_fields import ( + decimal_field, + nullable_decimal_field, + schema_field, +) +from moneywiz_api.schema.schema_profile_resolver import UnsupportedInvestmentSchemaError + + +def investment_profile(holding_shares, transaction_shares, price): + definitions = { + name: dict(fields) for name, fields in DEFAULT_SCHEMA_PROFILE.column_map.items() + } + definitions["InvestmentHolding"]["number_of_shares"] = nullable_decimal_field( + holding_shares + ) + for model in ("InvestmentBuyTransaction", "InvestmentSellTransaction"): + definitions[model]["number_of_shares"] = decimal_field(transaction_shares) + definitions[model]["price_per_share"] = decimal_field(price) + return SchemaProfile(definitions) + + +UNSUFFIXED_PROFILE = investment_profile( + "ZNUMBEROFSHARES", "ZNUMBEROFSHARES", "ZPRICEPERSHARE" +) +MIXED_PROFILE = investment_profile( + "ZNUMBEROFSHARES", "ZNUMBEROFSHARES1", "ZPRICEPERSHARE1" +) + + +def investment_transaction_row(ent: int, amount: float) -> dict: + return { + "Z_ENT": ent, + "ZOBJECTCREATIONDATE": 0.0, + "ZGID": f"investment-{ent}", + "Z_PK": ent, + "ZRECONCILED": 0, + "ZAMOUNT1": amount, + "ZDESC2": "Investment", + "ZDATE1": 0.0, + "ZNOTES1": None, + "ZACCOUNT2": 1, + "ZFEE2": 0.0, + "ZINVESTMENTHOLDING": 24, + "ZNUMBEROFSHARES": 2.0, + "ZNUMBEROFSHARES1": 9.0, + "ZPRICEPERSHARE": 10.0, + "ZPRICEPERSHARE1": 1.0, + } + + +def investment_holding_row() -> dict: + return { + "Z_ENT": 24, + "ZOBJECTCREATIONDATE": 0.0, + "ZGID": "holding", + "Z_PK": 24, + "ZINVESTMENTACCOUNT": 1, + "ZOPENNINGNUMBEROFSHARES": None, + "ZNUMBEROFSHARES": 2.0, + "ZNUMBEROFSHARES1": 9.0, + "ZSYMBOL": "ACME", + "ZHOLDINGTYPE": None, + "ZDESC": "Acme Corp.", + "ZISPRICEPERSHAREAVAILABLEONLINE": 0, + } + + +@pytest.mark.parametrize( + ("constructor", "row"), + [ + (InvestmentBuyTransaction, investment_transaction_row(40, -20.0)), + (InvestmentSellTransaction, investment_transaction_row(41, 20.0)), + ], +) +def test_transactions_use_selected_profile_alias(constructor, row) -> None: + transaction = UNSUFFIXED_PROFILE.create_record(row, constructor) + + assert transaction.number_of_shares == Decimal("2.0") + assert transaction.price_per_share == Decimal("10.0") + + +def test_holding_uses_selected_profile_alias() -> None: + holding = UNSUFFIXED_PROFILE.create_record( + investment_holding_row(), InvestmentHolding + ) + + assert holding.number_of_shares == Decimal("2.0") + + +@pytest.mark.parametrize( + ("constructor", "row"), + [ + (InvestmentBuyTransaction, investment_transaction_row(40, -9.0)), + (InvestmentSellTransaction, investment_transaction_row(41, 9.0)), + ], +) +def test_mixed_profile_uses_consumer_specific_share_aliases(constructor, row) -> None: + transaction = MIXED_PROFILE.create_record(row, constructor) + holding = MIXED_PROFILE.create_record(investment_holding_row(), InvestmentHolding) + + assert transaction.number_of_shares == Decimal("9.0") + assert transaction.price_per_share == Decimal("1.0") + assert holding.number_of_shares == Decimal("2.0") + + +@pytest.mark.parametrize( + ("constructor", "row"), + [ + (InvestmentBuyTransaction, investment_transaction_row(40, -20.0)), + (InvestmentSellTransaction, investment_transaction_row(41, 20.0)), + ], +) +def test_observed_store_profile_uses_suffixed_transaction_price( + constructor, row +) -> None: + transaction = DEFAULT_SCHEMA_PROFILE.create_record(row, constructor) + + assert transaction.number_of_shares == Decimal("2.0") + assert transaction.price_per_share == Decimal("1.0") + + +def create_investment_database(db_path, rows): + columns = sorted({column for row in rows for column in row}) + column_sql = ", ".join(f'"{column}"' for column in columns) + placeholders = ", ".join("?" for _ in columns) + with sqlite3.connect(db_path) as connection: + connection.execute(f"CREATE TABLE ZSYNCOBJECT ({column_sql})") + connection.executemany( + f"INSERT INTO ZSYNCOBJECT VALUES ({placeholders})", + [[row.get(column) for column in columns] for row in rows], + ) + connection.execute("CREATE TABLE Z_PRIMARYKEY (Z_ENT INTEGER, Z_NAME TEXT)") + connection.executemany( + "INSERT INTO Z_PRIMARYKEY VALUES (?, ?)", + [ + (24, "InvestmentHolding"), + (40, "InvestmentBuyTransaction"), + (41, "InvestmentSellTransaction"), + ] + + [ + (100 + index, name) + for index, name in enumerate(TransactionManager().ents) + if name not in {"InvestmentBuyTransaction", "InvestmentSellTransaction"} + ], + ) + connection.execute( + "CREATE TABLE Z_37TAGS (Z_3TRANSACTIONS INTEGER, Z_4TAGS INTEGER)" + ) + connection.close() + + +@pytest.mark.parametrize( + ("holding_shares", "transaction_shares", "price"), + [ + ("ZNUMBEROFSHARES", "ZNUMBEROFSHARES", "ZPRICEPERSHARE"), + ("ZNUMBEROFSHARES", "ZNUMBEROFSHARES1", "ZPRICEPERSHARE1"), + ("ZNUMBEROFSHARES", "ZNUMBEROFSHARES", "ZPRICEPERSHARE1"), + ("ZNUMBEROFSHARES1", "ZNUMBEROFSHARES1", "ZPRICEPERSHARE1"), + ], +) +def test_managers_and_direct_reads_use_same_resolved_profile( + tmp_path, holding_shares, transaction_shares, price +) -> None: + rows = [ + investment_holding_row(), + investment_transaction_row(40, -20.0), + investment_transaction_row(41, 20.0), + ] + for row in rows: + for column in ( + "ZNUMBEROFSHARES", + "ZNUMBEROFSHARES1", + "ZPRICEPERSHARE", + "ZPRICEPERSHARE1", + ): + row[column] = None + row[holding_shares if row["Z_ENT"] == 24 else transaction_shares] = 2.0 + if row["Z_ENT"] != 24: + row[price] = 10.0 + db_path = tmp_path / "moneywiz.sqlite" + create_investment_database(db_path, rows) + profile = SchemaProfileResolver(db_path).resolve() + accessor = DatabaseAccessor(db_path, profile) + try: + transaction_manager = TransactionManager() + transaction_manager.load(accessor) + holding_manager = InvestmentHoldingManager() + holding_manager.load(accessor) + + for pk, constructor in ( + (40, InvestmentBuyTransaction), + (41, InvestmentSellTransaction), + ): + loaded = transaction_manager.get(pk) + direct = accessor.get_record(pk, constructor) + by_gid = accessor.get_record_by_gid(f"investment-{pk}", constructor) + assert loaded == direct == by_gid + assert direct.number_of_shares == Decimal("2.0") + assert direct.price_per_share == Decimal("10.0") + holding = accessor.get_record(24, InvestmentHolding) + assert holding_manager.get(24) == holding + assert holding.number_of_shares == Decimal("2.0") + assert accessor.schema_profile is profile + assert transaction_manager.load_errors == holding_manager.load_errors == [] + finally: + accessor._con.close() + + +class ManagerAccessor: + schema_profile = UNSUFFIXED_PROFILE + + def __init__(self, typename, rows): + self.typename = typename + self.rows = rows + + def query_objects(self, _typenames): + return self.rows + + def typename_for(self, _ent_id): + return self.typename + + def get_category_assignment(self): + return {} + + def get_refund_maps(self): + return {} + + def get_tags_map(self): + return {} + + +@pytest.mark.parametrize("malformation", ["invalid", "null", "missing", "identity"]) +def test_managers_report_bad_rows_and_load_later_valid_rows(malformation) -> None: + bad_row = investment_transaction_row(40, -20.0) + if malformation == "missing": + del bad_row["ZPRICEPERSHARE"] + elif malformation == "identity": + bad_row["ZGID"] = None + else: + bad_row["ZPRICEPERSHARE"] = None if malformation == "null" else "invalid" + good_row = {**investment_transaction_row(40, -20.0), "Z_PK": 42, "ZGID": "valid"} + manager = TransactionManager() + + manager.load( + cast( + DatabaseAccessor, + ManagerAccessor("InvestmentBuyTransaction", [bad_row, good_row]), + ) + ) + + assert list(manager.records()) == [42] + error_type = "AssertionError" if malformation == "identity" else "RuntimeError" + assert manager.load_errors == [(40, "InvestmentBuyTransaction", error_type)] + manager.load_errors.clear() + assert len(manager.load_errors) == 1 + + +@pytest.mark.parametrize("identity", ["gid", "id", "both"]) +def test_manager_duplicate_refusal_preserves_existing_records(identity) -> None: + manager = TransactionManager() + row = investment_transaction_row(40, -20.0) + first = UNSUFFIXED_PROFILE.create_record(row, InvestmentBuyTransaction) + manager.add(first) + duplicate_row = { + **row, + "Z_PK": 41 if identity == "gid" else 40, + "ZGID": "duplicate" if identity == "id" else row["ZGID"], + } + duplicate = UNSUFFIXED_PROFILE.create_record( + duplicate_row, InvestmentBuyTransaction + ) + + with pytest.raises(RuntimeError, match="Duplicate"): + manager.add(duplicate) + + assert list(manager.records()) == [40] + assert manager.get(40) is first + assert manager.get_by_gid(row["ZGID"]) is first + assert manager.get_by_gid("duplicate") is None + assert manager.load_errors == [] + + +@pytest.mark.parametrize("identity", ["gid", "id", "both"]) +def test_manager_reports_duplicates_and_loads_later_rows(tmp_path, identity) -> None: + first_row = investment_transaction_row(40, -20.0) + first_row["ZNUMBEROFSHARES1"] = None + first_row["ZPRICEPERSHARE1"] = None + duplicate_row = { + **first_row, + "Z_PK": 41 if identity == "gid" else 40, + "ZGID": "duplicate" if identity == "id" else first_row["ZGID"], + "ZAMOUNT1": -40.0, + "ZNUMBEROFSHARES": 4.0, + } + later_row = {**first_row, "Z_PK": 42, "ZGID": "later"} + db_path = tmp_path / "moneywiz.sqlite" + create_investment_database(db_path, [first_row, duplicate_row, later_row]) + profile = SchemaProfileResolver(db_path, UNSUFFIXED_PROFILE).resolve() + accessor = DatabaseAccessor(db_path, profile) + manager = TransactionManager() + + try: + manager.load(accessor) + + assert list(manager.records()) == [40, 42] + first = manager.get(40) + assert first is not None + assert manager.get_by_gid(first_row["ZGID"]) is first + assert first.amount == Decimal("-20.0") + assert manager.get_by_gid("later") is manager.get(42) + assert manager.get_by_gid("duplicate") is None + assert manager.load_errors == [ + (duplicate_row["Z_PK"], "InvestmentBuyTransaction", "RuntimeError") + ] + assert ( + manager.category_assignment == manager.refund_maps == manager.tags_map == {} + ) + finally: + accessor._con.close() + + +def test_ambiguous_store_schema_stops_before_record_loading(tmp_path) -> None: + db_path = tmp_path / "moneywiz.sqlite" + create_investment_database(db_path, [investment_transaction_row(40, -20.0)]) + + with patch("moneywiz_api.moneywiz_api.DatabaseAccessor") as accessor: + with pytest.raises(UnsupportedInvestmentSchemaError): + MoneywizApi(db_path) + accessor.assert_not_called() + + +def test_manager_does_not_skip_unsupported_schema_errors() -> None: + accessor = Mock() + accessor.query_objects.return_value = [investment_transaction_row(40, -20.0)] + accessor.typename_for.return_value = "InvestmentBuyTransaction" + accessor.schema_profile.create_record.side_effect = ( + UnsupportedInvestmentSchemaError("unsupported schema") + ) + manager = TransactionManager() + + with pytest.raises(UnsupportedInvestmentSchemaError): + manager.load(accessor) + assert manager.load_errors == [] + + +def default_investment_rows(): + rows = [ + investment_holding_row(), + investment_transaction_row(40, -20.0), + investment_transaction_row(41, 20.0), + ] + for row in rows: + row["ZNUMBEROFSHARES1"] = None + row["ZPRICEPERSHARE"] = None + if row["Z_ENT"] != 24: + row["ZPRICEPERSHARE1"] = 10.0 + return rows + + +@pytest.mark.parametrize( + ("model", "field_name"), + [ + (InvestmentHolding, "number_of_shares"), + (InvestmentBuyTransaction, "number_of_shares"), + (InvestmentSellTransaction, "number_of_shares"), + (InvestmentBuyTransaction, "price_per_share"), + (InvestmentSellTransaction, "price_per_share"), + ], +) +def test_resolver_preserves_custom_investment_aliases(tmp_path, model, field_name): + definitions = { + name: dict(fields) for name, fields in DEFAULT_SCHEMA_PROFILE.column_map.items() + } + spec = DEFAULT_SCHEMA_PROFILE.specs_for(model)[field_name] + definitions[model.__name__][field_name] = schema_field( + "CUSTOM_FIELD", converter=spec.converter + ) + rows = default_investment_rows() + entity = { + InvestmentHolding: 24, + InvestmentBuyTransaction: 40, + InvestmentSellTransaction: 41, + }[model] + row = next(row for row in rows if row["Z_ENT"] == entity) + row["CUSTOM_FIELD"] = 7.0 if field_name == "number_of_shares" else 12.0 + # Ambiguous stock aliases on the customized consumer cannot override its mapping + # or invalidate inference for the other consumer. + row[ + "ZNUMBEROFSHARES1" if field_name == "number_of_shares" else "ZPRICEPERSHARE" + ] = 9.0 + if entity != 24: + amount = 70.0 if field_name == "number_of_shares" else 24.0 + row["ZAMOUNT1"] = -amount if entity == 40 else amount + db_path = tmp_path / "moneywiz.sqlite" + create_investment_database(db_path, rows) + baseline = SchemaProfile(definitions) + + profile = SchemaProfileResolver(db_path, baseline).resolve() + record = profile.create_record(row, model) + record.validate() + + assert getattr(record, field_name) == Decimal(str(row["CUSTOM_FIELD"])) + assert profile.specs_for(model)[field_name].aliases == ("CUSTOM_FIELD",) + assert profile.specs_for(model)[field_name].converter is spec.converter + assert baseline.column_map == definitions + + +@pytest.mark.parametrize("custom_aliases", [False, True]) +def test_resolver_preserves_inherited_investment_fields(tmp_path, custom_aliases): + definitions = { + name: dict(fields) for name, fields in DEFAULT_SCHEMA_PROFILE.column_map.items() + } + definitions["InvestmentTransaction"] = definitions["InvestmentBuyTransaction"] + definitions["InvestmentBuyTransaction"] = {} + definitions["InvestmentSellTransaction"] = {} + if custom_aliases: + definitions["InvestmentTransaction"]["number_of_shares"] = decimal_field( + "CUSTOM_SHARES" + ) + definitions["InvestmentTransaction"]["price_per_share"] = decimal_field( + "CUSTOM_PRICE" + ) + rows = default_investment_rows() + for row in rows[1:]: + row["CUSTOM_SHARES"] = 2.0 + row["CUSTOM_PRICE"] = 10.0 + row["ZNUMBEROFSHARES"] = None + row["ZNUMBEROFSHARES1"] = 2.0 + row["ZPRICEPERSHARE"] = 10.0 + row["ZPRICEPERSHARE1"] = None + db_path = tmp_path / "moneywiz.sqlite" + create_investment_database(db_path, rows) + baseline = SchemaProfile(definitions) + + profile = SchemaProfileResolver(db_path, baseline).resolve() + + for row, model in zip( + rows[1:], (InvestmentBuyTransaction, InvestmentSellTransaction) + ): + record = profile.create_record(row, model) + record.validate() + assert record.number_of_shares == Decimal("2.0") + assert record.price_per_share == Decimal("10.0") + assert baseline.column_map == definitions + + +def test_unrelated_custom_mapping_keeps_investment_detection_and_converter(tmp_path): + definitions = { + name: dict(fields) for name, fields in DEFAULT_SCHEMA_PROFILE.column_map.items() + } + definitions["Tag"]["name"] = schema_field("CUSTOM_TAG_NAME") + + def converter(value): + return Decimal(str(value)) + + definitions["InvestmentBuyTransaction"]["price_per_share"] = schema_field( + "ZPRICEPERSHARE1", converter=converter + ) + rows = default_investment_rows() + for row in rows[1:]: + row["ZPRICEPERSHARE"] = 10.0 + row["ZPRICEPERSHARE1"] = None + db_path = tmp_path / "moneywiz.sqlite" + create_investment_database(db_path, rows) + + profile = SchemaProfileResolver(db_path, SchemaProfile(definitions)).resolve() + record = profile.create_record(rows[1], InvestmentBuyTransaction) + record.validate() + + assert record.price_per_share == Decimal("10.0") + assert ( + profile.specs_for(InvestmentBuyTransaction)["price_per_share"].converter + is converter + ) + assert profile.column_map["Tag"]["name"].aliases == ("CUSTOM_TAG_NAME",) + + +def test_custom_price_does_not_require_stock_price_columns(tmp_path): + definitions = { + name: dict(fields) for name, fields in DEFAULT_SCHEMA_PROFILE.column_map.items() + } + definitions["InvestmentBuyTransaction"]["price_per_share"] = decimal_field( + "CUSTOM_PRICE" + ) + row = investment_transaction_row(40, -24.0) + row["ZNUMBEROFSHARES1"] = None + row["CUSTOM_PRICE"] = 12.0 + del row["ZPRICEPERSHARE"] + del row["ZPRICEPERSHARE1"] + db_path = tmp_path / "moneywiz.sqlite" + create_investment_database(db_path, [row]) + + profile = SchemaProfileResolver(db_path, SchemaProfile(definitions)).resolve() + record = profile.create_record(row, InvestmentBuyTransaction) + record.validate() + assert record.price_per_share == Decimal("12.0") + + +def test_custom_price_does_not_bypass_ambiguous_default_shares(tmp_path): + definitions = { + name: dict(fields) for name, fields in DEFAULT_SCHEMA_PROFILE.column_map.items() + } + definitions["InvestmentBuyTransaction"]["price_per_share"] = decimal_field( + "CUSTOM_PRICE" + ) + row = {**investment_transaction_row(40, -24.0), "CUSTOM_PRICE": 12.0} + db_path = tmp_path / "moneywiz.sqlite" + create_investment_database(db_path, [row]) + + with pytest.raises(UnsupportedInvestmentSchemaError): + SchemaProfileResolver(db_path, SchemaProfile(definitions)).resolve() diff --git a/tests/unit/test_schema_profiles.py b/tests/unit/test_schema_profiles.py new file mode 100644 index 0000000..291b17a --- /dev/null +++ b/tests/unit/test_schema_profiles.py @@ -0,0 +1,249 @@ +import sqlite3 + +import pytest + +from moneywiz_api import DEFAULT_SCHEMA_PROFILE, SchemaProfile, SchemaProfileResolver +from moneywiz_api.schema.raw_data_handler import RawDataHandler +from moneywiz_api.schema.schema_profile_resolver import UnsupportedInvestmentSchemaError + + +def make_connection(db_path, *columns: str) -> sqlite3.Connection: + connection = sqlite3.connect(db_path) + column_sql = ", ".join(f'"{column}" FLOAT' for column in columns) + connection.execute( + f"CREATE TABLE ZSYNCOBJECT (Z_PK INTEGER, Z_ENT INTEGER, {column_sql})" + ) + connection.execute("CREATE TABLE Z_PRIMARYKEY (Z_ENT INTEGER, Z_NAME TEXT)") + connection.executemany( + "INSERT INTO Z_PRIMARYKEY VALUES (?, ?)", + [ + (24, "InvestmentHolding"), + (40, "InvestmentBuyTransaction"), + (41, "InvestmentSellTransaction"), + ], + ) + connection.execute( + "CREATE TABLE Z_37TAGS (Z_3TRANSACTIONS INTEGER, Z_4TAGS INTEGER)" + ) + return connection + + +def insert_investment_row( + connection: sqlite3.Connection, entity: int, **values: float | None +) -> None: + columns = ["Z_ENT", *values] + placeholders = ", ".join("?" for _ in columns) + quoted_columns = ", ".join(f'"{column}"' for column in columns) + connection.execute( + f"INSERT INTO ZSYNCOBJECT ({quoted_columns}) VALUES ({placeholders})", + [entity, *values.values()], + ) + + +def resolve_profile(connection, db_path): + connection.commit() + connection.close() + return SchemaProfileResolver(db_path).resolve() + + +@pytest.mark.parametrize( + ("shares_column", "price_column"), + [ + ("ZNUMBEROFSHARES1", "ZPRICEPERSHARE1"), + ("ZNUMBEROFSHARES", "ZPRICEPERSHARE"), + ("ZNUMBEROFSHARES", "ZPRICEPERSHARE1"), + ], +) +def test_detects_investment_columns_when_aliases_are_absent( + tmp_path, shares_column, price_column +) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection(db_path, shares_column, price_column) + insert_investment_row(connection, 24, **{shares_column: 2.0}) + for entity in (40, 41): + insert_investment_row( + connection, entity, **{shares_column: 2.0, price_column: 10.0} + ) + + profile = resolve_profile(connection, db_path) + + assert profile.column_map["InvestmentHolding"]["number_of_shares"].aliases == ( + shares_column, + ) + for model in ("InvestmentBuyTransaction", "InvestmentSellTransaction"): + assert profile.column_map[model]["number_of_shares"].aliases == (shares_column,) + assert profile.column_map[model]["price_per_share"].aliases == (price_column,) + + +def test_detects_consumer_specific_mixed_share_columns(tmp_path) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection( + db_path, "ZNUMBEROFSHARES", "ZNUMBEROFSHARES1", "ZPRICEPERSHARE1" + ) + insert_investment_row(connection, 24, ZNUMBEROFSHARES=2.0, ZNUMBEROFSHARES1=None) + insert_investment_row( + connection, 40, ZNUMBEROFSHARES=None, ZNUMBEROFSHARES1=9.0, ZPRICEPERSHARE1=10.0 + ) + insert_investment_row( + connection, 41, ZNUMBEROFSHARES=None, ZNUMBEROFSHARES1=3.0, ZPRICEPERSHARE1=11.0 + ) + + profile = resolve_profile(connection, db_path) + + assert profile.column_map["InvestmentHolding"]["number_of_shares"].aliases == ( + "ZNUMBEROFSHARES", + ) + for model in ("InvestmentBuyTransaction", "InvestmentSellTransaction"): + assert profile.column_map[model]["number_of_shares"].aliases == ( + "ZNUMBEROFSHARES1", + ) + + +def test_detects_observed_store_profile_from_entity_rows(tmp_path) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection( + db_path, "ZNUMBEROFSHARES", "ZPRICEPERSHARE", "ZPRICEPERSHARE1" + ) + insert_investment_row( + connection, 24, ZNUMBEROFSHARES=None, ZPRICEPERSHARE=15.0, ZPRICEPERSHARE1=None + ) + for entity in (40, 41): + insert_investment_row( + connection, + entity, + ZNUMBEROFSHARES=2.0, + ZPRICEPERSHARE=None, + ZPRICEPERSHARE1=10.0, + ) + + profile = resolve_profile(connection, db_path) + + for model in ("InvestmentBuyTransaction", "InvestmentSellTransaction"): + assert profile.column_map[model]["number_of_shares"].aliases == ( + "ZNUMBEROFSHARES", + ) + assert profile.column_map[model]["price_per_share"].aliases == ( + "ZPRICEPERSHARE1", + ) + + +@pytest.mark.parametrize("prices", [(10.0, 1.0), (None, None)]) +def test_rejects_dual_price_columns_without_discriminating_rows( + tmp_path, prices +) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection( + db_path, "ZNUMBEROFSHARES", "ZPRICEPERSHARE", "ZPRICEPERSHARE1" + ) + insert_investment_row( + connection, + 40, + ZNUMBEROFSHARES=2.0, + ZPRICEPERSHARE=prices[0], + ZPRICEPERSHARE1=prices[1], + ) + + with pytest.raises(UnsupportedInvestmentSchemaError, match="transaction columns"): + resolve_profile(connection, db_path) + + +@pytest.mark.parametrize("shares", [(2.0, 9.0), (None, None)]) +def test_rejects_ambiguous_holding_share_columns(tmp_path, shares) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection( + db_path, "ZNUMBEROFSHARES", "ZNUMBEROFSHARES1", "ZPRICEPERSHARE1" + ) + insert_investment_row( + connection, 24, ZNUMBEROFSHARES=shares[0], ZNUMBEROFSHARES1=shares[1] + ) + + with pytest.raises(UnsupportedInvestmentSchemaError, match="holding shares column"): + resolve_profile(connection, db_path) + + +def test_rejects_conflicting_transaction_column_evidence(tmp_path) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection( + db_path, "ZNUMBEROFSHARES", "ZNUMBEROFSHARES1", "ZPRICEPERSHARE1" + ) + insert_investment_row(connection, 40, ZNUMBEROFSHARES=2.0, ZPRICEPERSHARE1=10.0) + insert_investment_row(connection, 41, ZNUMBEROFSHARES1=3.0, ZPRICEPERSHARE1=11.0) + + with pytest.raises(UnsupportedInvestmentSchemaError, match="transaction columns"): + resolve_profile(connection, db_path) + + +def test_rejects_missing_investment_columns(tmp_path) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection(db_path, "ZNUMBEROFSHARES") + insert_investment_row(connection, 40, ZNUMBEROFSHARES=2.0) + + with pytest.raises(UnsupportedInvestmentSchemaError, match="transaction columns"): + resolve_profile(connection, db_path) + + +def test_empty_investment_entities_do_not_require_row_evidence(tmp_path) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection( + db_path, + "ZNUMBEROFSHARES", + "ZNUMBEROFSHARES1", + "ZPRICEPERSHARE", + "ZPRICEPERSHARE1", + ) + + profile = resolve_profile(connection, db_path) + + assert profile.column_map == DEFAULT_SCHEMA_PROFILE.column_map + assert profile.tag_table_info is not None + + +@pytest.mark.parametrize("entity", [24, 40]) +def test_resolves_only_populated_investment_consumers(tmp_path, entity) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection( + db_path, + "ZNUMBEROFSHARES", + "ZNUMBEROFSHARES1", + "ZPRICEPERSHARE", + "ZPRICEPERSHARE1", + ) + if entity == 24: + insert_investment_row(connection, entity, ZNUMBEROFSHARES=0.0) + else: + insert_investment_row( + connection, entity, ZNUMBEROFSHARES1=2.0, ZPRICEPERSHARE1=10.0 + ) + + profile = resolve_profile(connection, db_path) + + model = "InvestmentHolding" if entity == 24 else "InvestmentBuyTransaction" + column = "ZNUMBEROFSHARES" if entity == 24 else "ZNUMBEROFSHARES1" + assert profile.column_map[model]["number_of_shares"].aliases == (column,) + + +def test_resolves_investment_fields_with_preconfigured_tag_table(tmp_path) -> None: + db_path = tmp_path / "moneywiz.sqlite" + connection = make_connection(db_path, "ZNUMBEROFSHARES1", "ZPRICEPERSHARE") + insert_investment_row(connection, 40, ZNUMBEROFSHARES1=2.0, ZPRICEPERSHARE=10.0) + default = resolve_profile(connection, db_path) + baseline = SchemaProfile(DEFAULT_SCHEMA_PROFILE.column_map, default.tag_table_info) + + profile = SchemaProfileResolver(db_path, baseline).resolve() + + assert profile.tag_table_info == baseline.tag_table_info + assert profile.column_map["InvestmentBuyTransaction"][ + "number_of_shares" + ].aliases == ("ZNUMBEROFSHARES1",) + assert profile.column_map["InvestmentBuyTransaction"][ + "price_per_share" + ].converter is ( + DEFAULT_SCHEMA_PROFILE.column_map["InvestmentBuyTransaction"][ + "price_per_share" + ].converter + ) + assert baseline.column_map == DEFAULT_SCHEMA_PROFILE.column_map + + +def test_filter_row_accepts_missing_optional_blob_columns() -> None: + assert RawDataHandler.filter_row({"Z_PK": 1}) == {"Z_PK": 1} diff --git a/tests/unit/test_transaction_compatibility.py b/tests/unit/test_transaction_compatibility.py new file mode 100644 index 0000000..960f215 --- /dev/null +++ b/tests/unit/test_transaction_compatibility.py @@ -0,0 +1,124 @@ +from decimal import Decimal +from typing import cast + +import pytest + +from moneywiz_api import DEFAULT_SCHEMA_PROFILE +from moneywiz_api.database_accessor import DatabaseAccessor +from moneywiz_api.managers.record_manager import RecordManager +from moneywiz_api.model.transaction import TransferWithdrawTransaction + + +def transfer_withdraw_row(**overrides): + row = { + "Z_ENT": 46, + "ZOBJECTCREATIONDATE": 0.0, + "ZGID": "transfer-withdraw", + "Z_PK": 1, + "ZRECONCILED": 0, + "ZAMOUNT1": -10.0, + "ZDESC2": "Transfer", + "ZDATE1": 0.0, + "ZNOTES1": None, + "ZACCOUNT2": 10, + "ZRECIPIENTACCOUNT1": 20, + "ZRECIPIENTTRANSACTION": 2, + "ZORIGINALAMOUNT": 0.0, + "ZORIGINALCURRENCY": "EUR", + "ZORIGINALRECIPIENTAMOUNT": 10.0, + "ZORIGINALRECIPIENTCURRENCY": "USD", + "ZORIGINALFEE": None, + "ZORIGINALFEECURRENCY": None, + "ZORIGINALEXCHANGERATE": 0.0, + } + row.update(overrides) + return row + + +@pytest.mark.parametrize( + ("original_amount", "recipient_amount"), [(0.0, 10.0), (-10.0, 10.0), (-10.0, 0.0)] +) +def test_transfer_withdraw_rejects_zero_rate(original_amount, recipient_amount) -> None: + with pytest.raises(ValueError, match="zero exchange rate"): + DEFAULT_SCHEMA_PROFILE.create_record( + transfer_withdraw_row( + ZORIGINALAMOUNT=original_amount, + ZORIGINALRECIPIENTAMOUNT=recipient_amount, + ), + TransferWithdrawTransaction, + ) + + +@pytest.mark.parametrize("original_amount", [0.0, -10.0]) +def test_transfer_withdraw_preserves_amount_with_nonzero_rate(original_amount) -> None: + transaction = DEFAULT_SCHEMA_PROFILE.create_record( + transfer_withdraw_row( + ZORIGINALAMOUNT=original_amount, ZORIGINALEXCHANGERATE=1.0 + ), + TransferWithdrawTransaction, + ) + transaction.validate() + + assert transaction.original_amount == Decimal("-10.0") + + +def test_transfer_withdraw_preserves_stored_amount_with_rounded_rate() -> None: + transaction = DEFAULT_SCHEMA_PROFILE.create_record( + transfer_withdraw_row( + ZAMOUNT1=-3.333, + ZORIGINALRECIPIENTAMOUNT=10.0, + ZORIGINALEXCHANGERATE=3.0, + ), + TransferWithdrawTransaction, + ) + + assert transaction.original_amount == Decimal("-3.333") + + +class TransferWithdrawManager(RecordManager): + @property + def ents(self): + return {"TransferWithdrawTransaction": TransferWithdrawTransaction} + + +class TransferWithdrawAccessor: + schema_profile = DEFAULT_SCHEMA_PROFILE + + def __init__(self, rows): + self.rows = rows + + def query_objects(self, _typenames): + return self.rows + + def typename_for(self, _ent_id): + return "TransferWithdrawTransaction" + + +@pytest.mark.parametrize( + ("original_amount", "recipient_amount"), [(0.0, 10.0), (-10.0, 10.0), (-10.0, 0.0)] +) +def test_transfer_manager_skips_zero_rate_and_loads_later_valid_row( + original_amount, recipient_amount +) -> None: + manager = TransferWithdrawManager() + manager.load( + cast( + DatabaseAccessor, + TransferWithdrawAccessor( + [ + transfer_withdraw_row( + ZORIGINALAMOUNT=original_amount, + ZORIGINALRECIPIENTAMOUNT=recipient_amount, + ), + transfer_withdraw_row( + Z_PK=2, + ZGID="valid-transfer-withdraw", + ZORIGINALEXCHANGERATE=1.0, + ), + ] + ), + ) + ) + + assert list(manager.records()) == [2] + assert manager.load_errors == [(1, "TransferWithdrawTransaction", "ValueError")]