| |
| |
| |
| import os, sys, json |
| os.environ.setdefault("GEMINI_API_KEY", "fake-key-for-tests") |
| os.environ.setdefault("FIREBASE", "{}") |
| os.environ.setdefault("aai_key", "fake") |
| os.environ.setdefault("whatsapp_token", "fake") |
| os.environ.setdefault("phone_number_id", "0000") |
| os.environ.setdefault("WHATSAPP_PROXY_URL", "") |
| os.environ.setdefault("VERIFY_TOKEN", "fake") |
| from unittest import mock |
|
|
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
|
|
| import firebase_admin |
| from firebase_admin import credentials, firestore |
|
|
| _fake_db = mock.MagicMock() |
| with mock.patch.object(credentials, "Certificate", return_value=mock.MagicMock()), \ |
| mock.patch.object(firebase_admin, "initialize_app", return_value=mock.MagicMock()), \ |
| mock.patch.object(firestore, "client", return_value=_fake_db): |
| import main |
| import utility |
|
|
| PASS = FAIL = 0 |
| def check(desc, got, want): |
| global PASS, FAIL |
| ok = got == want |
| PASS += ok; FAIL += (not ok) |
| print(("PASS" if ok else "FAIL") + f" {desc}\n got={got!r} want={want!r}" if not ok else f"PASS {desc}") |
|
|
| def check_true(desc, got): check(desc, bool(got), True) |
| def check_false(desc, got): check(desc, bool(got), False) |
|
|
| print("== Bug 2: price-format equivalence (_parse_price_reply) ==") |
| check("carrot: USD0.15", main._parse_price_reply("carrot: USD0.15"), {"carrot": 0.15}) |
| check("carrot: 0.15 dollars", main._parse_price_reply("carrot: 0.15 dollars"), {"carrot": 0.15}) |
| check("carrot: 0.15 USD", main._parse_price_reply("carrot: 0.15 USD"), {"carrot": 0.15}) |
| check("carrot: 15 cents", main._parse_price_reply("Carrot: 15 cents"), {"carrot": 0.15}) |
| check("carrot: 50c", main._parse_price_reply("carrot: 50c"), {"carrot": 0.5}) |
| check("peanut butter - $3", main._parse_price_reply("peanut butter - $3"), {"peanut butter": 3.0}) |
| check("multiline", main._parse_price_reply("jam: USD2\nbread 1.20 each"), {"jam": 2.0, "bread": 1.2}) |
| check("skip", main._parse_price_reply("skip"), {}) |
| check("CRUD not price", main._parse_price_reply("update transaction abc123 to USD12"), None) |
| check("question not price", main._parse_price_reply("what is on sale?"), None) |
|
|
| print("== Bug 2: bare money reply ==") |
| check("USD0.50", main._parse_bare_money_reply("USD0.50"), 0.5) |
| check("0.50 dollars", main._parse_bare_money_reply("0.50 dollars"), 0.5) |
| check("50c", main._parse_bare_money_reply("50c"), 0.5) |
| check("0.50 each", main._parse_bare_money_reply("0.50 each"), 0.5) |
| check("plain 2", main._parse_bare_money_reply("2"), 2.0) |
| check("not money", main._parse_bare_money_reply("2 jam"), None) |
| check("not money txt", main._parse_bare_money_reply("hello"), None) |
|
|
| print("== Bug 1: move-item extraction ==") |
| check("move dovi", main._extract_move_item("move dovi"), "dovi") |
| check("put dovi on sale", main._extract_move_item("put dovi on sale"), "dovi") |
| check("discount tomatoes", main._extract_move_item("discount tomatoes"), "tomatoes") |
| check("question guarded", main._extract_move_item("what is on sale?"), None) |
|
|
| print("== Bugs 4+5: debt query detection ==") |
| check_true(" who do I owe money?", utility.is_debt_query("Who do I owe money?")) |
| check_true(" who do I owe?", utility.is_debt_query("Who do I owe?")) |
| check_true(" who owes me money?", utility.is_debt_query("Who owes me money?")) |
| check_true(" show my payables", utility.is_debt_query("show my payables")) |
| check_true(" my debtors", utility.is_debt_query("list my debtors")) |
| check_false(" I owe John 50 (record, not ask)", utility.is_debt_query("I owe John 50")) |
| check_false(" Chipo owes me 20 for bread", utility.is_debt_query("Chipo owes me 20 for bread")) |
| check_false(" sold 2 jam", utility.is_debt_query("sold 2 jam")) |
|
|
| print("== Bugs 4+5: canonical report format ==") |
| custs = [{"name": "Supplier A", "payable": 50.0, "receivable": 0}, |
| {"name": "Supplier B", "payable": 75.0, "receivable": 0}, |
| {"name": "Chipo", "payable": 0, "receivable": 20.0}] |
| rep = utility._handle_customer_credit_query("who do i owe money?", custs, "USD") |
| print(rep); print("---") |
| check_true(" has You owe heading", rep.startswith("*You owe:*")) |
| check_true(" has em-dash lines", "Supplier A — USD50.00" in rep) |
| check_true(" has total", "*Total you owe:* USD125.00" in rep) |
| rep2 = utility._handle_customer_credit_query("who owes me money?", custs, "USD") |
| check_true(" receivables heading", rep2.startswith("*Who owes you:*")) |
| check_true(" receivable line", "Chipo — USD20.00" in rep2) |
| check_true(" receivable total", "*Total owed to you:* USD20.00" in rep2) |
| rep3 = utility._handle_customer_credit_query("who owes me and who do i owe", custs, "USD") |
| check_true(" combined both", ("*Who owes you:*" in rep3) and ("*You owe:*" in rep3)) |
|
|
| print("== Bug 3: money-mention provenance ==") |
| check_false(" sold 40 tomatoes to Chipo", utility._has_money_mention("I sold 40 tomatoes to Chipo")) |
| check_true(" sold 2 jam for 4", utility._has_money_mention("sold 2 jam for 4")) |
| check_true(" sold 2 jam at 3 each", utility._has_money_mention("sold 2 jam at 3 each")) |
| check_true(" USD4", utility._has_money_mention("sold jam USD4")) |
| check_true(" 40 dollars", utility._has_money_mention("sold tomatoes 40 dollars")) |
| check_true(" paid 20", utility._has_money_mention("sold jam, she paid 20")) |
|
|
| print("== Bug 3: promo price wins over guessed amount (resolve_sale_prices) ==") |
| class _Doc: |
| def __init__(self, d): self._d = d |
| def to_dict(self): return self._d |
| class _Q: |
| def __init__(self, docs): self._docs = docs |
| def where(self, *a, **k): return self |
| def get(self): return self._docs |
| class _Coll: |
| def __init__(self, name_map): self.name_map = name_map |
| def document(self, _): return self |
| def collection(self, name): return _Q(self.name_map.get(name, [])) |
| promo_db = mock.MagicMock() |
| promo_db.collection.return_value = _Coll({ |
| "stock_batches": [_Doc({"name": "tomato", "on_sale": True, "sale_price_each": 0.51, |
| "quantity_remaining": 200, "price_each": 0.51})], |
| "price_overrides": [], |
| }) |
| txns = [{"intent": "create", "transaction_type": "sale", |
| "details": {"items": [{"item": "tomato", "quantity": 40, "unit": "each"}], |
| "amount": 40.0, "customer": "Chipo"}}] |
| utility.resolve_sale_prices(promo_db, "263771", txns, raw_text="I sold 40 tomatoes to Chipo") |
| d = txns[0]["details"]; it = d["items"][0] |
| check(" promo ppu", it.get("price_per_unit"), 0.51) |
| check(" promo total", d.get("amount"), 20.4) |
| check_true(" promo flag survives _infer_prices", it.get("promo_applied")) |
|
|
| |
| txns2 = [{"intent": "create", "transaction_type": "sale", |
| "details": {"items": [{"item": "tomato", "quantity": 40, "unit": "each", |
| "price_per_unit": 1.0}], "amount": 40.0}}] |
| utility.resolve_sale_prices(promo_db, "263771", txns2, raw_text="I sold 40 tomatoes for $1 each") |
| check(" explicit price kept", txns2[0]["details"]["items"][0].get("price_per_unit"), 1.0) |
| check_true(" explicit flag set", txns2[0]["details"].get("explicit_price")) |
|
|
| print("== Bug 6: stock-in cost prompt + remembered-cost correction ==") |
| stock_txn = [{"intent": "create", "transaction_type": "stock_in", |
| "details": {"items": [{"item": "jam", "quantity": 100, "unit": "each"}]}}] |
| prompt = main._check_missing_details(stock_txn, currency="USD") |
| print(prompt); print("---") |
| check_true(" asks cost for jam", prompt is not None and "pay for the Jam" in prompt) |
| stock_ok = [{"intent": "create", "transaction_type": "stock_in", |
| "details": {"items": [{"item": "orange", "quantity": 100, "unit": "each", |
| "price_per_unit": 0.15}]}}] |
| check(" no prompt when cost known", main._check_missing_details(stock_ok, currency="USD"), None) |
|
|
| remembered = [{"intent": "create", "transaction_type": "stock_in", |
| "details": {"items": [{"item": "orange", "quantity": 100, "unit": "each", |
| "price_per_unit": 0.15, "cost_source": "remembered"}], |
| "amount": 15.0}}] |
| res = main._try_fill_missing_details("0.20 each", remembered, currency="USD", mobile="") |
| check_true(" correction accepted", res is not None) |
| if res: |
| filled, still = res |
| it = filled[0]["details"]["items"][0] |
| check(" corrected ppu", it.get("price_per_unit"), 0.2) |
| check(" corrected total", filled[0]["details"].get("amount"), 20.0) |
| check(" nothing still missing", still, None) |
|
|
| print("== card shows remembered-cost note ==") |
| card = utility.format_transaction_response([{ |
| "intent": "create", "transaction_type": "stock_in", |
| "details": {"items": [{"item": "orange", "quantity": 100, "unit": "each", |
| "price_per_unit": 0.15, "cost_source": "remembered"}], |
| "currency": "USD"}}]) |
| print(card); print("---") |
| check_true(" last-cost marker", "your last cost" in card) |
| check_true(" changed-cost hint", "reply with the new cost" in card) |
| card2 = utility.format_transaction_response([{ |
| "intent": "create", "transaction_type": "sale", |
| "details": {"items": [{"item": "tomato", "quantity": 40, "unit": "each", |
| "price_per_unit": 0.51, "promo_applied": True}], |
| "currency": "USD"}}]) |
| print(card2); print("---") |
| check_true(" on-sale marker", "on sale" in card2) |
| check_false(" no raw flag leak", "Promo Applied" in card2 or "Explicit Price" in card2) |
|
|
| print("== _money_to_float equivalence ==") |
| for v, want in [("USD0.15", 0.15), ("0.15 USD", 0.15), ("0.15 dollars", 0.15), |
| ("15 cents", 0.15), ("50c", 0.5), ("R50", 50.0), ("4", 4.0)]: |
| check(f" {v}", utility._money_to_float(v), want) |
|
|
| print("== misc: normalise, debt-regex safety ==") |
| check(" carrots→carrot", utility.normalise_item_name("Carrots"), "carrot") |
| check(" tomatoes→tomato", utility.normalise_item_name("tomatoes"), "tomato") |
|
|
| print(f"\n{'='*40}\nTOTAL: {PASS} passed, {FAIL} failed") |
| sys.exit(1 if FAIL else 0) |
|
|
