""" Property-based tests for Cache Invalidation After Posting (Property 9). Property 9: Cache Invalidation After Posting For any account and date `d`, if a new JournalEntry is posted for date `d`, then a financial report generated after the posting SHALL include the new amounts (i.e., the cached value before posting SHALL NOT be returned). The test follows this protocol: 1. Post entry 1 (amount A on the test account). 2. Prime the cache by calling `calculate_account_balance`. 3. Post entry 2 (amount B on the same account). 4. Call `calculate_account_balance` again. 5. Assert that the balance after step 4 equals the balance from step 2 plus the net contribution of entry 2, proving that the cache was cleared between steps 2 and 4. **Validates: Requirements 11.1, 11.2, 11.3, 11.4** """ import uuid from datetime import date from decimal import Decimal from hypothesis import given, settings, strategies as st from hypothesis.extra.django import TestCase from django.contrib.auth import get_user_model from accounting.models import ( Account, FiscalPeriod, 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, ) # --------------------------------------------------------------------------- # Test case # --------------------------------------------------------------------------- class CacheInvalidationAfterPostingPropertyTest(TestCase): """ Property 9: Cache Invalidation After Posting **Validates: Requirements 11.1, 11.2, 11.3, 11.4** For any account and date `d`, posting a new JournalEntry MUST cause the next call to `calculate_account_balance` to return a fresh value that incorporates the newly posted amounts, not a stale cached value. """ def setUp(self): """Create the minimal shared fixture for each hypothesis example.""" suffix = uuid.uuid4().hex[:8] phone_suffix = uuid.uuid4().int % 10 ** 9 # 9-digit integer phone_number = f"+254{phone_suffix:09d}" self.user = User.objects.create_user( username=f"cache_inv_user_{suffix}", email=f"cache_inv_{suffix}@example.com", password="testpass123", phone_number=phone_number, ) self.branch = Branch.objects.create( name=f"CacheInvBranch_{suffix}", code=f"CI{suffix[:5].upper()}", ) # Wide-open fiscal period so all generated dates fall inside. self.fiscal_period = FiscalPeriod.objects.create( name=f"Cache Invalidation Period {suffix}", period_type="monthly", start_date=date(2020, 1, 1), end_date=date(2030, 12, 31), status="open", ) # Asset account — the "focus" account whose balance we track. self.asset_account = Account.objects.create( code=f"1{suffix[:6]}", name="CacheInv Asset", account_type="asset", subtype="current_asset", description="Cache invalidation test — asset", is_active=True, created_by=self.user, ) # Counter-part account (income / credit side). self.income_account = Account.objects.create( code=f"3{suffix[:6]}", name="CacheInv Income", account_type="income", description="Cache invalidation test — income", is_active=True, created_by=self.user, ) self.accounting_service = AccountingService() # ----------------------------------------------------------------------- # Helpers # ----------------------------------------------------------------------- def _unique_ref(self, prefix: str = "CI") -> str: """Generate a unique journal-entry reference number.""" return f"{prefix}-{uuid.uuid4().hex[:12].upper()}" def _post_asset_entry( self, debit_amount: Decimal, tx_date: date = date(2024, 7, 15), ) -> JournalEntry: """ Post a balanced JournalEntry that debits `self.asset_account` and credits `self.income_account` by `debit_amount`. Returns the posted JournalEntry. """ ref = self._unique_ref() je = JournalEntry.objects.create( reference_number=ref, transaction_date=tx_date, description=f"Cache invalidation property test entry {ref}", branch=self.branch, created_by=self.user, status="draft", ) # Debit leg — asset account JournalEntryLine.objects.create( journal_entry=je, account=self.asset_account, description="Asset debit leg", debit_amount=debit_amount, credit_amount=Decimal("0.00"), line_number=1, ) # Credit leg — income account (balances the entry) JournalEntryLine.objects.create( journal_entry=je, account=self.income_account, description="Income credit leg", debit_amount=Decimal("0.00"), credit_amount=debit_amount, line_number=2, ) self.accounting_service.post_journal_entry(je, self.user) je.refresh_from_db() return je # ----------------------------------------------------------------------- # Property tests # ----------------------------------------------------------------------- @settings(max_examples=40, deadline=None) @given( amount_a=_amount_strategy(), amount_b=_amount_strategy(), ) def test_balance_reflects_second_posting_after_cache_primed( self, amount_a: Decimal, amount_b: Decimal ): """ Property 9: Cache Invalidation After Posting **Validates: Requirements 11.1, 11.2, 11.3, 11.4** Protocol: 1. Post entry 1 (debit amount_a on asset_account). 2. Prime the cache: call calculate_account_balance — gets balance_after_1. 3. Post entry 2 (debit amount_b on asset_account). 4. Call calculate_account_balance again — gets balance_after_2. 5. Assert balance_after_2 == balance_after_1 + amount_b. If the cache was NOT invalidated, balance_after_2 would still equal balance_after_1, making the assertion fail. """ tx_date = date(2024, 7, 15) # Step 1 — post first entry self._post_asset_entry(amount_a, tx_date) # Step 2 — prime the memory cache balance_after_1 = self.accounting_service.calculate_account_balance( self.asset_account, tx_date ) # Sanity-check: balance after entry 1 must be positive (debit account) assert balance_after_1 > Decimal("0.00"), ( f"balance_after_1 should be positive (asset account), got {balance_after_1}" ) # Step 3 — post second entry self._post_asset_entry(amount_b, tx_date) # Step 4 — read balance again; cache must have been invalidated balance_after_2 = self.accounting_service.calculate_account_balance( self.asset_account, tx_date ) # Step 5 — core assertion: fresh value incorporates entry 2 expected_balance = balance_after_1 + amount_b assert balance_after_2 == expected_balance, ( f"Cache invalidation failed: expected balance {expected_balance} " f"(balance_after_1={balance_after_1} + amount_b={amount_b}), " f"but got {balance_after_2}. " f"The cached value from before the second posting was returned." ) @settings(max_examples=30, deadline=None) @given(amount=_amount_strategy()) def test_single_posting_invalidates_stale_cache(self, amount: Decimal): """ Property 9 (stale-cache variant): if the cache is primed before a posting and then an entry is posted, the next read MUST return a value that incorporates the posted amount, not the stale cached value. Protocol: 1. Explicitly clear any pre-existing cache entry for this account. 2. Prime the cache by reading the balance (records baseline). 3. Post an entry (amount A). 4. Read again — balance must equal baseline + A. **Validates: Requirements 11.1, 11.4** """ from django.core.cache import cache as django_cache tx_date = date(2024, 8, 20) # Step 1 — ensure a clean slate by clearing any leftover cache key # from previous hypothesis examples (LocMemCache persists in-process). cache_key = ( f"balance_{self.asset_account.code}_{tx_date}_all" ) django_cache.delete(cache_key) # Step 2 — prime the cache, record baseline baseline = self.accounting_service.calculate_account_balance( self.asset_account, tx_date ) # Step 3 — post an entry self._post_asset_entry(amount, tx_date) # Step 4 — read again; cache must have been cleared by post_journal_entry balance_after_posting = self.accounting_service.calculate_account_balance( self.asset_account, tx_date ) expected = baseline + amount assert balance_after_posting == expected, ( f"Cache invalidation failed after posting: " f"expected baseline({baseline}) + amount({amount}) = {expected}, " f"but got {balance_after_posting}. " f"The stale cached value was returned." ) @settings(max_examples=30, deadline=None) @given( amount_a=_amount_strategy(), amount_b=_amount_strategy(), amount_c=_amount_strategy(), ) def test_balance_reflects_all_three_sequential_postings( self, amount_a: Decimal, amount_b: Decimal, amount_c: Decimal, ): """ Property 9 (three-posting chain): posting three entries sequentially MUST produce a final balance equal to the sum of all three amounts, proving that cache invalidation fires reliably on every post. **Validates: Requirements 11.1, 11.2, 11.3, 11.4** """ tx_date = date(2024, 9, 10) # Post entry 1, prime cache self._post_asset_entry(amount_a, tx_date) b1 = self.accounting_service.calculate_account_balance( self.asset_account, tx_date ) # Post entry 2, prime cache again self._post_asset_entry(amount_b, tx_date) b2 = self.accounting_service.calculate_account_balance( self.asset_account, tx_date ) # Post entry 3 self._post_asset_entry(amount_c, tx_date) b3 = self.accounting_service.calculate_account_balance( self.asset_account, tx_date ) # Each cache read must reflect the latest state assert b2 == b1 + amount_b, ( f"After second posting: expected b2={b1 + amount_b}, got {b2}. " f"Cache was not invalidated after entry 2." ) assert b3 == b2 + amount_c, ( f"After third posting: expected b3={b2 + amount_c}, got {b3}. " f"Cache was not invalidated after entry 3." ) assert b3 == amount_a + amount_b + amount_c, ( f"Final balance {b3} does not equal sum of all amounts " f"({amount_a} + {amount_b} + {amount_c} = {amount_a + amount_b + amount_c})." )