""" Property-based tests for Opening Balance Continuity (Property 7). Property 7: Opening Balance Continuity For any account and any date `d`, the balance at `d` SHALL equal the balance at `d − 1` plus the net GL activity that occurred on date `d`. Formally: calculate_account_balance(account, d) == calculate_account_balance(account, d − 1) + net_activity_on_date(account, d) where net_activity_on_date is: • For asset/expense accounts (debit-normal): sum(debits) − sum(credits) • For liability/equity/income accounts (credit-normal): sum(credits) − sum(debits) This guarantees there are no unexplained gaps between consecutive periods and that the opening balance of any period equals the closing balance of the immediately preceding period. **Validates: Requirements 8.1, 8.2** Uses Hypothesis to generate arbitrary accounts and date ranges, posts journal entries on a specific date `d`, and asserts the continuity invariant holds. """ import uuid from datetime import date, timedelta from decimal import Decimal from hypothesis import given, settings, strategies as st, assume from hypothesis.extra.django import TestCase from django.contrib.auth import get_user_model from django.core.cache import cache as django_cache from accounting.models import ( Account, FiscalPeriod, GeneralLedger, JournalEntry, JournalEntryLine, ) from accounting.services.accounting_service import AccountingService from users.models import Branch User = get_user_model() # --------------------------------------------------------------------------- # Hypothesis strategies # --------------------------------------------------------------------------- def _amount_strategy( min_val: str = "1.00", max_val: str = "9999.00" ) -> st.SearchStrategy: """Strategy for positive Decimals with exactly 2 decimal places.""" return st.decimals( min_value=Decimal(min_val), max_value=Decimal(max_val), places=2, allow_nan=False, allow_infinity=False, ) @st.composite def transaction_date_strategy(draw): """ Generate a transaction date `d` within 2024, constrained so that `d − 1` also falls within the open fiscal period (2020–2030). Min: 2020-01-02 (so d-1 >= 2020-01-01 is always within the period). """ start_day_offset = draw(st.integers(min_value=1, max_value=365)) base = date(2024, 1, 1) d = base + timedelta(days=start_day_offset) assume(d.year == 2024) return d @st.composite def entry_amounts_strategy(draw): """Generate a list of 1–4 amounts to post on date `d`.""" n = draw(st.integers(min_value=1, max_value=4)) return [draw(_amount_strategy()) for _ in range(n)] # --------------------------------------------------------------------------- # Helper: net activity for a given account type # --------------------------------------------------------------------------- def _compute_net_activity( account: Account, tx_date: date, branch=None ) -> Decimal: """ Return the net GL activity for `account` on exactly `tx_date`. For asset/expense (debit-normal): net = sum(debits) - sum(credits) For liability/equity/income (credit-normal): net = sum(credits) - sum(debits) """ qs = GeneralLedger.objects.filter( account=account, transaction_date=tx_date, ) if branch: qs = qs.filter(branch=branch) totals = qs.values("debit_amount", "credit_amount") total_debits = sum(r["debit_amount"] for r in totals) total_credits = sum(r["credit_amount"] for r in totals) if account.account_type in ("asset", "expense"): return total_debits - total_credits else: # liability, equity, income return total_credits - total_debits # --------------------------------------------------------------------------- # Test case # --------------------------------------------------------------------------- class OpeningBalanceContinuityPropertyTest(TestCase): """ Property 7: Opening Balance Continuity **Validates: Requirements 8.1, 8.2** For any account and date `d`: calculate_account_balance(account, d) == calculate_account_balance(account, d − 1) + net_activity_on_date(account, d) This ensures the opening balance of any period equals the closing balance of the immediately preceding period — with no unexplained gaps. """ def setUp(self): """Create the minimal shared fixture for each Hypothesis example.""" suffix = uuid.uuid4().hex[:8] phone_suffix = uuid.uuid4().int % 10 ** 9 phone_number = f"+254{phone_suffix:09d}" self.user = User.objects.create_user( username=f"ob_cont_user_{suffix}", email=f"ob_cont_{suffix}@example.com", password="testpass123", phone_number=phone_number, ) self.branch = Branch.objects.create( name=f"OBContBranch_{suffix}", code=f"OB{suffix[:5].upper()}", ) # Open fiscal period wide enough for any generated date to fall inside. self.fiscal_period, _ = FiscalPeriod.objects.get_or_create( name="OB Continuity Property Test Period", defaults=dict( period_type="monthly", start_date=date(2020, 1, 1), end_date=date(2030, 12, 31), status="open", ), ) # Four accounts — one per accounting normal (debit/credit) and type. self.asset_account = Account.objects.create( code=f"10{suffix[:5]}", name="OBCont Cash", account_type="asset", subtype="current_asset", description="Opening balance continuity test — asset", is_active=True, created_by=self.user, ) self.liability_account = Account.objects.create( code=f"20{suffix[:5]}", name="OBCont Payable", account_type="liability", subtype="current_liability", description="Opening balance continuity test — liability", is_active=True, created_by=self.user, ) self.income_account = Account.objects.create( code=f"40{suffix[:5]}", name="OBCont Interest Income", account_type="income", subtype="interest_income", description="Opening balance continuity test — income", is_active=True, created_by=self.user, ) self.expense_account = Account.objects.create( code=f"50{suffix[:5]}", name="OBCont Staff Cost", account_type="expense", subtype="staff_costs", description="Opening balance continuity test — expense", is_active=True, created_by=self.user, ) self.accounting_service = AccountingService() # ----------------------------------------------------------------------- # Helpers # ----------------------------------------------------------------------- def _ref(self, prefix: str = "JE") -> str: """Generate a unique journal-entry reference.""" return f"{prefix}-{uuid.uuid4().hex[:12].upper()}" def _post_asset_income_entry(self, amount: Decimal, tx_date: date) -> None: """ Post: Dr Asset (cash) amount Cr Income amount This increases both the asset and income accounts. """ je = JournalEntry.objects.create( reference_number=self._ref("AI"), transaction_date=tx_date, description="OBCont income receipt", branch=self.branch, created_by=self.user, status="draft", ) JournalEntryLine.objects.create( journal_entry=je, account=self.asset_account, description="Cash received", debit_amount=amount, credit_amount=Decimal("0.00"), line_number=1, ) JournalEntryLine.objects.create( journal_entry=je, account=self.income_account, description="Income earned", debit_amount=Decimal("0.00"), credit_amount=amount, line_number=2, ) self.accounting_service.post_journal_entry(je, self.user) def _post_expense_liability_entry(self, amount: Decimal, tx_date: date) -> None: """ Post: Dr Expense amount Cr Liability amount This increases both the expense and liability accounts. """ je = JournalEntry.objects.create( reference_number=self._ref("EL"), transaction_date=tx_date, description="OBCont expense accrual", branch=self.branch, created_by=self.user, status="draft", ) JournalEntryLine.objects.create( journal_entry=je, account=self.expense_account, description="Expense incurred", debit_amount=amount, credit_amount=Decimal("0.00"), line_number=1, ) JournalEntryLine.objects.create( journal_entry=je, account=self.liability_account, description="Accrued liability", debit_amount=Decimal("0.00"), credit_amount=amount, line_number=2, ) self.accounting_service.post_journal_entry(je, self.user) def _assert_continuity( self, account: Account, d: date, branch=None ) -> None: """ Core continuity assertion for a single account and date: balance(d) == balance(d − 1) + net_activity(d) Clears caches before each measurement so stale values do not interfere with the GL-based calculation. """ django_cache.clear() balance_d = self.accounting_service.calculate_account_balance( account, d, branch=branch ) django_cache.clear() balance_d_minus_1 = self.accounting_service.calculate_account_balance( account, d - timedelta(days=1), branch=branch ) django_cache.clear() net_activity = _compute_net_activity(account, d, branch=branch) expected = balance_d_minus_1 + net_activity assert abs(balance_d - expected) < Decimal("0.01"), ( f"Opening balance continuity violated for account " f"'{account.name}' ({account.account_type}) on date {d}.\n" f" balance({d}): {balance_d}\n" f" balance({d - timedelta(days=1)}): {balance_d_minus_1}\n" f" net_activity({d}): {net_activity}\n" f" Expected (prev + act): {expected}\n" f" Difference: {abs(balance_d - expected)}" ) # ----------------------------------------------------------------------- # Property tests # ----------------------------------------------------------------------- @settings(max_examples=50, deadline=None) @given(amounts=entry_amounts_strategy(), d=transaction_date_strategy()) def test_asset_balance_continuity(self, amounts, d): """ Property 7 — Asset accounts: For any asset account and date `d`, balance(d) == balance(d-1) + net_activity(d). **Validates: Requirements 8.1, 8.2** """ for amount in amounts: self._post_asset_income_entry(amount, d) self._assert_continuity(self.asset_account, d) @settings(max_examples=50, deadline=None) @given(amounts=entry_amounts_strategy(), d=transaction_date_strategy()) def test_income_balance_continuity(self, amounts, d): """ Property 7 — Income accounts: For any income account and date `d`, balance(d) == balance(d-1) + net_activity(d). **Validates: Requirements 8.1, 8.2** """ for amount in amounts: self._post_asset_income_entry(amount, d) self._assert_continuity(self.income_account, d) @settings(max_examples=50, deadline=None) @given(amounts=entry_amounts_strategy(), d=transaction_date_strategy()) def test_expense_balance_continuity(self, amounts, d): """ Property 7 — Expense accounts: For any expense account and date `d`, balance(d) == balance(d-1) + net_activity(d). **Validates: Requirements 8.1, 8.2** """ for amount in amounts: self._post_expense_liability_entry(amount, d) self._assert_continuity(self.expense_account, d) @settings(max_examples=50, deadline=None) @given(amounts=entry_amounts_strategy(), d=transaction_date_strategy()) def test_liability_balance_continuity(self, amounts, d): """ Property 7 — Liability accounts: For any liability account and date `d`, balance(d) == balance(d-1) + net_activity(d). **Validates: Requirements 8.1, 8.2** """ for amount in amounts: self._post_expense_liability_entry(amount, d) self._assert_continuity(self.liability_account, d) @settings(max_examples=40, deadline=None) @given(d=transaction_date_strategy()) def test_continuity_holds_with_no_activity_on_date_d(self, d): """ Property 7 — Empty day: When no GL entries exist on date `d`, balance(d) == balance(d-1) (net activity is zero, so the balance is unchanged). **Validates: Requirements 8.1, 8.2** """ # No postings on date d — verify the invariant holds trivially. for account in [ self.asset_account, self.income_account, self.expense_account, self.liability_account, ]: self._assert_continuity(account, d) @settings(max_examples=40, deadline=None) @given(amounts=entry_amounts_strategy(), d=transaction_date_strategy()) def test_continuity_with_prior_period_entries(self, amounts, d): """ Property 7 — Prior period isolation: Entries posted BEFORE date `d` affect balance(d-1) and balance(d) equally, so the continuity invariant still holds for date `d`. **Validates: Requirements 8.1, 8.2** """ prior_date = d - timedelta(days=7) assume(prior_date >= date(2020, 1, 2)) # Post entries one week before date d (pre-period activity) for amount in amounts: self._post_asset_income_entry(amount, prior_date) # Now post entries ON date d for amount in amounts: self._post_asset_income_entry(amount, d) # Continuity must hold for date d regardless of prior entries self._assert_continuity(self.asset_account, d) self._assert_continuity(self.income_account, d) @settings(max_examples=40, deadline=None) @given( amounts_before=entry_amounts_strategy(), amounts_on_d=entry_amounts_strategy(), d=transaction_date_strategy(), ) def test_closing_d_minus_1_equals_opening_of_d( self, amounts_before, amounts_on_d, d ): """ Property 7 — Consecutive period closing = next opening: The closing balance of date `d − 1` SHALL equal the opening balance for date `d` (which is balance(d) before any entries on `d` are applied). Verified by: 1. Posting entries on d-1 (closes d-1) 2. Capturing balance(d-1) [closing of d-1] 3. Verifying balance(d) before posting on d == balance(d-1) 4. Posting entries on d 5. Verifying continuity holds for d **Validates: Requirements 8.1, 8.2** """ prev_date = d - timedelta(days=1) assume(prev_date >= date(2020, 1, 1)) # Step 1: post entries on d-1 for amount in amounts_before: self._post_asset_income_entry(amount, prev_date) # Step 2: capture closing balance of d-1 django_cache.clear() closing_d_minus_1 = self.accounting_service.calculate_account_balance( self.asset_account, prev_date ) # Step 3: before posting anything on d, balance(d) should equal closing(d-1) # because the GL only has entries up to d-1. django_cache.clear() opening_of_d = self.accounting_service.calculate_account_balance( self.asset_account, d ) assert abs(opening_of_d - closing_d_minus_1) < Decimal("0.01"), ( f"Opening balance of date {d} ({opening_of_d}) does not equal " f"closing balance of date {prev_date} ({closing_d_minus_1}).\n" f" Difference: {abs(opening_of_d - closing_d_minus_1)}" ) # Step 4: post entries on d and verify full continuity for amount in amounts_on_d: self._post_asset_income_entry(amount, d) self._assert_continuity(self.asset_account, d) @settings(max_examples=30, deadline=None) @given(amount=_amount_strategy()) def test_single_entry_continuity_asset(self, amount): """ Property 7 — Single entry (asset): After posting one entry of `amount` on date `d`, the asset balance on `d` must equal the asset balance on `d − 1` plus `amount`. **Validates: Requirements 8.1, 8.2** """ d = date(2024, 6, 15) django_cache.clear() balance_before = self.accounting_service.calculate_account_balance( self.asset_account, d - timedelta(days=1) ) self._post_asset_income_entry(amount, d) django_cache.clear() balance_after = self.accounting_service.calculate_account_balance( self.asset_account, d ) expected = balance_before + amount # asset is debit-normal; debit increases it assert abs(balance_after - expected) < Decimal("0.01"), ( f"Asset continuity violated for single entry amount={amount}.\n" f" balance({d - timedelta(days=1)}): {balance_before}\n" f" balance({d}): {balance_after}\n" f" Expected: {expected}\n" f" Difference: {abs(balance_after - expected)}" ) @settings(max_examples=30, deadline=None) @given(amount=_amount_strategy()) def test_single_entry_continuity_income(self, amount): """ Property 7 — Single entry (income): After posting one credit entry of `amount` on date `d`, the income balance on `d` must equal the income balance on `d − 1` plus `amount`. **Validates: Requirements 8.1, 8.2** """ d = date(2024, 7, 20) django_cache.clear() balance_before = self.accounting_service.calculate_account_balance( self.income_account, d - timedelta(days=1) ) self._post_asset_income_entry(amount, d) django_cache.clear() balance_after = self.accounting_service.calculate_account_balance( self.income_account, d ) expected = balance_before + amount # income is credit-normal; credit increases it assert abs(balance_after - expected) < Decimal("0.01"), ( f"Income continuity violated for single entry amount={amount}.\n" f" balance({d - timedelta(days=1)}): {balance_before}\n" f" balance({d}): {balance_after}\n" f" Expected: {expected}\n" f" Difference: {abs(balance_after - expected)}" )