"""
Provider resilience audit tests (StockLab overhaul, Part 23): source-hierarchy conflict
resolution (real, tested — but see docs/AUDIT_PROVIDER_RESILIENCE.md for the finding that nothing
in this codebase currently calls it with two real sources) and the retry/backoff policy added to
`app/workers/ingest.py::ingest_security_task` this pass (not directly testable here — Celery
itself isn't installed in this sandbox, see the docs — so this file covers what IS testable
without Celery/DB: the conflict-resolution engine itself).
"""
from __future__ import annotations

from app.adapters.resolver import merge_statement_line_items, resolve_conflict, resolve_line_items


def test_resolve_conflict_picks_highest_priority_source():
    result = resolve_conflict([("SECONDARY", 100.0), ("OFFICIAL_FILING", 100.3), ("PRIMARY", 100.1)])
    assert result.selected_value == 100.3
    assert result.selected_source_tier == "OFFICIAL_FILING"
    assert result.is_conflicting is False  # all three within 0.5% of the selected value


def test_resolve_conflict_flags_real_disagreement():
    # OFFICIAL_FILING says 100, PRIMARY says 120 -> 20% apart, well over the 0.5% tolerance.
    result = resolve_conflict([("OFFICIAL_FILING", 100.0), ("PRIMARY", 120.0)])
    assert result.selected_value == 100.0
    assert result.is_conflicting is True


def test_resolve_conflict_tolerates_small_rounding_differences():
    # 100.0 vs 100.3 = 0.3% apart, under the default 0.5% tolerance -> not flagged as conflicting.
    result = resolve_conflict([("OFFICIAL_FILING", 100.0), ("PRIMARY", 100.3)])
    assert result.is_conflicting is False


def test_resolve_conflict_handles_zero_value_without_dividing_by_zero():
    result = resolve_conflict([("OFFICIAL_FILING", 0.0), ("PRIMARY", 0.0)])
    assert result.is_conflicting is False
    result2 = resolve_conflict([("OFFICIAL_FILING", 0.0), ("PRIMARY", 5.0)])
    assert result2.is_conflicting is True  # 0 vs non-zero is a real disagreement, not a /0 crash


def test_resolve_conflict_no_sources_available():
    result = resolve_conflict([("PRIMARY", None)])
    assert result.selected_value is None
    assert result.is_conflicting is False


def test_resolve_conflict_unknown_tier_sorts_last():
    result = resolve_conflict([("SOME_UNKNOWN_TIER", 999.0), ("SECONDARY", 50.0)])
    assert result.selected_value == 50.0  # SECONDARY, a known tier, wins over an unranked one


# --- app/adapters/resolver.py::resolve_line_items (StockLab overhaul, Part A6) -------------------
# The statement-level (multi-field) counterpart to resolve_conflict() above -- see that function's
# docstring for the full design. These tests are what makes A6's "real, tested" claim honest: the
# actual ingest.py wiring that CALLS this (fetching two live adapters, writing DataQuality rows)
# cannot be executed in this sandbox (no network, no sqlalchemy) and is DOCUMENTED / NOT TESTED --
# see docs/AUDIT_PROVIDER_CONFLICT_A6.md -- but this pure resolution logic has real coverage.

_FIELDS = ("revenue", "net_income", "ebitda")


def test_resolve_line_items_no_secondary_passes_through_primary_unchanged():
    # the default case: PROVIDER_SECONDARY=NONE -- must be byte-identical to single-source ingestion
    primary = {"revenue": 100.0, "net_income": 10.0}
    resolved, conflicts = resolve_line_items(primary, None, _FIELDS, primary_tier="PRIMARY")
    assert resolved == {"revenue": 100.0, "net_income": 10.0}
    assert conflicts == []


def test_resolve_line_items_field_absent_from_both_sources_is_omitted():
    primary = {"revenue": 100.0}  # net_income, ebitda absent from both
    secondary = {"revenue": 100.0}
    resolved, conflicts = resolve_line_items(primary, secondary, _FIELDS, "PRIMARY", "SECONDARY")
    assert "net_income" not in resolved
    assert "ebitda" not in resolved
    assert resolved == {"revenue": 100.0}


def test_resolve_line_items_agreeing_sources_no_conflicts():
    primary = {"revenue": 100.0, "net_income": 10.0, "ebitda": 20.0}
    secondary = {"revenue": 100.2, "net_income": 10.0, "ebitda": 20.05}  # all within 0.5% tolerance
    resolved, conflicts = resolve_line_items(primary, secondary, _FIELDS, "PRIMARY", "SECONDARY")
    assert resolved == {"revenue": 100.0, "net_income": 10.0, "ebitda": 20.0}  # primary wins tier order
    assert conflicts == []


def test_resolve_line_items_flags_real_disagreement_on_one_field_only():
    primary = {"revenue": 100.0, "net_income": 10.0}
    secondary = {"revenue": 100.0, "net_income": 50.0}  # net_income disagrees by 400%
    resolved, conflicts = resolve_line_items(primary, secondary, _FIELDS, "PRIMARY", "SECONDARY")
    assert resolved["net_income"] == 10.0  # PRIMARY still wins tier order despite the conflict
    assert len(conflicts) == 1
    assert conflicts[0].field == "net_income"
    assert conflicts[0].value_a == 10.0 and conflicts[0].value_b == 50.0


