Add comprehensive unit tests for all agent service components

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

tests/test_crm_tools.py

@@ -0,0 +1,192 @@
+"""Unit tests for CrmTools."""
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+from agent_service.tools.crm_tools import CrmTools
+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 CrmTools(odoo)
+
+
+# ── get_leads ────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_get_leads_default():
+    t = _make_tools()
+    result = await t.get_leads()
+    t._o.search_read.assert_awaited_once()
+    domain = t._o.search_read.call_args[0][1]
+    assert ('type', '=', 'lead') in domain
+    assert isinstance(result, list)
+
+
+@pytest.mark.asyncio
+async def test_get_leads_with_stage_filter():
+    t = _make_tools()
+    await t.get_leads(stage_id=3)
+    domain = t._o.search_read.call_args[0][1]
+    assert ('stage_id', '=', 3) in domain
+
+
+@pytest.mark.asyncio
+async def test_get_leads_with_user_filter():
+    t = _make_tools()
+    await t.get_leads(user_id=5)
+    domain = t._o.search_read.call_args[0][1]
+    assert ('user_id', '=', 5) in domain
+
+
+# ── get_opportunities ────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_get_opportunities_default():
+    t = _make_tools()
+    result = await t.get_opportunities()
+    domain = t._o.search_read.call_args[0][1]
+    assert ('type', '=', 'opportunity') in domain
+    assert isinstance(result, list)
+
+
+@pytest.mark.asyncio
+async def test_get_opportunities_filters():
+    t = _make_tools()
+    await t.get_opportunities(stage_id=2, user_id=7)
+    domain = t._o.search_read.call_args[0][1]
+    assert ('stage_id', '=', 2) in domain
+    assert ('user_id', '=', 7) in domain
+
+
+# ── get_pipeline_summary ─────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_get_pipeline_summary_empty():
+    t = _make_tools()
+    result = await t.get_pipeline_summary()
+    assert 'total_opportunities' in result
+    assert result['total_opportunities'] == 0
+    assert 'weighted_pipeline' in result
+
+
+@pytest.mark.asyncio
+async def test_get_pipeline_summary_aggregates_by_stage():
+    t = _make_tools()
+    t._o.search_read = AsyncMock(side_effect=[
+        [{'id': 1, 'name': 'New', 'sequence': 1}],  # stages
+        [
+            {'stage_id': [1, 'New'], 'expected_revenue': 10000.0, 'probability': 20},
+            {'stage_id': [1, 'New'], 'expected_revenue': 5000.0, 'probability': 50},
+        ],
+    ])
+    result = await t.get_pipeline_summary()
+    assert result['total_opportunities'] == 2
+    assert result['total_pipeline'] == 15000.0
+    assert len(result['stages']) == 1
+    assert result['stages'][0]['count'] == 2
+
+
+@pytest.mark.asyncio
+async def test_get_pipeline_summary_weighted_calculation():
+    t = _make_tools()
+    t._o.search_read = AsyncMock(side_effect=[
+        [],
+        [{'stage_id': [1, 'Stage'], 'expected_revenue': 1000.0, 'probability': 50}],
+    ])
+    result = await t.get_pipeline_summary()
+    assert result['weighted_pipeline'] == 500.0
+
+
+# ── update_lead_stage ─────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_update_lead_stage_success():
+    t = _make_tools()
+    result = await t.update_lead_stage(lead_id=1, stage_id=3)
+    assert result is True
+    t._o.write.assert_awaited_once_with('crm.lead', [1], {'stage_id': 3})
+
+
+@pytest.mark.asyncio
+async def test_update_lead_stage_failure():
+    t = _make_tools()
+    t._o.write = AsyncMock(return_value=WriteResult(
+        success=False, model='', record_id=None, action='write', error='permission denied'))
+    result = await t.update_lead_stage(lead_id=1, stage_id=3)
+    assert result is False
+
+
+# ── assign_lead ───────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_assign_lead_success():
+    t = _make_tools()
+    result = await t.assign_lead(lead_id=5, user_id=10)
+    assert result is True
+    t._o.write.assert_awaited_once_with('crm.lead', [5], {'user_id': 10})
+
+
+# ── log_activity ──────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_log_activity_without_type_record():
+    t = _make_tools()
+    t._o.search_read = AsyncMock(return_value=[])
+    result = await t.log_activity(lead_id=1, activity_type='call', note='Called customer')
+    assert result is True
+    t._o.call.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_log_activity_with_type_record():
+    t = _make_tools()
+    t._o.search_read = AsyncMock(return_value=[{'id': 7}])
+    await t.log_activity(lead_id=1, activity_type='email', note='Sent email', date_deadline='2026-06-01')
+    call_args = t._o.call.call_args[0]
+    vals = call_args[2][0]
+    assert vals['activity_type_id'] == 7
+    assert vals['date_deadline'] == '2026-06-01'
+    assert vals['note'] == 'Sent email'
+
+
+# ── get_won_lost_analysis ─────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_get_won_lost_analysis_counts():
+    t = _make_tools()
+    t._o.search_read = AsyncMock(side_effect=[
+        [{'expected_revenue': 5000.0}, {'expected_revenue': 3000.0}],  # won
+        [{'expected_revenue': 1000.0}],  # lost
+    ])
+    result = await t.get_won_lost_analysis()
+    assert result['won_count'] == 2
+    assert result['won_revenue'] == 8000.0
+    assert result['lost_count'] == 1
+    assert result['lost_revenue'] == 1000.0
+
+
+@pytest.mark.asyncio
+async def test_get_won_lost_analysis_with_dates():
+    t = _make_tools()
+    await t.get_won_lost_analysis(date_from='2026-01-01', date_to='2026-01-31')
+    first_call = t._o.search_read.call_args_list[0][0][1]
+    assert ('date_closed', '>=', '2026-01-01') in first_call
+
+
+# ── post_chatter_note ─────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_post_chatter_note_calls_message_post():
+    t = _make_tools()
+    result = await t.post_chatter_note('crm.lead', 1, 'Note text')
+    assert result is True
+    t._o.call.assert_awaited_once()
+    call_args = t._o.call.call_args[0]
+    assert call_args[0] == 'crm.lead'
+    assert call_args[1] == 'message_post'