Spaces:
Running on Zero
Running on Zero
File size: 10,138 Bytes
bfd6783 f8d827b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 | from datetime import date
import pytest
from finbot.extract import ExtractedDocument
from finbot.ingest import Ingestor, _coerce_amount, _draft_from_obj, _parse_any_date
from finbot.llm import LLMUnavailable
TODAY = date(2026, 8, 15)
class StubLLM:
"""Stands in for LLMClient without touching the network."""
def __init__(self, payload=None, enabled=True, raises=None):
self.payload = payload
self.enabled = enabled
self.raises = raises
self.calls = []
def chat_json(self, system, user, **kwargs):
self.calls.append(("chat_json", system, user))
if self.raises:
raise self.raises
return self.payload
def vision_json(self, system, user, image_bytes, mime_type, **kwargs):
self.calls.append(("vision_json", system, user))
if self.raises:
raise self.raises
return self.payload
class TestHelpers:
@pytest.mark.parametrize(
"raw,expected",
[(10, 10.0), (-10, 10.0), ("1,234.56", 1234.56), ("₹450", 450.0), ("", None), (None, None)],
)
def test_coerce_amount(self, raw, expected):
assert _coerce_amount(raw) == expected
@pytest.mark.parametrize(
"raw,expected",
[
("2026-08-01", date(2026, 8, 1)),
("01/08/2026", date(2026, 8, 1)), # day-first
("", TODAY),
("garbage", TODAY),
(None, TODAY),
],
)
def test_parse_any_date(self, raw, expected):
assert _parse_any_date(raw, TODAY) == expected
def test_draft_from_obj(self):
draft = _draft_from_obj(
{
"amount": 450,
"currency": "inr",
"description": "Swiggy order",
"date": "2026-08-01",
"category": "food_dining",
},
"USD",
TODAY,
)
assert draft.amount_minor == 45000
assert draft.currency == "INR"
assert draft.category == "food_dining"
def test_draft_rejects_zero_and_missing_amounts(self):
assert _draft_from_obj({"amount": 0}, "INR", TODAY) is None
assert _draft_from_obj({}, "INR", TODAY) is None
assert _draft_from_obj("not a dict", "INR", TODAY) is None
def test_draft_falls_back_to_keyword_category(self):
# Model gave an unusable category, but the text is obviously transport.
draft = _draft_from_obj(
{"amount": 100, "description": "uber ride", "category": "???"}, "INR", TODAY
)
assert draft.category == "transport"
def test_draft_defaults_currency(self):
draft = _draft_from_obj({"amount": 10}, "SGD", TODAY)
assert draft.currency == "SGD"
class TestFromText:
def test_deterministic_path_does_not_call_the_model(self):
llm = StubLLM()
draft = Ingestor(llm, "INR").from_text("coffee 250", TODAY)
assert draft.amount_minor == 25000
assert llm.calls == [] # the whole point of the fast path
def test_falls_back_to_the_model_without_an_amount(self):
llm = StubLLM(payload={"amount": 12, "currency": "USD", "description": "a coffee"})
draft = Ingestor(llm, "INR").from_text("bought a coffee for twelve dollars", TODAY)
assert draft is not None
assert draft.amount_minor == 1200
assert len(llm.calls) == 1
def test_returns_none_when_the_model_is_disabled(self):
llm = StubLLM(enabled=False)
assert Ingestor(llm, "INR").from_text("no numbers here", TODAY) is None
def test_model_failure_degrades_quietly(self):
llm = StubLLM(raises=LLMUnavailable("down"))
assert Ingestor(llm, "INR").from_text("no numbers here", TODAY) is None
class TestFromReceipt:
def test_reads_a_total(self):
llm = StubLLM(
payload={
"merchant": "Blue Tokai",
"amount": 480,
"currency": "INR",
"date": "2026-08-14",
"category": "food_dining",
"confidence": 0.9,
}
)
draft, error = Ingestor(llm, "INR").from_receipt(b"jpeg", "image/jpeg", TODAY)
assert error == ""
assert draft.amount_minor == 48000
assert draft.merchant == "Blue Tokai"
def test_explains_when_disabled(self):
llm = StubLLM(enabled=False)
draft, error = Ingestor(llm, "INR").from_receipt(b"x", "image/jpeg", TODAY)
assert draft is None
assert "HF_TOKEN" in error
def test_explains_an_unreadable_image(self):
llm = StubLLM(payload={"amount": 0})
draft, error = Ingestor(llm, "INR").from_receipt(b"x", "image/jpeg", TODAY)
assert draft is None
assert "couldn't find a total" in error
class TestFromDocument:
def _doc(self):
return ExtractedDocument(kind="csv", rows=[["Date", "Desc", "Amt"], ["01/08/2026", "Swiggy", "450"]])
def test_extracts_debits(self):
llm = StubLLM(
payload={
"transactions": [
{
"date": "2026-08-01",
"description": "Swiggy",
"amount": 450,
"currency": "INR",
"direction": "debit",
"category": "food_dining",
}
]
}
)
pending, error = Ingestor(llm, "INR").from_document(self._doc(), user_id=1, today=TODAY)
assert error == ""
assert len(pending.drafts) == 1
assert pending.total_by_currency() == {"INR": 45000}
def test_credits_are_counted_not_stored(self):
# Storing income alongside spend would corrupt every total.
llm = StubLLM(
payload={
"transactions": [
{"date": "2026-08-01", "description": "Swiggy", "amount": 450,
"currency": "INR", "direction": "debit"},
{"date": "2026-08-03", "description": "Salary", "amount": 85000,
"currency": "INR", "direction": "credit"},
]
}
)
pending, error = Ingestor(llm, "INR").from_document(self._doc(), user_id=1, today=TODAY)
assert len(pending.drafts) == 1
assert "credits skipped" in pending.label
def test_counts_unreadable_rows(self):
llm = StubLLM(
payload={
"transactions": [
{"date": "2026-08-01", "description": "Swiggy", "amount": 450,
"currency": "INR", "direction": "debit"},
{"description": "junk"}, # no amount
]
}
)
pending, _ = Ingestor(llm, "INR").from_document(self._doc(), user_id=1, today=TODAY)
assert len(pending.drafts) == 1
assert pending.skipped == 1
def test_accepts_a_bare_array(self):
llm = StubLLM(
payload=[{"date": "2026-08-01", "description": "Swiggy", "amount": 450,
"currency": "INR", "direction": "debit"}]
)
pending, error = Ingestor(llm, "INR").from_document(self._doc(), user_id=1, today=TODAY)
assert error == ""
assert len(pending.drafts) == 1
def test_reports_when_nothing_spendable_is_found(self):
llm = StubLLM(payload={"transactions": []})
pending, error = Ingestor(llm, "INR").from_document(self._doc(), user_id=1, today=TODAY)
assert pending is None
assert "no spending transactions" in error
def test_unsupported_document_short_circuits(self):
llm = StubLLM()
doc = ExtractedDocument(kind="unsupported", note="nope")
pending, error = Ingestor(llm, "INR").from_document(doc, user_id=1, today=TODAY)
assert pending is None
assert error == "nope"
assert llm.calls == [] # never spend a call on a file we cannot read
def test_flags_truncation(self):
llm = StubLLM(
payload={"transactions": [{"date": "2026-08-01", "description": "x",
"amount": 1, "currency": "INR", "direction": "debit"}]}
)
doc = self._doc()
doc.truncated = True
pending, _ = Ingestor(llm, "INR").from_document(doc, user_id=1, today=TODAY)
assert "truncated" in pending.label
def test_disabled_model_explains_itself(self):
llm = StubLLM(enabled=False)
pending, error = Ingestor(llm, "INR").from_document(self._doc(), user_id=1, today=TODAY)
assert pending is None
assert "HF_TOKEN" in error
class TestFailuresDoNotLeakInternals:
"""An infra failure is not the user's fault and must not surface tracebacks.
Regression: a missing torchvision made every receipt reply
"I couldn't read that receipt (local vision generation failed: 'ImportError').
Try a clearer photo?" -- which both leaked internals and blamed the photo.
"""
def _doc(self):
return ExtractedDocument(kind="csv", rows=[["Date", "Amt"], ["01/08/2026", "450"]])
def test_receipt_infra_failure_message(self):
llm = StubLLM(raises=LLMUnavailable("local vision generation failed: 'ImportError'"))
draft, error = Ingestor(llm, "INR").from_receipt(b"jpeg", "image/jpeg", TODAY)
assert draft is None
assert "ImportError" not in error
assert "clearer photo" not in error
assert "type the amount" in error
def test_statement_infra_failure_message(self):
llm = StubLLM(raises=LLMUnavailable("CUDA out of memory"))
pending, error = Ingestor(llm, "INR").from_document(self._doc(), user_id=1, today=TODAY)
assert pending is None
assert "CUDA" not in error
def test_unreadable_image_still_blames_the_image_not_the_bot(self):
# Model ran fine and found no total: here it IS about the picture.
llm = StubLLM(payload={"amount": 0})
draft, error = Ingestor(llm, "INR").from_receipt(b"jpeg", "image/jpeg", TODAY)
assert draft is None
assert "couldn't find a total" in error
|