194 lines
7.2 KiB
Python
194 lines
7.2 KiB
Python
"""Unit tests for AccountingTools."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from agent_service.tools.accounting_tools import AccountingTools
|
|
from agent_service.tools.odoo_client import WriteResult
|
|
|
|
|
|
def _make_tools():
|
|
odoo = MagicMock()
|
|
odoo.search_read = AsyncMock(return_value=[])
|
|
odoo.write = AsyncMock(return_value=WriteResult(
|
|
success=True, model='', record_id=None, action='write'))
|
|
odoo.create = AsyncMock(return_value=WriteResult(
|
|
success=True, model='', record_id=42, action='create'))
|
|
odoo.call = AsyncMock(return_value=True)
|
|
return AccountingTools(odoo)
|
|
|
|
|
|
# ── get_journal_entries ───────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_journal_entries_default():
|
|
t = _make_tools()
|
|
result = await t.get_journal_entries()
|
|
t._o.search_read.assert_awaited_once()
|
|
assert isinstance(result, list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_journal_entries_with_filters():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[{'id': 1, 'name': 'JE/001'}])
|
|
result = await t.get_journal_entries(journal_id=5, date_from='2026-01-01', state='draft')
|
|
assert len(result) == 1
|
|
call_args = t._o.search_read.call_args
|
|
domain = call_args[0][1]
|
|
assert ('journal_id', '=', 5) in domain
|
|
assert ('date', '>=', '2026-01-01') in domain
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_journal_entries_date_to_filter():
|
|
t = _make_tools()
|
|
await t.get_journal_entries(date_to='2026-01-31')
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('date', '<=', '2026-01-31') in domain
|
|
|
|
|
|
# ── get_chart_of_accounts ─────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_chart_of_accounts_default():
|
|
t = _make_tools()
|
|
result = await t.get_chart_of_accounts()
|
|
t._o.search_read.assert_awaited_once()
|
|
assert isinstance(result, list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_chart_of_accounts_with_type():
|
|
t = _make_tools()
|
|
await t.get_chart_of_accounts(account_type='asset_current')
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('account_type', '=', 'asset_current') in domain
|
|
|
|
|
|
# ── get_account_balance ───────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_account_balance_found():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'id': 1, 'code': '1000', 'name': 'Cash', 'balance': 5000.0}
|
|
])
|
|
result = await t.get_account_balance(account_id=1)
|
|
assert result['balance'] == 5000.0
|
|
assert result['code'] == '1000'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_account_balance_not_found():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[])
|
|
result = await t.get_account_balance(account_id=999)
|
|
assert result == {}
|
|
|
|
|
|
# ── get_trial_balance ─────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_trial_balance_aggregates_lines():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'account_id': [1, 'Cash'], 'debit': 1000.0, 'credit': 0.0, 'balance': 1000.0},
|
|
{'account_id': [1, 'Cash'], 'debit': 500.0, 'credit': 0.0, 'balance': 500.0},
|
|
{'account_id': [2, 'AP'], 'debit': 0.0, 'credit': 2000.0, 'balance': -2000.0},
|
|
])
|
|
result = await t.get_trial_balance()
|
|
assert isinstance(result, list)
|
|
assert len(result) == 2
|
|
cash_entry = next(r for r in result if r['account_id'] == 1)
|
|
assert cash_entry['debit'] == 1500.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_trial_balance_balance_computed():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'account_id': [1, 'Cash'], 'debit': 1000.0, 'credit': 200.0, 'balance': 800.0},
|
|
])
|
|
result = await t.get_trial_balance()
|
|
assert result[0]['balance'] == 800.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_trial_balance_empty_returns_empty():
|
|
t = _make_tools()
|
|
result = await t.get_trial_balance()
|
|
assert result == []
|
|
|
|
|
|
# ── get_tax_summary ───────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_tax_summary_returns_dict():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'tax_ids': [1], 'debit': 100.0, 'credit': 0.0, 'balance': 100.0},
|
|
{'tax_ids': [2], 'debit': 50.0, 'credit': 0.0, 'balance': 50.0},
|
|
])
|
|
result = await t.get_tax_summary()
|
|
assert 'total_tax_lines' in result
|
|
assert result['total_tax_lines'] == 2
|
|
assert 'total_tax_amount' in result
|
|
assert result['total_tax_amount'] == 150.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_tax_summary_with_date_range():
|
|
t = _make_tools()
|
|
await t.get_tax_summary(date_from='2026-01-01', date_to='2026-01-31')
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('date', '>=', '2026-01-01') in domain
|
|
assert ('date', '<=', '2026-01-31') in domain
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_tax_summary_includes_period():
|
|
t = _make_tools()
|
|
result = await t.get_tax_summary(date_from='2026-01-01', date_to='2026-01-31')
|
|
assert result['period_from'] == '2026-01-01'
|
|
assert result['period_to'] == '2026-01-31'
|
|
|
|
|
|
# ── flag_for_review ───────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flag_for_review_calls_message_post():
|
|
t = _make_tools()
|
|
result = await t.flag_for_review('account.move', 1, 'Test reason', severity='high')
|
|
assert result is True
|
|
t._o.call.assert_awaited_once()
|
|
call_args = t._o.call.call_args
|
|
assert 'message_post' in str(call_args)
|
|
kwargs = call_args[0][3]
|
|
assert '[AI FLAG - HIGH]' in kwargs['body']
|
|
assert 'Test reason' in kwargs['body']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flag_for_review_default_severity():
|
|
t = _make_tools()
|
|
await t.flag_for_review('account.move', 1, 'reason')
|
|
kwargs = t._o.call.call_args[0][3]
|
|
assert 'MEDIUM' in kwargs['body']
|
|
|
|
|
|
# ── post_chatter_note ─────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_chatter_note_returns_true():
|
|
t = _make_tools()
|
|
result = await t.post_chatter_note('account.move', 1, 'Test note')
|
|
assert result is True
|
|
t._o.call.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_chatter_note_includes_body():
|
|
t = _make_tools()
|
|
await t.post_chatter_note('account.move', 5, 'My note text')
|
|
kwargs = t._o.call.call_args[0][3]
|
|
assert kwargs['body'] == 'My note text'
|