from datetime import date from decimal import Decimal import pytest from app.models.finance import Account, JournalEntry from app.models.accounting import AccountBalance, AccountingPeriod from app.models.audit import AuditLog from app.models.user import Role, User from app.services.accounting import PostingService, AccountingError, seed_chart_of_accounts from app.services.auth import hash_password, create_access_token def _acct(db, code): return db.query(Account).filter(Account.code == code).first() def _id(db, code): return _acct(db, code).id _counter = [0] def _post(db, user_id, debit_code, credit_code, amount, entry_date=None): _counter[0] += 1 svc = PostingService(db) entry = svc.post_event( source="journal", reference_type="test", reference_id=f"T-{_counter[0]}", entry_date=entry_date or date(2026, 9, 10), branch_id=None, lines=[ {"account_id": _id(db, debit_code), "debit": amount, "credit": 0}, {"account_id": _id(db, credit_code), "debit": 0, "credit": amount}, ], description=f"{debit_code} <-> {credit_code} {amount}", created_by=user_id, ) db.commit() return entry class TestPosting: def test_1_balanced_entry_posts(self, seeded_db, db): entry = _post(db, seeded_db["admin"].id, "1010", "4010", "100.00") db.refresh(entry) assert entry.status == "posted" assert entry.entry_number.startswith("JV-") balance = db.query(AccountBalance).filter( AccountBalance.account_id == _id(db, "1010") ).first() assert Decimal(str(balance.total_debit)) == Decimal("100.00") assert Decimal(str(balance.closing_balance)) == Decimal("100.00") def test_2_unbalanced_entry_rejected(self, seeded_db, db): svc = PostingService(db) with pytest.raises(AccountingError): svc.create_draft( entry_date=date(2026, 9, 10), lines=[ {"account_id": _id(db, "1010"), "debit": 100, "credit": 0}, {"account_id": _id(db, "4010"), "debit": 0, "credit": 90}, ], description="unbalanced", created_by=seeded_db["admin"].id, ) def test_3_duplicate_posting_noop(self, seeded_db, db): svc = PostingService(db) kwargs = dict( source="sale", reference_type="sale", reference_id="S1", entry_date=date(2026, 9, 10), branch_id=None, created_by=seeded_db["admin"].id, ) first = svc.post_event( lines=[ {"account_id": _id(db, "1010"), "debit": 50, "credit": 0}, {"account_id": _id(db, "4010"), "debit": 0, "credit": 50}, ], description="sale 1", **kwargs, ) db.commit() second = svc.post_event( lines=[ {"account_id": _id(db, "1010"), "debit": 50, "credit": 0}, {"account_id": _id(db, "4010"), "debit": 0, "credit": 50}, ], description="sale 1 again", **kwargs, ) assert first.id == second.id assert db.query(JournalEntry).filter(JournalEntry.reference_id == "S1").count() == 1 def test_4_repost_idempotent_via_api(self, seeded_db, db, client, auth_headers): payload = { "entry_date": date(2026, 9, 10), "description": "cash sale", "source": "sale", "reference_type": "sale", "reference_id": "SALE-1", "lines": [ {"account_id": _id(db, "1010"), "debit": 120.0, "credit": 0}, {"account_id": _id(db, "4010"), "debit": 0, "credit": 120.0}, ], } svc = PostingService(db) entry = svc.post_event(**payload, branch_id=None, created_by=seeded_db["admin"].id) db.commit() again = svc.post_event(**payload, branch_id=None, created_by=seeded_db["admin"].id) assert entry.id == again.id assert entry.status == "posted" def test_14_reversal_mirrors_and_locks_original(self, seeded_db, db, client, auth_headers): entry = _post(db, seeded_db["admin"].id, "1010", "4010", "80.00") resp = client.post( f"/api/accounting/journal/{entry.id}/reverse", json={"reason": "test reversal"}, headers=auth_headers, ) assert resp.status_code == 200 reversal = resp.json() assert reversal["is_reversal"] is True assert reversal["reversal_of_id"] == entry.id db.refresh(entry) assert entry.status == "reversed" cl = db.query(AccountBalance).filter( AccountBalance.account_id == _id(db, "1010") ).all() net = sum((Decimal(str(b.total_debit)) - Decimal(str(b.total_credit)) for b in cl), Decimal("0")) assert net == Decimal("0.00") edit = client.put( f"/api/accounting/journal/{entry.id}", json={"description": "hack"}, headers=auth_headers, ) assert edit.status_code == 409 def test_16_closed_period_rejects_posting(self, seeded_db, db, client, auth_headers): _post(db, seeded_db["admin"].id, "1010", "4010", "10.00") period = db.query(AccountingPeriod).first() assert period is not None resp = client.post(f"/api/accounting/periods/{period.id}/close", headers=auth_headers) assert resp.status_code == 200 resp = client.post( "/api/accounting/journal", json={ "entry_date": date(2026, 9, 20).isoformat(), "description": "late entry", "lines": [ {"account_id": _id(db, "1010"), "debit": 10, "credit": 0}, {"account_id": _id(db, "4010"), "debit": 0, "credit": 10}, ], }, headers=auth_headers, ) assert resp.status_code == 201 entry_id = resp.json()["id"] post = client.post(f"/api/accounting/journal/{entry_id}/post", headers=auth_headers) assert post.status_code == 422 def test_17_trial_balance_correct(self, seeded_db, db, client, auth_headers): _post(db, seeded_db["admin"].id, "1010", "4010", "100.00") _post(db, seeded_db["admin"].id, "6010", "1010", "30.00") resp = client.get("/api/accounting/trial-balance", headers=auth_headers) assert resp.status_code == 200 data = resp.json() assert data["balanced"] is True assert data["total_debit"] == pytest.approx(130.0) assert data["total_credit"] == pytest.approx(130.0) cash = next(row for row in data["lines"] if row["code"] == "1010") assert cash["debit"] == pytest.approx(100.0) assert cash["credit"] == pytest.approx(30.0) def test_17b_balance_sheet_and_pnl(self, seeded_db, db, client, auth_headers): _post(db, seeded_db["admin"].id, "1010", "4010", "200.00") _post(db, seeded_db["admin"].id, "6010", "1010", "50.00") pnl = client.get("/api/accounting/profit-loss", headers=auth_headers).json() assert pnl["total_revenue"] == pytest.approx(200.0) assert pnl["total_expenses"] == pytest.approx(50.0) assert pnl["net_profit"] == pytest.approx(150.0) bs = client.get("/api/accounting/balance-sheet", headers=auth_headers).json() assert bs["total_assets"] == pytest.approx(150.0) assert bs["balanced"] is True def test_20_audit_logs_written(self, seeded_db, db, client, auth_headers): entry = _post(db, seeded_db["admin"].id, "1010", "4010", "60.00") client.post(f"/api/accounting/journal/{entry.id}/reverse", json={}, headers=auth_headers) logs = db.query(AuditLog).filter(AuditLog.module == "accounting").all() actions = {log.action for log in logs} assert "post_journal" in actions assert "reverse_journal" in actions class TestOpeningAndPermissions: def test_opening_balances_auto_balances_via_3010(self, seeded_db, db, client, auth_headers): cash = _id(db, "1010") payable = _id(db, "2010") resp = client.post( "/api/accounting/opening-balances", json={ "items": [ {"account_id": cash, "amount": 5000.0}, {"account_id": payable, "amount": 800.0}, ], "entry_date": "2026-09-01", }, headers=auth_headers, ) assert resp.status_code == 201, resp.text data = resp.json() assert data["source"] == "opening" assert data["status"] == "posted" obe = _acct(db, "3010") assert obe is not None tb = client.get("/api/accounting/trial-balance", headers=auth_headers).json() assert tb["balanced"] is True assert tb["total_debit"] == pytest.approx(5000.0) assert tb["total_credit"] == pytest.approx(5000.0) def test_recompute_reconciles_balances(self, seeded_db, db): _post(db, seeded_db["admin"].id, "1010", "4010", "75.00") _post(db, seeded_db["admin"].id, "6010", "1010", "25.00") from app.services.balances import recompute_account_balances recompute_account_balances(db) rows = db.query(AccountBalance).all() assert len(rows) >= 2 cash = next(r for r in rows if r.account_id == _id(db, "1010")) assert Decimal(str(cash.closing_balance)) == Decimal("50.00") def test_unauthorized_post_forbidden(self, seeded_db, db, client): staff = User( username="staff1", email="staff1@test.com", password_hash=hash_password("staff123"), full_name="Staff One", is_active=True, branch_id=seeded_db["branch"].id, ) viewer_role = None for role in db.query(Role).all(): if role.name == "viewer": viewer_role = role if viewer_role is None: viewer_role = Role(name="viewer", description="read only") db.add(viewer_role) db.flush() staff.roles = [viewer_role] db.add(staff) db.commit() headers = {"Authorization": f"Bearer {create_access_token({'sub': staff.id})}"} resp = client.post( "/api/accounting/journal", json={ "entry_date": "2026-09-10", "description": "denied", "lines": [ {"account_id": _id(db, "1010"), "debit": 1, "credit": 0}, {"account_id": _id(db, "4010"), "debit": 0, "credit": 1}, ], }, headers=headers, ) assert resp.status_code == 403 class TestPermissionGating: def _no_perm_headers(self, seeded_db, db): viewer_role = db.query(Role).filter(Role.name == "viewer_no_perms").first() if viewer_role is None: viewer_role = Role(name="viewer_no_perms", description="no permissions") db.add(viewer_role) db.flush() user = User( username="viewer_no_perms", email="viewer_no_perms@test.com", password_hash=hash_password("viewer123"), full_name="No Permissions", is_active=True, branch_id=seeded_db["branch"].id, ) user.roles = [viewer_role] db.add(user) db.commit() return {"Authorization": f"Bearer {create_access_token({'sub': user.id})}"} def test_finance_endpoints_deny_without_permission(self, seeded_db, db, client): headers = self._no_perm_headers(seeded_db, db) denied = [ ("GET", "/api/finance/accounts"), ("POST", "/api/finance/accounts"), ("GET", "/api/finance/expenses"), ("POST", "/api/finance/expenses"), ("GET", "/api/finance/income"), ("POST", "/api/finance/income"), ("GET", "/api/finance/payments"), ("POST", "/api/finance/payments"), ] for method, url in denied: resp = client.request(method, url, headers=headers, json={} if method == "POST" else None) assert resp.status_code == 403, f"{method} {url} -> {resp.status_code}" def test_accounting_endpoints_deny_without_permission(self, seeded_db, db, client): headers = self._no_perm_headers(seeded_db, db) denied = [ ("GET", "/api/accounting/accounts"), ("POST", "/api/accounting/accounts"), ("POST", "/api/accounting/journal"), ("GET", "/api/accounting/trial-balance"), ("GET", "/api/accounting/cashbook"), ("POST", "/api/accounting/cashbook/deposit"), ("GET", "/api/accounting/loans"), ("GET", "/api/accounting/production-costs"), ("POST", "/api/accounting/production-costs"), ("GET", "/api/accounting/batch-profitability"), ("GET", "/api/accounting/reports/trial-balance/export.csv"), ("GET", "/api/accounting/reports/trial-balance/export.xlsx"), ("GET", "/api/accounting/reports/trial-balance/export.pdf"), ] for method, url in denied: resp = client.request(method, url, headers=headers, json={} if method == "POST" else None) assert resp.status_code == 403, f"{method} {url} -> {resp.status_code}"