"""Prepaid wallet: every price keeps the full margin after GST and the gateway fee, a job runs inside what was reserved for it, failures cost nothing, and a PhonePe top-up is credited once and only when PhonePe says it is paid in full.""" import os import tempfile import pytest os.environ.setdefault("USERS_DB", os.path.join(tempfile.mkdtemp(), "users.db")) os.environ.setdefault("SKIP_MODEL_LOAD", "1") import wallet as W # noqa: E402 RUN = __import__("uuid").uuid4().hex[:8] # fresh customers for every run (the test database is shared) @pytest.mark.parametrize("seconds,llm", [(0, 0), (5, 0), (90, 0.20), (210, 0.04), (700, 1.5), (3000, 4.0)]) def test_every_price_keeps_the_margin_after_gst_and_fees(seconds, llm): b = W.breakdown(seconds, llm) cost_inr = b["cost_usd"] * W.USD_INR kept = b["inr"] - b["gst_inr"] - b["fee_inr"] assert kept >= cost_inr * (1 + W.MARGIN / 100) - 0.01 # margin on top of the full cost assert b["profit_inr"] >= cost_inr * W.MARGIN / 100 - 0.01 assert b["inr"] >= W.MIN_INR def test_a_reservation_covers_the_worst_case_of_its_job(): for kind, words, budget in (("fast", 2500, 600), ("smart", 4000, 1500), ("detector", 3000, 300)): p = W.plan_job(kind, words, 10 ** 7, budget) worst = W.gpu_usd(p["deadline_s"] + p["tail_s"]) + p["groq_cap"] assert W.price_paise(worst) <= p["hold"] + 100 # the job cannot cost more than it reserved assert p["groq_cap"] >= W.usual(kind, words)[1] # ... and has room for its usual Groq use def test_a_short_balance_asks_for_money_instead_of_running(): p = W.plan_job("fast", 2500, 500, 600) # ₹5 in the wallet assert "short" in p and p["short"] > 500 def test_reserve_settle_release(): e = f"wallet-test-{RUN}@example.com" W.adjust(e, 50000, "test credit") # ₹500 ref = W.reserve(e, 20000, "Smart Humanize") assert ref and W.available(e) == 30000 and W.balance(e) == 50000 assert W.reserve(e, 40000) is None # more than is free right now got = W.settle(ref, 210, 0.04, 4000, "paper.docx") assert 0 < got["charged"] <= 20000 and W.balance(e) == 50000 - got["charged"] and W.held(e) == 0 assert W.settle(ref, 210, 0.04)["charged"] == 0 # settled once only ref2 = W.reserve(e, 10000) W.release(ref2) # a failed run assert W.balance(e) == 50000 - got["charged"] and W.held(e) == 0 def test_a_charge_never_exceeds_the_reservation(): e = f"wallet-cap-{RUN}@example.com" W.adjust(e, 100000, "test credit") ref = W.reserve(e, 1500) # ₹15 reserved got = W.settle(ref, 3000, 5.0) # an impossible run far above it assert got["charged"] == 1500 def test_topup_is_credited_once_and_only_in_full(monkeypatch): import phonepe as PP e = f"wallet-topup-{RUN}@example.com" calls = {"state": "PENDING", "amount": 25000} monkeypatch.setattr(PP, "configured", lambda: True) monkeypatch.setattr(PP, "create_payment", lambda oid, paise, url, note="", message="": f"https://pay.example/{oid}") monkeypatch.setattr(PP, "order_status", lambda oid: {"state": calls["state"], "amount": calls["amount"]}) url = W.start_topup(e, 250) oid = url.rsplit("/", 1)[-1] assert W.resolve_topup(oid)[0] == "PENDING" and W.balance(e) == 0 calls.update(state="COMPLETED", amount=100) # paid, but not the full amount assert W.resolve_topup(oid)[0] == "MISMATCH" and W.balance(e) == 0 calls.update(amount=25000) assert W.resolve_topup(oid)[0] == "COMPLETED" and W.balance(e) == 25000 assert W.resolve_topup(oid)[0] == "COMPLETED" and W.balance(e) == 25000 # the return page and the poller with pytest.raises(ValueError): W.start_topup(e, 5) # below the smallest top-up def test_the_engines_cap_follows_the_wallet(): import groqai as G tok = G.WALLET_CAP.set(0.05) try: assert G._cap() == 0.05 finally: G.WALLET_CAP.reset(tok) def test_phonepe_webhook_checks_its_password_and_only_prompts_a_status_check(monkeypatch): import hashlib from fastapi import FastAPI from fastapi.testclient import TestClient import phonepe as PP monkeypatch.setattr(PP, "WEBHOOK_USER", "cswebhook") monkeypatch.setattr(PP, "WEBHOOK_PASSWORD", "s3cret-test") good = hashlib.sha256(b"cswebhook:s3cret-test").hexdigest() asked = [] app = FastAPI() PP.install_webhook(app, asked.append) c = TestClient(app) body = {"event": "checkout.order.completed", "payload": {"merchantOrderId": "CSW123ABC", "state": "COMPLETED"}} assert c.post(PP.WEBHOOK_PATH, json=body, headers={"Authorization": "wrong"}).status_code == 401 assert c.post(PP.WEBHOOK_PATH, json=body, headers={"Authorization": good}).status_code == 200 assert c.post(PP.WEBHOOK_PATH, json=body, headers={"Authorization": "SHA256 " + good.upper()}).status_code == 200 import time time.sleep(0.3) assert asked == ["CSW123ABC", "CSW123ABC"] # each call only asks PhonePe; money moves on its answer def test_report_order_is_paid_once_in_full_and_used_up_after_delivery(monkeypatch): import phonepe as PP e = f"report-order-{RUN}@example.com" st = {"state": "PENDING", "amount": 0} monkeypatch.setattr(PP, "configured", lambda: True) monkeypatch.setattr(PP, "create_payment", lambda oid, paise, url, note="", message="": f"https://pay.example/{oid}") monkeypatch.setattr(PP, "order_status", lambda oid: dict(st)) order, paise, url = W.start_report_order(e, 2, "filekey") assert paise == W.usual_price("turnitin", 0) * 2 and url.endswith(order) assert W.resolve_report_order(order)[0] == "PENDING" and W.find_paid_report(e, "filekey", 2) is None st.update(state="COMPLETED", amount=paise - 100) assert W.resolve_report_order(order)[0] == "MISMATCH" st["amount"] = paise assert W.resolve_report_order(order)[0] == "PAID" and W.balance(e) == 0 # the wallet is not touched assert W.find_paid_report(e, "filekey", 2) == order and W.find_paid_report(e, "other", 2) is None assert W.finish_report_order(order, 1) == paise // 2 # one part failed: half is owed back assert W.report_order(order)[3] == "USED" and W.find_paid_report(e, "filekey", 2) is None