302 lines
12 KiB
Python
302 lines
12 KiB
Python
"""Unit tests for ElearningTools."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, call
|
|
from agent_service.tools.elearning_tools import ElearningTools
|
|
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='slide.channel', record_id=10, action='create'))
|
|
odoo.call = AsyncMock(return_value=True)
|
|
return ElearningTools(odoo)
|
|
|
|
|
|
# ── get_courses ───────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_courses_default():
|
|
t = _make_tools()
|
|
result = await t.get_courses()
|
|
t._o.search_read.assert_awaited_once()
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('active', '=', True) in domain
|
|
assert isinstance(result, list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_courses_inactive():
|
|
t = _make_tools()
|
|
await t.get_courses(active=False)
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('active', '=', False) in domain
|
|
|
|
|
|
# ── get_course_stats ──────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_course_stats_found():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(side_effect=[
|
|
[{'name': 'Python 101', 'total_slides': 10, 'members_count': 5,
|
|
'completion_rate': 80.0, 'total_time': 300}], # channel
|
|
[{'name': 'Slide 1', 'slide_type': 'video', 'completion_rate': 90.0,
|
|
'likes': 3, 'dislikes': 0, 'view_count': 50}], # slides
|
|
])
|
|
result = await t.get_course_stats(channel_id=1)
|
|
assert 'channel' in result
|
|
assert result['slide_count'] == 1
|
|
assert 'avg_slide_completion' in result
|
|
assert 'total_views' in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_course_stats_not_found():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[])
|
|
result = await t.get_course_stats(channel_id=999)
|
|
assert result == {}
|
|
|
|
|
|
# ── get_enrolled_users ────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_enrolled_users():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'partner_id': [1, 'Alice'], 'completion': 80, 'channel_completion': 80.0}
|
|
])
|
|
result = await t.get_enrolled_users(channel_id=1)
|
|
assert len(result) == 1
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('channel_id', '=', 1) in domain
|
|
|
|
|
|
# ── get_slide_completion ──────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_slide_completion_default():
|
|
t = _make_tools()
|
|
await t.get_slide_completion(channel_id=1)
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('channel_id', '=', 1) in domain
|
|
assert ('channel_completion', '>=', 0.0) in domain
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_slide_completion_min_filter():
|
|
t = _make_tools()
|
|
await t.get_slide_completion(channel_id=1, min_completion=50.0)
|
|
domain = t._o.search_read.call_args[0][1]
|
|
assert ('channel_completion', '>=', 50.0) in domain
|
|
|
|
|
|
# ── get_course_details ────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_course_details_combines_data():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(side_effect=[
|
|
[{'name': 'Python 101', 'total_slides': 5, 'members_count': 10,
|
|
'completion_rate': 70.0, 'total_time': 120}],
|
|
[{'name': 'Slide A', 'slide_type': 'video', 'completion_rate': 80.0,
|
|
'likes': 1, 'dislikes': 0, 'view_count': 20}],
|
|
[{'partner_id': [1, 'Alice'], 'completion': 80, 'channel_completion': 80.0}],
|
|
[{'partner_id': [1, 'Alice'], 'channel_completion': 80.0}],
|
|
])
|
|
result = await t.get_course_details(channel_id=1)
|
|
assert 'channel' in result
|
|
assert 'enrolled_users' in result
|
|
assert 'slide_completion' in result
|
|
|
|
|
|
# ── get_learning_summary ──────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_learning_summary_structure():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'name': 'Course A', 'members_count': 10, 'completion_rate': 80.0},
|
|
{'name': 'Course B', 'members_count': 5, 'completion_rate': 20.0},
|
|
])
|
|
result = await t.get_learning_summary()
|
|
assert result['total_courses'] == 2
|
|
assert result['total_enrollments'] == 15
|
|
assert 'avg_completion' in result
|
|
assert 'low_completion_courses' in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_learning_summary_flags_low_completion():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[
|
|
{'name': 'Bad Course', 'members_count': 5, 'completion_rate': 10.0},
|
|
])
|
|
result = await t.get_learning_summary()
|
|
assert len(result['low_completion_courses']) == 1
|
|
|
|
|
|
# ── create_course ─────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_course_success():
|
|
t = _make_tools()
|
|
result = await t.create_course(name='New Course')
|
|
assert result['success'] is True
|
|
assert result['id'] == 10
|
|
assert result['name'] == 'New Course'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_course_failure():
|
|
t = _make_tools()
|
|
t._o.create = AsyncMock(return_value=WriteResult(
|
|
success=False, model='', record_id=None, action='create', error='DB error'))
|
|
result = await t.create_course(name='Bad Course')
|
|
assert result['success'] is False
|
|
assert 'error' in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_course_with_options():
|
|
t = _make_tools()
|
|
await t.create_course(name='Paid Course', enroll_policy='paid', website_published=True)
|
|
vals = t._o.create.call_args[0][1]
|
|
assert vals['enroll_policy'] == 'paid'
|
|
assert vals['website_published'] is True
|
|
|
|
|
|
# ── update_course ─────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_course_name():
|
|
t = _make_tools()
|
|
result = await t.update_course(channel_id=1, name='Updated Name')
|
|
assert result['success'] is True
|
|
t._o.write.assert_awaited_once_with('slide.channel', [1], {'name': 'Updated Name'})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_course_publish():
|
|
t = _make_tools()
|
|
await t.update_course(channel_id=1, website_published=True)
|
|
vals = t._o.write.call_args[0][2]
|
|
assert vals['website_published'] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_course_no_values():
|
|
t = _make_tools()
|
|
result = await t.update_course(channel_id=1)
|
|
assert result['success'] is False
|
|
assert 'error' in result
|
|
|
|
|
|
# ── add_section ───────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_section_success():
|
|
t = _make_tools()
|
|
t._o.create = AsyncMock(return_value=WriteResult(
|
|
success=True, model='slide.slide', record_id=20, action='create'))
|
|
result = await t.add_section(channel_id=1, name='Section 1', sequence=5)
|
|
assert result['success'] is True
|
|
assert result['id'] == 20
|
|
vals = t._o.create.call_args[0][1]
|
|
assert vals['is_category'] is True
|
|
assert vals['sequence'] == 5
|
|
|
|
|
|
# ── create_slide ──────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_slide_success():
|
|
t = _make_tools()
|
|
t._o.create = AsyncMock(return_value=WriteResult(
|
|
success=True, model='slide.slide', record_id=30, action='create'))
|
|
result = await t.create_slide(channel_id=1, name='My Slide', slide_type='video')
|
|
assert result['success'] is True
|
|
assert result['id'] == 30
|
|
assert result['slide_type'] == 'video'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_slide_with_content():
|
|
t = _make_tools()
|
|
await t.create_slide(channel_id=1, name='HTML Slide', html_content='<p>hello</p>')
|
|
vals = t._o.create.call_args[0][1]
|
|
assert vals['html_content'] == '<p>hello</p>'
|
|
|
|
|
|
# ── enroll_user ───────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enroll_user_new():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[])
|
|
t._o.create = AsyncMock(return_value=WriteResult(
|
|
success=True, model='slide.channel.partner', record_id=5, action='create'))
|
|
result = await t.enroll_user(channel_id=1, partner_id=10)
|
|
assert result['success'] is True
|
|
t._o.create.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enroll_user_already_enrolled():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(return_value=[{'id': 5}])
|
|
result = await t.enroll_user(channel_id=1, partner_id=10)
|
|
assert result['success'] is True
|
|
assert result['already_enrolled'] is True
|
|
t._o.create.assert_not_awaited()
|
|
|
|
|
|
# ── suggest_next_course ───────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_suggest_next_course_excludes_completed():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(side_effect=[
|
|
[{'channel_id': [1, 'Python 101']}], # completed
|
|
[{'id': 2, 'name': 'Advanced Python'}], # suggestions
|
|
])
|
|
result = await t.suggest_next_course(partner_id=5)
|
|
domain = t._o.search_read.call_args_list[1][0][1]
|
|
assert ('id', 'not in', [1]) in domain
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_suggest_next_course_no_completed():
|
|
t = _make_tools()
|
|
t._o.search_read = AsyncMock(side_effect=[
|
|
[],
|
|
[{'id': 1, 'name': 'Python 101'}],
|
|
])
|
|
result = await t.suggest_next_course(partner_id=5)
|
|
assert isinstance(result, list)
|
|
|
|
|
|
# ── post_chatter_note ─────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_chatter_note():
|
|
t = _make_tools()
|
|
result = await t.post_chatter_note('slide.channel', 1, 'Test note')
|
|
assert result is True
|
|
t._o.call.assert_awaited_once()
|
|
|
|
|
|
# ── flag_low_completion ───────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flag_low_completion():
|
|
t = _make_tools()
|
|
result = await t.flag_low_completion(channel_id=1, reason='Only 10% completion')
|
|
assert result is True
|
|
call_kwargs = t._o.call.call_args[0][3]
|
|
assert '[AI FLAG]' in call_kwargs['body']
|