248 lines
9.9 KiB
Python
248 lines
9.9 KiB
Python
"""Unit tests for AccountingAgent — plan, gather, reason, report, peer_bus, sweep."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from agent_service.agents.accounting_agent import AccountingAgent, ACCOUNTING_TOOLS
|
|
from agent_service.agents.base_agent import AgentDirective, AgentReport, SweepReport
|
|
|
|
|
|
def _directive(intent='', context=None):
|
|
return AgentDirective(
|
|
directive_id='test-d1', user_id='1', intent=intent,
|
|
context=context or {}, agent_name='accounting_agent',
|
|
)
|
|
|
|
|
|
def _make_agent():
|
|
odoo = MagicMock()
|
|
llm = MagicMock()
|
|
agent = AccountingAgent(odoo=odoo, llm=llm)
|
|
agent._at = MagicMock()
|
|
agent._at.get_trial_balance = AsyncMock(return_value=[])
|
|
agent._at.get_tax_summary = AsyncMock(return_value={})
|
|
agent._at.get_journal_entries = AsyncMock(return_value=[])
|
|
agent._at.get_chart_of_accounts = AsyncMock(return_value=[])
|
|
agent._at.get_account_balance = AsyncMock(return_value={'debit': 0, 'credit': 0})
|
|
agent._at.flag_for_review = AsyncMock(return_value=True)
|
|
agent._at.post_chatter_note = AsyncMock(return_value=True)
|
|
return agent
|
|
|
|
|
|
# ── Meta ────────────────────────────────────────────────────────────────────
|
|
|
|
def test_tool_count():
|
|
assert len(ACCOUNTING_TOOLS) <= 8
|
|
|
|
def test_agent_name():
|
|
assert AccountingAgent.name == 'accounting_agent'
|
|
|
|
def test_hipaa_locked():
|
|
from agent_service.llm.llm_router import HIPAA_LOCKED_AGENTS
|
|
assert 'accounting_agent' in HIPAA_LOCKED_AGENTS
|
|
|
|
|
|
# ── _plan ───────────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plan_trial_balance_intent():
|
|
agent = _make_agent()
|
|
plan = await agent._plan(_directive(intent='show trial balance'))
|
|
assert plan['fetch_trial_balance'] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plan_tax_intent():
|
|
agent = _make_agent()
|
|
plan = await agent._plan(_directive(intent='get tax summary for Q1'))
|
|
assert plan['fetch_tax'] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plan_journal_entries_intent():
|
|
agent = _make_agent()
|
|
plan = await agent._plan(_directive(intent='list journal entries'))
|
|
assert plan['fetch_entries'] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plan_unknown_intent_all_false():
|
|
agent = _make_agent()
|
|
plan = await agent._plan(_directive(intent='random question'))
|
|
assert plan['fetch_trial_balance'] is False
|
|
assert plan['fetch_tax'] is False
|
|
assert plan['fetch_entries'] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plan_propagates_dates():
|
|
agent = _make_agent()
|
|
plan = await agent._plan(_directive(context={'date_from': '2026-01-01', 'date_to': '2026-03-31'}))
|
|
assert plan['date_from'] == '2026-01-01'
|
|
assert plan['date_to'] == '2026-03-31'
|
|
|
|
|
|
# ── _gather ─────────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gather_fetches_trial_balance():
|
|
agent = _make_agent()
|
|
agent._at.get_trial_balance = AsyncMock(return_value=[{'account_name': 'Cash', 'balance': 5000}])
|
|
ctx = {'plan': {'fetch_trial_balance': True, 'fetch_tax': False, 'fetch_entries': False,
|
|
'date_from': None, 'date_to': None}}
|
|
data = await agent._gather(ctx)
|
|
assert 'trial_balance' in data
|
|
agent._at.get_trial_balance.assert_awaited_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gather_fetches_tax_summary():
|
|
agent = _make_agent()
|
|
agent._at.get_tax_summary = AsyncMock(return_value={'total_tax_amount': 1500.0})
|
|
ctx = {'plan': {'fetch_trial_balance': False, 'fetch_tax': True, 'fetch_entries': False,
|
|
'date_from': None, 'date_to': None}}
|
|
data = await agent._gather(ctx)
|
|
assert 'tax_summary' in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gather_falls_back_to_entries_when_nothing_else():
|
|
agent = _make_agent()
|
|
ctx = {'plan': {'fetch_trial_balance': False, 'fetch_tax': False, 'fetch_entries': False,
|
|
'date_from': None, 'date_to': None}}
|
|
data = await agent._gather(ctx)
|
|
assert 'entries' in data
|
|
agent._at.get_journal_entries.assert_awaited()
|
|
|
|
|
|
# ── _reason ─────────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reason_flags_large_balance():
|
|
agent = _make_agent()
|
|
agent._gathered_data = {
|
|
'trial_balance': [{'account_name': 'Suspense', 'balance': 200000.0}]
|
|
}
|
|
analysis = await agent._reason({})
|
|
assert len(analysis['flags']) == 1
|
|
assert analysis['flags'][0]['balance'] == 200000.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reason_no_flags_below_threshold():
|
|
agent = _make_agent()
|
|
agent._gathered_data = {'trial_balance': [{'account_name': 'Cash', 'balance': 50000.0}]}
|
|
analysis = await agent._reason({})
|
|
assert analysis['flags'] == []
|
|
|
|
|
|
# ── _report ─────────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_report_mentions_trial_balance_count():
|
|
agent = _make_agent()
|
|
agent._gathered_data = {
|
|
'trial_balance': [{'account_name': 'A', 'balance': 1000}, {'account_name': 'B', 'balance': 2000}]
|
|
}
|
|
agent._escalations_list = []
|
|
report = await agent._report({})
|
|
assert isinstance(report, AgentReport)
|
|
assert '2' in report.summary
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_report_mentions_tax_total():
|
|
agent = _make_agent()
|
|
agent._gathered_data = {'tax_summary': {'total_tax_amount': 4500.0, 'total_tax_lines': 12}}
|
|
agent._escalations_list = []
|
|
report = await agent._report({})
|
|
assert '4500' in report.summary
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_report_fallback_message():
|
|
agent = _make_agent()
|
|
agent._gathered_data = {}
|
|
agent._escalations_list = []
|
|
report = await agent._report({})
|
|
assert 'complete' in report.summary.lower()
|
|
|
|
|
|
# ── _dispatch_tool ───────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_tool_get_journal_entries():
|
|
agent = _make_agent()
|
|
agent._at.get_journal_entries = AsyncMock(return_value=[{'id': 1}])
|
|
result = await agent._dispatch_tool('get_journal_entries', {'limit': 5})
|
|
agent._at.get_journal_entries.assert_awaited_once_with(limit=5)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_tool_unknown_raises():
|
|
agent = _make_agent()
|
|
with pytest.raises(ValueError, match='Unknown tool'):
|
|
await agent._dispatch_tool('nonexistent', {})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_tool_flag_for_review():
|
|
agent = _make_agent()
|
|
await agent._dispatch_tool('flag_for_review',
|
|
{'model': 'account.move', 'record_id': 1, 'reason': 'anomaly'})
|
|
agent._at.flag_for_review.assert_awaited_once()
|
|
|
|
|
|
# ── handle_peer_request ──────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_peer_request_trial_balance():
|
|
agent = _make_agent()
|
|
agent._at.get_trial_balance = AsyncMock(return_value=[{'account_name': 'Cash'}])
|
|
result = await agent.handle_peer_request('trial_balance', {}, 'dir-1')
|
|
assert 'trial_balance' in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_peer_request_account_balance():
|
|
agent = _make_agent()
|
|
agent._at.get_account_balance = AsyncMock(return_value={'debit': 100, 'credit': 50})
|
|
result = await agent.handle_peer_request('account_balance', {'account_id': 42}, 'dir-1')
|
|
assert 'debit' in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_peer_request_tax_summary():
|
|
agent = _make_agent()
|
|
agent._at.get_tax_summary = AsyncMock(return_value={'total_tax_amount': 0})
|
|
result = await agent.handle_peer_request('tax_summary', {}, 'dir-1')
|
|
assert 'total_tax_amount' in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_peer_request_unknown_returns_error():
|
|
agent = _make_agent()
|
|
result = await agent.handle_peer_request('nonexistent', {}, 'dir-1')
|
|
assert 'error' in result
|
|
|
|
|
|
# ── sweep ────────────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sweep_returns_sweep_report():
|
|
agent = _make_agent()
|
|
agent._at.get_trial_balance = AsyncMock(return_value=[])
|
|
result = await agent.sweep()
|
|
assert isinstance(result, SweepReport)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sweep_finds_large_balance():
|
|
agent = _make_agent()
|
|
agent._at.get_trial_balance = AsyncMock(return_value=[
|
|
{'account_name': 'Suspense', 'balance': 600000.0}
|
|
])
|
|
result = await agent.sweep()
|
|
assert len(result.findings) == 1
|
|
assert result.findings[0]['type'] == 'large_balance'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sweep_below_threshold_no_findings():
|
|
agent = _make_agent()
|
|
agent._at.get_trial_balance = AsyncMock(return_value=[
|
|
{'account_name': 'Cash', 'balance': 100.0}
|
|
])
|
|
result = await agent.sweep()
|
|
assert result.findings == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sweep_handles_exception():
|
|
agent = _make_agent()
|
|
agent._at.get_trial_balance = AsyncMock(side_effect=Exception('DB error'))
|
|
result = await agent.sweep()
|
|
assert isinstance(result, SweepReport)
|
|
assert result.error is not None
|