diff --git a/README.md b/README.md index cbf93dd..ffd61f4 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,7 @@ A Python API to access MoneyWiz Sqlite database. - [Get Started](#get-started) - [Tests](#tests) - [Contribution](#contribution) +- [Read completeness](#read-completeness) ## Get Started @@ -64,3 +65,26 @@ MONEYWIZ_TEST_DB_PATH=/absolute/path/to/test.sqlite uv run pytest tests ## Contribution This project is in very early stage, all contributions are welcomed! + +## Read completeness + +```python +from contextlib import closing +from pathlib import Path + +from moneywiz_api import MoneywizApi + +with closing(MoneywizApi(Path("moneywiz.sqlite"), managers=("accounts",))) as api: + report = api.completeness() + if not report.complete: + raise RuntimeError("incomplete read") + accounts = api.account_manager.records() +``` + +By default, all managers load. Select from `accounts`, `payees`, `categories`, +`transactions`, `investment_holdings`, and `tags`. `load(managers=...)` refreshes +the selection and previously loaded managers in one read transaction. A failed +reload invalidates those managers. Reports contain source and parsed IDs, +identity-only skipped row diagnostics, and transaction relationship status. +Unknown layouts prevent a complete result. Reopen the read-only API after a +schema change. diff --git a/src/moneywiz_api/database_accessor.py b/src/moneywiz_api/database_accessor.py index 9f8a6b3..14a0a8b 100644 --- a/src/moneywiz_api/database_accessor.py +++ b/src/moneywiz_api/database_accessor.py @@ -1,18 +1,39 @@ import re import sqlite3 from collections import defaultdict +from contextlib import contextmanager from decimal import Decimal from pathlib import Path -from typing import Any, Callable, Dict, List, Tuple +from typing import Any, Callable, Dict, List, Tuple, cast from moneywiz_api.model.raw_data_handler import RawDataHandler as RDH from moneywiz_api.model.record import Record +from moneywiz_api.model.schema_mapped_row import mapped_row +from moneywiz_api.read_result import ( + LoadErrorKind, + RelationshipLoadReport, + RelationshipStorage, + SkippedRecord, +) +from moneywiz_api.schema_profile import SchemaProfile, detect_schema_profile from moneywiz_api.types import ENT_ID, GID, ID +class DatabaseSchemaError(ValueError): + """Raised when a readable SQLite file is not a supported MoneyWiz store.""" + + +def require_integer_identity(value: object) -> None: + """Require an uncoerced integer identity.""" + if type(value) is not int: + raise AssertionError() + + class DatabaseAccessor: def __init__(self, db_path: Path): - self._con = sqlite3.connect(db_path, uri=True) + self._con = sqlite3.connect( + f"{Path(db_path).expanduser().resolve().as_uri()}?mode=ro", uri=True + ) def dict_factory(cursor, row): record = {} @@ -20,30 +41,89 @@ def dict_factory(cursor, row): record[col[0]] = row[idx] return record - self._con.row_factory = dict_factory + try: + self._con.row_factory = dict_factory + self._initialize_schema_cache() + except BaseException: + self._con.close() + raise - self._ent_to_typename: Dict[ENT_ID, str] = self._load_primarykey() - self._typename_to_ent: Dict[str, ENT_ID] = { - v: k for k, v in self._ent_to_typename.items() - } + def _initialize_schema_cache(self) -> None: + """Bind all schema-dependent caches from one SQLite snapshot.""" + self._admission_depth = 0 + with self._raw_read_transaction(): + if not self._table_exists("Z_PRIMARYKEY") or not self._table_exists( + "ZSYNCOBJECT" + ): + raise DatabaseSchemaError( + "database is missing required MoneyWiz schema tables" + ) + required_metadata_columns = {"Z_ENT", "Z_NAME"} + if not required_metadata_columns.issubset( + self._table_columns("Z_PRIMARYKEY") + ): + raise DatabaseSchemaError( + "Z_PRIMARYKEY is missing required entity metadata columns" + ) + schema_identity = self._read_schema_identity() + schema_profile = detect_schema_profile(self._con) - def _load_primarykey(self) -> Dict[int, str]: + self._schema_identity = schema_identity + self._schema_profile = schema_profile + metadata = schema_identity[1] + self._ent_to_typename = {ent_id: name for ent_id, name in metadata} + self._typename_to_ent = {name: ent_id for ent_id, name in metadata} + + def _read_entity_metadata(self) -> tuple[tuple[int, str], ...]: cur = self._con.cursor() res = cur.execute( """ - SELECT * FROM "Z_PRIMARYKEY" ORDER BY "Z_ENT" LIMIT 1000 OFFSET 0; + SELECT Z_ENT, Z_NAME + FROM "Z_PRIMARYKEY" + ORDER BY Z_ENT """ ) - ent_to_typename: Dict[int, str] = {} + rows: list[tuple[int, str]] = [] + ent_ids: set[int] = set() + typenames: set[str] = set() for row in res.fetchall(): - ent_to_typename[row["Z_ENT"]] = row["Z_NAME"] - return ent_to_typename + ent_id = row["Z_ENT"] + typename = row["Z_NAME"] + if ( + not isinstance(ent_id, int) + or not isinstance(typename, str) + or not typename + ): + raise DatabaseSchemaError("Z_PRIMARYKEY contains invalid metadata") + if ent_id in ent_ids: + raise DatabaseSchemaError("Z_PRIMARYKEY contains duplicate entity IDs") + if typename in typenames: + raise DatabaseSchemaError( + "Z_PRIMARYKEY contains duplicate entity names" + ) + ent_ids.add(ent_id) + typenames.add(typename) + rows.append((ent_id, typename)) + return tuple(rows) + + def _read_schema_identity( + self, + ) -> tuple[int, tuple[tuple[int, str], ...]]: + schema_version = self._con.execute("PRAGMA schema_version").fetchone()[ + "schema_version" + ] + return schema_version, self._read_entity_metadata() def __repr__(self): return "\n".join( f"{key}: {value}" for key, value in self._ent_to_typename.items() ) + @property + def schema_profile(self) -> SchemaProfile: + """Return the physical-column compatibility profile for this store.""" + return self._schema_profile + def typename_for(self, ent_id: ENT_ID) -> str: typename = self._ent_to_typename.get(ent_id) assert typename is not None, f"Unknown ent_id {ent_id}" @@ -54,117 +134,375 @@ 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 _table_columns(self, table_name: str) -> set[str]: + return { + str(row["name"]) + for row in self._con.execute( + f'PRAGMA table_info("{table_name}")' + ).fetchall() + } + + def _relationship_storage( + self, + *, + entity_name: str, + table_name: str, + required_columns: tuple[str, ...], + ) -> RelationshipStorage: + metadata_present = entity_name in self._typename_to_ent + table_present = self._table_exists(table_name) + if not metadata_present and not table_present: + return RelationshipStorage.ABSENT + if not metadata_present or not table_present: + return RelationshipStorage.UNKNOWN + if not set(required_columns).issubset(self._table_columns(table_name)): + return RelationshipStorage.UNKNOWN + return RelationshipStorage.PRESENT + + def _transaction_tag_storage( + self, + ) -> tuple[RelationshipStorage, str | None, str | None, str | None]: + transaction_ent = self._typename_to_ent.get("Transaction") + tag_ent = self._typename_to_ent.get("Tag") + table_name = f"Z_{transaction_ent}TAGS" if transaction_ent is not None else None + transaction_column = ( + f"Z_{transaction_ent}TRANSACTIONS" if transaction_ent is not None else None + ) + tag_column = f"Z_{tag_ent}TAGS" if tag_ent is not None else None + candidates: list[tuple[str, set[str]]] = [] + # Core Data can create several *_TAGS tables; select only the Transaction–Tag link. + for row in self._con.execute( + "SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name" + ): + candidate = row["name"] + match = ( + re.fullmatch(r"Z_(\d+)TAGS", candidate) + if isinstance(candidate, str) + else None + ) + if match is None: + continue + columns = self._table_columns(candidate) + if ( + candidate == table_name + or int(match.group(1)) not in self._ent_to_typename + or any( + re.fullmatch(r"Z_\d+TRANSACTIONS\d*", column) for column in columns + ) + ): + candidates.append((candidate, columns)) + + if transaction_ent is None and tag_ent is None: + storage = ( + RelationshipStorage.UNKNOWN + if candidates + else RelationshipStorage.ABSENT + ) + return ( + storage, + candidates[0][0] if len(candidates) == 1 else None, + None, + None, + ) + if transaction_ent is None or tag_ent is None: + return RelationshipStorage.UNKNOWN, None, None, None + + storage = ( + RelationshipStorage.PRESENT + if candidates == [(table_name, {transaction_column, tag_column})] + else RelationshipStorage.UNKNOWN + ) + return storage, table_name, transaction_column, tag_column + + @staticmethod + def _relationship_error(exc: Exception) -> LoadErrorKind: + if isinstance(exc, KeyError): + return LoadErrorKind.MISSING_FIELD + if isinstance(exc, AssertionError): + return LoadErrorKind.VALIDATION + return LoadErrorKind.INVALID_VALUE + + def _skipped_relationship( + self, record_id: ID | str | None, entity: str, exc: Exception + ) -> SkippedRecord: + return SkippedRecord( + record_id=record_id, + entity=entity, + error=self._relationship_error(exc), + exception_type=type(exc).__name__, + ) + def query_objects(self, typenames: List[str]) -> List[Any]: + """Query live rows against the verified cached entity mapping.""" + with self.read_transaction(): + return self._query_objects(typenames) + + def _query_objects(self, typenames: List[str]) -> List[Any]: + ent_ids = [self._typename_to_ent.get(name) for name in typenames] + ent_ids = [ent_id for ent_id in ent_ids if ent_id is not None] + if not ent_ids: + return [] cur = self._con.cursor() res = cur.execute( """ SELECT * FROM ZSYNCOBJECT WHERE Z_ENT in (%s) """ - % (",".join("?" * len(typenames))), - [self.ent_for(x) for x in typenames], + % (",".join("?" * len(ent_ids))), + ent_ids, ) return res.fetchall() + def close(self) -> None: + """Close the read-only SQLite connection.""" + self._con.close() + + @contextmanager + def _raw_read_transaction(self): + """Provide snapshot ownership without consulting schema caches.""" + owns_transaction = not self._con.in_transaction + if owns_transaction: + self._con.execute("BEGIN") + try: + yield + finally: + if owns_transaction and self._con.in_transaction: + self._con.rollback() + + def _require_active_read(self) -> None: + if not self._con.in_transaction: + raise DatabaseSchemaError( + "database read transaction was interrupted; close and reopen the accessor" + ) + + def _verify_schema_identity(self) -> None: + current = self._read_schema_identity() + if current != self._schema_identity or ( + detect_schema_profile(self._con) != self._schema_profile + ): + raise DatabaseSchemaError( + "database schema changed; close and reopen the accessor" + ) + + @contextmanager + def read_transaction(self): + """Keep cache-dependent reads on one verified SQLite snapshot.""" + if self._admission_depth: + self._require_active_read() + + with self._raw_read_transaction(): + if not self._admission_depth: + # Cached entity IDs and column profiles must still match the live schema. + self._verify_schema_identity() + self._require_active_read() + self._admission_depth += 1 + try: + yield + self._require_active_read() + finally: + self._admission_depth -= 1 + + def _construct_record(self, row, constructor: Callable): + if isinstance(constructor, type) and issubclass(constructor, Record): + model_constructor = cast(Callable[..., Any], constructor) + return model_constructor( + mapped_row( + row, cast(type, constructor), schema_profile=self.schema_profile + ) + ) + raw_constructor = cast(Callable[..., Any], constructor) + return raw_constructor(row) + def get_record(self, pk_id: ID, constructor: Callable = Record): - cur = self._con.cursor() - res = cur.execute( - """ + with self.read_transaction(): + cur = self._con.cursor() + res = cur.execute( + """ SELECT * FROM ZSYNCOBJECT WHERE Z_PK = ? """, - [pk_id], - ) + [pk_id], + ) - return constructor(res.fetchone()) + return self._construct_record(res.fetchone(), constructor) def get_record_by_gid(self, gid: GID, constructor: Callable = Record): - cur = self._con.cursor() - res = cur.execute( - """ + with self.read_transaction(): + cur = self._con.cursor() + res = cur.execute( + """ SELECT * FROM ZSYNCOBJECT WHERE ZGID = ? """, - [gid], - ) + [gid], + ) - return constructor(res.fetchone()) + return self._construct_record(res.fetchone(), constructor) - def get_category_assignment(self) -> Dict[ID, List[Tuple[ID, Decimal]]]: + def read_category_assignments( + self, + ) -> tuple[Dict[ID, List[Tuple[ID, Decimal]]], RelationshipLoadReport]: + with self.read_transaction(): + return self._read_category_assignments() + + def _read_category_assignments( + self, + ) -> tuple[Dict[ID, List[Tuple[ID, Decimal]]], RelationshipLoadReport]: transaction_map: Dict[ID, List[Tuple[ID, Decimal]]] = defaultdict(list) - cur = self._con.cursor() - res = cur.execute( - """ - SELECT ZCATEGORY, ZTRANSACTION, ZAMOUNT FROM ZCATEGORYASSIGMENT WHERE ZTRANSACTION IS NOT NULL - - """ + table_name = "ZCATEGORYASSIGMENT" + columns = ("Z_PK", "ZCATEGORY", "ZTRANSACTION", "ZAMOUNT") + storage = self._relationship_storage( + entity_name="CategoryAssigment", + table_name=table_name, + required_columns=columns, ) - for row in res.fetchall(): - transaction_map[row["ZTRANSACTION"]].append( - (row["ZCATEGORY"], RDH.get_decimal(row["ZAMOUNT"])) + if storage != RelationshipStorage.PRESENT: + return transaction_map, RelationshipLoadReport( + storage=storage, + storage_name=table_name, ) - return transaction_map - def get_refund_maps(self) -> Dict[ID, ID]: - refund_to_withdraw: Dict[ID, ID] = {} - cur = self._con.cursor() - res = cur.execute( - """ - SELECT ZREFUNDTRANSACTION, ZWITHDRAWTRANSACTION FROM ZWITHDRAWREFUNDTRANSACTIONLINK - - """ + skipped: list[SkippedRecord] = [] + rows = self._con.execute( + f'SELECT {", ".join(columns)} FROM "{table_name}" ' + "WHERE ZTRANSACTION IS NOT NULL" + ).fetchall() + for row in rows: + raw_source_id = row.get("Z_PK") + source_id = raw_source_id if type(raw_source_id) is int else None + try: + require_integer_identity(raw_source_id) + category_id = row["ZCATEGORY"] + transaction_id = row["ZTRANSACTION"] + require_integer_identity(category_id) + require_integer_identity(transaction_id) + amount = RDH.get_decimal(row["ZAMOUNT"]) + transaction_map[transaction_id].append((category_id, amount)) + except (AssertionError, KeyError, ValueError) as exc: + skipped.append( + self._skipped_relationship(source_id, "CategoryAssigment", exc) + ) + continue + return transaction_map, RelationshipLoadReport( + storage=storage, + storage_name=table_name, + source_count=len(rows), + skipped=tuple(skipped), ) - for row in res.fetchall(): - refund_to_withdraw[row["ZREFUNDTRANSACTION"]] = row["ZWITHDRAWTRANSACTION"] - return refund_to_withdraw - def _get_tags_table_info(self) -> Tuple[str, str, str]: - cur = self._con.cursor() - res = cur.execute( - """ - SELECT name FROM sqlite_master WHERE type = 'table' + def get_category_assignment(self) -> Dict[ID, List[Tuple[ID, Decimal]]]: + """Return category assignments without completeness metadata.""" + return self.read_category_assignments()[0] - """ + def read_refund_maps( + self, + ) -> tuple[Dict[ID, ID], RelationshipLoadReport]: + with self.read_transaction(): + return self._read_refund_maps() + + def _read_refund_maps( + self, + ) -> tuple[Dict[ID, ID], RelationshipLoadReport]: + refund_to_withdraw: Dict[ID, ID] = {} + table_name = "ZWITHDRAWREFUNDTRANSACTIONLINK" + columns = ("Z_PK", "ZREFUNDTRANSACTION", "ZWITHDRAWTRANSACTION") + storage = self._relationship_storage( + entity_name="WithdrawRefundTransactionLink", + table_name=table_name, + required_columns=columns, ) - tag_tables = [] - for row in res.fetchall(): - match = re.fullmatch(r"Z_(\d+)TAGS", row["name"]) - if match: - tag_tables.append((int(match.group(1)), row["name"])) + if storage != RelationshipStorage.PRESENT: + return refund_to_withdraw, RelationshipLoadReport( + storage=storage, + storage_name=table_name, + ) - if not tag_tables: - raise ValueError("Could not find a tags join table matching Z_TAGS") + skipped: list[SkippedRecord] = [] + rows = self._con.execute( + f'SELECT {", ".join(columns)} FROM "{table_name}"' + ).fetchall() + for row in rows: + raw_source_id = row.get("Z_PK") + source_id = raw_source_id if type(raw_source_id) is int else None + try: + require_integer_identity(raw_source_id) + refund_id = row["ZREFUNDTRANSACTION"] + withdraw_id = row["ZWITHDRAWTRANSACTION"] + require_integer_identity(refund_id) + require_integer_identity(withdraw_id) + if refund_id in refund_to_withdraw: + raise ValueError() + refund_to_withdraw[refund_id] = withdraw_id + except (AssertionError, KeyError, ValueError) as exc: + skipped.append( + self._skipped_relationship( + source_id, "WithdrawRefundTransactionLink", exc + ) + ) + continue + return refund_to_withdraw, RelationshipLoadReport( + storage=storage, + storage_name=table_name, + source_count=len(rows), + skipped=tuple(skipped), + ) - tags_table_name = max(tag_tables)[1] - res = cur.execute(f'PRAGMA table_info("{tags_table_name}")') - columns = [row["name"] for row in res.fetchall()] + def get_refund_maps(self) -> Dict[ID, ID]: + """Return refund links without completeness metadata.""" + return self.read_refund_maps()[0] - transactions_columns = [ - column for column in columns if re.fullmatch(r"Z_\d+TRANSACTIONS", column) - ] - tags_columns = [ - column for column in columns if re.fullmatch(r"Z_\d+TAGS", column) - ] + def read_tags_map( + self, + ) -> tuple[Dict[ID, List[ID]], RelationshipLoadReport]: + with self.read_transaction(): + return self._read_tags_map() - if len(transactions_columns) != 1 or len(tags_columns) != 1: - raise ValueError( - f"Could not find expected tag columns in {tags_table_name}" + def _read_tags_map( + self, + ) -> tuple[Dict[ID, List[ID]], RelationshipLoadReport]: + transactions_to_tags: Dict[ID, List[ID]] = defaultdict(list) + storage, table_name, transaction_column, tag_column = ( + self._transaction_tag_storage() + ) + if storage != RelationshipStorage.PRESENT: + return transactions_to_tags, RelationshipLoadReport( + storage=storage, + storage_name=table_name, ) + columns = (cast(str, transaction_column), cast(str, tag_column)) - return tags_table_name, transactions_columns[0], tags_columns[0] + skipped: list[SkippedRecord] = [] + rows = self._con.execute( + f'SELECT {", ".join(columns)} FROM "{table_name}"' + ).fetchall() + for position, row in enumerate(rows): + transaction_id = row.get(transaction_column) + tag_id = row.get(tag_column) + try: + require_integer_identity(transaction_id) + require_integer_identity(tag_id) + transactions_to_tags[transaction_id].append(tag_id) + except (AssertionError, KeyError, ValueError) as exc: + skipped.append( + self._skipped_relationship(f"row:{position}", "TransactionTag", exc) + ) + continue + return transactions_to_tags, RelationshipLoadReport( + storage=storage, + storage_name=table_name, + source_count=len(rows), + skipped=tuple(skipped), + ) def get_tags_map(self) -> Dict[ID, List[ID]]: - transactions_to_tags: Dict[ID, List[ID]] = defaultdict(list) - cur = self._con.cursor() - tags_table_name, transactions_column, tags_column = self._get_tags_table_info() - res = cur.execute( - f""" - SELECT {transactions_column}, {tags_column} FROM "{tags_table_name}" - - """ - ) - for row in res.fetchall(): - transactions_to_tags[row[transactions_column]].append(row[tags_column]) - return transactions_to_tags + """Return transaction tags without completeness metadata.""" + return self.read_tags_map()[0] def get_users(self) -> Dict[ID, str]: users_map: Dict[ID, str] = {} diff --git a/src/moneywiz_api/managers/record_manager.py b/src/moneywiz_api/managers/record_manager.py index 07a98aa..e4c3007 100644 --- a/src/moneywiz_api/managers/record_manager.py +++ b/src/moneywiz_api/managers/record_manager.py @@ -1,37 +1,132 @@ from abc import ABC, abstractmethod -from typing import Dict, Generic, TypeVar +import logging +from typing import Callable, Dict, Generic, TypeVar, cast from moneywiz_api.database_accessor import DatabaseAccessor from moneywiz_api.model.record import Record from moneywiz_api.model.schema_mapped_row import mapped_row +from moneywiz_api.schema_profile import UnsupportedInvestmentSchemaError +from moneywiz_api.read_result import ( + LoadErrorKind, + ManagerLoadReport, + SkippedRecord, +) from moneywiz_api.types import GID, ID T = TypeVar("T", bound=Record) +logger = logging.getLogger(__name__) + + +class DuplicateRecordIdError(ValueError): + """Raised when two source rows use the same primary key.""" + + +class DuplicateRecordGidError(ValueError): + """Raised when two source rows use the same global identifier.""" class RecordManager(ABC, Generic[T]): def __init__(self) -> None: self._records: Dict[ID, T] = {} self._gid_to_id: Dict[GID, ID] = {} + self._load_report = ManagerLoadReport.unloaded() @property @abstractmethod def ents(self) -> Dict[str, type[T]]: raise NotImplementedError() - def load(self, db_accessor: DatabaseAccessor) -> None: - records = db_accessor.query_objects(list(self.ents.keys())) + def load(self, db_accessor: DatabaseAccessor) -> ManagerLoadReport: + self._discard_incomplete_load() + try: + with db_accessor.read_transaction(): + report = self._load_in_transaction(db_accessor) + except BaseException: + self._discard_incomplete_load() + raise + self._load_report = report + return report + + def _load_in_transaction(self, db_accessor: DatabaseAccessor) -> ManagerLoadReport: + """Load all manager state inside the caller-owned read snapshot.""" + ents = self.ents + # Completeness is scoped to this manager's explicitly supported entities. + records = db_accessor.query_objects(list(ents)) + + source_ids: list[ID | None] = [] + parsed_ids: list[ID] = [] + skipped: list[SkippedRecord] = [] for record in records: - typename = db_accessor.typename_for(record["Z_ENT"]) - assert typename in self.ents, ( - f"Unknown typename {typename} for record {record}" - ) + raw_record_id = record.get("Z_PK") + record_id = raw_record_id if type(raw_record_id) is int else None + source_ids.append(record_id) + typename = None + try: + typename = db_accessor.typename_for(record["Z_ENT"]) + obj = self.construct_record(ents[typename], record, db_accessor) + obj.validate() + if obj.id in self._records: + raise DuplicateRecordIdError() + if obj.gid in self._gid_to_id: + raise DuplicateRecordGidError() + self.add(obj) + except UnsupportedInvestmentSchemaError: + raise + except (AssertionError, KeyError, ValueError, RuntimeError) as exc: + error = self._error_kind(exc) + skipped.append( + SkippedRecord( + record_id=record_id, + entity=typename, + error=error, + exception_type=type(exc).__name__, + ) + ) + logger.debug( + "Skipping unreadable %s record %s: %s", + typename, + record_id, + error.value, + ) + continue + parsed_ids.append(obj.id) + + return ManagerLoadReport( + source_ids=tuple(source_ids), + parsed_ids=tuple(parsed_ids), + skipped=tuple(skipped), + ) - model_cls = self.ents[typename] - obj = model_cls(mapped_row(record, model_cls)) - obj.validate() - self.add(obj) + def _discard_incomplete_load(self) -> None: + """Clear state when a direct manager load did not finish.""" + self._records = {} + self._gid_to_id = {} + self._load_report = ManagerLoadReport.unloaded() + + @staticmethod + def _error_kind(exc: Exception) -> LoadErrorKind: + if isinstance(exc, DuplicateRecordIdError): + return LoadErrorKind.DUPLICATE_ID + if isinstance(exc, DuplicateRecordGidError): + return LoadErrorKind.DUPLICATE_GID + if isinstance(exc, KeyError): + return LoadErrorKind.MISSING_FIELD + if isinstance(exc, AssertionError): + return LoadErrorKind.VALIDATION + return LoadErrorKind.INVALID_VALUE + + def construct_record( + self, constructor: Callable, record, db_accessor: DatabaseAccessor + ): + """Construct a record; subclasses can supply schema-specific context.""" + return constructor( + mapped_row( + record, + cast(type, constructor), + schema_profile=getattr(db_accessor, "schema_profile", None), + ) + ) def add(self, record: T) -> None: self._records[record.id] = record @@ -54,5 +149,18 @@ 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 | None, str | None, str]]: + """Return records skipped during best-effort read parsing.""" + return [ + (item.record_id, item.entity, item.exception_type) + for item in self._load_report.skipped + ] + + @property + def load_report(self) -> ManagerLoadReport: + """Return structured completeness evidence from the latest load.""" + return self._load_report + def __repr__(self): return "\n".join(f"{key}: {value}" for key, value in self.records().items()) diff --git a/src/moneywiz_api/managers/transaction_manager.py b/src/moneywiz_api/managers/transaction_manager.py index a8bd139..36163b8 100644 --- a/src/moneywiz_api/managers/transaction_manager.py +++ b/src/moneywiz_api/managers/transaction_manager.py @@ -1,4 +1,5 @@ from datetime import datetime +from dataclasses import replace from decimal import Decimal from typing import Dict, List, Tuple, Protocol, cast @@ -17,6 +18,7 @@ TransferWithdrawTransaction, WithdrawTransaction, ) +from moneywiz_api.read_result import ManagerLoadReport from moneywiz_api.types import ID @@ -47,11 +49,29 @@ def ents(self) -> Dict[str, type[Transaction]]: "WithdrawTransaction": WithdrawTransaction, } - def load(self, db_accessor: DatabaseAccessor) -> None: - super().load(db_accessor) - self.category_assignment = db_accessor.get_category_assignment() - self.refund_maps = db_accessor.get_refund_maps() - self.tags_map = db_accessor.get_tags_map() + def _load_in_transaction(self, db_accessor: DatabaseAccessor) -> ManagerLoadReport: + report = super()._load_in_transaction(db_accessor) + category_assignment, category_report = db_accessor.read_category_assignments() + refund_maps, refund_report = db_accessor.read_refund_maps() + tags_map, tags_report = db_accessor.read_tags_map() + self.category_assignment = category_assignment + self.refund_maps = refund_maps + self.tags_map = tags_map + return replace( + report, + relationships={ + "category_assignments": category_report, + "refund_links": refund_report, + "transaction_tags": tags_report, + }, + ) + + def _discard_incomplete_load(self) -> None: + """Clear records and relationships after an incomplete load.""" + super()._discard_incomplete_load() + self.category_assignment = {} + self.refund_maps = {} + self.tags_map = {} def category_for_transaction( self, transaction_id: ID diff --git a/src/moneywiz_api/model/account.py b/src/moneywiz_api/model/account.py index afb0fcb..0b8a6f0 100644 --- a/src/moneywiz_api/model/account.py +++ b/src/moneywiz_api/model/account.py @@ -47,6 +47,7 @@ def __init__(self, row): def validate(self) -> None: super().validate() assert self.display_order is not None, self.as_dict() + assert isinstance(self.display_order, (int, float)) assert self.group_id is not None, self.as_dict() assert self.name is not None, self.as_dict() assert self.currency is not None, self.as_dict() diff --git a/src/moneywiz_api/model/investment_holding.py b/src/moneywiz_api/model/investment_holding.py index c7513d4..9c079ee 100644 --- a/src/moneywiz_api/model/investment_holding.py +++ b/src/moneywiz_api/model/investment_holding.py @@ -8,9 +8,7 @@ is_one_field, mapped_row, nullable_decimal_field, -) -from moneywiz_api.model.schema_mapped_row import ( - schema_field as schema_field, + schema_field, ) from moneywiz_api.types import ID @@ -20,7 +18,9 @@ class InvestmentHolding(Record): FIELDS = { "account": schema_field("ZINVESTMENTACCOUNT"), "opening_number_of_shares": nullable_decimal_field("ZOPENNINGNUMBEROFSHARES"), - "number_of_shares": nullable_decimal_field("ZNUMBEROFSHARES"), + "number_of_shares": nullable_decimal_field( + "ZNUMBEROFSHARES", profile_column="holding_number_of_shares_column" + ), "symbol": schema_field("ZSYMBOL"), "holding_type": schema_field("ZHOLDINGTYPE"), "description": schema_field("ZDESC"), @@ -78,7 +78,8 @@ def __init__(self, row): ) # Fixes - self.number_of_shares = self.number_of_shares or Decimal(0) + if self.number_of_shares is None: + self.number_of_shares = Decimal(0) def validate(self) -> None: super().validate() diff --git a/src/moneywiz_api/model/raw_data_handler.py b/src/moneywiz_api/model/raw_data_handler.py index a8b3527..eb53ac3 100644 --- a/src/moneywiz_api/model/raw_data_handler.py +++ b/src/moneywiz_api/model/raw_data_handler.py @@ -27,13 +27,57 @@ def get_decimal(raw_value: Any) -> Decimal: ) return Decimal(str(raw_value)) + @staticmethod + def get_decimal_alias(row: Dict[str, Any], *keys: str) -> Decimal: + """Read a required decimal field from the first available column alias.""" + for key in keys: + if key in row and row[key] is not None: + return RawDataHandler.get_decimal(row[key]) + raise KeyError(f"none of the schema aliases are present: {', '.join(keys)}") + + @staticmethod + def get_profile_decimal( + row: Dict[str, Any], selected_key: Optional[str], *fallback_keys: str + ) -> Decimal: + """Read the schema-selected value, using aliases only for unknown profiles.""" + if selected_key is not None: + return RawDataHandler.get_decimal(row[selected_key]) + return RawDataHandler.get_decimal_alias(row, *fallback_keys) + + @staticmethod + def get_nullable_decimal_alias( + row: Dict[str, Any], *keys: str + ) -> Optional[Decimal]: + """Read an optional decimal field from the first available column alias.""" + found_alias = False + for key in keys: + if key in row: + found_alias = True + if row[key] is not None: + return RawDataHandler.get_nullable_decimal(row[key]) + if found_alias: + return None + raise KeyError(f"none of the schema aliases are present: {', '.join(keys)}") + + @staticmethod + def get_profile_nullable_decimal( + row: Dict[str, Any], selected_key: Optional[str], *fallback_keys: str + ) -> Optional[Decimal]: + """Read an optional schema-selected value for a known profile.""" + if selected_key is not None: + return RawDataHandler.get_nullable_decimal(row[selected_key]) + return RawDataHandler.get_nullable_decimal_alias(row, *fallback_keys) + @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/model/record.py b/src/moneywiz_api/model/record.py index 040fdd2..dba6220 100644 --- a/src/moneywiz_api/model/record.py +++ b/src/moneywiz_api/model/record.py @@ -43,10 +43,10 @@ def ent(self) -> ENT_ID: def validate(self) -> None: assert self._raw - assert self._ent + assert type(self._ent) is int and bool(self._ent) assert self._created_at is not None - assert self.gid - assert self.id + assert type(self.gid) is str and bool(self.gid) + assert type(self.id) is int and bool(self.id) def filtered(self) -> Dict[str, Any]: """ diff --git a/src/moneywiz_api/model/schema_mapped_row.py b/src/moneywiz_api/model/schema_mapped_row.py index e18371e..d2675f1 100644 --- a/src/moneywiz_api/model/schema_mapped_row.py +++ b/src/moneywiz_api/model/schema_mapped_row.py @@ -3,6 +3,7 @@ from typing import Any, Callable, Dict from moneywiz_api.model.raw_data_handler import RawDataHandler as RDH +from moneywiz_api.schema_profile import SchemaProfile Converter = Callable[[Any], Any] @@ -11,12 +12,27 @@ class FieldSpec: aliases: tuple[str, ...] converter: Converter | None = None + profile_column: str | None = None class SchemaMappedRow: - def __init__(self, raw_row: Dict[str, Any], model_cls: type): + def __init__( + self, + raw_row: Dict[str, Any], + model_cls: type, + schema_profile: SchemaProfile | None = None, + ): self.raw_row = raw_row self.fields = self._fields_for(model_cls) + self._field_overrides: Dict[str, FieldSpec] = {} + self.schema_profile = schema_profile + profile_columns = { + field.profile_column + for field in self.fields.values() + if field.profile_column is not None + } + if schema_profile is not None and profile_columns: + schema_profile.require_columns(*sorted(profile_columns)) @classmethod def from_row(cls, row: Any, model_cls: type) -> "SchemaMappedRow": @@ -26,7 +42,12 @@ def from_row(cls, row: Any, model_cls: type) -> "SchemaMappedRow": def get(self, field_name: str) -> Any: spec = self.fields[field_name] - for alias in spec.aliases: + aliases = spec.aliases + if self.schema_profile is not None and spec.profile_column is not None: + profile_alias = getattr(self.schema_profile, spec.profile_column) + if profile_alias is not None: + aliases = (profile_alias,) + for alias in aliases: if alias in self.raw_row: raw_value = self.raw_row[alias] if spec.converter is not None: @@ -43,7 +64,7 @@ def get(self, field_name: str) -> Any: raise KeyError( f"Could not resolve field {field_name}. " - f"Tried {list(spec.aliases)}. " + f"Tried {list(aliases)}. " f"Available columns: {list(self.raw_row.keys())}" ) @@ -65,8 +86,14 @@ def _fields_for(model_cls: type) -> Dict[str, FieldSpec]: # fields -def schema_field(*aliases: str, converter: Converter | None = None) -> FieldSpec: - return FieldSpec(aliases=aliases, converter=converter) +def schema_field( + *aliases: str, + converter: Converter | None = None, + profile_column: str | None = None, +) -> FieldSpec: + return FieldSpec( + aliases=aliases, converter=converter, profile_column=profile_column + ) def datetime_field(*aliases: str, value_if_null: datetime | None = None) -> FieldSpec: @@ -78,17 +105,45 @@ def converter(raw_value: Any) -> datetime | None: return schema_field(*aliases, converter=converter) -def decimal_field(*aliases: str) -> FieldSpec: - return schema_field(*aliases, converter=RDH.get_decimal) +def decimal_field(*aliases: str, profile_column: str | None = None) -> FieldSpec: + return schema_field( + *aliases, converter=RDH.get_decimal, profile_column=profile_column + ) -def nullable_decimal_field(*aliases: str) -> FieldSpec: - return schema_field(*aliases, converter=RDH.get_nullable_decimal) +def nullable_decimal_field( + *aliases: str, profile_column: str | None = None +) -> FieldSpec: + return schema_field( + *aliases, converter=RDH.get_nullable_decimal, profile_column=profile_column + ) def is_one_field(*aliases: str) -> FieldSpec: - return schema_field(*aliases, converter=lambda raw_value: raw_value == 1) + def converter(raw_value: Any) -> bool: + if type(raw_value) is not int or raw_value not in (0, 1): + raise ValueError("boolean field must be stored as zero or one") + return raw_value == 1 + + return schema_field(*aliases, converter=converter) -def mapped_row(row: Any, model_cls: type) -> SchemaMappedRow: - return SchemaMappedRow.from_row(row, model_cls) +def mapped_row( + row: Any, + model_cls: type, + field_overrides: Dict[str, FieldSpec] | None = None, + schema_profile: SchemaProfile | None = None, +) -> SchemaMappedRow: + if isinstance(row, SchemaMappedRow): + if schema_profile is None: + mapped = row + else: + mapped = SchemaMappedRow(row.raw_row, model_cls, schema_profile) + mapped.fields = {**mapped.fields, **row._field_overrides} + mapped._field_overrides = dict(row._field_overrides) + else: + mapped = SchemaMappedRow(row, model_cls, schema_profile) + if field_overrides: + mapped.fields = {**mapped.fields, **field_overrides} + mapped._field_overrides.update(field_overrides) + return mapped diff --git a/src/moneywiz_api/model/transaction.py b/src/moneywiz_api/model/transaction.py index 24ad286..85c0ffb 100644 --- a/src/moneywiz_api/model/transaction.py +++ b/src/moneywiz_api/model/transaction.py @@ -35,7 +35,7 @@ class Transaction(Record, ABC): reconciled: bool amount: Decimal - description: str + description: Optional[str] datetime: datetime notes: Optional[str] @@ -52,9 +52,11 @@ def __init__(self, row): def validate(self) -> None: super().validate() - assert self.reconciled is not None, self.as_dict() + assert type(self.reconciled) is bool + if hasattr(self, "account") and self.account is not None: + assert type(self.account) is int assert self.amount is not None, self.as_dict() - assert self.description is not None, self.as_dict() + assert self.description is None or type(self.description) is str assert self.datetime is not None, self.as_dict() # self.notes can be None @@ -183,8 +185,12 @@ class InvestmentBuyTransaction(InvestmentTransaction): "account": schema_field("ZACCOUNT2"), "fee": decimal_field("ZFEE2"), "investment_holding": schema_field("ZINVESTMENTHOLDING"), - "number_of_shares": decimal_field("ZNUMBEROFSHARES"), - "price_per_share": decimal_field("ZPRICEPERSHARE1"), + "number_of_shares": decimal_field( + "ZNUMBEROFSHARES", profile_column="transaction_number_of_shares_column" + ), + "price_per_share": decimal_field( + "ZPRICEPERSHARE1", profile_column="price_per_share_column" + ), } account: ID @@ -209,7 +215,8 @@ def __init__(self, row): self.price_per_share = row.get("price_per_share") # Fixes - self.fee = max(self.fee, 0) + if self.fee < 0: + self.fee = Decimal(0) def validate(self) -> None: super().validate() @@ -237,8 +244,12 @@ class InvestmentSellTransaction(InvestmentTransaction): "account": schema_field("ZACCOUNT2"), "fee": decimal_field("ZFEE2"), "investment_holding": schema_field("ZINVESTMENTHOLDING"), - "number_of_shares": decimal_field("ZNUMBEROFSHARES"), - "price_per_share": decimal_field("ZPRICEPERSHARE1"), + "number_of_shares": decimal_field( + "ZNUMBEROFSHARES", profile_column="transaction_number_of_shares_column" + ), + "price_per_share": decimal_field( + "ZPRICEPERSHARE1", profile_column="price_per_share_column" + ), } account: ID @@ -263,7 +274,8 @@ def __init__(self, row): self.price_per_share = row.get("price_per_share") # Fixes - self.fee = max(self.fee, 0) + if self.fee < 0: + self.fee = Decimal(0) def validate(self) -> None: super().validate() @@ -525,6 +537,15 @@ def __init__(self, row): 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), + ): + if self.original_exchange_rate == 0: + raise ValueError( + "cannot reconstruct a transfer amount with a zero exchange rate" + ) + self.original_amount = self.amount def validate(self) -> None: super().validate() diff --git a/src/moneywiz_api/moneywiz_api.py b/src/moneywiz_api/moneywiz_api.py index 9057aeb..8f5e727 100644 --- a/src/moneywiz_api/moneywiz_api.py +++ b/src/moneywiz_api/moneywiz_api.py @@ -1,5 +1,6 @@ import logging from pathlib import Path +from typing import Iterable from moneywiz_api.database_accessor import DatabaseAccessor from moneywiz_api.managers.account_manager import AccountManager @@ -8,14 +9,16 @@ InvestmentHoldingManager, ) from moneywiz_api.managers.payee_manager import PayeeManager +from moneywiz_api.managers.record_manager import RecordManager from moneywiz_api.managers.tag_manager import TagManager from moneywiz_api.managers.transaction_manager import TransactionManager +from moneywiz_api.read_result import ApiCompleteness logger = logging.getLogger(__name__) class MoneywizApi: - def __init__(self, db_file: Path): + def __init__(self, db_file: Path, managers: Iterable[str] | None = None): self.accessor = DatabaseAccessor(db_file) self.account_manager = AccountManager() self.payee_manager = PayeeManager() @@ -23,13 +26,47 @@ def __init__(self, db_file: Path): self.transaction_manager = TransactionManager() self.investment_holding_manager = InvestmentHoldingManager() self.tag_manager = TagManager() + self._managers: dict[str, RecordManager] = { + "accounts": self.account_manager, + "payees": self.payee_manager, + "categories": self.category_manager, + "transactions": self.transaction_manager, + "investment_holdings": self.investment_holding_manager, + "tags": self.tag_manager, + } - self.load() + self.load(managers) - def load(self): - self.account_manager.load(self.accessor) - self.payee_manager.load(self.accessor) - self.category_manager.load(self.accessor) - self.transaction_manager.load(self.accessor) - self.investment_holding_manager.load(self.accessor) - self.tag_manager.load(self.accessor) + def load(self, managers: Iterable[str] | None = None) -> None: + """Load requested and previously loaded managers in one read transaction.""" + requested = tuple(self._managers) if managers is None else tuple(managers) + unknown = [name for name in requested if name not in self._managers] + if unknown: + raise ValueError(f"unknown manager names: {', '.join(unknown)}") + names = tuple( + name + for name, manager in self._managers.items() + if manager.load_report.observed or name in requested + ) + try: + with self.accessor.read_transaction(): + for name in names: + self._managers[name].load(self.accessor) + except BaseException: + for name in names: + self._managers[name]._discard_incomplete_load() + raise + + def completeness(self) -> ApiCompleteness: + """Return completeness evidence without reloading the database.""" + return ApiCompleteness( + managers={ + name: manager.load_report + for name, manager in self._managers.items() + if manager.load_report.observed + } + ) + + def close(self) -> None: + """Close the underlying read-only database connection.""" + self.accessor.close() diff --git a/src/moneywiz_api/read_result.py b/src/moneywiz_api/read_result.py new file mode 100644 index 0000000..c3590c7 --- /dev/null +++ b/src/moneywiz_api/read_result.py @@ -0,0 +1,182 @@ +"""Structured results for bounded MoneyWiz reads.""" + +from dataclasses import dataclass, field +from enum import Enum +from types import MappingProxyType +from typing import Any, Mapping + +from moneywiz_api.types import ID + + +class LoadErrorKind(str, Enum): + """Bounded classifications for records omitted from a read.""" + + MISSING_FIELD = "missing_field" + INVALID_VALUE = "invalid_value" + VALIDATION = "validation" + DUPLICATE_ID = "duplicate_id" + DUPLICATE_GID = "duplicate_gid" + + +class RelationshipStorage(str, Enum): + """Whether relationship storage is available and understood.""" + + PRESENT = "present" + ABSENT = "absent" + UNKNOWN = "unknown" + + +@dataclass(frozen=True) +class SkippedRecord: + """Identity-only diagnostic for one source row that was not parsed.""" + + record_id: ID | str | None + entity: str | None + error: LoadErrorKind + exception_type: str + + def as_dict(self) -> dict[str, Any]: + return { + "record_id": self.record_id, + "entity": self.entity, + "error": self.error.value, + "exception_type": self.exception_type, + } + + +@dataclass(frozen=True) +class RelationshipLoadReport: + """Completeness evidence for one relationship storage layout.""" + + storage: RelationshipStorage + storage_name: str | None = None + source_count: int = 0 + skipped: tuple[SkippedRecord, ...] = () + + @property + def parsed_count(self) -> int: + return self.source_count - len(self.skipped) + + @property + def complete(self) -> bool: + return self.storage != RelationshipStorage.UNKNOWN and not self.skipped + + @property + def status(self) -> str: + if self.storage == RelationshipStorage.ABSENT: + return "absent" + if self.complete: + return "complete" + if self.parsed_count: + return "partial" + return "error" + + def as_dict(self) -> dict[str, Any]: + return { + "status": self.status, + "complete": self.complete, + "storage": self.storage.value, + "storage_name": self.storage_name, + "source_count": self.source_count, + "parsed_count": self.parsed_count, + "skipped": [item.as_dict() for item in self.skipped], + } + + +@dataclass(frozen=True) +class ManagerLoadReport: + """Completeness evidence for one manager load.""" + + source_ids: tuple[ID | None, ...] = () + parsed_ids: tuple[ID, ...] = () + skipped: tuple[SkippedRecord, ...] = () + relationships: Mapping[str, RelationshipLoadReport] = field(default_factory=dict) + observed: bool = True + + def __post_init__(self) -> None: + object.__setattr__( + self, "relationships", MappingProxyType(dict(self.relationships)) + ) + + @classmethod + def unloaded(cls) -> "ManagerLoadReport": + """Return evidence that no complete manager observation was published.""" + return cls(observed=False) + + @property + def source_count(self) -> int: + return len(self.source_ids) + + @property + def parsed_count(self) -> int: + return len(self.parsed_ids) + + @property + def complete(self) -> bool: + return ( + self.observed + and self.source_count == self.parsed_count + and not self.skipped + and all(report.complete for report in self.relationships.values()) + ) + + @property + def status(self) -> str: + if not self.observed: + return "unloaded" + if self.complete: + return "complete" + if self.parsed_count or any( + report.parsed_count for report in self.relationships.values() + ): + return "partial" + return "error" + + def as_dict(self) -> dict[str, Any]: + return { + "status": self.status, + "complete": self.complete, + "source_count": self.source_count, + "parsed_count": self.parsed_count, + "source_ids": list(self.source_ids), + "parsed_ids": list(self.parsed_ids), + "skipped": [item.as_dict() for item in self.skipped], + "relationships": { + name: report.as_dict() for name, report in self.relationships.items() + }, + } + + +@dataclass(frozen=True) +class ApiCompleteness: + """Aggregate completeness for the managers included in a read.""" + + managers: Mapping[str, ManagerLoadReport] + + def __post_init__(self) -> None: + object.__setattr__(self, "managers", MappingProxyType(dict(self.managers))) + + @property + def complete(self) -> bool: + return bool(self.managers) and all( + report.complete for report in self.managers.values() + ) + + @property + def status(self) -> str: + if self.complete: + return "complete" + if self.managers and all( + report.status == "error" for report in self.managers.values() + ): + return "error" + return "partial" + + def as_dict(self) -> dict[str, Any]: + return { + "status": self.status, + "complete": self.complete, + "managers": { + name: report.as_dict() for name, report in self.managers.items() + }, + } diff --git a/src/moneywiz_api/schema_profile.py b/src/moneywiz_api/schema_profile.py new file mode 100644 index 0000000..dbd1f9d --- /dev/null +++ b/src/moneywiz_api/schema_profile.py @@ -0,0 +1,143 @@ +"""Schema capability profiles for MoneyWiz SQLite stores.""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass +from typing import Any, Iterable + + +class UnsupportedInvestmentSchemaError(ValueError): + """Raised when investment column aliases cannot be mapped safely.""" + + +@dataclass(frozen=True) +class SchemaProfile: + """Capabilities inferred from physical columns and investment rows.""" + + profile_id: str + holding_number_of_shares_column: str | None + transaction_number_of_shares_column: str | None + price_per_share_column: str | None + + @property + def is_known(self) -> bool: + return self.profile_id != "unknown" + + def require_columns(self, *columns: str) -> None: + """Raise when this model's profile-selected columns are unresolved.""" + if any(getattr(self, column) is None for column in columns): + raise UnsupportedInvestmentSchemaError( + "unsupported investment schema profile" + ) + + +def _value(row: Any, name: str, index: int) -> Any: + if isinstance(row, dict): + return row[name] + return row[index] + + +def _column_for_entities( + connection: sqlite3.Connection, + columns: set[str], + unsuffixed_column: str, + suffixed_column: str, + entity_names: Iterable[str], +) -> str | None: + has_unsuffixed = unsuffixed_column in columns + has_suffixed = suffixed_column in columns + if has_unsuffixed != has_suffixed: + return unsuffixed_column if has_unsuffixed else suffixed_column + if not has_unsuffixed: + return None + + tables = { + str(_value(row, "name", 0)) + for row in connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ).fetchall() + } + if "Z_PRIMARYKEY" not in tables: + return None + + names = tuple(entity_names) + placeholders = ", ".join("?" for _ in names) + entity_rows = connection.execute( + f'SELECT "Z_ENT", "Z_NAME" FROM "Z_PRIMARYKEY" ' + f'WHERE "Z_NAME" IN ({placeholders})', + names, + ).fetchall() + entity_ids = tuple(_value(row, "Z_ENT", 0) for row in entity_rows) + if not entity_ids: + return None + + placeholders = ", ".join("?" for _ in entity_ids) + counts = connection.execute( + f'SELECT COUNT("{unsuffixed_column}") AS "unsuffixed_count", ' + f'COUNT("{suffixed_column}") AS "suffixed_count" ' + f'FROM "ZSYNCOBJECT" WHERE "Z_ENT" IN ({placeholders})', + entity_ids, + ).fetchone() + unsuffixed_count = int(_value(counts, "unsuffixed_count", 0)) + suffixed_count = int(_value(counts, "suffixed_count", 1)) + if unsuffixed_count and not suffixed_count: + return unsuffixed_column + if suffixed_count and not unsuffixed_count: + return suffixed_column + return None + + +def detect_schema_profile(connection: sqlite3.Connection) -> SchemaProfile: + """Select investment aliases using columns and entity-specific row evidence.""" + columns = { + str(_value(row, "name", 1)) + for row in connection.execute("PRAGMA table_info(ZSYNCOBJECT)").fetchall() + } + + holding_shares = _column_for_entities( + connection, + columns, + "ZNUMBEROFSHARES", + "ZNUMBEROFSHARES1", + ("InvestmentHolding",), + ) + transaction_entities = ("InvestmentBuyTransaction", "InvestmentSellTransaction") + transaction_shares = _column_for_entities( + connection, + columns, + "ZNUMBEROFSHARES", + "ZNUMBEROFSHARES1", + transaction_entities, + ) + price_per_share = _column_for_entities( + connection, + columns, + "ZPRICEPERSHARE", + "ZPRICEPERSHARE1", + transaction_entities, + ) + + if not holding_shares or not transaction_shares or not price_per_share: + profile_id = "unknown" + elif ( + len( + { + column.endswith("1") + for column in (holding_shares, transaction_shares, price_per_share) + } + ) + > 1 + ): + profile_id = "mixed-investment-columns" + elif holding_shares.endswith("1"): + profile_id = "suffixed-investment-columns" + else: + profile_id = "unsuffixed-investment-columns" + + return SchemaProfile( + profile_id=profile_id, + holding_number_of_shares_column=holding_shares, + transaction_number_of_shares_column=transaction_shares, + price_per_share_column=price_per_share, + ) diff --git a/tests/unit/test_api_scoped_generation.py b/tests/unit/test_api_scoped_generation.py new file mode 100644 index 0000000..6fb58c6 --- /dev/null +++ b/tests/unit/test_api_scoped_generation.py @@ -0,0 +1,116 @@ +import sqlite3 +from contextlib import closing + +import pytest + +from moneywiz_api import MoneywizApi + + +def create_read_schema(path) -> None: + connection = sqlite3.connect(path) + connection.execute( + "CREATE TABLE Z_PRIMARYKEY " + "(Z_ENT INTEGER PRIMARY KEY, Z_NAME TEXT, Z_SUPER INTEGER, Z_MAX INTEGER)" + ) + connection.executemany( + "INSERT INTO Z_PRIMARYKEY VALUES (?, ?, ?, ?)", + [ + (8, "SyncObject", 0, 0), + (9, "Account", 8, 0), + (10, "CashAccount", 9, 0), + (35, "Tag", 8, 0), + (36, "Transaction", 8, 0), + (37, "DepositTransaction", 36, 0), + ], + ) + connection.execute( + "CREATE TABLE ZSYNCOBJECT (" + "Z_PK INTEGER PRIMARY KEY, Z_ENT INTEGER, ZOBJECTCREATIONDATE FLOAT, " + "ZGID TEXT, ZDISPLAYORDER INTEGER, ZGROUPID INTEGER, ZNAME TEXT, " + "ZCURRENCYNAME TEXT, ZOPENINGBALANCE FLOAT, ZINFO TEXT, ZUSER INTEGER, " + "ZRECONCILED INTEGER, ZAMOUNT1 FLOAT, ZDESC2 TEXT, ZDATE1 FLOAT, " + "ZNOTES1 TEXT, ZACCOUNT2 INTEGER, ZPAYEE2 INTEGER, " + "ZORIGINALCURRENCY TEXT, ZORIGINALAMOUNT FLOAT, " + "ZORIGINALEXCHANGERATE FLOAT)" + ) + connection.execute( + "CREATE TABLE Z_36TAGS (Z_36TRANSACTIONS INTEGER, Z_35TAGS INTEGER)" + ) + connection.execute( + "INSERT INTO ZSYNCOBJECT " + "(Z_PK, Z_ENT, ZOBJECTCREATIONDATE, ZGID, ZDISPLAYORDER, ZGROUPID, " + "ZNAME, ZCURRENCYNAME, ZOPENINGBALANCE, ZINFO, ZUSER) " + "VALUES (1, 10, 0, 'account-1', 1, 1, 'Before', 'EUR', 0, NULL, 1)" + ) + connection.commit() + connection.close() + + +def mutate_between_loads(path) -> None: + connection = sqlite3.connect(path) + connection.execute("UPDATE ZSYNCOBJECT SET ZNAME = 'After' WHERE Z_PK = 1") + connection.execute( + "INSERT INTO ZSYNCOBJECT " + "(Z_PK, Z_ENT, ZOBJECTCREATIONDATE, ZGID, ZRECONCILED, ZAMOUNT1, " + "ZDESC2, ZDATE1, ZNOTES1, ZACCOUNT2, ZPAYEE2, ZORIGINALCURRENCY, " + "ZORIGINALAMOUNT, ZORIGINALEXCHANGERATE) " + "VALUES (2, 37, 0, 'transaction-2', 0, 5, 'Income', 0, NULL, 1, " + "NULL, 'EUR', 5, 1)" + ) + connection.commit() + connection.close() + + +@pytest.mark.parametrize( + ("initial_managers", "expected_managers"), + [ + (("accounts",), ("accounts", "transactions")), + (None, None), + ], +) +def test_scoped_reload_refreshes_loaded_union_atomically( + tmp_path, initial_managers, expected_managers +) -> None: + path = tmp_path / "generation.sqlite" + create_read_schema(path) + + with closing(MoneywizApi(path, managers=initial_managers)) as api: + account = api.account_manager.get(1) + assert account is not None + assert account.name == "Before" + assert account.info is None + assert api.account_manager.load_report.complete + mutate_between_loads(path) + + api.load(("transactions",)) + completeness = api.completeness() + assert tuple(completeness.managers) == ( + expected_managers or tuple(api._managers) + ) + account = api.account_manager.get(1) + transaction = api.transaction_manager.get(2) + assert account is not None + assert transaction is not None + assert account.name == "After" + assert transaction.description == "Income" + + +def test_failed_scoped_reload_invalidates_loaded_managers( + tmp_path, monkeypatch +) -> None: + path = tmp_path / "reload.sqlite" + create_read_schema(path) + + with closing(MoneywizApi(path, managers=("accounts",))) as api: + + def fail(_accessor): + raise RuntimeError("transaction load failed") + + monkeypatch.setattr(api.transaction_manager, "load", fail) + with pytest.raises(RuntimeError, match="transaction load failed"): + api.load(("transactions",)) + + assert api.account_manager.records() == {} + assert api.account_manager.load_report.status == "unloaded" + assert api.transaction_manager.load_report.status == "unloaded" + assert not api.completeness().complete diff --git a/tests/unit/test_database_accessor.py b/tests/unit/test_database_accessor.py index 2ef0669..df80741 100644 --- a/tests/unit/test_database_accessor.py +++ b/tests/unit/test_database_accessor.py @@ -1,37 +1,81 @@ +from contextlib import closing import sqlite3 import pytest -from moneywiz_api.database_accessor import DatabaseAccessor +from moneywiz_api.database_accessor import ( + DatabaseAccessor, + DatabaseSchemaError, +) -def _create_accessor(db_path): - con = sqlite3.connect(db_path) - cur = con.cursor() - cur.execute("CREATE TABLE Z_PRIMARYKEY (Z_ENT INTEGER, Z_NAME TEXT)") - con.commit() - con.close() +def create_schema(path) -> None: + connection = sqlite3.connect(path) + connection.execute( + "CREATE TABLE Z_PRIMARYKEY " + "(Z_ENT INTEGER PRIMARY KEY, Z_NAME TEXT, Z_SUPER INTEGER, Z_MAX INTEGER)" + ) + connection.execute("CREATE TABLE ZSYNCOBJECT (Z_PK INTEGER, Z_ENT INTEGER)") + connection.executemany( + "INSERT INTO Z_PRIMARYKEY VALUES (?, ?, ?, ?)", + [ + (8, "SyncObject", 0, 0), + (9, "Account", 8, 0), + (10, "CashAccount", 9, 0), + (11, "FutureAccount", 9, 0), + (12, "OnlineAccount", 8, 0), + ], + ) + connection.commit() + connection.close() - return DatabaseAccessor(db_path) +def test_accessor_requires_an_existing_database(tmp_path) -> None: + path = tmp_path / "missing.sqlite" + with pytest.raises(sqlite3.OperationalError): + DatabaseAccessor(path) + assert not path.exists() -def test_get_tags_map_uses_highest_numbered_tags_table(tmp_path): - db_path = tmp_path / "moneywiz.sqlite" - accessor = _create_accessor(db_path) - cur = accessor._con.cursor() - cur.execute("CREATE TABLE Z_36TAGS (Z_1TRANSACTIONS INTEGER, Z_2TAGS INTEGER)") - cur.execute("CREATE TABLE Z_37TAGS (Z_3TRANSACTIONS INTEGER, Z_4TAGS INTEGER)") - cur.execute("INSERT INTO Z_36TAGS VALUES (1, 10)") - cur.execute("INSERT INTO Z_37TAGS VALUES (2, 20)") - cur.execute("INSERT INTO Z_37TAGS VALUES (2, 21)") - accessor._con.commit() - assert accessor.get_tags_map() == {2: [20, 21]} +def test_accessor_rejects_non_moneywiz_sqlite(tmp_path) -> None: + path = tmp_path / "other.sqlite" + sqlite3.connect(path).close() + with pytest.raises(DatabaseSchemaError): + DatabaseAccessor(path) -def test_get_tags_map_raises_when_tags_table_is_missing(tmp_path): - db_path = tmp_path / "moneywiz.sqlite" - accessor = _create_accessor(db_path) - with pytest.raises(ValueError, match="Z_TAGS"): - accessor.get_tags_map() +def test_accessor_is_read_only(tmp_path) -> None: + path = tmp_path / "moneywiz.sqlite" + create_schema(path) + + with closing(DatabaseAccessor(path)) as accessor: + with pytest.raises(sqlite3.OperationalError, match="readonly"): + accessor._con.execute("CREATE TABLE mutation (id INTEGER)") + + +def test_get_record_preserves_callable_constructors(tmp_path) -> None: + path = tmp_path / "moneywiz.sqlite" + create_schema(path) + with sqlite3.connect(path) as connection: + connection.execute("ALTER TABLE ZSYNCOBJECT ADD COLUMN ZGID TEXT") + connection.execute("INSERT INTO ZSYNCOBJECT VALUES (1, 10, 'record-gid')") + + def get_gid(row): + return row["ZGID"] + + with closing(DatabaseAccessor(path)) as accessor: + assert accessor.get_record(1, get_gid) == "record-gid" + assert accessor.get_record_by_gid("record-gid", get_gid) == "record-gid" + + +@pytest.mark.parametrize("columns", ["Z_NAME TEXT", "Z_ENT INTEGER"]) +def test_incomplete_entity_metadata_is_rejected(tmp_path, columns) -> None: + path = tmp_path / "malformed.sqlite" + with closing(sqlite3.connect(path)) as connection: + connection.execute(f"CREATE TABLE Z_PRIMARYKEY ({columns})") + connection.execute("CREATE TABLE ZSYNCOBJECT (Z_PK INTEGER, Z_ENT INTEGER)") + connection.commit() + + with pytest.raises(DatabaseSchemaError): + DatabaseAccessor(path) diff --git a/tests/unit/test_manager_transaction_lifecycle.py b/tests/unit/test_manager_transaction_lifecycle.py new file mode 100644 index 0000000..e4531a2 --- /dev/null +++ b/tests/unit/test_manager_transaction_lifecycle.py @@ -0,0 +1,100 @@ +from contextlib import closing +from decimal import Decimal +import sqlite3 + +from moneywiz_api.database_accessor import DatabaseAccessor +from moneywiz_api.managers.transaction_manager import TransactionManager + + +def _create_wal_transaction_store(path) -> None: + connection = sqlite3.connect(path) + connection.execute("PRAGMA journal_mode=WAL") + connection.execute( + "CREATE TABLE Z_PRIMARYKEY " + "(Z_ENT INTEGER, Z_NAME TEXT, Z_SUPER INTEGER, Z_MAX INTEGER)" + ) + connection.executemany( + "INSERT INTO Z_PRIMARYKEY VALUES (?, ?, ?, 0)", + [ + (2, "CategoryAssigment", 0), + (8, "SyncObject", 0), + (35, "Tag", 8), + (36, "Transaction", 8), + (37, "DepositTransaction", 36), + (50, "WithdrawRefundTransactionLink", 0), + ], + ) + connection.execute( + "CREATE TABLE ZSYNCOBJECT (" + "Z_PK INTEGER, Z_ENT INTEGER, ZOBJECTCREATIONDATE FLOAT, ZGID TEXT, " + "ZRECONCILED INTEGER, ZAMOUNT1 FLOAT, ZDESC2 TEXT, ZDATE1 FLOAT, " + "ZNOTES1 TEXT, ZACCOUNT2 INTEGER, ZPAYEE2 INTEGER, " + "ZORIGINALCURRENCY TEXT, ZORIGINALAMOUNT FLOAT, " + "ZORIGINALEXCHANGERATE FLOAT)" + ) + connection.execute( + "INSERT INTO ZSYNCOBJECT VALUES " + "(100, 37, 0, 'transaction-100', 0, 5, 'old', 0, NULL, 1, NULL, " + "'EUR', 5, 1)" + ) + connection.execute( + "CREATE TABLE ZCATEGORYASSIGMENT " + "(Z_PK INTEGER, ZCATEGORY INTEGER, ZTRANSACTION INTEGER, ZAMOUNT FLOAT)" + ) + connection.execute("INSERT INTO ZCATEGORYASSIGMENT VALUES (1, 7, 100, 5)") + connection.execute( + "CREATE TABLE ZWITHDRAWREFUNDTRANSACTIONLINK " + "(Z_PK INTEGER, ZREFUNDTRANSACTION INTEGER, ZWITHDRAWTRANSACTION INTEGER)" + ) + connection.execute("INSERT INTO ZWITHDRAWREFUNDTRANSACTIONLINK VALUES (2, 100, 90)") + connection.execute( + "CREATE TABLE Z_36TAGS (Z_36TRANSACTIONS INTEGER, Z_35TAGS INTEGER)" + ) + connection.execute("INSERT INTO Z_36TAGS VALUES (100, 9)") + connection.commit() + connection.close() + + +def _mutate_wal_transaction_store(path) -> None: + connection = sqlite3.connect(path) + connection.execute("UPDATE ZSYNCOBJECT SET ZAMOUNT1 = 6, ZORIGINALAMOUNT = 6") + connection.execute("UPDATE ZCATEGORYASSIGMENT SET ZAMOUNT = 7") + connection.execute( + "UPDATE ZWITHDRAWREFUNDTRANSACTIONLINK SET ZWITHDRAWTRANSACTION = 91" + ) + connection.execute("UPDATE Z_36TAGS SET Z_35TAGS = 10") + connection.commit() + connection.close() + + +def test_direct_transaction_manager_load_uses_one_wal_generation( + tmp_path, monkeypatch +) -> None: + path = tmp_path / "manager-generation.sqlite" + _create_wal_transaction_store(path) + + with closing(DatabaseAccessor(path)) as accessor: + method_name = "read_category_assignments" + original = getattr(accessor, method_name) + mutated = False + + def read_then_mutate(*args, **kwargs): + nonlocal mutated + result = original(*args, **kwargs) + if not mutated: + _mutate_wal_transaction_store(path) + mutated = True + return result + + monkeypatch.setattr(accessor, method_name, read_then_mutate) + manager = TransactionManager() + report = manager.load(accessor) + + assert report.complete + assert all(item.complete for item in report.relationships.values()) + transaction = manager.get(100) + assert transaction is not None + assert transaction.amount == Decimal("5.0") + assert manager.category_assignment == {100: [(7, Decimal("5.0"))]} + assert manager.refund_maps == {100: 90} + assert manager.tags_map == {100: [9]} diff --git a/tests/unit/test_profiled_investment_values.py b/tests/unit/test_profiled_investment_values.py new file mode 100644 index 0000000..04b05c4 --- /dev/null +++ b/tests/unit/test_profiled_investment_values.py @@ -0,0 +1,350 @@ +from contextlib import contextmanager +from decimal import Decimal +from typing import Any, cast + +import pytest + +from moneywiz_api.managers.investment_holding_manager import InvestmentHoldingManager +from moneywiz_api.managers.transaction_manager import TransactionManager +from moneywiz_api.database_accessor import DatabaseAccessor +from moneywiz_api.model.investment_holding import InvestmentHolding +from moneywiz_api.model.schema_mapped_row import mapped_row +from moneywiz_api.model.transaction import ( + InvestmentBuyTransaction, + InvestmentSellTransaction, +) +from moneywiz_api.read_result import RelationshipLoadReport, RelationshipStorage +from moneywiz_api.schema_profile import ( + SchemaProfile, + UnsupportedInvestmentSchemaError, +) + + +UNSUFFIXED_PROFILE = SchemaProfile( + profile_id="unsuffixed-investment-columns", + holding_number_of_shares_column="ZNUMBEROFSHARES", + transaction_number_of_shares_column="ZNUMBEROFSHARES", + price_per_share_column="ZPRICEPERSHARE", +) +MIXED_PROFILE = SchemaProfile( + profile_id="mixed-investment-columns", + holding_number_of_shares_column="ZNUMBEROFSHARES", + transaction_number_of_shares_column="ZNUMBEROFSHARES1", + price_per_share_column="ZPRICEPERSHARE1", +) +OBSERVED_STORE_PROFILE = SchemaProfile( + profile_id="mixed-investment-columns", + holding_number_of_shares_column="ZNUMBEROFSHARES", + transaction_number_of_shares_column="ZNUMBEROFSHARES", + price_per_share_column="ZPRICEPERSHARE1", +) +UNKNOWN_PROFILE = SchemaProfile("unknown", None, None, None) +HOLDING_ONLY_PROFILE = SchemaProfile("unknown", "ZNUMBEROFSHARES", None, None) + + +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": 2, + "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, + "ZINVESTMENTOBJECTTYPE": 0, + "ZCOSTBASISOFMISSINGOBSHARES": 0.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 = constructor( + mapped_row(row, constructor, schema_profile=UNSUFFIXED_PROFILE) + ) + + 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 = InvestmentHolding( + mapped_row( + investment_holding_row(), + InvestmentHolding, + schema_profile=UNSUFFIXED_PROFILE, + ) + ) + + assert holding.number_of_shares == Decimal("2.0") + + +@pytest.mark.parametrize(("raw_quantity", "rendered"), [(0.0, "0.0"), (None, "0")]) +def test_holding_zero_quantity_fallback_preserves_scale(raw_quantity, rendered) -> None: + row = investment_holding_row() + row["ZNUMBEROFSHARES"] = raw_quantity + + holding = InvestmentHolding( + mapped_row(row, InvestmentHolding, schema_profile=UNSUFFIXED_PROFILE) + ) + + assert str(holding.number_of_shares) == rendered + + +@pytest.mark.parametrize( + ("constructor", "transaction_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, transaction_row +) -> None: + holding_row = investment_holding_row() + + transaction = constructor( + mapped_row(transaction_row, constructor, schema_profile=MIXED_PROFILE) + ) + holding = InvestmentHolding( + mapped_row(holding_row, InvestmentHolding, schema_profile=MIXED_PROFILE) + ) + + 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: + row = {**row, "ZAMOUNT1": -2.0 if constructor is InvestmentBuyTransaction else 2.0} + transaction = constructor( + mapped_row(row, constructor, schema_profile=OBSERVED_STORE_PROFILE) + ) + + assert transaction.number_of_shares == Decimal("2.0") + assert transaction.price_per_share == Decimal("1.0") + + +class ProfileAccessor: + schema_profile = UNSUFFIXED_PROFILE + + +def test_managers_pass_profile_to_investment_constructors() -> None: + accessor = ProfileAccessor() + transaction = TransactionManager().construct_record( + InvestmentBuyTransaction, + investment_transaction_row(40, -20.0), + cast(DatabaseAccessor, accessor), + ) + holding = InvestmentHoldingManager().construct_record( + InvestmentHolding, + investment_holding_row(), + cast(DatabaseAccessor, accessor), + ) + + assert transaction.price_per_share == Decimal("10.0") + assert holding.number_of_shares == Decimal("2.0") + + +class ManagerAccessor(ProfileAccessor): + def __init__( + self, + typename: str, + row: dict, + schema_profile: SchemaProfile = UNSUFFIXED_PROFILE, + ): + self.typename = typename + self.row = row + self.schema_profile = schema_profile + + @contextmanager + def read_transaction(self): + yield + + def read_category_assignments(self): + return {}, RelationshipLoadReport(storage=RelationshipStorage.ABSENT) + + read_refund_maps = read_category_assignments + read_tags_map = read_category_assignments + + def query_objects(self, _typenames): + return [self.row] + + def typename_for(self, _ent_id): + return self.typename + + +def test_managers_load_profiled_investment_records() -> None: + transaction_manager = TransactionManager() + transaction_manager.load( + cast( + DatabaseAccessor, + ManagerAccessor( + "InvestmentBuyTransaction", investment_transaction_row(40, -20.0) + ), + ) + ) + holding_manager = InvestmentHoldingManager() + holding_manager.load( + cast( + DatabaseAccessor, + ManagerAccessor("InvestmentHolding", investment_holding_row()), + ) + ) + + transaction = transaction_manager.get(40) + holding = holding_manager.get(24) + assert isinstance(transaction, InvestmentBuyTransaction) + assert holding is not None + assert transaction.price_per_share == Decimal("10.0") + assert holding.number_of_shares == Decimal("2.0") + + +class StaticCursor: + def __init__(self, row: dict): + self.row = row + + def execute(self, _query, _parameters): + return self + + def fetchone(self): + return self.row + + +class StaticConnection: + def __init__(self, row: dict): + self.row = row + + def cursor(self): + return StaticCursor(self.row) + + +def test_accessor_public_constructors_receive_schema_profile(monkeypatch) -> None: + @contextmanager + def static_read_transaction(_accessor): + yield + + monkeypatch.setattr(DatabaseAccessor, "read_transaction", static_read_transaction) + transaction_accessor = DatabaseAccessor.__new__(DatabaseAccessor) + transaction_accessor._con = cast( + Any, StaticConnection(investment_transaction_row(40, -20.0)) + ) + transaction_accessor._schema_profile = UNSUFFIXED_PROFILE + holding_accessor = DatabaseAccessor.__new__(DatabaseAccessor) + holding_accessor._con = cast(Any, StaticConnection(investment_holding_row())) + holding_accessor._schema_profile = UNSUFFIXED_PROFILE + + transaction = transaction_accessor.get_record(40, InvestmentBuyTransaction) + holding = holding_accessor.get_record_by_gid("holding", InvestmentHolding) + + assert transaction.price_per_share == Decimal("10.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)), + (InvestmentHolding, investment_holding_row()), + ], +) +def test_investment_models_reject_ambiguous_profile(constructor, row) -> None: + with pytest.raises( + UnsupportedInvestmentSchemaError, + match="unsupported investment schema profile", + ): + constructor(mapped_row(row, constructor, schema_profile=UNKNOWN_PROFILE)) + + +def test_holding_load_works_when_transaction_profile_is_unresolved() -> None: + manager = InvestmentHoldingManager() + manager.load( + cast( + DatabaseAccessor, + ManagerAccessor( + "InvestmentHolding", investment_holding_row(), HOLDING_ONLY_PROFILE + ), + ) + ) + + holding = manager.get(24) + assert holding is not None + assert holding.number_of_shares == Decimal("2.0") + + +def test_transaction_load_rejects_its_unresolved_profile_columns() -> None: + with pytest.raises( + UnsupportedInvestmentSchemaError, + match="unsupported investment schema profile", + ): + InvestmentBuyTransaction( + mapped_row( + investment_transaction_row(40, -20.0), + InvestmentBuyTransaction, + schema_profile=HOLDING_ONLY_PROFILE, + ) + ) + + +@pytest.mark.parametrize( + ("manager", "typename", "row"), + [ + ( + TransactionManager(), + "InvestmentBuyTransaction", + investment_transaction_row(40, -20.0), + ), + (InvestmentHoldingManager(), "InvestmentHolding", investment_holding_row()), + ], +) +def test_managers_propagate_unsupported_investment_schema_errors( + manager, typename, row +) -> None: + with pytest.raises( + UnsupportedInvestmentSchemaError, + match="unsupported investment schema profile", + ): + manager.load( + cast(DatabaseAccessor, ManagerAccessor(typename, row, UNKNOWN_PROFILE)) + ) diff --git a/tests/unit/test_read_completeness.py b/tests/unit/test_read_completeness.py new file mode 100644 index 0000000..2768d74 --- /dev/null +++ b/tests/unit/test_read_completeness.py @@ -0,0 +1,239 @@ +import json +from typing import cast +from moneywiz_api.database_accessor import DatabaseAccessor +from contextlib import contextmanager + +import pytest + +from moneywiz_api.managers.record_manager import RecordManager +from moneywiz_api.model.record import Record +from moneywiz_api.read_result import ( + ApiCompleteness, + LoadErrorKind, + ManagerLoadReport, + RelationshipLoadReport, + RelationshipStorage, +) + + +class ExampleRecord(Record): + pass + + +class ExampleManager(RecordManager): + @property + def ents(self): + return {"ExampleRecord": ExampleRecord} + + +def record_row(record_id=1, gid="record-1", ent=1): + return { + "Z_ENT": ent, + "ZOBJECTCREATIONDATE": 0.0, + "ZGID": gid, + "Z_PK": record_id, + } + + +@pytest.mark.parametrize( + "field,value", + [ + ("ZGID", b"binary-id"), + ("Z_PK", "1"), + ("Z_ENT", True), + ], +) +def test_record_validation_rejects_coerced_identity(field, value) -> None: + row = record_row() + row[field] = value + with pytest.raises(AssertionError): + ExampleRecord(row).validate() + + +class RecordAccessor: + def __init__(self, rows, typenames=None): + self.rows = rows + self.typenames = typenames or {1: "ExampleRecord"} + + @contextmanager + def read_transaction(self): + yield + + def query_objects(self, typenames): + return [ + row + for row in self.rows + if "Z_ENT" not in row or self.typenames.get(row["Z_ENT"]) in typenames + ] + + def typename_for(self, ent_id): + return self.typenames.get(ent_id) + + +def test_report_counts_duplicate_rows_without_partial_mutation() -> None: + manager = ExampleManager() + rows = [ + record_row(), + record_row(gid="duplicate-id"), + record_row(record_id=2), + record_row(record_id=3, gid="record-3", ent=2), + ] + + report = manager.load( + cast( + DatabaseAccessor, + RecordAccessor(rows, {1: "ExampleRecord", 2: "FutureRecord"}), + ) + ) + + assert report.source_ids == (1, 1, 2) + assert report.parsed_ids == (1,) + assert [item.error for item in report.skipped] == [ + LoadErrorKind.DUPLICATE_ID, + LoadErrorKind.DUPLICATE_GID, + ] + assert report.observed + assert report.status == "partial" + assert list(manager.records()) == [1] + + +@pytest.mark.parametrize("exception_type", [ValueError, RuntimeError]) +def test_construction_error_is_bounded_without_raw_value( + caplog, exception_type +) -> None: + class BrokenRecord(Record): + def __init__(self, _row): + raise exception_type("PRIVATE_PAYLOAD") + + class BrokenManager(RecordManager): + @property + def ents(self): + return {"ExampleRecord": BrokenRecord} + + caplog.set_level("DEBUG") + report = BrokenManager().load( + cast(DatabaseAccessor, RecordAccessor([record_row()])) + ) + + assert report.skipped[0].error == LoadErrorKind.INVALID_VALUE + assert report.skipped[0].exception_type == exception_type.__name__ + assert "PRIVATE_PAYLOAD" not in json.dumps(report.as_dict()) + assert "PRIVATE_PAYLOAD" not in caplog.text + + +def test_report_mappings_cannot_change_completeness_after_publication() -> None: + relationships = { + "transaction_tags": RelationshipLoadReport(RelationshipStorage.UNKNOWN) + } + report = ManagerLoadReport(relationships=relationships) + managers = {"transactions": report} + completeness = ApiCompleteness(managers) + + relationships.clear() + managers.clear() + assert not report.complete + assert not completeness.complete + mutable_relationships = cast( + dict[str, RelationshipLoadReport], report.relationships + ) + mutable_managers = cast(dict[str, ManagerLoadReport], completeness.managers) + with pytest.raises(TypeError): + mutable_relationships["transaction_tags"] = RelationshipLoadReport( + RelationshipStorage.ABSENT + ) + with pytest.raises(TypeError): + mutable_managers["transactions"] = ManagerLoadReport() + + +def test_missing_field_is_reported() -> None: + row = record_row() + del row["Z_ENT"] + + report = ExampleManager().load(cast(DatabaseAccessor, RecordAccessor([row]))) + + assert report.skipped[0].error == LoadErrorKind.MISSING_FIELD + assert report.status == "error" + assert not report.complete + + +def test_each_load_resets_records_and_diagnostics() -> None: + manager = ExampleManager() + manager.load( + cast( + DatabaseAccessor, + RecordAccessor([record_row(), record_row(gid="duplicate-id")]), + ) + ) + + report = manager.load( + cast( + DatabaseAccessor, RecordAccessor([record_row(record_id=4, gid="record-4")]) + ) + ) + + assert report.complete + assert report.source_ids == (4,) + assert report.parsed_ids == (4,) + assert list(manager.records()) == [4] + assert manager.load_errors == [] + + +def test_failed_direct_reload_discards_state_and_can_retry() -> None: + manager = ExampleManager() + assert manager.load_report.as_dict()["status"] == "unloaded" + manager.load(cast(DatabaseAccessor, RecordAccessor([record_row()]))) + + class FailingAccessor(RecordAccessor): + def query_objects(self, _typenames): + raise RuntimeError("synthetic reload failure") + + with pytest.raises(RuntimeError, match="synthetic reload failure"): + manager.load(cast(DatabaseAccessor, FailingAccessor([]))) + + assert manager.load_report.status == "unloaded" + assert not manager.load_report.complete + assert manager.records() == {} + + report = manager.load(cast(DatabaseAccessor, RecordAccessor([]))) + assert report.complete + + +@pytest.mark.parametrize("interruption_type", [KeyboardInterrupt, SystemExit]) +def test_interrupted_reload_discards_partial_state_and_reraises( + interruption_type, +) -> None: + manager = ExampleManager() + manager.load( + cast( + DatabaseAccessor, RecordAccessor([record_row(record_id=9, gid="record-9")]) + ) + ) + interruption = interruption_type("synthetic interruption") + + class InterruptingAccessor(RecordAccessor): + def __init__(self): + super().__init__([record_row(), record_row(record_id=2, gid="record-2")]) + self.lookups = 0 + + def typename_for(self, ent_id): + self.lookups += 1 + if self.lookups == 2: + raise interruption + return super().typename_for(ent_id) + + with pytest.raises(interruption_type) as error: + manager.load(cast(DatabaseAccessor, InterruptingAccessor())) + + assert error.value is interruption + assert manager.records() == {} + assert manager._gid_to_id == {} + assert manager.load_report.status == "unloaded" + + report = manager.load( + cast( + DatabaseAccessor, RecordAccessor([record_row(record_id=3, gid="record-3")]) + ) + ) + + assert report.complete + assert list(manager.records()) == [3] diff --git a/tests/unit/test_relationship_completeness.py b/tests/unit/test_relationship_completeness.py new file mode 100644 index 0000000..e49d96c --- /dev/null +++ b/tests/unit/test_relationship_completeness.py @@ -0,0 +1,330 @@ +from contextlib import closing +import json +import sqlite3 + +import pytest + +from moneywiz_api.database_accessor import DatabaseAccessor +from moneywiz_api.managers.transaction_manager import TransactionManager +from moneywiz_api.read_result import RelationshipStorage + + +def create_relationship_schema(path, *, include_tables=True) -> None: + create_custom_relationship_schema( + path, + metadata=( + (2, "CategoryAssigment", 0, 0), + (8, "SyncObject", 0, 0), + (36, "Tag", 8, 0), + (37, "Transaction", 8, 0), + (50, "WithdrawRefundTransactionLink", 0, 0), + ), + tables=( + "CREATE TABLE ZCATEGORYASSIGMENT " + "(Z_PK INTEGER PRIMARY KEY, ZCATEGORY INTEGER, ZAMOUNT FLOAT, " + "ZTRANSACTION INTEGER)", + "CREATE TABLE ZWITHDRAWREFUNDTRANSACTIONLINK " + "(Z_PK INTEGER PRIMARY KEY, ZREFUNDTRANSACTION INTEGER, " + "ZWITHDRAWTRANSACTION INTEGER)", + "CREATE TABLE Z_37TAGS (Z_37TRANSACTIONS INTEGER, Z_36TAGS INTEGER)", + ) + if include_tables + else (), + ) + if include_tables: + connection = sqlite3.connect(path) + connection.executemany( + "INSERT INTO ZCATEGORYASSIGMENT VALUES (?, ?, ?, ?)", + [ + (1, 10, 5.0, 100), + (2, 11, None, 101), + (3, 12, 2.0, None), + (6, "PRIVATE_PAYLOAD", 3.0, 102), + ], + ) + connection.executemany( + "INSERT INTO ZWITHDRAWREFUNDTRANSACTIONLINK VALUES (?, ?, ?)", + [(4, 102, 103), (5, 104, None)], + ) + connection.executemany( + "INSERT INTO Z_37TAGS VALUES (?, ?)", + [(100, 200), (101, None)], + ) + connection.commit() + connection.close() + + +def create_custom_relationship_schema(path, metadata=(), tables=()) -> None: + connection = sqlite3.connect(path) + connection.execute( + "CREATE TABLE Z_PRIMARYKEY " + "(Z_ENT INTEGER PRIMARY KEY, Z_NAME TEXT, Z_SUPER INTEGER, Z_MAX INTEGER)" + ) + connection.execute("CREATE TABLE ZSYNCOBJECT (Z_PK INTEGER, Z_ENT INTEGER)") + connection.executemany("INSERT INTO Z_PRIMARYKEY VALUES (?, ?, ?, ?)", metadata) + for statement in tables: + connection.execute(statement) + connection.commit() + connection.close() + + +def read_tags(path): + with closing(DatabaseAccessor(path)) as accessor: + tags, report = accessor.read_tags_map() + return tags, report + + +def test_shifted_relationship_layouts_report_valid_and_skipped_rows(tmp_path) -> None: + path = tmp_path / "relationships.sqlite" + create_relationship_schema(path) + + with closing(DatabaseAccessor(path)) as accessor: + categories, category_report = accessor.read_category_assignments() + refunds, refund_report = accessor.read_refund_maps() + tags, tag_report = accessor.read_tags_map() + + assert categories == {100: [(10, 5)]} + assert category_report.source_count == 3 + assert category_report.parsed_count == 1 + assert category_report.status == "partial" + assert "PRIVATE_PAYLOAD" not in json.dumps(category_report.as_dict()) + assert refunds == {102: 103} + assert refund_report.source_count == 2 + assert refund_report.parsed_count == 1 + assert refund_report.status == "partial" + assert tags == {100: [200]} + assert tag_report.storage_name == "Z_37TAGS" + assert tag_report.source_count == 2 + assert tag_report.parsed_count == 1 + assert tag_report.skipped[0].record_id == "row:1" + assert tag_report.status == "partial" + + manager_report = TransactionManager().load(accessor) + + assert manager_report.status == "partial" + assert not manager_report.complete + assert set(manager_report.relationships) == { + "category_assignments", + "refund_links", + "transaction_tags", + } + assert ( + manager_report.as_dict()["relationships"]["transaction_tags"]["storage_name"] + == "Z_37TAGS" + ) + + +def test_category_counts_exclude_supported_nontransaction_owners(tmp_path) -> None: + path = tmp_path / "nontransaction-category-owners.sqlite" + create_custom_relationship_schema( + path, + metadata=((2, "CategoryAssigment", 0, 0),), + tables=( + "CREATE TABLE ZCATEGORYASSIGMENT " + "(Z_PK INTEGER, ZCATEGORY INTEGER, ZAMOUNT FLOAT, " + "ZTRANSACTION INTEGER, ZBUDGET INTEGER, " + "ZSCHEDULEDTRANSACITION INTEGER, ZSTRINGHISTORYITEM INTEGER)", + ), + ) + with sqlite3.connect(path) as connection: + connection.executemany( + "INSERT INTO ZCATEGORYASSIGMENT VALUES (?, ?, ?, ?, ?, ?, ?)", + [ + (1, 10, 5.0, 100, None, None, None), + (2, 11, 2.0, None, 200, None, None), + (3, 12, 3.0, None, None, 300, None), + (4, 13, 4.0, None, None, None, 400), + ], + ) + + with closing(DatabaseAccessor(path)) as accessor: + categories, report = accessor.read_category_assignments() + + assert categories == {100: [(10, 5)]} + assert report.source_count == 1 + assert report.parsed_count == 1 + assert report.complete + + +def test_known_entity_with_missing_storage_is_unknown(tmp_path) -> None: + path = tmp_path / "missing-storage.sqlite" + create_relationship_schema(path, include_tables=False) + + with closing(DatabaseAccessor(path)) as accessor: + results = ( + accessor.read_category_assignments(), + accessor.read_refund_maps(), + accessor.read_tags_map(), + ) + manager_report = TransactionManager().load(accessor) + + assert all(data == {} for data, _ in results) + assert all( + report.storage == RelationshipStorage.UNKNOWN and not report.complete + for _, report in results + ) + assert manager_report.status == "error" + assert not manager_report.complete + + +def test_missing_optional_relationship_entities_are_absent(tmp_path) -> None: + path = tmp_path / "absent-relationships.sqlite" + create_custom_relationship_schema(path) + + with closing(DatabaseAccessor(path)) as accessor: + reports = [ + accessor.read_category_assignments()[1], + accessor.read_refund_maps()[1], + accessor.read_tags_map()[1], + ] + manager_report = TransactionManager().load(accessor) + + assert all(report.storage == RelationshipStorage.ABSENT for report in reports) + assert all(report.complete for report in reports) + assert manager_report.complete + + +@pytest.mark.parametrize( + ("reader_name", "table_statement"), + [ + ( + "read_category_assignments", + "CREATE TABLE ZCATEGORYASSIGMENT " + "(Z_PK INTEGER, ZCATEGORY INTEGER, ZTRANSACTION INTEGER, ZAMOUNT FLOAT)", + ), + ( + "read_tags_map", + "CREATE TABLE Z_37TAGS (Z_37TRANSACTIONS INTEGER, Z_36TAGS INTEGER)", + ), + ], +) +def test_physical_storage_without_metadata_is_unknown( + tmp_path, reader_name, table_statement +) -> None: + path = tmp_path / f"{reader_name}-physical-only.sqlite" + create_custom_relationship_schema(path, tables=(table_statement,)) + + with closing(DatabaseAccessor(path)) as accessor: + data, report = getattr(accessor, reader_name)() + + assert data == {} + assert report.storage == RelationshipStorage.UNKNOWN + assert report.status == "error" + assert not report.complete + + +@pytest.mark.parametrize( + ("reader_name", "metadata", "table_statement", "expected_storage"), + [ + ( + "read_category_assignments", + ((2, "CategoryAssigment", 0, 0),), + "CREATE TABLE ZCATEGORYASSIGMENT " + "(Z_PK INTEGER, ZCATEGORY INTEGER, ZTRANSACTION INTEGER, ZAMOUNT FLOAT)", + RelationshipStorage.PRESENT, + ), + ( + "read_tags_map", + ((36, "Tag", 0, 0), (37, "Transaction", 0, 0)), + "CREATE TABLE Z_37TAGS (Z_37TRANSACTIONS INTEGER, Z_36TAGS INTEGER)", + RelationshipStorage.PRESENT, + ), + ( + "read_category_assignments", + ((2, "CategoryAssigment", 0, 0),), + "CREATE TABLE ZCATEGORYASSIGMENT " + "(Z_PK INTEGER, ZCATEGORY INTEGER, ZTRANSACTION INTEGER)", + RelationshipStorage.UNKNOWN, + ), + ( + "read_tags_map", + ((36, "Tag", 0, 0), (37, "Transaction", 0, 0)), + "CREATE TABLE Z_37TAGS (Z_37TRANSACTIONS INTEGER)", + RelationshipStorage.UNKNOWN, + ), + ], +) +def test_empty_relationship_storage_requires_expected_columns( + tmp_path, reader_name, metadata, table_statement, expected_storage +) -> None: + path = tmp_path / f"{reader_name}-{expected_storage.value}.sqlite" + create_custom_relationship_schema( + path, metadata=metadata, tables=(table_statement,) + ) + + with closing(DatabaseAccessor(path)) as accessor: + data, report = getattr(accessor, reader_name)() + + assert data == {} + assert report.storage == expected_storage + assert report.complete == (expected_storage == RelationshipStorage.PRESENT) + + +@pytest.mark.parametrize( + ("alternate_table", "include_expected"), + [ + ( + "CREATE TABLE Z_36TAGS (Z_36TRANSACTIONS INTEGER, Z_35TAGS INTEGER)", + False, + ), + ( + "CREATE TABLE Z_99TAGS (Z_99TRANSACTIONS INTEGER, Z_36TAGS INTEGER)", + True, + ), + ], +) +def test_wrong_or_conflicting_transaction_tag_layout_is_unknown( + tmp_path, alternate_table, include_expected +) -> None: + path = tmp_path / "conflicting-tags.sqlite" + expected_table = ( + "CREATE TABLE Z_37TAGS (Z_37TRANSACTIONS INTEGER, Z_36TAGS INTEGER)" + ) + tables = ( + (expected_table, alternate_table) if include_expected else (alternate_table,) + ) + create_custom_relationship_schema( + path, + metadata=((36, "Tag", 0, 0), (37, "Transaction", 0, 0)), + tables=tables, + ) + + with closing(DatabaseAccessor(path)) as accessor: + tags, report = accessor.read_tags_map() + + assert tags == {} + assert report.storage == RelationshipStorage.UNKNOWN + assert not report.complete + + +@pytest.mark.parametrize( + ("info_id", "scheduled_id", "tag_id", "transaction_id"), + [(23, 31, 35, 36), (24, 32, 36, 37)], +) +def test_unrelated_tag_tables_do_not_hide_transaction_tags( + tmp_path, info_id, scheduled_id, tag_id, transaction_id +) -> None: + path = tmp_path / "unrelated-tags.sqlite" + create_custom_relationship_schema( + path, + metadata=( + (info_id, "InfoCard", 0, 0), + (scheduled_id, "ScheduledTransaction", 0, 0), + (tag_id, "Tag", 0, 0), + (transaction_id, "Transaction", 0, 0), + ), + tables=( + f"CREATE TABLE Z_{info_id}TAGS (Z_{info_id}INFOCARDS5 INTEGER, Z_{tag_id}TAGS1 INTEGER)", + f"CREATE TABLE Z_{scheduled_id}TAGS (Z_{scheduled_id}SCHEDULEDTRANSACTIONS1 INTEGER, Z_{tag_id}TAGS2 INTEGER)", + f"CREATE TABLE Z_{transaction_id}TAGS (Z_{transaction_id}TRANSACTIONS INTEGER, Z_{tag_id}TAGS INTEGER)", + ), + ) + with sqlite3.connect(path) as connection: + connection.execute(f"INSERT INTO Z_{info_id}TAGS VALUES (101, 201)") + connection.execute(f"INSERT INTO Z_{scheduled_id}TAGS VALUES (102, 202)") + connection.execute(f"INSERT INTO Z_{transaction_id}TAGS VALUES (103, 203)") + tags, report = read_tags(path) + assert tags == {103: [203]} + assert report.storage == RelationshipStorage.PRESENT + assert report.storage_name == f"Z_{transaction_id}TAGS" + assert report.complete diff --git a/tests/unit/test_schema_cache_binding.py b/tests/unit/test_schema_cache_binding.py new file mode 100644 index 0000000..852d279 --- /dev/null +++ b/tests/unit/test_schema_cache_binding.py @@ -0,0 +1,319 @@ +from contextlib import closing +from pathlib import Path +import sqlite3 + +import pytest + +from moneywiz_api import MoneywizApi +from moneywiz_api.database_accessor import DatabaseAccessor, DatabaseSchemaError +from moneywiz_api.schema_profile import detect_schema_profile + + +SCHEMA_CHANGED = "database schema changed; close and reopen the accessor" + + +def open_writer(path: Path): + connection = sqlite3.connect(path) + connection.execute("PRAGMA journal_mode=WAL") + return connection + + +def create_account_store(path: Path, *, extra_metadata=()) -> None: + connection = open_writer(path) + connection.execute( + "CREATE TABLE Z_PRIMARYKEY " + "(Z_ENT INTEGER, Z_NAME TEXT, Z_SUPER INTEGER, Z_MAX INTEGER)" + ) + connection.executemany( + "INSERT INTO Z_PRIMARYKEY VALUES (?, ?, ?, ?)", + [ + (8, "SyncObject", 0, 0), + (9, "Account", 8, 0), + (10, "CashAccount", 9, 1), + *extra_metadata, + ], + ) + connection.execute( + "CREATE TABLE ZSYNCOBJECT (" + "Z_PK INTEGER PRIMARY KEY, Z_ENT INTEGER, ZOBJECTCREATIONDATE FLOAT, " + "ZGID TEXT, ZDISPLAYORDER INTEGER, ZGROUPID INTEGER, ZNAME TEXT, " + "ZCURRENCYNAME TEXT, ZOPENINGBALANCE FLOAT, ZINFO TEXT, ZUSER INTEGER)" + ) + connection.execute( + "INSERT INTO ZSYNCOBJECT VALUES " + "(1, 10, 0, 'account-1', 1, 1, 'Before', 'EUR', 0, NULL, 1)" + ) + connection.commit() + connection.close() + + +def remap_cash_account(path: Path) -> None: + connection = open_writer(path) + connection.execute( + "UPDATE Z_PRIMARYKEY SET Z_NAME = 'RetiredCashAccount' WHERE Z_ENT = 10" + ) + connection.execute("INSERT INTO Z_PRIMARYKEY VALUES (11, 'CashAccount', 9, 1)") + connection.execute( + "UPDATE ZSYNCOBJECT SET Z_ENT = 11, ZNAME = 'After' WHERE Z_PK = 1" + ) + connection.commit() + connection.close() + + +def create_holding_store(path: Path) -> None: + connection = open_writer(path) + connection.execute( + "CREATE TABLE Z_PRIMARYKEY " + "(Z_ENT INTEGER, Z_NAME TEXT, Z_SUPER INTEGER, Z_MAX INTEGER)" + ) + connection.executemany( + "INSERT INTO Z_PRIMARYKEY VALUES (?, ?, ?, ?)", + [(8, "SyncObject", 0, 0), (24, "InvestmentHolding", 8, 1)], + ) + connection.execute( + "CREATE TABLE ZSYNCOBJECT (" + "Z_PK INTEGER PRIMARY KEY, Z_ENT INTEGER, ZOBJECTCREATIONDATE FLOAT, " + "ZGID TEXT, ZINVESTMENTACCOUNT INTEGER, ZOPENNINGNUMBEROFSHARES FLOAT, " + "ZNUMBEROFSHARES FLOAT, ZPRICEPERSHARE FLOAT, ZSYMBOL TEXT, " + "ZHOLDINGTYPE TEXT, ZDESC TEXT, ZISPRICEPERSHAREAVAILABLEONLINE INTEGER, " + "ZINVESTMENTOBJECTTYPE INTEGER, ZCOSTBASISOFMISSINGOBSHARES FLOAT)" + ) + connection.execute( + "INSERT INTO ZSYNCOBJECT VALUES " + "(1, 24, 0, 'holding-1', 5, NULL, 2, 10, 'ACME', NULL, 'Acme', 0, 0, 0)" + ) + connection.commit() + connection.close() + + +def test_metadata_remap_refuses_and_invalidates_loaded_manager(tmp_path) -> None: + path = tmp_path / "metadata-remap.sqlite" + create_account_store(path) + with closing(MoneywizApi(path, managers=("accounts",))) as api: + manager = api.account_manager + remap_cash_account(path) + + with pytest.raises(DatabaseSchemaError, match=f"^{SCHEMA_CHANGED}$"): + api.load(("accounts",)) + + assert api.account_manager is manager + assert manager.records() == {} + assert not manager.load_report.observed + assert not api.completeness().complete + assert not api.accessor._con.in_transaction + + with closing(MoneywizApi(path, managers=("accounts",))) as reopened: + account = reopened.account_manager.get(1) + assert account is not None + assert account.name == "After" + assert reopened.completeness().managers["accounts"].complete + + +@pytest.mark.parametrize("duplicate", ["id", "name"]) +def test_duplicate_entity_map_keys_are_rejected(tmp_path, duplicate) -> None: + path = tmp_path / f"duplicate-{duplicate}.sqlite" + create_account_store(path) + connection = open_writer(path) + if duplicate == "id": + connection.execute( + "INSERT INTO Z_PRIMARYKEY VALUES (10, 'OtherCashAccount', 9, 0)" + ) + else: + connection.execute("INSERT INTO Z_PRIMARYKEY VALUES (11, 'CashAccount', 9, 0)") + connection.commit() + connection.close() + + with pytest.raises(DatabaseSchemaError, match="duplicate entity"): + DatabaseAccessor(path) + + +def test_relationship_schema_change_refuses_with_equal_profile(tmp_path) -> None: + path = tmp_path / "relationship-ddl.sqlite" + create_account_store(path) + with closing(DatabaseAccessor(path)) as accessor: + profile = accessor.schema_profile + connection = open_writer(path) + connection.execute( + "CREATE TABLE Z_99TAGS (Z_99TRANSACTIONS INTEGER, Z_36TAGS INTEGER)" + ) + connection.commit() + connection.close() + assert detect_schema_profile(accessor._con) == profile + + with pytest.raises(DatabaseSchemaError, match=f"^{SCHEMA_CHANGED}$"): + accessor.read_tags_map() + + +def test_normal_data_and_zmax_growth_reload_successfully(tmp_path) -> None: + path = tmp_path / "growth.sqlite" + create_account_store(path) + with closing(MoneywizApi(path, managers=("accounts",))) as api: + manager = api.account_manager + connection = open_writer(path) + connection.execute("UPDATE Z_PRIMARYKEY SET Z_MAX = 2 WHERE Z_ENT = 10") + connection.execute("UPDATE ZSYNCOBJECT SET ZNAME = 'Edited' WHERE Z_PK = 1") + connection.execute( + "INSERT INTO ZSYNCOBJECT VALUES " + "(2, 10, 0, 'account-2', 2, 1, 'Added', 'EUR', 0, NULL, 1)" + ) + connection.commit() + connection.close() + + api.load(("accounts",)) + report = api.account_manager.load_report + + assert api.account_manager is manager + assert report.complete + assert report.source_ids == (1, 2) + edited = manager.get(1) + added = manager.get(2) + assert edited is not None + assert added is not None + assert edited.name == "Edited" + assert added.name == "Added" + + +def test_initialization_race_binds_one_old_snapshot_then_refuses(tmp_path, monkeypatch): + import moneywiz_api.database_accessor as accessor_module + + path = tmp_path / "initialization-race.sqlite" + create_holding_store(path) + original_detect = accessor_module.detect_schema_profile + migration_committed = False + + def migrate_then_detect(connection): + nonlocal migration_committed + writer = open_writer(path) + writer.execute( + "ALTER TABLE ZSYNCOBJECT RENAME COLUMN ZNUMBEROFSHARES TO ZNUMBEROFSHARES1" + ) + writer.execute( + "ALTER TABLE ZSYNCOBJECT RENAME COLUMN ZPRICEPERSHARE TO ZPRICEPERSHARE1" + ) + writer.execute( + "UPDATE Z_PRIMARYKEY SET Z_NAME = 'RetiredHolding' WHERE Z_ENT = 24" + ) + writer.execute( + "INSERT INTO Z_PRIMARYKEY VALUES (25, 'InvestmentHolding', 8, 1)" + ) + writer.execute("UPDATE ZSYNCOBJECT SET Z_ENT = 25") + writer.commit() + writer.close() + migration_committed = True + return original_detect(connection) + + monkeypatch.setattr(accessor_module, "detect_schema_profile", migrate_then_detect) + accessor = DatabaseAccessor(path) + + assert migration_committed + assert accessor.schema_profile.profile_id == "unsuffixed-investment-columns" + assert accessor.ent_for("InvestmentHolding") == 24 + assert (24, "InvestmentHolding") in accessor._schema_identity[1] + assert (25, "InvestmentHolding", 8) not in accessor._schema_identity[1] + with accessor._raw_read_transaction(): + assert accessor._read_schema_identity() != accessor._schema_identity + with pytest.raises(DatabaseSchemaError, match=f"^{SCHEMA_CHANGED}$"): + accessor.query_objects(["InvestmentHolding"]) + accessor.close() + + monkeypatch.setattr(accessor_module, "detect_schema_profile", original_detect) + with closing(DatabaseAccessor(path)) as reopened: + assert reopened.schema_profile.profile_id == "suffixed-investment-columns" + assert reopened.ent_for("InvestmentHolding") == 25 + + +def test_reload_race_reads_old_snapshot_then_next_load_refuses( + tmp_path, monkeypatch +) -> None: + path = tmp_path / "reload-race.sqlite" + create_account_store(path) + with closing(MoneywizApi(path, managers=("accounts",))) as api: + original_verify = api.accessor._verify_schema_identity + migration_committed = False + + def verify_then_migrate(): + nonlocal migration_committed + original_verify() + if not migration_committed: + remap_cash_account(path) + migration_committed = True + + monkeypatch.setattr( + api.accessor, "_verify_schema_identity", verify_then_migrate + ) + + api.load(("accounts",)) + report = api.account_manager.load_report + + assert migration_committed + assert report.complete + assert report.source_ids == (1,) + account = api.account_manager.get(1) + assert account is not None + assert account.name == "Before" + with pytest.raises(DatabaseSchemaError, match=f"^{SCHEMA_CHANGED}$"): + api.load(("accounts",)) + assert api.account_manager.records() == {} + assert not api.account_manager.load_report.observed + + +def test_interrupted_nested_read_refuses_and_recovers(tmp_path) -> None: + path = tmp_path / "interrupted-read.sqlite" + create_account_store(path) + + with closing(DatabaseAccessor(path)) as accessor: + with pytest.raises( + DatabaseSchemaError, match="read transaction was interrupted" + ): + with accessor.read_transaction(): + accessor._con.rollback() + with accessor.read_transaction(): + pytest.fail("nested read must refuse the lost snapshot") + + assert accessor.query_objects(["CashAccount"]) + + +def test_data_only_investment_alias_change_requires_reopening(tmp_path): + from tests.unit.test_profiled_investment_values import investment_transaction_row + from moneywiz_api.managers.transaction_manager import TransactionManager + from moneywiz_api.model.transaction import InvestmentBuyTransaction + + row = investment_transaction_row(40, -20.0) + row.update( + ZNUMBEROFSHARES=2.0, + ZPRICEPERSHARE=10.0, + ZNUMBEROFSHARES1=None, + ZPRICEPERSHARE1=None, + ) + path = tmp_path / "data-only-alias.sqlite" + with sqlite3.connect(path) as connection: + connection.execute( + "CREATE TABLE Z_PRIMARYKEY (Z_ENT INTEGER, Z_NAME TEXT, Z_SUPER INTEGER)" + ) + connection.execute( + "INSERT INTO Z_PRIMARYKEY VALUES (40, 'InvestmentBuyTransaction', 0)" + ) + columns = ", ".join(f'"{column}"' for column in row) + connection.execute(f"CREATE TABLE ZSYNCOBJECT ({columns})") + placeholders = ", ".join("?" for _ in row) + connection.execute( + f"INSERT INTO ZSYNCOBJECT VALUES ({placeholders})", tuple(row.values()) + ) + with closing(DatabaseAccessor(path)) as accessor: + original_schema = accessor._read_schema_identity() + assert accessor.get_record(40, InvestmentBuyTransaction).number_of_shares == 2 + writer = sqlite3.connect(path) + writer.execute( + "UPDATE ZSYNCOBJECT SET ZNUMBEROFSHARES=NULL, ZPRICEPERSHARE=NULL, " + "ZNUMBEROFSHARES1=2.0, ZPRICEPERSHARE1=10.0" + ) + writer.commit() + writer.close() + assert accessor._read_schema_identity() == original_schema + assert detect_schema_profile(accessor._con) != accessor.schema_profile + with pytest.raises(DatabaseSchemaError, match="close and reopen"): + accessor.get_record(40, InvestmentBuyTransaction) + with pytest.raises(DatabaseSchemaError, match="close and reopen"): + accessor.get_record_by_gid("investment-40", InvestmentBuyTransaction) + with pytest.raises(DatabaseSchemaError, match="close and reopen"): + TransactionManager().load(accessor) diff --git a/tests/unit/test_schema_profiles.py b/tests/unit/test_schema_profiles.py new file mode 100644 index 0000000..471238b --- /dev/null +++ b/tests/unit/test_schema_profiles.py @@ -0,0 +1,188 @@ +import sqlite3 + +from moneywiz_api.model.raw_data_handler import RawDataHandler +from moneywiz_api.schema_profile import detect_schema_profile + + +def make_connection(*columns: str) -> sqlite3.Connection: + connection = sqlite3.connect(":memory:") + column_sql = ", ".join(f'"{column}" FLOAT' for column in columns) + connection.execute( + f"CREATE TABLE ZSYNCOBJECT (Z_PK INTEGER, Z_ENT INTEGER, {column_sql})" + ) + return connection + + +def add_investment_entities(connection: sqlite3.Connection) -> None: + 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"), + ], + ) + + +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 test_detects_suffixed_fixture_profile() -> None: + connection = make_connection("ZNUMBEROFSHARES1", "ZPRICEPERSHARE1") + + profile = detect_schema_profile(connection) + + assert profile.profile_id == "suffixed-investment-columns" + assert profile.holding_number_of_shares_column == "ZNUMBEROFSHARES1" + assert profile.transaction_number_of_shares_column == "ZNUMBEROFSHARES1" + assert profile.price_per_share_column == "ZPRICEPERSHARE1" + + +def test_detects_unsuffixed_profile_when_aliases_are_absent() -> None: + connection = make_connection("ZNUMBEROFSHARES", "ZPRICEPERSHARE") + + profile = detect_schema_profile(connection) + + assert profile.profile_id == "unsuffixed-investment-columns" + assert profile.holding_number_of_shares_column == "ZNUMBEROFSHARES" + assert profile.transaction_number_of_shares_column == "ZNUMBEROFSHARES" + assert profile.price_per_share_column == "ZPRICEPERSHARE" + + +def test_detects_mixed_investment_columns() -> None: + connection = make_connection("ZNUMBEROFSHARES", "ZPRICEPERSHARE1") + + profile = detect_schema_profile(connection) + + assert profile.profile_id == "mixed-investment-columns" + assert profile.holding_number_of_shares_column == "ZNUMBEROFSHARES" + assert profile.transaction_number_of_shares_column == "ZNUMBEROFSHARES" + assert profile.price_per_share_column == "ZPRICEPERSHARE1" + + +def test_detects_consumer_specific_mixed_share_columns() -> None: + connection = make_connection( + "ZNUMBEROFSHARES", "ZNUMBEROFSHARES1", "ZPRICEPERSHARE1" + ) + add_investment_entities(connection) + insert_investment_row(connection, 24, ZNUMBEROFSHARES=2.0, ZNUMBEROFSHARES1=None) + insert_investment_row(connection, 40, ZNUMBEROFSHARES=None, ZNUMBEROFSHARES1=9.0) + insert_investment_row(connection, 41, ZNUMBEROFSHARES=None, ZNUMBEROFSHARES1=3.0) + + profile = detect_schema_profile(connection) + + assert profile.profile_id == "mixed-investment-columns" + assert profile.holding_number_of_shares_column == "ZNUMBEROFSHARES" + assert profile.transaction_number_of_shares_column == "ZNUMBEROFSHARES1" + assert profile.price_per_share_column == "ZPRICEPERSHARE1" + + +def test_detects_observed_store_profile_from_entity_rows() -> None: + connection = make_connection("ZNUMBEROFSHARES", "ZPRICEPERSHARE", "ZPRICEPERSHARE1") + add_investment_entities(connection) + insert_investment_row( + connection, + 24, + ZNUMBEROFSHARES=None, + ZPRICEPERSHARE=15.0, + ZPRICEPERSHARE1=None, + ) + insert_investment_row( + connection, + 40, + ZNUMBEROFSHARES=2.0, + ZPRICEPERSHARE=None, + ZPRICEPERSHARE1=10.0, + ) + insert_investment_row( + connection, + 41, + ZNUMBEROFSHARES=3.0, + ZPRICEPERSHARE=None, + ZPRICEPERSHARE1=11.0, + ) + + profile = detect_schema_profile(connection) + + assert profile.profile_id == "mixed-investment-columns" + assert profile.holding_number_of_shares_column == "ZNUMBEROFSHARES" + assert profile.transaction_number_of_shares_column == "ZNUMBEROFSHARES" + assert profile.price_per_share_column == "ZPRICEPERSHARE1" + + +def test_rejects_dual_price_columns_without_discriminating_rows() -> None: + connection = make_connection("ZNUMBEROFSHARES", "ZPRICEPERSHARE", "ZPRICEPERSHARE1") + add_investment_entities(connection) + insert_investment_row( + connection, 40, ZNUMBEROFSHARES=2.0, ZPRICEPERSHARE=10.0, ZPRICEPERSHARE1=1.0 + ) + + profile = detect_schema_profile(connection) + + assert profile.profile_id == "unknown" + assert not profile.is_known + + +def test_rejects_ambiguous_investment_column_profile() -> None: + connection = make_connection( + "ZNUMBEROFSHARES", + "ZNUMBEROFSHARES1", + "ZPRICEPERSHARE", + "ZPRICEPERSHARE1", + ) + + profile = detect_schema_profile(connection) + + assert profile.profile_id == "unknown" + assert not profile.is_known + + +def test_decimal_alias_reads_both_profiles() -> None: + assert ( + RawDataHandler.get_decimal_alias( + {"ZNUMBEROFSHARES": 2.5}, "ZNUMBEROFSHARES1", "ZNUMBEROFSHARES" + ) + == 2.5 + ) + assert ( + RawDataHandler.get_decimal_alias( + {"ZNUMBEROFSHARES1": 3.5}, "ZNUMBEROFSHARES1", "ZNUMBEROFSHARES" + ) + == 3.5 + ) + assert ( + RawDataHandler.get_decimal_alias( + {"ZNUMBEROFSHARES1": None, "ZNUMBEROFSHARES": 4.5}, + "ZNUMBEROFSHARES1", + "ZNUMBEROFSHARES", + ) + == 4.5 + ) + assert ( + RawDataHandler.get_nullable_decimal_alias( + {"ZNUMBEROFSHARES": None}, "ZNUMBEROFSHARES1", "ZNUMBEROFSHARES" + ) + is None + ) + assert ( + RawDataHandler.get_nullable_decimal_alias( + {"ZNUMBEROFSHARES1": None, "ZNUMBEROFSHARES": 5.5}, + "ZNUMBEROFSHARES1", + "ZNUMBEROFSHARES", + ) + == 5.5 + ) + + +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..1a6ea29 --- /dev/null +++ b/tests/unit/test_transaction_compatibility.py @@ -0,0 +1,126 @@ +from contextlib import contextmanager +from decimal import Decimal +from typing import Any, cast + +import pytest + +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("reconciled", [None, 2, "1", True]) +def test_transaction_validation_rejects_nonbinary_reconciled(reconciled) -> None: + with pytest.raises(RuntimeError, match="Failed to convert field reconciled"): + TransferWithdrawTransaction( + transfer_withdraw_row(ZORIGINALEXCHANGERATE=1.0, ZRECONCILED=reconciled) + ) + + +def test_transaction_validation_preserves_nullable_description() -> None: + transaction = TransferWithdrawTransaction( + transfer_withdraw_row(ZORIGINALEXCHANGERATE=1.0, ZDESC2=None) + ) + transaction.validate() + assert transaction.description is None + + +def test_transaction_validation_rejects_binary_description() -> None: + transaction = TransferWithdrawTransaction( + transfer_withdraw_row(ZORIGINALEXCHANGERATE=1.0, ZDESC2=b"binary") + ) + with pytest.raises(AssertionError): + transaction.validate() + + +def test_transfer_withdraw_rejects_zero_rate_reconstruction() -> None: + with pytest.raises(ValueError, match="zero exchange rate"): + TransferWithdrawTransaction(transfer_withdraw_row()) + + +def test_transfer_withdraw_reconstructs_with_nonzero_rate() -> None: + transaction = TransferWithdrawTransaction( + transfer_withdraw_row(ZORIGINALEXCHANGERATE=1.0) + ) + + assert transaction.original_amount == Decimal("-10.0") + + +def test_transfer_withdraw_preserves_stored_amount_with_rounded_rate() -> None: + transaction = TransferWithdrawTransaction( + transfer_withdraw_row( + ZAMOUNT1=-3.333, + ZORIGINALRECIPIENTAMOUNT=10.0, + ZORIGINALEXCHANGERATE=3.0, + ) + ) + + assert transaction.original_amount == Decimal("-3.333") + + +class TransferWithdrawManager(RecordManager): + @property + def ents(self): + return {"TransferWithdrawTransaction": TransferWithdrawTransaction} + + +class TransferWithdrawAccessor: + def __init__(self, rows): + self.rows = rows + + @contextmanager + def read_transaction(self): + yield + + def query_objects(self, _typenames): + return self.rows + + def typename_for(self, _ent_id): + return "TransferWithdrawTransaction" + + +def test_transfer_manager_skips_zero_rate_and_loads_later_valid_row() -> None: + manager = TransferWithdrawManager() + manager.load( + cast( + Any, + TransferWithdrawAccessor( + [ + transfer_withdraw_row(), + 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")]