"""Cash & Bank module — comprehensive backend pytest suite.""" import os import pytest import requests BASE_URL = os.environ.get("REACT_APP_BACKEND_URL", "https://offline-billing-pro-2.preview.emergentagent.com").rstrip("/") ADMIN = {"email": "regalmarketing2024@gmail.com", "password": "Rvasa@#9955"} STAFF = {"email": "staff@rmregal.com", "password": "Staff@123"} # ---------- Session fixtures ---------- @pytest.fixture(scope="session") def admin_session(): s = requests.Session() r = s.post(f"{BASE_URL}/api/auth/login", json=ADMIN, timeout=20) assert r.status_code == 200, f"Admin login failed: {r.status_code} {r.text}" return s @pytest.fixture(scope="session") def staff_session(): s = requests.Session() r = s.post(f"{BASE_URL}/api/auth/login", json=STAFF, timeout=20) if r.status_code != 200: pytest.skip(f"staff login failed {r.status_code}") return s @pytest.fixture(scope="session") def company_id(admin_session): r = admin_session.get(f"{BASE_URL}/api/companies", timeout=20) assert r.status_code == 200 cs = r.json() assert len(cs) > 0 return cs[0]["id"] @pytest.fixture(scope="session") def state(admin_session, company_id): """Shared scratch state to share ids between tests.""" return {"company_id": company_id, "created_ids": {}} # ---------- Health/Dashboard ---------- class TestDashboard: def test_dashboard_shape(self, admin_session, company_id): r = admin_session.get(f"{BASE_URL}/api/cash-bank/dashboard", params={"company_id": company_id}) assert r.status_code == 200, r.text d = r.json() for k in ["cash_balance", "bank_balance", "bank_breakdown", "pending_cheques_in", "pending_cheques_out", "loan_outstanding", "loan_count"]: assert k in d assert isinstance(d["bank_breakdown"], list) assert "total" in d["pending_cheques_in"] and "count" in d["pending_cheques_in"] # ---------- Bank Accounts CRUD ---------- class TestBankAccounts: def test_create_bank(self, admin_session, company_id, state): payload = { "bank_name": "TEST_HDFC", "holder_name": "TEST Holder", "account_no": "TEST123456", "ifsc": "HDFC0000001", "account_type": "current", "opening_balance": 50000.0, "upi_id": "test@hdfc", } r = admin_session.post(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}, json=payload) assert r.status_code == 200, r.text b = r.json() assert b["bank_name"] == "TEST_HDFC" assert "id" in b state["created_ids"]["bank_id"] = b["id"] def test_create_bank2(self, admin_session, company_id, state): payload = {"bank_name": "TEST_ICICI", "holder_name": "TEST", "account_no": "TEST999", "opening_balance": 0.0} r = admin_session.post(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}, json=payload) assert r.status_code == 200 state["created_ids"]["bank_id2"] = r.json()["id"] def test_list_banks_shows_current_balance(self, admin_session, company_id, state): r = admin_session.get(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}) assert r.status_code == 200 banks = r.json() b = next((x for x in banks if x["id"] == state["created_ids"]["bank_id"]), None) assert b is not None assert b.get("current_balance") == 50000.0 def test_update_bank(self, admin_session, state): bid = state["created_ids"]["bank_id"] payload = {"bank_name": "TEST_HDFC_UPDATED", "holder_name": "TEST", "account_no": "TEST123456", "opening_balance": 50000.0, "account_type": "current"} r = admin_session.put(f"{BASE_URL}/api/cash-bank/banks/{bid}", json=payload) assert r.status_code == 200 assert r.json()["bank_name"] == "TEST_HDFC_UPDATED" def test_invalid_bank_id_returns_400(self, admin_session): r = admin_session.put(f"{BASE_URL}/api/cash-bank/banks/not-an-oid", json={"bank_name": "x", "holder_name": "x", "account_no": "x"}) assert r.status_code == 400 # ---------- Cash transactions ---------- class TestCash: def test_cash_deposit(self, admin_session, company_id, state): r = admin_session.post(f"{BASE_URL}/api/cash-bank/cash/transactions", params={"company_id": company_id}, json={"type": "deposit", "amount": 15000, "notes": "TEST deposit"}) assert r.status_code == 200 state["created_ids"]["cash_txn"] = r.json()["id"] def test_cash_balance(self, admin_session, company_id): r = admin_session.get(f"{BASE_URL}/api/cash-bank/cash/balance", params={"company_id": company_id}) assert r.status_code == 200 assert r.json()["balance"] >= 15000.0 # may have prior data def test_cash_to_bank_transfer_mirrors(self, admin_session, company_id, state): bank_id = state["created_ids"]["bank_id"] r = admin_session.post(f"{BASE_URL}/api/cash-bank/cash/transactions", params={"company_id": company_id}, json={"type": "transfer_out", "amount": 2000, "counter_bank_id": bank_id, "notes": "TEST cash->bank"}) assert r.status_code == 200 state["created_ids"]["cash_transfer_out"] = r.json()["id"] # bank list — current_balance for bank_id should be 50000 + 2000 = 52000 r2 = admin_session.get(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}) b = next(x for x in r2.json() if x["id"] == bank_id) assert b["current_balance"] == 52000.0 # ---------- Bank transactions / transfers ---------- class TestBankTxn: def test_create_bank_txn_deposit(self, admin_session, company_id, state): r = admin_session.post(f"{BASE_URL}/api/cash-bank/banks/transactions", params={"company_id": company_id}, json={"bank_account_id": state["created_ids"]["bank_id"], "type": "interest", "amount": 100}) assert r.status_code == 200 state["created_ids"]["bank_txn1"] = r.json()["id"] def test_bank_to_bank_transfer_mirrors(self, admin_session, company_id, state): b1 = state["created_ids"]["bank_id"] b2 = state["created_ids"]["bank_id2"] r = admin_session.post(f"{BASE_URL}/api/cash-bank/banks/transactions", params={"company_id": company_id}, json={"bank_account_id": b1, "type": "transfer_out", "amount": 1000, "counter_bank_id": b2}) assert r.status_code == 200 # b2 should show +1000 r2 = admin_session.get(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}) bb2 = next(x for x in r2.json() if x["id"] == b2) assert bb2["current_balance"] == 1000.0 def test_unified_transactions(self, admin_session, company_id): r = admin_session.get(f"{BASE_URL}/api/cash-bank/transactions", params={"company_id": company_id, "source": "all"}) assert r.status_code == 200 items = r.json() assert isinstance(items, list) # each should have source if items: assert all("source" in i for i in items) # ---------- Cheques ---------- class TestCheques: def test_create_cheque(self, admin_session, company_id, state): payload = { "direction": "received", "cheque_no": "TEST001", "bank_name": "SBI", "amount": 25000, "bank_account_id": state["created_ids"]["bank_id"], } r = admin_session.post(f"{BASE_URL}/api/cash-bank/cheques", params={"company_id": company_id}, json=payload) assert r.status_code == 200 state["created_ids"]["cheque_id"] = r.json()["id"] assert r.json()["status"] == "pending" def test_cheque_cleared_creates_bank_deposit(self, admin_session, company_id, state): cid = state["created_ids"]["cheque_id"] bank_id = state["created_ids"]["bank_id"] # current balance before clear r0 = admin_session.get(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}) before = next(x for x in r0.json() if x["id"] == bank_id)["current_balance"] r = admin_session.post(f"{BASE_URL}/api/cash-bank/cheques/{cid}/status", params={"new_status": "cleared"}) assert r.status_code == 200 assert r.json()["status"] == "cleared" r1 = admin_session.get(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}) after = next(x for x in r1.json() if x["id"] == bank_id)["current_balance"] assert round(after - before, 2) == 25000.0 def test_filter_cheques_by_direction(self, admin_session, company_id): r = admin_session.get(f"{BASE_URL}/api/cash-bank/cheques", params={"company_id": company_id, "direction": "received"}) assert r.status_code == 200 for c in r.json(): assert c["direction"] == "received" def test_delete_cheque_cascades(self, admin_session, company_id, state): cid = state["created_ids"]["cheque_id"] bank_id = state["created_ids"]["bank_id"] r0 = admin_session.get(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}) before = next(x for x in r0.json() if x["id"] == bank_id)["current_balance"] r = admin_session.delete(f"{BASE_URL}/api/cash-bank/cheques/{cid}") assert r.status_code == 200 r1 = admin_session.get(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}) after = next(x for x in r1.json() if x["id"] == bank_id)["current_balance"] assert round(before - after, 2) == 25000.0 # ---------- Loans + Payments ---------- class TestLoans: def test_create_loan(self, admin_session, company_id, state): r = admin_session.post(f"{BASE_URL}/api/cash-bank/loans", params={"company_id": company_id}, json={"provider_name": "TEST_BANK_LOAN", "principal": 500000, "emi_amount": 10500, "interest_rate": 9.5, "tenure_months": 60}) assert r.status_code == 200 state["created_ids"]["loan_id"] = r.json()["id"] def test_loan_payment_creates_bank_outflow(self, admin_session, company_id, state): lid = state["created_ids"]["loan_id"] bid = state["created_ids"]["bank_id"] r0 = admin_session.get(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}) before = next(x for x in r0.json() if x["id"] == bid)["current_balance"] r = admin_session.post(f"{BASE_URL}/api/cash-bank/loans/{lid}/payments", params={"company_id": company_id}, json={"loan_id": lid, "amount": 10500, "principal_paid": 8000, "interest_paid": 2500, "paid_from": "bank", "bank_account_id": bid}) assert r.status_code == 200 state["created_ids"]["loan_payment"] = r.json()["id"] r1 = admin_session.get(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}) after = next(x for x in r1.json() if x["id"] == bid)["current_balance"] assert round(before - after, 2) == 10500.0 def test_list_loans_computed_fields(self, admin_session, company_id, state): lid = state["created_ids"]["loan_id"] r = admin_session.get(f"{BASE_URL}/api/cash-bank/loans", params={"company_id": company_id}) assert r.status_code == 200 ln = next(x for x in r.json() if x["id"] == lid) assert ln["principal_paid_total"] == 8000 assert ln["remaining_balance"] == 492000 assert ln["payment_count"] == 1 def test_delete_loan_blocked_with_payments(self, admin_session, state): lid = state["created_ids"]["loan_id"] r = admin_session.delete(f"{BASE_URL}/api/cash-bank/loans/{lid}") assert r.status_code == 400 def test_delete_loan_payment_cascades(self, admin_session, company_id, state): lid = state["created_ids"]["loan_id"] pid = state["created_ids"]["loan_payment"] r = admin_session.delete(f"{BASE_URL}/api/cash-bank/loans/{lid}/payments/{pid}") assert r.status_code == 200 # ---------- Ledger ---------- class TestLedger: def test_bank_ledger(self, admin_session, company_id, state): bid = state["created_ids"]["bank_id"] r = admin_session.get(f"{BASE_URL}/api/cash-bank/ledger", params={"company_id": company_id, "source": bid}) assert r.status_code == 200 d = r.json() assert d["opening_balance"] == 50000.0 assert "rows" in d and "closing_balance" in d def test_cash_ledger(self, admin_session, company_id): r = admin_session.get(f"{BASE_URL}/api/cash-bank/ledger", params={"company_id": company_id, "source": "cash"}) assert r.status_code == 200 d = r.json() assert "rows" in d and "closing_balance" in d # ---------- Reports ---------- class TestReports: def test_cashflow(self, admin_session, company_id): r = admin_session.get(f"{BASE_URL}/api/cash-bank/reports/cashflow", params={"company_id": company_id}) assert r.status_code == 200 for k in ["inflow", "outflow", "net", "by_category"]: assert k in r.json() def test_daily_collection(self, admin_session, company_id): r = admin_session.get(f"{BASE_URL}/api/cash-bank/reports/daily-collection", params={"company_id": company_id}) assert r.status_code == 200 for k in ["rows", "total_cash", "total_bank"]: assert k in r.json() # ---------- Role enforcement ---------- class TestRoles: def test_staff_can_read_dashboard(self, staff_session, company_id): r = staff_session.get(f"{BASE_URL}/api/cash-bank/dashboard", params={"company_id": company_id}) assert r.status_code == 200 def test_staff_cannot_create_bank(self, staff_session, company_id): r = staff_session.post(f"{BASE_URL}/api/cash-bank/banks", params={"company_id": company_id}, json={"bank_name": "STAFF_X", "holder_name": "X", "account_no": "X"}) assert r.status_code == 403 def test_staff_cannot_create_cheque(self, staff_session, company_id): r = staff_session.post(f"{BASE_URL}/api/cash-bank/cheques", params={"company_id": company_id}, json={"direction": "received", "cheque_no": "STAFFX", "bank_name": "x", "amount": 100}) assert r.status_code == 403 def test_staff_cannot_create_loan(self, staff_session, company_id): r = staff_session.post(f"{BASE_URL}/api/cash-bank/loans", params={"company_id": company_id}, json={"provider_name": "X", "principal": 1}) assert r.status_code == 403 # ---------- Cleanup ---------- class TestZCleanup: def test_cleanup(self, admin_session, company_id, state): # delete loan (should now succeed after payment removed) lid = state["created_ids"].get("loan_id") if lid: admin_session.delete(f"{BASE_URL}/api/cash-bank/loans/{lid}") # cash txns for k in ("cash_txn", "cash_transfer_out"): tid = state["created_ids"].get(k) if tid: admin_session.delete(f"{BASE_URL}/api/cash-bank/cash/transactions/{tid}") # delete bank txns by fetching list (linked ones may be gone) for k in ("bank_id", "bank_id2"): bid = state["created_ids"].get(k) if not bid: continue r = admin_session.get(f"{BASE_URL}/api/cash-bank/banks/{bid}/transactions") if r.status_code == 200: for d in r.json(): admin_session.delete(f"{BASE_URL}/api/cash-bank/banks/transactions/{d['id']}") admin_session.delete(f"{BASE_URL}/api/cash-bank/banks/{bid}") assert True