LedgerLine/tests/test_seed_data.py

"""G4 -- the seed data is large, skewed, and contains the intended poison rows."""
import os
import subprocess
from pathlib import Path

import psycopg
import pytest

from harness.config import POSTGRES_DSN, REPO_ROOT

pytestmark = pytest.mark.seed

TARGET_COUNTS = {
    "accounts": 800,
    "transactions": 250_000,
    "merchant_categories": 45,
    "sanctions_entries": 12_000,
}


@pytest.fixture
def db_cursor(dsn, seeded):
    with psycopg.connect(dsn, autocommit=True) as conn:
        with conn.cursor() as cur:
            yield cur


def _count(cur, table):
    cur.execute(f"SELECT count(*) FROM {table}")
    return cur.fetchone()[0]


def test_row_counts_match_manifest_and_targets(manifest, db_cursor):
    """G4: row counts match the manifest and the spec's targets within +-2%.

    `transactions` is compared with >= rather than == : `make test-contract`
    runs contract tests (which legitimately create new transactions) ahead of
    this one in the same session (alphabetical file collection order), so its
    count can only grow past the seed-time manifest snapshot, never shrink or
    diverge below it. The other three tables are never mutated by any test,
    so they stay an exact match.
    """
    for table, target in TARGET_COUNTS.items():
        actual = _count(db_cursor, table)
        manifest_count = manifest["counts"][table]
        if table == "transactions":
            assert actual >= manifest_count, f"{table}: db={actual} manifest={manifest_count}"
        else:
            assert actual == manifest_count, f"{table}: db={actual} manifest={manifest_count}"
        tolerance = target * 0.02
        assert abs(actual - target) <= tolerance, f"{table}: {actual} not within 2% of {target}"


def test_exactly_five_hot_accounts_matching_manifest(manifest, db_cursor):
    """G4: exactly 5 accounts have is_hot = true, and they match hot_account_ids."""
    db_cursor.execute("SELECT id FROM accounts WHERE is_hot = true")
    hot_ids = {str(r[0]) for r in db_cursor.fetchall()}
    assert len(hot_ids) == 5, hot_ids
    assert hot_ids == set(manifest["hot_account_ids"])


def test_hot_accounts_own_20_to_40_percent_of_transactions(manifest, db_cursor):
    """G4: the 5 hot accounts own between 20% and 40% of all transactions."""
    hot_ids = list(manifest["hot_account_ids"])
    db_cursor.execute("SELECT count(*) FROM transactions WHERE account_id = ANY(%s)", (hot_ids,))
    hot_count = db_cursor.fetchone()[0]
    total = _count(db_cursor, "transactions")
    share = hot_count / total
    assert 0.20 <= share <= 0.40, f"hot accounts own {share:.1%} of transactions"


def test_transactions_span_at_least_110_distinct_days(db_cursor):
    """G4: transactions span >= 110 distinct days."""
    db_cursor.execute("SELECT count(DISTINCT created_at::date) FROM transactions")
    (days,) = db_cursor.fetchone()
    assert days >= 110, days


def test_currency_distribution_roughly_70_20_10(db_cursor):
    """G4: currency distribution is roughly 70/20/10 EUR/USD/GBP (+-10 points each)."""
    db_cursor.execute("SELECT currency, count(*) FROM transactions GROUP BY currency")
    rows = dict(db_cursor.fetchall())
    total = sum(rows.values())
    expected = {"EUR": 0.70, "USD": 0.20, "GBP": 0.10}
    for currency, target in expected.items():
        actual = rows.get(currency, 0) / total
        assert abs(actual - target) <= 0.10, f"{currency}: {actual:.1%} vs target {target:.1%}"


def test_status_distribution_roughly_92_5_3(db_cursor):
    """G4: status distribution is roughly 92/5/3 settled/rejected/pending (+-5 points)."""
    db_cursor.execute("SELECT status, count(*) FROM transactions GROUP BY status")
    rows = dict(db_cursor.fetchall())
    total = sum(rows.values())
    expected = {"settled": 0.92, "rejected": 0.05, "pending": 0.03}
    for status, target in expected.items():
        actual = rows.get(status, 0) / total
        assert abs(actual - target) <= 0.05, f"{status}: {actual:.1%} vs target {target:.1%}"


def test_malformed_rows_cover_all_five_varieties(db_cursor):
    """G4: malformed rows exist in all five described varieties, 1%-4% overall."""
    total = _count(db_cursor, "transactions")

    db_cursor.execute("SELECT count(*) FROM transactions WHERE currency IS NULL")
    null_currency = db_cursor.fetchone()[0]
    assert null_currency >= 1, "no null-currency rows found"

    db_cursor.execute("SELECT count(*) FROM transactions WHERE kind = 'deposit' AND amount_minor < 0")
    negative_deposit = db_cursor.fetchone()[0]
    assert negative_deposit >= 1, "no negative-amount deposit rows found"

    db_cursor.execute(r"SELECT count(*) FROM transactions WHERE merchant_name ~ '[^\x00-\x7F]'")
    non_ascii = db_cursor.fetchone()[0]
    assert non_ascii >= 1, "no non-ASCII merchant_name rows found"

    db_cursor.execute("SELECT count(*) FROM transactions WHERE created_at > now()")
    future = db_cursor.fetchone()[0]
    assert future >= 1, "no future-dated rows found"

    db_cursor.execute(
        """
        SELECT count(*) FROM transactions t
        WHERE t.category_id IS NOT NULL
          AND NOT EXISTS (SELECT 1 FROM merchant_categories c WHERE c.id = t.category_id)
        """
    )
    orphan_category = db_cursor.fetchone()[0]
    assert orphan_category >= 1, "no orphan category_id rows found"

    malformed_total = null_currency + negative_deposit + non_ascii + future + orphan_category
    share = malformed_total / total
    assert 0.01 <= share <= 0.04, f"malformed rows are {share:.1%} of total (want 1%-4%)"


def test_sanctions_entries_match_seeded_holders(db_cursor):
    """G4: at least 30 sanctions entries match names of seeded account holders."""
    db_cursor.execute(
        """
        SELECT count(DISTINCT s.id) FROM sanctions_entries s
        JOIN accounts a ON a.holder_name = s.normalized_name
        """
    )
    (matches,) = db_cursor.fetchone()
    assert matches >= 30, matches


def test_manifest_records_the_configured_seed(manifest):
    """G4 (determinism proxy): the manifest records the SEED value used to
    generate it. A full re-seed-and-diff is a manual check -- see
    tests/README.md, "Seed determinism" -- because it takes up to 60s and would
    blow the <20s seed-marker / <60s test-contract budgets on every run."""
    expected_seed = int(os.environ.get("SEED", "42"))
    assert manifest["seed"] == expected_seed


def test_no_missing_indexes_present_after_fresh_seed(db_cursor):
    """G4: neither transactions(account_id, created_at) nor
    transactions(category_id) exists in pg_indexes after a fresh seed."""
    db_cursor.execute("SELECT indexdef FROM pg_indexes WHERE tablename = 'transactions'")
    defs = [row[0].lower() for row in db_cursor.fetchall()]
    for d in defs:
        assert not ("account_id" in d and "created_at" in d), d
        assert "(category_id)" not in d.replace(" ", ""), d