291 lines
10 KiB
Python
291 lines
10 KiB
Python
"""Unit tests for ExpensesTools."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from agent_service.tools.expenses_tools import ExpensesTools
|
|
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 ExpensesTools(odoo)
|
|
|
|
|
|
# ── get_expenses ──────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expenses_default():
|
|
t = _make_tools()
|
|
result = await t.get_expenses()
|
|
t._o.search_read.assert_awaited_once()
|
|
assert isinstance(result, list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expenses_with_employee():
|
|
t = _make_tools()
|
|
await t.get_expenses(employee_id=5)
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('employee_id', '=', 5) in domain
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expenses_with_state():
|
|
t = _make_tools()
|
|
await t.get_expenses(state='draft')
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('state', '=', 'draft') in domain
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expenses_with_date_range():
|
|
t = _make_tools()
|
|
await t.get_expenses(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
|
|
|
|
|
|
# ── get_expense_sheets ────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expense_sheets_default():
|
|
t = _make_tools()
|
|
result = await t.get_expense_sheets()
|
|
t._o.search_read.assert_awaited_once()
|
|
assert isinstance(result, list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expense_sheets_with_state():
|
|
t = _make_tools()
|
|
await t.get_expense_sheets(state='submit')
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('state', '=', 'submit') in domain
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expense_sheets_with_employee():
|
|
t = _make_tools()
|
|
await t.get_expense_sheets(employee_id=3)
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('employee_id', '=', 3) in domain
|
|
|
|
|
|
# ── get_pending_approvals ─────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_pending_approvals_uses_submit_state():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'id': 1, 'name': 'Sheet 1', 'total_amount': 500.0}
|
|
])
|
|
result = await t.get_pending_approvals()
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('state', '=', 'submit') in domain
|
|
assert len(result) == 1
|
|
|
|
|
|
# ── approve_expense_sheet ─────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_expense_sheet_success():
|
|
t = _make_tools()
|
|
result = await t.approve_expense_sheet(sheet_id=1)
|
|
assert result is True
|
|
t._o.call.assert_awaited_once_with(
|
|
'hr.expense.sheet', 'approve_expense_sheets', [[1]])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_expense_sheet_handles_exception():
|
|
t = _make_tools()
|
|
t._o.call = AsyncMock(side_effect=Exception('permission denied'))
|
|
result = await t.approve_expense_sheet(sheet_id=1)
|
|
assert result is False
|
|
|
|
|
|
# ── get_expenses_summary ──────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expenses_summary_structure():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(side_effect=[
|
|
[{'total_amount': 100.0}, {'total_amount': 200.0}], # expenses
|
|
[], # pending_approvals
|
|
])
|
|
result = await t.get_expenses_summary()
|
|
assert 'total_expenses' in result
|
|
assert result['total_expenses'] == 2
|
|
assert result['total_amount'] == 300.0
|
|
assert 'pending_approval_count' in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expenses_summary_with_dates():
|
|
t = _make_tools()
|
|
await t.get_expenses_summary(date_from='2026-01-01', date_to='2026-01-31')
|
|
domain = t._o.search_read.call_args_list[0][0][1]
|
|
assert ('date', '>=', '2026-01-01') in domain
|
|
|
|
|
|
# ── get_expense_by_employee ───────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expense_by_employee():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'id': 1, 'name': 'Flight', 'total_amount': 500.0}
|
|
])
|
|
result = await t.get_expense_by_employee(employee_id=5)
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('employee_id', '=', 5) in domain
|
|
assert len(result) == 1
|
|
|
|
|
|
# ── get_employee_id_for_user ──────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_employee_id_for_user_found():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[{'id': 7, 'name': 'Alice'}])
|
|
result = await t.get_employee_id_for_user(user_id=3)
|
|
assert result == 7
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_employee_id_for_user_not_found():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[])
|
|
result = await t.get_employee_id_for_user(user_id=99)
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_employee_id_for_user_none_input():
|
|
t = _make_tools()
|
|
result = await t.get_employee_id_for_user(user_id=None)
|
|
assert result is None
|
|
|
|
|
|
# ── get_default_expense_product ───────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_default_expense_product_found():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[{'id': 99, 'name': 'Expense'}])
|
|
result = await t.get_default_expense_product()
|
|
assert result == 99
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_default_expense_product_not_found():
|
|
t = _make_tools()
|
|
result = await t.get_default_expense_product()
|
|
assert result is None
|
|
|
|
|
|
# ── get_expense_products ──────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expense_products():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'id': 1, 'name': 'Travel'}, {'id': 2, 'name': 'Meals'}
|
|
])
|
|
result = await t.get_expense_products()
|
|
assert len(result) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_expense_products_returns_empty_on_exception():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(side_effect=Exception('error'))
|
|
result = await t.get_expense_products()
|
|
assert result == []
|
|
|
|
|
|
# ── create_expense_sheet ──────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_expense_sheet():
|
|
t = _make_tools()
|
|
result = await t.create_expense_sheet(name='Q1 Expenses', employee_id=5)
|
|
t._o.create.assert_awaited_once_with('hr.expense.sheet', {
|
|
'name': 'Q1 Expenses', 'employee_id': 5,
|
|
})
|
|
|
|
|
|
# ── create_expense ────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_expense_basic():
|
|
t = _make_tools()
|
|
await t.create_expense(
|
|
sheet_id=1, employee_id=5, name='Flight',
|
|
total_amount=500.0, date='2026-01-15',
|
|
)
|
|
vals = t._o.create.call_args[0][1]
|
|
assert vals['name'] == 'Flight'
|
|
assert vals['total_amount'] == 500.0
|
|
assert vals['date'] == '2026-01-15'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_expense_with_product():
|
|
t = _make_tools()
|
|
await t.create_expense(
|
|
sheet_id=1, employee_id=5, name='Hotel',
|
|
total_amount=200.0, date='2026-01-16', product_id=10,
|
|
)
|
|
vals = t._o.create.call_args[0][1]
|
|
assert vals['product_id'] == 10
|
|
|
|
|
|
# ── attach_receipt ────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_attach_receipt_success():
|
|
t = _make_tools()
|
|
result = await t.attach_receipt(
|
|
model='hr.expense', record_id=1,
|
|
filename='receipt.pdf', file_b64='base64data', mimetype='application/pdf',
|
|
)
|
|
assert result is True
|
|
t._o.create.assert_awaited_once()
|
|
vals = t._o.create.call_args[0][1]
|
|
assert vals['name'] == 'receipt.pdf'
|
|
assert vals['res_model'] == 'hr.expense'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_attach_receipt_handles_exception():
|
|
t = _make_tools()
|
|
t._o.create = AsyncMock(side_effect=Exception('error'))
|
|
result = await t.attach_receipt('hr.expense', 1, 'f.pdf', 'b64', 'application/pdf')
|
|
assert result is False
|
|
|
|
|
|
# ── flag_for_review / post_chatter_note ──────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flag_for_review():
|
|
t = _make_tools()
|
|
result = await t.flag_for_review('hr.expense', 1, 'Suspicious amount')
|
|
assert result is True
|
|
call_args = t._o.call.call_args[0]
|
|
assert 'MEDIUM' in call_args[3]['body']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_chatter_note():
|
|
t = _make_tools()
|
|
result = await t.post_chatter_note('hr.expense', 1, 'Expense verified')
|
|
assert result is True
|
|
t._o.call.assert_awaited_once()
|