def test_resolve_line_items_official_filing_secondary_overrides_primary_tier():
    # if the "secondary" adapter is actually the higher-priority tier for this field (e.g. an
    # OFFICIAL_FILING source plugged in as `secondary`), its value must still win -- tier order,
    # not argument position, decides.
    primary = {"revenue": 100.0}  # tier PRIMARY
    secondary = {"revenue": 105.0}  # tier OFFICIAL_FILING -- higher priority
    resolved, conflicts = resolve_line_items(primary, secondary, ("revenue",), "PRIMARY", "OFFICIAL_FILING")
    assert resolved["revenue"] == 105.0


def test_resolve_line_items_one_source_missing_a_field_other_has_it():
    primary = {"revenue": 100.0}  # net_income missing from primary entirely
    secondary = {"revenue": 100.0, "net_income": 10.0}
    resolved, conflicts = resolve_line_items(primary, secondary, _FIELDS, "PRIMARY", "SECONDARY")
    assert resolved["net_income"] == 10.0
    assert len(conflicts) == 1
    assert conflicts[0].field == "net_income"
    assert conflicts[0].value_a is None
    assert conflicts[0].value_b == 10.0
    assert conflicts[0].selected_value == 10.0
    assert conflicts[0].selection_reason == "Primary field missing; value supplied by secondary source"


# --- Part A9: merge_statement_line_items (the critical ingestion bug fix) ---


def test_merge_combines_all_three_statements():
    """The bug this guards: before Part A9, only the income dict reached the BalanceSheet/CashFlow
    tables, so every balance-sheet and cash-flow column was NULL for FMP-sourced data."""
    merged = merge_statement_line_items(
        {"revenue": 1000.0, "net_income": 100.0},
        {"total_assets": 5000.0, "shareholders_equity": 2000.0},
        {"operating_cash_flow": 150.0, "capital_expenditure": 40.0},
    )
    assert merged == {
        "revenue": 1000.0, "net_income": 100.0,
        "total_assets": 5000.0, "shareholders_equity": 2000.0,
        "operating_cash_flow": 150.0, "capital_expenditure": 40.0,
    }


def test_merge_income_statement_wins_on_a_shared_field():
    """FMP's cash-flow response also carries netIncome. The income statement is authoritative."""
    merged = merge_statement_line_items(
        {"net_income": 100.0}, None, {"net_income": 999.0, "operating_cash_flow": 150.0},
    )
    assert merged["net_income"] == 100.0
    assert merged["operating_cash_flow"] == 150.0


def test_merge_later_none_never_erases_an_earlier_real_value():
    merged = merge_statement_line_items({"net_income": 100.0}, {"net_income": None}, None)
    assert merged["net_income"] == 100.0


def test_merge_fills_a_field_the_income_statement_left_none():
    merged = merge_statement_line_items({"net_income": None}, {"net_income": 42.0}, None)
    assert merged["net_income"] == 42.0


def test_merge_drops_none_valued_keys_entirely():
    """A key present with a None value must not appear in the merged dict -- resolve_line_items
    treats an absent field and a None field differently only in that a None-valued key would
    otherwise be offered as a candidate."""
    merged = merge_statement_line_items({"revenue": 1000.0, "cogs": None}, None, None)
    assert "cogs" not in merged


def test_merge_with_only_an_income_statement_is_identity():
    """The DEMO adapter returns the same fully-merged dict from all three statement methods, so
    the Part A9 fix must be a no-op for demo data. This is the degenerate form of that."""
    income = {"revenue": 1000.0, "total_assets": 5000.0, "operating_cash_flow": 150.0}
    assert merge_statement_line_items(income, None, None) == income


def test_merge_of_three_identical_dicts_is_idempotent():
    """The exact DEMO shape: get_income_statements/get_balance_sheets/get_cash_flows all return
    the same dict. Merging it with itself twice must return it unchanged -- proving the ingestion
    fix cannot alter demo-mode output."""
    d = {"revenue": 1000.0, "total_assets": 5000.0, "operating_cash_flow": 150.0,
         "diluted_shares": 100.0, "net_income": 90.0}
    assert merge_statement_line_items(d, dict(d), dict(d)) == d


def test_merge_of_all_empty_inputs_is_empty():
    assert merge_statement_line_items({}, None, None) == {}


ALL_TESTS = [obj for name, obj in list(globals().items()) if name.startswith("test_") and callable(obj)]

if __name__ == "__main__":
    passed, failed = 0, []
    for fn in ALL_TESTS:
        try:
            fn()
            passed += 1
            print(f"PASS  {fn.__name__}")
        except AssertionError as e:
            failed.append(fn.__name__)
            print(f"FAIL  {fn.__name__}: {e}")
    print(f"\n{passed}/{len(ALL_TESTS)} passed")
    if failed:
        raise SystemExit(1)