""" Known-case tests for the shared provider-response validation layer (Part D). Pure, dependency-free — genuinely executed. Dual-mode: pytest, or `PYTHONPATH=. python3 tests/test_validation.py`. """ from __future__ import annotations from app.adapters.validation import ( SEVERITY_ERROR, SEVERITY_WARNING, ValidationReport, coerce_float, validate_line_items, validate_price_bar, ) # --- coerce_float --- def test_coerce_accepts_int_and_float(): r = ValidationReport() assert coerce_float(5, "x", r) == 5.0 assert coerce_float(5.5, "x", r) == 5.5 assert r.issues == [] def test_coerce_accepts_a_numeric_string(): """Providers really do return `"12.5"` -- rejecting it would be its own data loss.""" r = ValidationReport() assert coerce_float("12.5", "x", r) == 12.5 assert coerce_float(" 12.5 ", "x", r) == 12.5 assert r.issues == [] def test_coerce_rejects_a_non_numeric_string(): r = ValidationReport() assert coerce_float("n/a", "x", r) is None assert "not parseable" in r.issues[0].reason def test_coerce_rejects_a_boolean(): """True would otherwise become 1.0 and produce a silently wrong number that no downstream check could catch.""" r = ValidationReport() assert coerce_float(True, "x", r) is None assert "boolean" in r.issues[0].reason def test_coerce_rejects_nan_and_infinity(): r = ValidationReport() assert coerce_float(float("nan"), "x", r) is None assert coerce_float(float("inf"), "x", r) is None assert coerce_float(float("-inf"), "x", r) is None assert len(r.issues) == 3 def test_coerce_rejects_a_list(): r = ValidationReport() assert coerce_float([1, 2], "x", r) is None assert "unsupported type list" in r.issues[0].reason def test_coerce_treats_none_and_empty_string_as_absent_without_an_issue(): r = ValidationReport() assert coerce_float(None, "x", r) is None assert coerce_float("", "x", r) is None assert r.issues == [] def test_coerce_records_an_issue_for_an_absent_required_field(): r = ValidationReport() assert coerce_float(None, "close", r, required=True) is None assert "required field is missing" in r.issues[0].reason def test_coerce_without_a_report_still_returns_the_right_value(): assert coerce_float("3.5", "x") == 3.5 assert coerce_float("bad", "x") is None # --- validate_price_bar --- def _bar(**kw): base = {"date": "2025-09-30", "close": 100.0} base.update(kw) return base def test_valid_bar_is_accepted_with_every_field(): r = ValidationReport() out = validate_price_bar( _bar(open=99.0, high=101.0, low=98.0, adjusted_close=100.0, volume=1000), r) assert out["close"] == 100.0 and out["high"] == 101.0 and out["volume"] == 1000.0 assert r.rows_accepted == 1 and r.ok def test_bar_without_a_date_is_rejected(): r = ValidationReport() assert validate_price_bar({"close": 100.0}, r) is None assert r.rows_accepted == 0 and not r.ok def test_bar_without_a_close_is_rejected(): r = ValidationReport() assert validate_price_bar({"date": "2025-09-30"}, r) is None assert any("required field is missing" in i.reason for i in r.errors) def test_bar_with_a_non_positive_close_is_rejected(): r = ValidationReport() for bad in (0.0, -1.0): assert validate_price_bar(_bar(close=bad), r) is None assert len(r.errors) == 2 def test_a_bad_optional_field_does_not_reject_the_bar(): r = ValidationReport() out = validate_price_bar(_bar(open="junk", volume=-5), r) assert out is not None and out["close"] == 100.0 assert out["open"] is None and out["volume"] is None assert r.rows_accepted == 1 def test_high_below_low_drops_both_and_warns(): """Impossible, and it means the two are swapped or mismapped. Guessing which one is wrong would be worse than dropping both -- `close` is what downstream code needs anyway.""" r = ValidationReport() out = validate_price_bar(_bar(high=90.0, low=110.0), r) assert out["high"] is None and out["low"] is None assert out["close"] == 100.0 assert any("high is below low" in i.reason and i.severity == SEVERITY_WARNING for i in r.issues) def test_close_above_high_is_warned_but_kept(): r = ValidationReport() out = validate_price_bar(_bar(close=120.0, high=110.0, low=100.0), r) assert out["close"] == 120.0 assert any("above high" in i.reason for i in r.warnings) def test_close_below_low_is_warned_but_kept(): r = ValidationReport() out = validate_price_bar(_bar(close=90.0, high=110.0, low=100.0), r) assert out["close"] == 90.0 assert any("below low" in i.reason for i in r.warnings) def test_report_counts_and_rejection_rate(): r = ValidationReport() validate_price_bar(_bar(), r) validate_price_bar(_bar(), r) validate_price_bar({"close": 1.0}, r) # no date -> rejected assert r.rows_seen == 3 and r.rows_accepted == 2 and r.rows_rejected == 1 assert abs(r.rejection_rate - 1 / 3) < 1e-12 assert "2/3 rows accepted" in r.summary() def test_an_empty_report_has_a_zero_rejection_rate_not_a_division_error(): assert ValidationReport().rejection_rate == 0.0 def test_issues_carry_the_row_key_so_they_can_be_traced(): r = ValidationReport() validate_price_bar(_bar(open="junk"), r) assert r.issues[0].row_key == "2025-09-30" assert "2025-09-30" in str(r.issues[0]) # --- validate_line_items --- def test_line_items_coerces_numeric_strings(): r = ValidationReport() out = validate_line_items({"revenue": "1000", "net_income": "100.5"}, r) assert out == {"revenue": 1000.0, "net_income": 100.5} def test_line_items_drops_a_bad_field_without_losing_the_others(): """One impossible field does not invalidate the other thirty.""" r = ValidationReport() out = validate_line_items({"revenue": 1000.0, "inventory": "junk", "net_income": 100.0}, r) assert out == {"revenue": 1000.0, "net_income": 100.0} assert r.rows_accepted == 1 def test_negative_revenue_is_dropped_as_impossible(): r = ValidationReport() out = validate_line_items({"revenue": -500.0}, r) assert "revenue" not in out assert any("cannot be negative" in i.reason for i in r.warnings) def test_legitimately_negative_fields_are_kept(): """Equity, net income, operating income and FCF are all legitimately negative. Treating them as impossible would delete real data about distressed and loss-making companies.""" r = ValidationReport() out = validate_line_items({ "shareholders_equity": -200.0, "net_income": -50.0, "operating_income": -30.0, "operating_cash_flow": -10.0, }, r) assert out["shareholders_equity"] == -200.0 assert out["net_income"] == -50.0 assert out["operating_income"] == -30.0 assert out["operating_cash_flow"] == -10.0 def test_gross_profit_above_revenue_is_warned(): r = ValidationReport() validate_line_items({"revenue": 100.0, "gross_profit": 150.0}, r) assert any("exceeds revenue" in i.reason for i in r.warnings) def test_current_assets_above_total_assets_is_warned(): r = ValidationReport() validate_line_items({"total_assets": 100.0, "current_assets": 150.0}, r) assert any("exceeds total assets" in i.reason for i in r.warnings) def test_diluted_below_outstanding_is_warned(): """Diluted counts every share that COULD exist, so it can never be below outstanding.""" r = ValidationReport() validate_line_items({"shares_outstanding": 100.0, "diluted_shares": 90.0}, r) assert any("below shares outstanding" in i.reason for i in r.warnings) def test_consistent_line_items_produce_no_issues(): r = ValidationReport() out = validate_line_items({ "revenue": 1000.0, "gross_profit": 400.0, "total_assets": 5000.0, "current_assets": 1500.0, "shares_outstanding": 100.0, "diluted_shares": 102.0, }, r) assert r.issues == [] assert len(out) == 6 def test_a_warning_only_report_is_still_ok(): """`ok` means no ERRORs. Warnings describe suspicious data, not unusable data.""" r = ValidationReport() validate_line_items({"revenue": 100.0, "gross_profit": 150.0}, r) assert r.ok and r.warnings and not r.errors assert r.warnings[0].severity == SEVERITY_WARNING assert all(i.severity != SEVERITY_ERROR for i in r.issues) ALL_TESTS = [v for k, v in sorted(globals().items()) if k.startswith("test_")] if __name__ == "__main__": passed = failed = 0 for t in ALL_TESTS: try: t() print(f"PASS {t.__name__}") passed += 1 except Exception as exc: # noqa: BLE001 print(f"FAIL {t.__name__}: {exc}") failed += 1 print(f"\n{passed}/{passed + failed} passed") raise SystemExit(1 if failed else 0)