Spaces:
Paused
Paused
| """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) | |
| 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 | |