"""Tests for the fusion bank-rec HTTP controller (Task 26). Uses ``HttpCase`` so we exercise the full Werkzeug stack -- the JSON-RPC dispatcher, auth check, and route resolution all run for real. Tests authenticate as a Fusion Accounting administrator (the realistic role for a user driving the bank-rec widget); a separate test confirms the ``auth='user'`` decorator rejects anonymous traffic. """ import json from odoo.tests.common import HttpCase, new_test_user, tagged from . import _factories as f @tagged('post_install', '-at_install') class TestBankRecController(HttpCase): """End-to-end coverage of the 10 JSON-RPC endpoints.""" USER_LOGIN = 'ctrl_test_user' USER_PASSWORD = 'ctrl_test_user' def setUp(self): super().setUp() # group_account_user grants accounting write perms AND auto-implies # fusion_accounting_user via the security XML's implied_ids hook; # group_fusion_accounting_admin grants full CRUD on the fusion # suggestion / precedent / pattern models the engine writes to. self.user = new_test_user( self.env, login=self.USER_LOGIN, password=self.USER_PASSWORD, groups=( 'base.group_user,' 'account.group_account_user,' 'fusion_accounting_core.group_fusion_accounting_admin' ), ) self.partner = self.env['res.partner'].create( {'name': 'Controller Test Partner'}) self.journal = f.make_bank_journal( self.env, name='Ctrl Bank', code='CBNK') # ------------------------------------------------------------------ # helpers # ------------------------------------------------------------------ def _jsonrpc(self, endpoint, params, *, authenticate=True): """POST a JSON-RPC envelope to ``/fusion/bank_rec/``. Returns the ``result`` dict on success and fails the test if the body has an ``error`` key (so endpoint test failures show the actual server-side exception, not just the HTTP status). """ if authenticate: self.authenticate(self.USER_LOGIN, self.USER_PASSWORD) url = f'/fusion/bank_rec/{endpoint}' body = { 'jsonrpc': '2.0', 'method': 'call', 'params': params, 'id': 1, } response = self.url_open( url, data=json.dumps(body), headers={'Content-Type': 'application/json'}, ) self.assertEqual( response.status_code, 200, f"Endpoint {endpoint} returned {response.status_code}: " f"{response.text[:300]}") payload = response.json() if 'error' in payload: self.fail( f"Endpoint {endpoint} errored: " f"{json.dumps(payload['error'])[:600]}") return payload.get('result', {}) # ------------------------------------------------------------------ # 1. get_state # ------------------------------------------------------------------ def test_get_state(self): result = self._jsonrpc('get_state', { 'journal_id': self.journal.id, 'company_id': self.env.company.id, }) self.assertIn('journal', result) self.assertEqual(result['journal']['id'], self.journal.id) self.assertIn('unreconciled_count', result) self.assertIn('total_pending_amount', result) self.assertIn('last_statement_date', result) # ------------------------------------------------------------------ # 2. list_unreconciled # ------------------------------------------------------------------ def test_list_unreconciled(self): # Reuse a single statement so we don't trip the # (journal_id, name) uniqueness or hit the parent-move autocreate # path twice in the same flush window. statement = f.make_bank_statement( self.env, journal=self.journal, name='List Stmt') f.make_bank_line( self.env, journal=self.journal, statement=statement, amount=100, partner=self.partner, memo='List 1') f.make_bank_line( self.env, journal=self.journal, statement=statement, amount=200, partner=self.partner, memo='List 2') result = self._jsonrpc('list_unreconciled', { 'journal_id': self.journal.id, 'limit': 50, 'offset': 0, 'company_id': self.env.company.id, }) self.assertIn('lines', result) self.assertGreaterEqual(len(result['lines']), 2) self.assertGreaterEqual(result['total'], 2) first = result['lines'][0] for key in ('id', 'amount', 'fusion_top_suggestion_id', 'fusion_confidence_band', 'attachment_count'): self.assertIn(key, first) # ------------------------------------------------------------------ # 3. get_line_detail # ------------------------------------------------------------------ def test_get_line_detail(self): line = f.make_bank_line( self.env, journal=self.journal, amount=100, partner=self.partner) f.make_suggestion( self.env, statement_line=line, confidence=0.85) result = self._jsonrpc( 'get_line_detail', {'statement_line_id': line.id}) self.assertEqual(result['line']['id'], line.id) self.assertEqual(result['line']['amount'], 100.0) self.assertGreaterEqual(len(result['suggestions']), 1) sug = result['suggestions'][0] for key in ('id', 'candidate_ids', 'confidence', 'rank', 'reasoning', 'scores'): self.assertIn(key, sug) # ------------------------------------------------------------------ # 4. suggest_matches # ------------------------------------------------------------------ def test_suggest_matches(self): f.make_invoice(self.env, partner=self.partner, amount=300) line = f.make_bank_line( self.env, journal=self.journal, amount=300, partner=self.partner) result = self._jsonrpc('suggest_matches', { 'statement_line_ids': [line.id], 'limit_per_line': 3, }) self.assertIn('suggestions', result) self.assertIsInstance(result['suggestions'], dict) # ------------------------------------------------------------------ # 5. accept_suggestion # ------------------------------------------------------------------ def test_accept_suggestion(self): invoice = f.make_invoice( self.env, partner=self.partner, amount=400) recv = invoice.line_ids.filtered( lambda l: l.account_id.account_type == 'asset_receivable') line = f.make_bank_line( self.env, journal=self.journal, amount=400, partner=self.partner) sug = f.make_suggestion( self.env, statement_line=line, candidate_move_lines=recv, confidence=0.92) result = self._jsonrpc( 'accept_suggestion', {'suggestion_id': sug.id}) self.assertEqual(result['status'], 'accepted') self.assertGreater(len(result['partial_ids']), 0) self.assertIn('unreconciled_count_after', result) # ------------------------------------------------------------------ # 6. reconcile_manual # ------------------------------------------------------------------ def _make_pair(self, *, amount, statement=None): """Inline reconcile-able pair against ``self.journal``. The shared ``make_reconcileable_pair`` factory creates a fresh bank journal per call (default code 'TEST'), which collides with the unique (code, company) constraint when used multiple times in one test. Reusing ``self.journal`` (and optionally a shared statement) keeps every pair on the same journal. """ invoice = f.make_invoice( self.env, partner=self.partner, amount=amount) recv = invoice.line_ids.filtered( lambda l: l.account_id.account_type == 'asset_receivable') line = f.make_bank_line( self.env, journal=self.journal, statement=statement, amount=amount, partner=self.partner) return line, recv def test_reconcile_manual(self): line, recv = self._make_pair(amount=550) result = self._jsonrpc('reconcile_manual', { 'statement_line_id': line.id, 'against_move_line_ids': recv.ids, }) self.assertEqual(result['status'], 'reconciled') self.assertGreater(len(result['partial_ids']), 0) # ------------------------------------------------------------------ # 7. unreconcile # ------------------------------------------------------------------ def test_unreconcile(self): line, recv = self._make_pair(amount=625) rec = self.env['fusion.reconcile.engine'].reconcile_one( line, against_lines=recv) result = self._jsonrpc('unreconcile', { 'partial_reconcile_ids': rec['partial_ids'], }) self.assertEqual(result['status'], 'unreconciled') self.assertGreater(len(result['unreconciled_line_ids']), 0) # ------------------------------------------------------------------ # 8. write_off -- smoke only (Task 12 deferred full coverage) # ------------------------------------------------------------------ def test_write_off_smoke(self): line = f.make_bank_line( self.env, journal=self.journal, amount=12.34, partner=self.partner) # Pick any expense-type account that exists in the chart. wo_account = self.env['account.account'].search([ ('account_type', '=', 'expense'), ('company_ids', 'in', self.env.company.id), ], limit=1) if not wo_account: self.skipTest("No expense account available for write-off smoke") # Endpoint must respond without 500-erroring; engine may legitimately # raise a ValidationError for an over-allocation, in which case the # JSON-RPC response will include an 'error' key. We accept either # success or a structured error -- what we are guarding against is a # routing-layer regression (NameError, missing import, etc.). url = '/fusion/bank_rec/write_off' self.authenticate(self.USER_LOGIN, self.USER_PASSWORD) body = { 'jsonrpc': '2.0', 'method': 'call', 'id': 1, 'params': { 'statement_line_id': line.id, 'account_id': wo_account.id, 'amount': line.amount, 'label': 'Smoke write-off', }, } response = self.url_open( url, data=json.dumps(body), headers={'Content-Type': 'application/json'}) self.assertEqual( response.status_code, 200, f"write_off returned {response.status_code}: " f"{response.text[:300]}") # ------------------------------------------------------------------ # 9. bulk_reconcile # ------------------------------------------------------------------ def test_bulk_reconcile(self): statement = f.make_bank_statement( self.env, journal=self.journal, name='Bulk Stmt') line_ids = [] for amt in (110, 220, 330): line, _recv = self._make_pair(amount=amt, statement=statement) line_ids.append(line.id) result = self._jsonrpc('bulk_reconcile', { 'statement_line_ids': line_ids, 'strategy': 'auto', }) self.assertIn('reconciled_count', result) self.assertGreaterEqual(result['reconciled_count'], 3) # ------------------------------------------------------------------ # 10. get_partner_history # ------------------------------------------------------------------ def test_get_partner_history(self): for d in (5, 12, 20): f.make_precedent( self.env, partner=self.partner, days_ago=d, amount=1000) f.make_pattern( self.env, partner=self.partner, reconcile_count=3) result = self._jsonrpc('get_partner_history', { 'partner_id': self.partner.id, 'limit': 10, }) self.assertEqual(result['partner']['id'], self.partner.id) self.assertGreaterEqual(len(result['recent_reconciles']), 3) self.assertIsNotNone(result['pattern']) self.assertEqual(result['pattern']['reconcile_count'], 3) # ------------------------------------------------------------------ # 11. unauthenticated traffic is blocked # ------------------------------------------------------------------ def test_unauthenticated_request_blocked(self): # Use a fresh session by creating a new opener -- self.url_open # reuses the test session, which `authenticate()` would mutate. url = '/fusion/bank_rec/get_state' body = { 'jsonrpc': '2.0', 'method': 'call', 'id': 1, 'params': { 'journal_id': self.journal.id, 'company_id': self.env.company.id, }, } # No call to self.authenticate() -> session has no uid. response = self.url_open( url, data=json.dumps(body), headers={'Content-Type': 'application/json'}, allow_redirects=False, ) # Odoo's auth='user' on a JSON-RPC route returns a 200 with an # error envelope (SessionExpiredException) when not authenticated; # what must NOT happen is the handler running and returning our # success payload. if response.status_code == 200: payload = response.json() self.assertIn( 'error', payload, "Unauthenticated request should not return a success result") else: # 3xx redirect or 4xx are also acceptable rejections. self.assertGreaterEqual(response.status_code, 300)