from datetime import date from decimal import Decimal import pytest from app.models.finance import JournalEntry, Account, Expense from app.models.accounting import ( CashbookMovement, Loan, LoanSchedule, BankReconciliation, ) def _codes(db): return {a.id: a.code for a in db.query(Account).all()} def _entry_lines(db, entry): codes = _codes(db) return {(codes[ln.account_id], "debit" if Decimal(str(ln.debit)) > 0 else "credit", Decimal(str(ln.debit)) if Decimal(str(ln.debit)) > 0 else Decimal(str(ln.credit))) for ln in entry.lines} class TestExpensePosting: def test_recorded_expense_is_unposted_until_post(self, client, auth_headers, db): r = client.post("/api/finance/expenses", json={ "description": "Fertilizer", "amount": 5000, "category": "fertilizer", "date": str(date.today()), "payment_method": "cash", }, headers=auth_headers) assert r.status_code == 201, r.text exp = r.json() posting = client.get(f"/api/finance/expenses/{exp['id']}/posting", headers=auth_headers).json() assert posting["posted"] is False assert posting["approval_status"] == "approved" r = client.post(f"/api/finance/expenses/{exp['id']}/post", headers=auth_headers) assert r.status_code == 200, r.text posting = client.get(f"/api/finance/expenses/{exp['id']}/posting", headers=auth_headers).json() assert posting["posted"] is True entry = db.query(JournalEntry).filter( JournalEntry.source == "expense", JournalEntry.reference_id == exp["id"], JournalEntry.is_reversal == False, # noqa: E712 ).first() assert entry is not None lines = _entry_lines(db, entry) assert ("6100", "debit", Decimal("5000.00")) in lines # Fertilizer & Nutrients assert ("1010", "credit", Decimal("5000.00")) in lines # cash mv = db.query(CashbookMovement).filter( CashbookMovement.reference_type == "expense", CashbookMovement.reference_id == exp["id"], ).first() assert mv is not None assert mv.type == "payment" def test_pending_expense_requires_approval_then_posts(self, client, auth_headers, db): r = client.post("/api/finance/expenses", json={ "description": "Rent", "amount": 20000, "category": "rent", "date": str(date.today()), "payment_method": "bank", "approval_status": "pending", "recurring": True, }, headers=auth_headers) assert r.status_code == 201, r.text exp = r.json() assert exp["approval_status"] == "pending" r = client.post(f"/api/finance/expenses/{exp['id']}/post", headers=auth_headers) assert r.status_code == 400 r = client.post(f"/api/finance/expenses/{exp['id']}/approve", headers=auth_headers) assert r.status_code == 200, r.text posting = client.get(f"/api/finance/expenses/{exp['id']}/posting", headers=auth_headers).json() assert posting["posted"] is True assert posting["approval_status"] == "approved" def test_expense_reverse(self, client, auth_headers, db): r = client.post("/api/finance/expenses", json={ "description": "Utility", "amount": 1000, "category": "electricity", "date": str(date.today()), }, headers=auth_headers) exp = r.json() client.post(f"/api/finance/expenses/{exp['id']}/post", headers=auth_headers) r = client.post(f"/api/finance/expenses/{exp['id']}/reverse", params={"reason": "wrong entry"}, headers=auth_headers) assert r.status_code == 200, r.text entry = db.query(JournalEntry).filter( JournalEntry.source == "expense", JournalEntry.reference_id == exp["id"], JournalEntry.is_reversal == False, # noqa: E712 ).first() assert entry.status == "reversed" posting = client.get(f"/api/finance/expenses/{exp['id']}/posting", headers=auth_headers).json() assert posting["posted"] is False def test_recurring_generation_creates_pending_instance_once(self, client, auth_headers, db): client.post("/api/finance/expenses", json={ "description": "Office", "amount": 3000, "category": "overhead", "date": "2026-01-31", "recurring": True, "recurring_frequency": "monthly", }, headers=auth_headers) r = client.post("/api/finance/expenses/recurring/generate", headers=auth_headers) assert r.status_code == 200, r.text body = r.json() assert body["generated"] >= 1 child = db.query(Expense).filter(Expense.reference.like("RX-%")).first() assert child is not None assert child.approval_status == "pending" assert child.recurring is False r = client.post("/api/finance/expenses/recurring/generate", headers=auth_headers) assert r.status_code == 200 assert r.json()["generated"] == 0 class TestIncomePosting: def test_income_posts_to_revenue_and_reverses(self, client, auth_headers, db): r = client.post("/api/finance/income", json={ "description": "Garden services", "amount": 8000, "category": "services", "date": str(date.today()), "payment_method": "cash", }, headers=auth_headers) assert r.status_code == 201, r.text inc = r.json() posting = client.get(f"/api/finance/income/{inc['id']}/posting", headers=auth_headers).json() assert posting["posted"] is False r = client.post(f"/api/finance/income/{inc['id']}/post", headers=auth_headers) assert r.status_code == 200, r.text entry = db.query(JournalEntry).filter( JournalEntry.source == "income", JournalEntry.reference_id == inc["id"], JournalEntry.is_reversal == False, # noqa: E712 ).first() assert entry is not None lines = _entry_lines(db, entry) assert ("1010", "debit", Decimal("8000.00")) in lines assert ("4060", "credit", Decimal("8000.00")) in lines # Other Income / Services r = client.post(f"/api/finance/income/{inc['id']}/reverse", headers=auth_headers) assert r.status_code == 200 db.expire_all() # drop stale identity-map snapshot entry = db.query(JournalEntry).filter( JournalEntry.source == "income", JournalEntry.reference_id == inc["id"], JournalEntry.is_reversal == False, # noqa: E712 ).first() assert entry.status == "reversed" class TestCashbook: def test_deposit_transfer_and_book_balance(self, client, auth_headers, db): r = client.post("/api/accounting/cashbook/deposit", json={ "amount": 5000, "date": str(date.today()), "notes": "Bank deposit", }, headers=auth_headers) assert r.status_code == 201, r.text book = client.get("/api/accounting/cashbook", headers=auth_headers).json() accts = {a["code"]: a for a in book["accounts"]} assert accts["1010"]["closing_balance"] == pytest.approx(-5000.0) assert accts["1020"]["closing_balance"] == pytest.approx(5000.0) mvs = client.get("/api/accounting/cashbook?account_code=1020", headers=auth_headers).json()["movements"] assert any(m["type"] == "deposit" and m["amount"] == 5000.0 for m in mvs) def test_generic_transfer_between_accounts(self, client, auth_headers, db): cash = client.get("/api/accounting/cashbook/accounts", headers=auth_headers).json() cash_acct = next(a for a in cash if a["code"] == "1010") bank_acct = next(a for a in cash if a["code"] == "1020") r = client.post("/api/accounting/cashbook/transfer", json={ "from_account_id": bank_acct["id"], "to_account_id": cash_acct["id"], "amount": 1500, "date": str(date.today()), "notes": "ATM withdrawal", }, headers=auth_headers) assert r.status_code == 201, r.text book = client.get("/api/accounting/cashbook/accounts", headers=auth_headers).json() assert all(a["code"] not in ("1010", "1020") or True for a in book) def test_bank_reconciliation(self, client, auth_headers, db): client.post("/api/accounting/cashbook/deposit", json={ "amount": 1000, "date": str(date.today()), }, headers=auth_headers) bank = client.get("/api/accounting/cashbook/accounts", headers=auth_headers).json() bank_acct = next(a for a in bank if a["code"] == "1020") r = client.post("/api/accounting/bank-reconciliations", json={ "account_id": bank_acct["id"], "statement_balance": 1000, "as_of": str(date.today()), "lines": [], }, headers=auth_headers) assert r.status_code == 201, r.text rec = r.json() assert rec["difference"] == 0.0 r = client.post(f"/api/accounting/bank-reconciliations/{rec['id']}/resolve", json={"status": "matched"}, headers=auth_headers) assert r.status_code == 200, r.text assert r.json()["status"] == "matched" # mismatched statement balance cannot be resolved as adjusted r = client.post("/api/accounting/bank-reconciliations", json={ "account_id": bank_acct["id"], "statement_balance": 500, "as_of": str(date.today()), }, headers=auth_headers) rec = r.json() r = client.post(f"/api/accounting/bank-reconciliations/{rec['id']}/resolve", json={"status": "adjusted"}, headers=auth_headers) assert r.status_code == 400 class TestLoans: def _mk_loan(self, client, auth_headers, principal=12000, rate=12.0, tenure=12): r = client.post("/api/accounting/loans", json={ "lender_name": "Agro Bank", "lender_type": "bank", "principal": principal, "interest_rate_nominal": rate, "interest_rate_period": "year", "tenure_months": tenure, "start_date": str(date.today()), "repayment_frequency": "monthly", }, headers=auth_headers) assert r.status_code == 201, r.text return r.json() def test_create_loan_builds_schedule_and_verifies(self, client, auth_headers, db): loan = self._mk_loan(client, auth_headers) assert loan["loan_number"].startswith("LN-") assert loan["status"] == "draft" assert len(loan["schedules"]) == 12 total = sum(s["total_due"] for s in loan["schedules"]) assert total == pytest.approx(loan["principal"] + loan["total_interest"]) verify = client.get(f"/api/accounting/loans/{loan['id']}/verify", headers=auth_headers).json() assert verify["valid"] is True def test_activate_loan_posts_liability(self, client, auth_headers, db): loan = self._mk_loan(client, auth_headers, principal=10000) r = client.post(f"/api/accounting/loans/{loan['id']}/activate", json={"method": "bank"}, headers=auth_headers) assert r.status_code == 200, r.text loan = r.json()["loan"] assert loan["status"] == "active" entry = db.query(JournalEntry).filter( JournalEntry.source == "loan", JournalEntry.reference_type == "loan", JournalEntry.reference_id == loan["id"], JournalEntry.is_reversal == False, # noqa: E712 ).first() assert entry is not None lines = _entry_lines(db, entry) assert ("2030", "credit", Decimal("10000.00")) in lines assert any(code == "1020" and side == "debit" and amt == Decimal("10000.00") for code, side, amt in lines) def test_pay_installments_then_close(self, client, auth_headers, db): loan = self._mk_loan(client, auth_headers, principal=12000, rate=12.0, tenure=6) client.post(f"/api/accounting/loans/{loan['id']}/activate", json={"method": "cash"}, headers=auth_headers) first = loan["schedules"][0] r = client.post(f"/api/accounting/loans/{loan['id']}/pay", json={ "installment_ids": [first["id"]], "pay_date": str(date.today()), "method": "cash", }, headers=auth_headers) assert r.status_code == 200, r.text body = r.json() assert body["paid_amount"] == pytest.approx(first["total_due"]) assert body["remaining_installments"] == 5 schedule = db.query(LoanSchedule).filter(LoanSchedule.id == first["id"]).first() assert schedule.status == "paid" entry = db.query(JournalEntry).filter( JournalEntry.source == "loan", JournalEntry.reference_type == "loan_payment", JournalEntry.reference_id == first["id"], ).first() assert entry is not None lines = _entry_lines(db, entry) assert any(code == "2030" and side == "debit" for code, side, _ in lines) assert any(code == "6150" and side == "debit" for code, side, _ in lines) # cannot pay the same installment twice r = client.post(f"/api/accounting/loans/{loan['id']}/pay", json={ "installment_ids": [first["id"]], "pay_date": str(date.today()), }, headers=auth_headers) assert r.status_code == 400 # pay the rest → loan closes rest = [s["id"] for s in loan["schedules"][1:]] r = client.post(f"/api/accounting/loans/{loan['id']}/pay", json={ "installment_ids": rest, "pay_date": str(date.today()), "method": "bank", }, headers=auth_headers) assert r.status_code == 200, r.text loan_after = r.json()["loan"] assert loan_after["status"] == "closed" assert loan_after["remaining_principal"] == 0.0 def test_delete_only_draft_loans(self, client, auth_headers): loan = self._mk_loan(client, auth_headers) r = client.delete(f"/api/accounting/loans/{loan['id']}", headers=auth_headers) assert r.status_code == 200 client.post("/api/accounting/loans", json={ "lender_name": "B", "principal": 5000, "tenure_months": 12, "start_date": str(date.today()), }, headers=auth_headers) active_id = client.get("/api/accounting/loans", headers=auth_headers).json()[0]["id"] client.post(f"/api/accounting/loans/{active_id}/activate", headers=auth_headers) r = client.delete(f"/api/accounting/loans/{active_id}", headers=auth_headers) assert r.status_code == 400 class TestPhase4Reports: def test_reports_reflect_postings(self, client, auth_headers, db): client.post("/api/finance/expenses", json={ "description": "Feed", "amount": 2000, "category": "fertilizer", "date": str(date.today()), "payment_method": "cash", }, headers=auth_headers) exp = client.get("/api/finance/expenses", headers=auth_headers).json()[-1] client.post(f"/api/finance/expenses/{exp['id']}/post", headers=auth_headers) client.post("/api/finance/income", json={ "description": "Consulting", "amount": 4000, "category": "services", "date": str(date.today()), "payment_method": "cash", }, headers=auth_headers) inc = client.get("/api/finance/income", headers=auth_headers).json()[-1] client.post(f"/api/finance/income/{inc['id']}/post", headers=auth_headers) tb = client.get("/api/accounting/trial-balance", headers=auth_headers).json() assert tb["balanced"] is True assert any(ln["code"] == "6100" and ln["debit"] == 2000.0 for ln in tb["lines"]) cf = client.get("/api/accounting/cash-flow", headers=auth_headers).json() assert cf["net_cash_change"] == pytest.approx(cf["operating"] + cf["investing"] + cf["financing"]) pnl = client.get("/api/accounting/profit-loss", headers=auth_headers).json() assert pnl["total_expenses"] == pytest.approx(2000.0) assert pnl["total_revenue"] == pytest.approx(4000.0) branches = client.get("/api/accounting/pnl-by-branch", headers=auth_headers).json() assert len(branches["branches"]) >= 1 metrics = client.get("/api/accounting/dashboard-metrics", headers=auth_headers).json() for key in ("cash_and_bank", "receivables", "payables", "net_profit", "sales_count"): assert key in metrics assert metrics["net_profit"] == pytest.approx(2000.0) def test_csv_export(self, client, auth_headers): r = client.get(f"/api/accounting/reports/trial-balance/export.csv?account_code=1010", headers=auth_headers) assert r.status_code == 200 assert "text/csv" in r.headers["content-type"] assert r.content.decode("utf-8-sig").startswith("Code") def test_xlsx_export(self, client, auth_headers): r = client.get("/api/accounting/reports/cash-flow/export.xlsx", headers=auth_headers) assert r.status_code == 200 assert "spreadsheetml" in r.headers["content-type"] def test_pdf_export(self, client, auth_headers): r = client.get("/api/accounting/reports/trial-balance/export.pdf", headers=auth_headers) assert r.status_code == 200 assert r.headers["content-type"] == "application/pdf" assert r.content.startswith(b"%PDF") assert "filename=trial-balance.pdf" in r.headers["content-disposition"]