oca/oca-ocb-mail
2
0
Fork 0
mirror of https://github.com/bringout/oca-ocb-mail.git synced 2026-04-22 10:42:00 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

19.0 vanilla

Browse source
This commit is contained in:
Ernad Husremovic 2026-03-09 09:31:39 +01:00
parent 5df8c07b59
commit daa394e8b0
2114 changed files with 564841 additions and 299642 deletions
Download patch file Download diff file Expand all files Collapse all files

16
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/__init__.py
View file

@ -2,9 +2,23 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from . import chatbot_common
from . import test_call
from . import test_chatbot_form_ui
from . import test_chatbot_internals
from . import test_get_mail_channel
from . import test_digest
from . import test_discuss_channel
from . import test_cors_livechat
from . import test_get_discuss_channel
from . import test_get_operator
from . import test_im_livechat_calls
from . import test_im_livechat_channel
from . import test_im_livechat_report
from . import test_im_livechat_support_page
from . import test_js
from . import test_member_history
from . import test_message
from . import test_upload_attachment
from . import test_transcript
from . import test_res_users
from . import test_session_views
from . import test_user_livechat_username

54
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/chatbot_common.py
View file

@ -1,14 +1,14 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.tests import common
class ChatbotCase(common.TransactionCase):
class ChatbotCase(common.HttpCase):
@classmethod
def setUpClass(cls):
super(ChatbotCase, cls).setUpClass()
cls.maxDiff = None
cls.chatbot_script = cls.env['chatbot.script'].create({
'title': 'Testing Bot',
@ -38,8 +38,9 @@ class ChatbotCase(common.TransactionCase):
cls.step_dispatch_buy_software,
cls.step_dispatch_pricing,
cls.step_dispatch_operator,
cls.step_dispatch_documentation,
] = cls.env['chatbot.script.answer'].sudo().create([{
'name': 'I want to buy the software',
'name': 'I\'d like to buy the software',
'script_step_id': cls.step_dispatch.id,
}, {
'name': 'Pricing Question',
@ -47,6 +48,9 @@ class ChatbotCase(common.TransactionCase):
}, {
'name': "I want to speak with an operator",
'script_step_id': cls.step_dispatch.id,
}, {
'name': "Other & Documentation",
'script_step_id': cls.step_dispatch.id,
}])
[
@ -56,6 +60,7 @@ class ChatbotCase(common.TransactionCase):
cls.step_forward_operator,
cls.step_no_one_available,
cls.step_no_operator_dispatch,
cls.step_documentation_validated,
] = ChatbotScriptStep.create([{
'step_type': 'text',
'message': 'For any pricing question, feel free ton contact us at pricing@mycompany.com',
@ -86,6 +91,11 @@ class ChatbotCase(common.TransactionCase):
'message': 'So... What can I do to help you?',
'triggering_answer_ids': [(4, cls.step_dispatch_operator.id)],
'chatbot_script_id': cls.chatbot_script.id,
}, {
'step_type': 'text',
'message': 'Please find documentation at https://www.odoo.com/documentation/latest/',
'triggering_answer_ids': [(4, cls.step_dispatch_documentation.id)],
'chatbot_script_id': cls.chatbot_script.id,
}])
cls.step_no_operator_just_leaving = cls.env['chatbot.script.answer'].sudo().create({
@ -133,13 +143,33 @@ class ChatbotCase(common.TransactionCase):
})]
})
@classmethod
def _post_answer_and_trigger_next_step(cls, mail_channel, answer, chatbot_script_answer=False):
mail_message = mail_channel.message_post(body=answer)
def _post_answer_and_trigger_next_step(
self, discuss_channel, body=None, email=None, chatbot_script_answer=None
):
data = self.make_jsonrpc_request(
"/mail/message/post",
{
"thread_model": "discuss.channel",
"thread_id": discuss_channel.id,
"post_data": {
"body": body or email or chatbot_script_answer.name,
"message_type": "comment",
"subtype_xmlid": "mail.mt_comment",
},
},
)
if email:
self.make_jsonrpc_request(
"/chatbot/step/validate_email", {"channel_id": discuss_channel.id}
)
if chatbot_script_answer:
cls.env['chatbot.message'].search([
('mail_message_id', '=', mail_message.id)
], limit=1).user_script_answer_id = chatbot_script_answer.id
# sudo: chatbot.script.step - members of a channel can access the current chatbot step
next_step = mail_channel.chatbot_current_step_id.sudo()._process_answer(mail_channel, mail_message.body)
next_step._process_step(mail_channel)
message = self.env["mail.message"].browse(data["message_id"])
self.make_jsonrpc_request(
"/chatbot/answer/save",
{
"channel_id": discuss_channel.id,
"message_id": message.id,
"selected_answer_id": chatbot_script_answer.id,
},
)
self.make_jsonrpc_request("/chatbot/step/trigger", {"channel_id": discuss_channel.id})

66
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/common.py
View file

@ -1,18 +1,23 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.tests.common import TransactionCase
from odoo import Command, fields
from odoo.tests.common import HttpCase, new_test_user
from odoo.addons.bus.tests.common import BusCase
class TestImLivechatCommon(TransactionCase):
class TestImLivechatCommon(HttpCase, BusCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.password = 'Pl1bhD@2!kXZ'
cls.operators = cls.env['res.users'].create([{
'name': 'Michel',
'login': 'michel',
'password': cls.password,
'livechat_username': "Michel Operator",
'email': 'michel@example.com',
'group_ids': cls.env.ref('im_livechat.im_livechat_group_user'),
}, {
'name': 'Paul',
'login': 'paul'
@ -42,7 +47,58 @@ class TestImLivechatCommon(TransactionCase):
def setUp(self):
super().setUp()
def get_available_users(_):
return self.operators
def _compute_available_operator_ids(channel_self):
for record in channel_self:
record.available_operator_ids = record.user_ids
self.patch(type(self.env['im_livechat.channel']), '_get_available_users', get_available_users)
self.patch(type(self.env['im_livechat.channel']), '_compute_available_operator_ids', _compute_available_operator_ids)
class TestGetOperatorCommon(HttpCase):
def setUp(self):
super().setUp()
self.operator_id = 0
def _create_conversation(self, livechat, operator, in_call=False):
channel = self.env["discuss.channel"].create(
{
"name": "Visitor 1",
"channel_type": "livechat",
"livechat_channel_id": livechat.id,
"livechat_operator_id": operator.partner_id.id,
"channel_member_ids": [Command.create({"partner_id": operator.partner_id.id})],
"last_interest_dt": fields.Datetime.now(),
}
)
channel.with_user(operator).message_post(body="Hello, how can I help you?")
if in_call:
member = self.env["discuss.channel.member"].search(
[("partner_id", "=", operator.partner_id.id), ("channel_id", "=", channel.id)]
)
self.env["discuss.channel.rtc.session"].sudo().create(
{"channel_id": channel.id, "channel_member_id": member.id}
)
return channel
def _create_operator(self, lang_code=None, country_code=None, expertises=None):
self.env["res.lang"].with_context(active_test=False).search(
[("code", "=", lang_code)]
).sudo().active = True
operator = new_test_user(
self.env(su=True),
login=f"operator_{lang_code or country_code}_{self.operator_id}",
groups="im_livechat.im_livechat_group_user",
)
operator.res_users_settings_id.livechat_expertise_ids = expertises
operator.partner_id = self.env["res.partner"].create(
{
"name": f"Operator {lang_code or country_code}",
"lang": lang_code,
"country_id": self.env["res.country"].search([("code", "=", country_code)]).id
if country_code
else None,
}
)
self.env["mail.presence"]._update_presence(operator)
self.operator_id += 1
return operator

39
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_call.py Normal file
View file

@ -0,0 +1,39 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.tests.common import new_test_user, tagged, HttpCase, JsonRpcException
@tagged("post_install", "-at_install")
class TestCall(HttpCase):
def test_visitor_cannot_start_call(self):
self.authenticate(None, None)
operator = self.env["res.users"].create({"name": "Operator", "login": "operator"})
self.env["mail.presence"]._update_presence(operator)
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Test Livechat Channel", "user_ids": [operator.id]}
)
for pseudo_user in [
new_test_user(self.env, "portal_user", groups="base.group_portal"),
self.env["mail.guest"].create({"name": "Guest"}),
]:
user = pseudo_user if pseudo_user._name == "res.users" else self.env["res.users"]
guest = pseudo_user if pseudo_user._name == "mail.guest" else self.env["mail.guest"]
if user:
self.authenticate(user.login, user.login)
else:
self.authenticate(None, None)
if guest:
self.opener.cookies[guest._cookie_name] = guest._format_auth_cookie()
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"anonymous_name": "Visitor",
"channel_id": livechat_channel.id,
"persisted": True,
},
)
with self.assertRaises(JsonRpcException, msg="werkzeug.exceptions.NotFound"):
self.make_jsonrpc_request(
"/mail/rtc/channel/join_call",
{"channel_id": data["channel_id"]},
)

26
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_chatbot_form_ui.py
View file

@ -5,28 +5,27 @@ from odoo import tests
from odoo.addons.base.tests.common import HttpCaseWithUserDemo
@tests.tagged('post_install', '-at_install')
@tests.tagged("post_install", "-at_install")
class TestLivechatChatbotFormUI(HttpCaseWithUserDemo):
def test_chatbot_steps_sequence_ui(self):
""" As sequences are *critical* for the chatbot_script script, let us a run a little tour that
creates a few steps, then verify sequences are properly applied. """
self.start_tour(
'/web',
'/odoo',
'im_livechat_chatbot_steps_sequence_tour',
login='admin',
step_delay=1000
)
chatbot_script = self.env['chatbot.script'].search([('title', '=', 'Test Chatbot Sequence')])
self.assertEqual(len(chatbot_script.script_step_ids), 3)
self.assertEqual(chatbot_script.script_step_ids[0].message, 'Step 1')
self.assertEqual(chatbot_script.script_step_ids[0].message, "<p>Step 1</p>")
self.assertEqual(chatbot_script.script_step_ids[0].sequence, 0)
self.assertEqual(chatbot_script.script_step_ids[1].message, 'Step 2')
self.assertEqual(chatbot_script.script_step_ids[1].message, "<p>Step 2</p>")
self.assertEqual(chatbot_script.script_step_ids[1].sequence, 1)
self.assertEqual(chatbot_script.script_step_ids[2].message, 'Step 3')
self.assertEqual(chatbot_script.script_step_ids[2].message, "<p>Step 3</p>")
self.assertEqual(chatbot_script.script_step_ids[2].sequence, 2)
def test_chatbot_steps_sequence_with_move_ui(self):
@ -36,10 +35,9 @@ class TestLivechatChatbotFormUI(HttpCaseWithUserDemo):
move records around. """
self.start_tour(
'/web',
'/odoo',
'im_livechat_chatbot_steps_sequence_with_move_tour',
login='admin',
step_delay=1000
)
chatbot_script = self.env['chatbot.script'].search([('title', '=', 'Test Chatbot Sequence')])
@ -49,15 +47,15 @@ class TestLivechatChatbotFormUI(HttpCaseWithUserDemo):
# during the test, we create the steps normally and then move 'Step 5'
# in second position -> check order is correct
self.assertEqual(chatbot_script.script_step_ids[0].message, 'Step 1')
self.assertEqual(chatbot_script.script_step_ids[0].message, "<p>Step 1</p>")
self.assertEqual(chatbot_script.script_step_ids[0].sequence, 0)
self.assertEqual(chatbot_script.script_step_ids[1].message, 'Step 5')
self.assertEqual(chatbot_script.script_step_ids[1].message, "<p>Step 5</p>")
self.assertEqual(chatbot_script.script_step_ids[1].sequence, 1)
self.assertEqual(chatbot_script.script_step_ids[2].message, 'Step 2')
self.assertEqual(chatbot_script.script_step_ids[2].message, "<p>Step 2</p>")
self.assertEqual(chatbot_script.script_step_ids[2].sequence, 2)
self.assertEqual(chatbot_script.script_step_ids[3].message, 'Step 3')
self.assertEqual(chatbot_script.script_step_ids[3].message, "<p>Step 3</p>")
self.assertEqual(chatbot_script.script_step_ids[3].sequence, 3)
self.assertEqual(chatbot_script.script_step_ids[4].message, 'Step 4')
self.assertEqual(chatbot_script.script_step_ids[4].message, "<p>Step 4</p>")
self.assertEqual(chatbot_script.script_step_ids[4].sequence, 4)
self.assertEqual(chatbot_script.script_step_ids[5].message, 'Step 6')
self.assertEqual(chatbot_script.script_step_ids[5].message, "<p>Step 6</p>")
self.assertEqual(chatbot_script.script_step_ids[5].sequence, 5)

467
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_chatbot_internals.py
View file

@ -1,11 +1,17 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from freezegun import freeze_time
from odoo import Command, fields
from odoo.addons.im_livechat.tests import chatbot_common
from odoo.exceptions import ValidationError
from odoo.tests.common import JsonRpcException, new_test_user, tagged
from odoo.tools.misc import mute_logger
from odoo.addons.mail.tests.common import freeze_all_time, MailCommon
from odoo.addons.mail.tools.discuss import Store
class ChatbotCase(chatbot_common.ChatbotCase):
@tagged("post_install", "-at_install")
class ChatbotCase(MailCommon, chatbot_common.ChatbotCase):
def test_chatbot_duplicate(self):
""" In this test we make sure that 'triggering_answer_ids' are correctly duplicated and
@ -27,12 +33,12 @@ class ChatbotCase(chatbot_common.ChatbotCase):
self.assertNotEqual(step_email_copy, self.step_email)
self.assertEqual(len(step_email_copy.triggering_answer_ids), 1)
self.assertEqual(step_email_copy.triggering_answer_ids.name, 'I want to buy the software')
self.assertEqual(step_email_copy.triggering_answer_ids.name, 'I\'d like to buy the software')
self.assertNotEqual(step_email_copy.triggering_answer_ids, self.step_dispatch_buy_software)
def test_chatbot_is_forward_operator_child(self):
self.assertEqual([step.is_forward_operator_child for step in self.chatbot_script.script_step_ids],
[False, False, False, False, False, False, False, True, True, True, False, False, False, False],
[False, False, False, False, False, False, False, True, True, False, True, False, False, False, False],
"Steps 'step_no_one_available', 'step_no_operator_dispatch', 'step_just_leaving'"
"should be flagged as forward operator child.")
@ -40,28 +46,28 @@ class ChatbotCase(chatbot_common.ChatbotCase):
self.chatbot_script.script_step_ids.invalidate_recordset(['is_forward_operator_child'])
self.assertEqual([step.is_forward_operator_child for step in self.chatbot_script.script_step_ids],
[False, False, False, False, False, False, False, True, False, False, False, False, False, False],
[False, False, False, False, False, False, False, True, False, False, False, False, False, False, False],
"Only step 'step_no_one_available' should be flagged as forward operator child.")
def test_chatbot_steps(self):
channel_info = self.livechat_channel._open_livechat_mail_channel(
anonymous_name='Test Visitor', chatbot_script=self.chatbot_script)
mail_channel = self.env['mail.channel'].browse(channel_info['id'])
data = self.make_jsonrpc_request("/im_livechat/get_session", {
'chatbot_script_id': self.chatbot_script.id,
'channel_id': self.livechat_channel.id,
})
discuss_channel = self.env["discuss.channel"].browse(data["channel_id"])
self.assertEqual(mail_channel.chatbot_current_step_id, self.step_dispatch)
self.assertEqual(discuss_channel.chatbot_current_step_id, self.step_dispatch)
self._post_answer_and_trigger_next_step(
mail_channel,
self.step_dispatch_buy_software.name,
chatbot_script_answer=self.step_dispatch_buy_software
discuss_channel, chatbot_script_answer=self.step_dispatch_buy_software
)
self.assertEqual(mail_channel.chatbot_current_step_id, self.step_email)
self.assertEqual(discuss_channel.chatbot_current_step_id, self.step_email)
with self.assertRaises(ValidationError, msg="Should raise an error since it's not a valid email"):
self._post_answer_and_trigger_next_step(mail_channel, 'test')
with self.assertRaises(JsonRpcException, msg='odoo.exceptions.ValidationError'), mute_logger("odoo.http"):
self._post_answer_and_trigger_next_step(discuss_channel, email="test")
self._post_answer_and_trigger_next_step(mail_channel, 'test@example.com')
self.assertEqual(mail_channel.chatbot_current_step_id, self.step_email_validated)
self._post_answer_and_trigger_next_step(discuss_channel, email="test@example.com")
self.assertEqual(discuss_channel.chatbot_current_step_id, self.step_email_validated)
def test_chatbot_steps_sequence(self):
""" Ensure sequence is correct when creating chatbots and adding steps to an existing one.
@ -105,3 +111,428 @@ class ChatbotCase(chatbot_common.ChatbotCase):
welcome_steps = self.chatbot_script._get_welcome_steps()
self.assertEqual(len(welcome_steps), 1)
self.assertEqual(welcome_steps, self.chatbot_script.script_step_ids[0])
def test_chatbot_not_invited_to_rtc_calls(self):
with freeze_all_time():
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"channel_id": self.livechat_channel.id,
"chatbot_script_id": self.chatbot_script.id,
},
)
discuss_channel = (
self.env["discuss.channel"].sudo().browse(data["channel_id"])
)
self.assertEqual(discuss_channel.livechat_operator_id, self.chatbot_script.operator_partner_id)
discuss_channel._add_members(users=self.env.user)
self_member = discuss_channel.channel_member_ids.filtered(lambda m: m.is_self)
bot_member = discuss_channel.channel_member_ids.filtered(
lambda m: m.partner_id == self.chatbot_script.operator_partner_id
)
guest_member = discuss_channel.channel_member_ids.filtered(lambda m: bool(m.guest_id))
self.env["mail.presence"]._update_presence(guest_member.guest_id)
self_member._rtc_join_call()
self.assertTrue(guest_member.rtc_inviting_session_id)
self.assertFalse(bot_member.rtc_inviting_session_id)
@freeze_time("2020-03-22 10:42:06")
def test_forward_to_specific_operator(self):
"""Test _forward_operator takes into account the given users as candidates."""
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"channel_id": self.livechat_channel.id,
"chatbot_script_id": self.chatbot_script.id,
},
)
discuss_channel = (
self.env["discuss.channel"].sudo().browse(data["channel_id"])
)
discuss_channel._forward_human_operator(self.step_forward_operator)
self.assertEqual(
discuss_channel.livechat_operator_id, self.chatbot_script.operator_partner_id
)
self.assertEqual(discuss_channel.name, "Testing Bot")
member_bot = discuss_channel.channel_member_ids.filtered(
lambda m: m.partner_id == self.chatbot_script.operator_partner_id
)
member_bot_data = {
"create_date": fields.Datetime.to_string(member_bot.create_date),
"fetched_message_id": False,
"id": member_bot.id,
"livechat_member_type": "bot",
"last_seen_dt": False,
"partner_id": member_bot.partner_id.id,
"seen_message_id": False,
"channel_id": {"id": discuss_channel.id, "model": "discuss.channel"},
}
def get_forward_op_bus_params():
messages = self.env["mail.message"].search([], order="id desc", limit=3)
# only data relevant to the test are asserted for simplicity
transfer_message_data = Store(bus_channel=discuss_channel).add(messages[1]).get_result()
transfer_message_data["mail.message"][0].update(
{
"author_id": self.chatbot_script.operator_partner_id.id,
"body": ["markup", "<p>I will transfer you to a human.</p>"],
# thread not renamed yet at this step
"default_subject": "Testing Bot",
"record_name": "Testing Bot",
}
)
transfer_message_data["mail.thread"][0]["display_name"] = "Testing Bot"
joined_message_data = Store(bus_channel=discuss_channel).add(messages[0]).get_result()
joined_message_data["mail.message"][0].update(
{
"author_id": self.chatbot_script.operator_partner_id.id,
"body": [
"markup",
(
'<div class="o_mail_notification" data-oe-type="channel-joined">invited <a href="#" data-oe-model="res.partner" data-oe-id="'
f'{self.partner_employee.id}">@Ernest Employee</a> to the channel</div>'
),
],
# thread not renamed yet at this step
"default_subject": "Testing Bot",
"record_name": "Testing Bot",
}
)
joined_message_data["mail.thread"][0]["display_name"] = "Testing Bot"
member_emp = discuss_channel.channel_member_ids.filtered(
lambda m: m.partner_id == self.partner_employee
)
# data in-between join and leave
channel_data_join = (
Store(bus_channel=member_emp._bus_channel()).add(discuss_channel).get_result()
)
channel_data_join["discuss.channel"][0]["invited_member_ids"] = [["ADD", []]]
channel_data_join["discuss.channel"][0]["rtc_session_ids"] = [["ADD", []]]
channel_data_join["discuss.channel"][0]["livechat_outcome"] = "no_agent"
channel_data_join["discuss.channel"][0]["chatbot"]["currentStep"]["message"] = messages[1].id
channel_data_join["discuss.channel"][0]["chatbot"]["steps"][0]["message"] = messages[1].id
channel_data_join["discuss.channel"][0]["livechat_operator_id"] = self.chatbot_script.operator_partner_id.id
channel_data_join["discuss.channel"][0]["member_count"] = 3
channel_data_join["discuss.channel"][0]["name"] = "Testing Bot"
channel_data_join["discuss.channel.member"].insert(0, member_bot_data)
channel_data_join["discuss.channel.member"][2]["fetched_message_id"] = False
channel_data_join["discuss.channel.member"][2]["last_seen_dt"] = False
channel_data_join["discuss.channel.member"][2]["seen_message_id"] = False
channel_data_join["discuss.channel.member"][2]["unpin_dt"] = False
del channel_data_join["res.partner"][1]
channel_data_join["res.partner"].insert(
0,
{
"active": False,
"avatar_128_access_token": self.chatbot_script.operator_partner_id._get_avatar_128_access_token(),
"country_id": False,
"id": self.chatbot_script.operator_partner_id.id,
"im_status": "im_partner",
"im_status_access_token": self.chatbot_script.operator_partner_id._get_im_status_access_token(),
"is_public": False,
"mention_token": self.chatbot_script.operator_partner_id._get_mention_token(),
"name": "Testing Bot",
"user_livechat_username": False,
"write_date": fields.Datetime.to_string(
self.chatbot_script.operator_partner_id.write_date
),
},
)
channel_data = Store().add(discuss_channel).get_result()
channel_data["discuss.channel"][0]["message_needaction_counter_bus_id"] = 0
channel_data_emp = Store().add(discuss_channel.with_user(self.user_employee)).get_result()
channel_data_emp["discuss.channel"][0]["message_needaction_counter_bus_id"] = 0
channel_data_emp["discuss.channel.member"][1]["message_unread_counter_bus_id"] = 0
channel_data = Store().add(discuss_channel).get_result()
channel_data["discuss.channel"][0]["message_needaction_counter_bus_id"] = 0
channels, message_items = (
[
(self.cr.dbname, "discuss.channel", discuss_channel.id),
(self.cr.dbname, "res.partner", self.partner_employee.id),
(self.cr.dbname, "discuss.channel", discuss_channel.id),
(self.cr.dbname, "discuss.channel", discuss_channel.id),
(self.cr.dbname, "discuss.channel", discuss_channel.id),
(self.cr.dbname, "discuss.channel", discuss_channel.id),
(self.cr.dbname, "discuss.channel", discuss_channel.id),
(self.cr.dbname, "res.partner", self.partner_employee.id),
(self.cr.dbname, "res.partner", self.env.user.partner_id.id),
],
[
{
"type": "discuss.channel/new_message",
"payload": {
"data": transfer_message_data,
"id": discuss_channel.id,
},
},
{
"type": "discuss.channel/joined",
"payload": {
"channel_id": discuss_channel.id,
"invite_to_rtc_call": False,
"data": channel_data_join,
"invited_by_user_id": self.env.user.id,
},
},
{
"type": "discuss.channel/new_message",
"payload": {
"data": joined_message_data,
"id": discuss_channel.id,
},
},
{
"type": "mail.record/insert",
"payload": {
"discuss.channel": [{"id": discuss_channel.id, "member_count": 3}],
"discuss.channel.member": [
{
"create_date": fields.Datetime.to_string(
member_emp.create_date
),
"fetched_message_id": False,
"id": member_emp.id,
"livechat_member_type": "agent",
"last_seen_dt": fields.Datetime.to_string(
member_emp.last_seen_dt
),
"partner_id": self.partner_employee.id,
"seen_message_id": False,
"channel_id": {
"id": discuss_channel.id,
"model": "discuss.channel",
},
}
],
"res.country": [
{"code": "BE", "id": self.env.ref("base.be").id, "name": "Belgium"}
],
"res.partner": self._filter_partners_fields(
{
"active": True,
"avatar_128_access_token": self.partner_employee._get_avatar_128_access_token(),
"country_id": self.env.ref("base.be").id,
"id": self.partner_employee.id,
"im_status": "offline",
"im_status_access_token": self.partner_employee._get_im_status_access_token(),
"is_public": False,
"mention_token": self.partner_employee._get_mention_token(),
"name": "Ernest Employee",
"user_livechat_username": False,
"write_date": fields.Datetime.to_string(
self.partner_employee.write_date
),
}
),
},
},
{
"type": "mail.record/insert",
"payload": {
"discuss.channel": [
{
"channel_member_ids": [["DELETE", [member_bot.id]]],
"id": discuss_channel.id,
"member_count": 2,
}
]
},
},
{
"type": "mail.record/insert",
"payload": {
"discuss.channel": [
{
"id": discuss_channel.id,
"livechat_operator_id": self.partner_employee.id,
"name": "OdooBot Ernest Employee",
},
],
"res.partner": self._filter_partners_fields(
{
"avatar_128_access_token": self.partner_employee._get_avatar_128_access_token(),
"id": self.partner_employee.id,
"name": "Ernest Employee",
"user_livechat_username": False,
"write_date": fields.Datetime.to_string(
self.partner_employee.write_date
),
}
),
},
},
{"type": "mail.record/insert", "payload": channel_data_emp},
{"type": "mail.record/insert", "payload": channel_data},
],
)
return (channels, message_items)
with self.assertBus(get_params=get_forward_op_bus_params):
discuss_channel._forward_human_operator(self.step_forward_operator, users=self.user_employee)
self.assertEqual(discuss_channel.name, "OdooBot Ernest Employee")
self.assertEqual(discuss_channel.livechat_operator_id, self.partner_employee)
self.assertEqual(discuss_channel.livechat_outcome, "no_answer")
self.assertTrue(
discuss_channel.channel_member_ids.filtered(
lambda m: m.partner_id == self.partner_employee
and m.livechat_member_type == "agent"
)
)
def test_chatbot_multiple_rules_on_same_url(self):
bob_user = new_test_user(
self.env, login="bob_user", groups="im_livechat.im_livechat_group_user,base.group_user"
)
chatbot_no_operator = self.env["chatbot.script"].create(
{
"title": "Chatbot operators not available",
"script_step_ids": [
Command.create(
{
"step_type": "text",
"message": "I'm shown because there is no operator available",
}
)
],
}
)
chatbot_operator = self.env["chatbot.script"].create(
{
"title": "Chatbot operators available",
"script_step_ids": [
Command.create(
{
"step_type": "text",
"message": "I'm shown because there is an operator available",
}
)
],
}
)
self.livechat_channel.user_ids += bob_user
self.livechat_channel.rule_ids = self.env["im_livechat.channel.rule"].create(
[
{
"channel_id": self.livechat_channel.id,
"chatbot_script_id": chatbot_no_operator.id,
"chatbot_enabled_condition": "only_if_no_operator",
"regex_url": "/",
"sequence": 1,
},
{
"channel_id": self.livechat_channel.id,
"chatbot_script_id": chatbot_operator.id,
"regex_url": "/",
"sequence": 2,
},
]
)
self.assertFalse(self.livechat_channel.available_operator_ids)
self.assertEqual(
self.env["im_livechat.channel.rule"]
.match_rule(self.livechat_channel.id, "/")
.chatbot_script_id,
chatbot_no_operator,
)
self.env["mail.presence"]._update_presence(bob_user)
# Force the recomputation of `available_operator_ids` after bob becomes online
self.livechat_channel.invalidate_recordset(["available_operator_ids"])
self.assertTrue(self.livechat_channel.available_operator_ids)
self.assertEqual(
self.env["im_livechat.channel.rule"]
.match_rule(self.livechat_channel.id, "/")
.chatbot_script_id,
chatbot_operator,
)
def test_chatbot_enabled_condition(self):
cases = [
# condition - operator_available - expected_result
("only_if_no_operator", False, True),
("only_if_no_operator", True, False),
("only_if_operator", True, True),
("only_if_operator", False, False),
("always", False, True),
("always", True, True),
]
for condition, operator_available, expected_result in cases:
self.livechat_channel.user_ids.unlink()
if operator_available:
operator_user = new_test_user(
self.env,
login=f"operator_user_{condition}_{operator_available}_{expected_result}",
groups="im_livechat.im_livechat_group_user,base.group_user",
)
self.env["mail.presence"]._update_presence(operator_user)
self.livechat_channel.user_ids = operator_user
self.livechat_channel.rule_ids = self.env["im_livechat.channel.rule"].create(
{
"channel_id": self.livechat_channel.id,
"chatbot_script_id": self.chatbot_script.id,
"chatbot_enabled_condition": condition,
"regex_url": "/",
"sequence": 1,
}
)
matching_rule = (
self.env["im_livechat.channel.rule"].match_rule(self.livechat_channel.id, "/")
or self.env["im_livechat.channel.rule"]
)
self.assertEqual(
matching_rule.chatbot_script_id,
self.chatbot_script if expected_result else self.env["chatbot.script"],
f"Condition: {condition}, Operator available: {operator_available}, Expected result: {expected_result}",
)
def test_chatbot_member_type(self):
"""Ensure livechat_member_type are correctly set when using chatbot with a logged in user."""
self.authenticate(self.user_employee.login, self.user_employee.login)
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"chatbot_script_id": self.chatbot_script.id,
"channel_id": self.livechat_channel.id,
},
)
discuss_channel = self.env["discuss.channel"].browse(data["channel_id"])
self.assertEqual(
discuss_channel.channel_member_ids.mapped("livechat_member_type"),
["bot", "visitor"],
)
def test_chatbot_clear_answers_on_step_type_change(self):
chatbot = self.env['chatbot.script'].create({
'title': 'Clear Answer Test Bot',
'script_step_ids': [Command.create({
'step_type': 'question_selection',
'message': 'What do you want to do?',
'answer_ids': [
Command.create({'name': 'Buy'}),
Command.create({'name': 'Support'}),
]
})]
})
step = chatbot.script_step_ids[0]
answers = {a.name: a for a in step.answer_ids}
[step_2, step_3] = self.env['chatbot.script.step'].create([
{
'chatbot_script_id': chatbot.id,
'step_type': 'text',
'message': 'Great! Let me help you with buying.',
'sequence': 2,
'triggering_answer_ids': [Command.set(answers['Buy'].ids)],
},
{
'chatbot_script_id': chatbot.id,
'step_type': 'text',
'message': 'Sure! I can assist you with support.',
'sequence': 3,
'triggering_answer_ids': [Command.set(answers['Support'].ids)],
},
])
action = self.env.ref('im_livechat.chatbot_script_action')
self.start_tour(f"/odoo/action-{action.id}", 'change_chatbot_step_type', login='admin')
self.assertFalse(step.answer_ids, "Answers were not cleared after step_type was changed.")
self.assertFalse(step_2.triggering_answer_ids, "Step 2 still has stale triggering answers.")
self.assertFalse(step_3.triggering_answer_ids, "Step 3 still has stale triggering answers.")

84
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_cors_livechat.py Normal file
View file

@ -0,0 +1,84 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.tests import tagged, HttpCase, JsonRpcException
@tagged("post_install", "-at_install")
class TestCorsLivechat(HttpCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.operator = cls.env["res.users"].create(
{
"name": "Operator",
"login": "operator",
}
)
cls.env["mail.presence"]._update_presence(cls.operator)
cls.livechat_channel = cls.env["im_livechat.channel"].create(
{"name": "Test Livechat Channel", "user_ids": [cls.operator.id]}
)
def test_ignore_user_cookie(self):
self.authenticate("admin", "admin")
data = self.make_jsonrpc_request(
"/im_livechat/cors/get_session",
{
"channel_id": self.livechat_channel.id,
"persisted": True,
},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
self.assertEqual(channel.channel_member_ids[0].partner_id, self.operator.partner_id)
self.assertFalse(channel.channel_member_ids[1].partner_id)
self.assertTrue(channel.channel_member_ids[1].guest_id)
def test_ignore_guest_cookie(self):
guest = self.env["mail.guest"].create({"name": "Visitor"})
data = self.make_jsonrpc_request(
"/im_livechat/cors/get_session",
{
"channel_id": self.livechat_channel.id,
"persisted": True,
},
cookies={guest._cookie_name: f'{guest.id}{guest._cookie_separator}{guest.access_token}'}
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
channel_guest = channel.channel_member_ids.filtered(lambda member: member.guest_id).guest_id
self.assertNotEqual(channel_guest, guest)
def test_access_routes_with_valid_guest_token(self):
data = self.make_jsonrpc_request(
"/im_livechat/cors/get_session",
{
"channel_id": self.livechat_channel.id,
"persisted": True,
},
)
self.authenticate(None, None)
self.make_jsonrpc_request(
"/im_livechat/cors/channel/messages",
{
"guest_token": data["store_data"]["Store"]["guest_token"],
"channel_id": data["channel_id"],
},
)
def test_access_denied_for_wrong_channel(self):
data = self.make_jsonrpc_request(
"/im_livechat/cors/get_session",
{
"channel_id": self.livechat_channel.id,
"persisted": True,
},
)
guest = self.env["mail.guest"].create({"name": "Visitor"})
self.authenticate(None, None)
with self.assertRaises(JsonRpcException, msg="werkzeug.exceptions.NotFound"):
self.make_jsonrpc_request(
"/im_livechat/cors/channel/messages",
{
"guest_token": guest.access_token,
"channel_id": data["channel_id"],
},
)

63
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_digest.py Normal file
View file

@ -0,0 +1,63 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.addons.digest.tests.common import TestDigestCommon
from odoo.tools import mute_logger
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestLiveChatDigest(TestDigestCommon):
@classmethod
@mute_logger('odoo.models.unlink')
def setUpClass(cls):
super().setUpClass()
other_partner = cls.env['res.partner'].create({'name': 'Other Partner'})
cls.channels = cls.env['discuss.channel'].create([{
'name': 'Channel 1',
'livechat_operator_id': cls.env.user.partner_id.id,
'channel_type': 'livechat',
}, {
'name': 'Channel 2',
'livechat_operator_id': cls.env.user.partner_id.id,
'channel_type': 'livechat',
}, {
'name': 'Channel 3',
'livechat_operator_id': other_partner.id,
'channel_type': 'livechat',
}])
cls.env['rating.rating'].search([]).unlink()
cls.env['rating.rating'].create([{
'rated_partner_id': cls.env.user.partner_id.id,
'res_id': cls.channels[0].id,
'res_model_id': cls.env['ir.model']._get('discuss.channel').id,
'consumed': True,
'rating': 5,
}, {
'rated_partner_id': cls.env.user.partner_id.id,
'res_id': cls.channels[0].id,
'res_model_id': cls.env['ir.model']._get('discuss.channel').id,
'consumed': True,
'rating': 0,
}, {
'rated_partner_id': cls.env.user.partner_id.id,
'res_id': cls.channels[0].id,
'res_model_id': cls.env['ir.model']._get('discuss.channel').id,
'consumed': True,
'rating': 3,
}, {
'rated_partner_id': cls.env.user.partner_id.id,
'res_id': cls.channels[0].id,
'res_model_id': cls.env['ir.model']._get('discuss.channel').id,
'consumed': True,
'rating': 3,
}])
def test_kpi_livechat_rating_value(self):
# 1/3 of the ratings have 5/5 note (0 are ignored)
self.assertEqual(round(self.digest_1.kpi_livechat_rating_value, 2), 33.33)

322
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_discuss_channel.py Normal file
View file

@ -0,0 +1,322 @@
import json
from datetime import timedelta
from freezegun import freeze_time
from markupsafe import Markup
from odoo import Command, fields
from odoo.tests import new_test_user, tagged, users
from odoo.addons.im_livechat.tests.common import TestImLivechatCommon, TestGetOperatorCommon
from odoo.addons.mail.tests.common import MailCase
@tagged("-at_install", "post_install")
class TestDiscussChannel(TestImLivechatCommon, TestGetOperatorCommon, MailCase):
def test_unfollow_from_non_member_does_not_close_livechat(self):
bob_user = new_test_user(
self.env, "bob_user", groups="base.group_user,im_livechat.im_livechat_group_manager"
)
data = self.make_jsonrpc_request(
"/im_livechat/get_session", {"channel_id": self.livechat_channel.id}
)
chat = self.env["discuss.channel"].browse(data["channel_id"])
self.assertFalse(chat.livechat_end_dt)
chat.with_user(bob_user).action_unfollow()
self.assertFalse(chat.livechat_end_dt)
chat.with_user(chat.livechat_operator_id.main_user_id).action_unfollow()
self.assertTrue(chat.livechat_end_dt)
def test_human_operator_failure_states(self):
data = self.make_jsonrpc_request(
"/im_livechat/get_session", {"channel_id": self.livechat_channel.id}
)
chat = self.env["discuss.channel"].browse(data["channel_id"])
self.assertFalse(chat.chatbot_current_step_id) # assert there is no chatbot
self.assertEqual(chat.livechat_failure, "no_answer")
chat.with_user(chat.livechat_operator_id.main_user_id).message_post(
body="I am here to help!",
message_type="comment",
subtype_xmlid="mail.mt_comment",
)
self.assertEqual(chat.livechat_failure, "no_failure")
def test_chatbot_failure_states(self):
chatbot_script = self.env["chatbot.script"].create({"title": "Testing Bot"})
self.livechat_channel.rule_ids = [(0, 0, {"chatbot_script_id": chatbot_script.id})]
self.env["chatbot.script.step"].create({
"step_type": "forward_operator",
"message": "I will transfer you to a human.",
"chatbot_script_id": chatbot_script.id,
})
bob_operator = new_test_user(
self.env, "bob_user", groups="im_livechat.im_livechat_group_user"
)
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"chatbot_script_id": chatbot_script.id, "channel_id": self.livechat_channel.id},
)
chat = self.env["discuss.channel"].browse(data["channel_id"])
self.assertTrue(chat.chatbot_current_step_id) # assert there is a chatbot
self.assertEqual(chat.livechat_failure, "no_failure")
self.livechat_channel.user_ids = False # remove operators so forwarding will fail
chat._forward_human_operator(chat.chatbot_current_step_id)
self.assertEqual(chat.livechat_failure, "no_agent")
self.livechat_channel.user_ids += bob_operator
self.assertTrue(self.livechat_channel.available_operator_ids)
chat._forward_human_operator(chat.chatbot_current_step_id)
self.assertEqual(chat.livechat_operator_id, bob_operator.partner_id)
self.assertEqual(chat.livechat_failure, "no_answer")
chat.with_user(bob_operator).message_post(
body="I am here to help!",
message_type="comment",
subtype_xmlid="mail.mt_comment",
)
self.assertEqual(chat.livechat_failure, "no_failure")
def test_livechat_description_sync_to_internal_user_bus(self):
"""Test the description of a livechat conversation is sent to the internal user bus."""
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": self.livechat_channel.id},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
with self.assertBus(
[(self.cr.dbname, "discuss.channel", channel.id, "internal_users")],
[
{
"type": "mail.record/insert",
"payload": {
"discuss.channel": [
{
"id": channel.id,
"description": "Description of the conversation",
}
]
},
}
],
):
channel.description = "Description of the conversation"
def test_livechat_note_sync_to_internal_user_bus(self):
"""Test that a livechat note is sent to the internal user bus."""
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": self.livechat_channel.id},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
with self.assertBus(
[(self.cr.dbname, "discuss.channel", channel.id, "internal_users")],
[
{
"type": "mail.record/insert",
"payload": {
"discuss.channel": [
{
"id": channel.id,
"livechat_note": [
"markup",
"<p>This is a note for the internal user.</p>",
],
}
]
},
}
],
):
channel.livechat_note = "This is a note for the internal user."
def test_livechat_status_sync_to_internal_user_bus(self):
"""Test that a livechat status is sent to the internal user bus."""
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": self.livechat_channel.id},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
with self.assertBus(
[(self.cr.dbname, "discuss.channel", channel.id, "internal_users")],
[
{
"type": "mail.record/insert",
"payload": {
"discuss.channel": [
{
"id": channel.id,
"livechat_status": "waiting",
}
]
},
}
],
):
channel.livechat_status = "waiting"
def test_livechat_status_switch_on_operator_joined_batch(self):
"""Test that the livechat status switches to 'in_progress' when an operator joins multiple channels in a batch,
and ensure re-adding the same member does not change the status."""
channel_1 = self.env["discuss.channel"].create({
"name": "Livechat Channel 1",
"channel_type": "livechat",
"livechat_operator_id": self.operators[0].partner_id.id,
})
channel_2 = self.env["discuss.channel"].create({
"name": "Livechat Channel 2",
"channel_type": "livechat",
"livechat_operator_id": self.operators[0].partner_id.id,
})
bob_operator = new_test_user(self.env, "bob_user", groups="im_livechat.im_livechat_group_user")
channel_1.livechat_status = "need_help"
channel_2.livechat_status = "need_help"
self.assertEqual(channel_1.livechat_status, "need_help")
self.assertEqual(channel_2.livechat_status, "need_help")
self.assertFalse(channel_1.livechat_end_dt)
self.assertFalse(channel_2.livechat_end_dt)
# Add the operator to both channels in a batch, which should switch their status to 'in_progress'
(channel_1 | channel_2).with_user(channel_1.livechat_operator_id.main_user_id).add_members(
partner_ids=bob_operator.partner_id.ids
)
self.assertEqual(channel_1.livechat_status, "in_progress")
self.assertEqual(channel_2.livechat_status, "in_progress")
# Re-add the same operator and ensure the status does not change
channel_1.livechat_status = "need_help"
self.assertEqual(channel_1.livechat_status, "need_help")
channel_1.with_user(channel_1.livechat_operator_id.main_user_id).add_members(
partner_ids=bob_operator.partner_id.ids
)
self.assertEqual(channel_1.livechat_status, "need_help")
def test_join_livechat_needing_help(self):
bob = self._create_operator()
john = self._create_operator()
jane = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Livechat Channel", "user_ids": (bob + jane + john).ids},
)
chat = self._create_conversation(livechat_channel, bob)
chat.livechat_status = "need_help"
has_joined = chat.with_user(john).livechat_join_channel_needing_help()
self.assertTrue(has_joined)
self.assertIn(john.partner_id, chat.channel_member_ids.partner_id)
self.assertEqual(chat.livechat_status, "in_progress")
has_joined = chat.with_user(jane).livechat_join_channel_needing_help()
self.assertFalse(has_joined)
self.assertNotIn(jane.partner_id, chat.channel_member_ids.partner_id)
@users("michel")
def test_livechat_conversation_history(self):
"""Test livechat conversation history formatting"""
def _convert_attachment_to_html(attachment):
attachment_data = {
"id": attachment.id,
"access_token": attachment.access_token,
"checksum": attachment.checksum,
"extension": "txt",
"mimetype": attachment.mimetype,
"filename": attachment.display_name,
"url": attachment.url,
}
return Markup(
"<div data-embedded='file' data-oe-protected='true' contenteditable='false' data-embedded-props='%s'/>",
) % json.dumps({"fileData": attachment_data})
channel = self.env["discuss.channel"].create(
{
"name": "test",
"channel_type": "livechat",
"livechat_operator_id": self.operators[0].partner_id.id,
"channel_member_ids": [
Command.create({"partner_id": self.operators[0].partner_id.id}),
Command.create({"partner_id": self.visitor_user.partner_id.id}),
],
}
)
attachment1 = self.env["ir.attachment"].create({"name": "test.txt"})
attachment2 = self.env["ir.attachment"].with_user(self.visitor_user).create({"name": "test2.txt"})
channel.message_post(body="Operator Here", message_type="comment")
channel.message_post(body="", message_type="comment", attachment_ids=[attachment1.id])
channel.with_user(self.visitor_user).message_post(body="Visitor Here", message_type="comment")
channel.with_user(self.visitor_user).message_post(body="", message_type="comment", attachment_ids=[attachment2.id])
channel.message_post(body="Some notification", message_type="notification")
channel_history = channel.with_user(self.visitor_user)._get_channel_history()
self.assertEqual(
channel_history,
"<br/><strong>Michel Operator:</strong><br/>Operator Here<br/>%(attachment_1)s<br/>"
"<br/><strong>Rajesh:</strong><br/>Visitor Here<br/>%(attachment_2)s<br/>"
% {
"attachment_1": _convert_attachment_to_html(attachment1),
"attachment_2": _convert_attachment_to_html(attachment2),
},
)
def test_gc_bot_sessions_after_one_day_inactivity(self):
chatbot_script = self.env["chatbot.script"].create({"title": "Testing Bot"})
self.livechat_channel.rule_ids = [Command.create({"chatbot_script_id": chatbot_script.id})]
self.env["chatbot.script.step"].create({
"chatbot_script_id": chatbot_script.id,
"message": "Hello joey, how you doing?",
"step_type": "text",
})
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"anonymous_name": "Thomas",
"chatbot_script_id": chatbot_script.id,
"channel_id": self.livechat_channel.id,
},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
with freeze_time(fields.Datetime.to_string(fields.Datetime.now() + timedelta(hours=23))):
self.assertFalse(channel.livechat_end_dt)
with freeze_time(fields.Datetime.to_string(fields.Datetime.now() + timedelta(days=1))):
channel._gc_bot_only_ongoing_sessions()
self.assertTrue(channel.livechat_end_dt)
def test_expertises_added_from_discuss_are_kept(self):
bob = self._create_operator()
jane = self._create_operator()
dog_expertise = self.env["im_livechat.expertise"].create({"name": "Dog"})
operator_expertise_ids = dog_expertise
chatbot_script = self.env["chatbot.script"].create({"title": "Testing Bot"})
self.env["chatbot.script.step"].create(
[
{
"chatbot_script_id": chatbot_script.id,
"message": "Hello, how can I help you?",
"step_type": "free_input_single",
},
{
"chatbot_script_id": chatbot_script.id,
"operator_expertise_ids": operator_expertise_ids,
"step_type": "forward_operator",
},
]
)
self.livechat_channel.user_ids = jane
self.livechat_channel.rule_ids = [Command.create({"chatbot_script_id": chatbot_script.id})]
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"chatbot_script_id": chatbot_script.id,
"channel_id": self.livechat_channel.id,
},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
self.make_jsonrpc_request(
"/chatbot/step/trigger",
{"channel_id": channel.id, "chatbot_script_id": chatbot_script.id},
)
self.assertIn(jane.partner_id, channel.livechat_agent_history_ids.partner_id)
self.assertEqual(channel.livechat_expertise_ids, operator_expertise_ids)
cat_expertise = self.env["im_livechat.expertise"].create({"name": "Cat"})
self.authenticate(jane.login, jane.login)
self.make_jsonrpc_request(
"/im_livechat/conversation/write_expertises",
{
"channel_id": channel.id,
"orm_commands": [Command.link(cat_expertise.id)],
},
)
self.assertEqual(channel.livechat_expertise_ids, operator_expertise_ids | cat_expertise)
channel._add_members(users=bob)
self.assertEqual(channel.livechat_expertise_ids, operator_expertise_ids | cat_expertise)

435
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_get_discuss_channel.py Normal file
View file

@ -0,0 +1,435 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from datetime import timedelta
from freezegun import freeze_time
from unittest.mock import patch, PropertyMock
from odoo import fields
from odoo.addons.im_livechat.tests.common import TestImLivechatCommon
from odoo.addons.mail.tests.common import MailCommon
from odoo.tests import new_test_user, tagged
@tagged("post_install", "-at_install")
class TestGetDiscussChannel(TestImLivechatCommon, MailCommon):
def test_get_discuss_channel(self):
"""For a livechat with 5 available operators, we open 5 channels 5 times (25 channels total).
For every 5 channels opening, we check that all operators were assigned.
"""
for _i in range(5):
discuss_channels = self._open_livechat_discuss_channel()
channel_operator_ids = [
channel_info["livechat_operator_id"] for channel_info in discuss_channels
]
self.assertTrue(all(partner_id in channel_operator_ids for partner_id in self.operators.mapped('partner_id').ids))
def test_channel_get_livechat_visitor_info(self):
self.maxDiff = None
belgium = self.env.ref('base.be')
test_user = self.env['res.users'].create({'name': 'Roger', 'login': 'roger', 'password': self.password, 'country_id': belgium.id})
# ensure visitor info are correct with anonymous
operator = self.operators[0]
with patch('odoo.http.GeoIP.country_code', new_callable=PropertyMock(return_value=belgium.code)):
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"previous_operator_id": operator.partner_id.id,
"channel_id": self.livechat_channel.id,
},
)["store_data"]
channel_info = data["discuss.channel"][0]
self.assertEqual(channel_info["name"], "Visitor Michel Operator")
self.assertEqual(channel_info["country_id"], belgium.id)
self.assertEqual(data["res.country"], [{"code": "BE", "id": belgium.id, "name": "Belgium"}])
# ensure persona info are hidden (in particular email and real name when livechat username is present)
channel = self.env["discuss.channel"].browse(channel_info["id"])
guest = channel.channel_member_ids.guest_id[0]
self.assertEqual(
data["mail.guest"],
[
{
"avatar_128_access_token": guest._get_avatar_128_access_token(),
"country_id": belgium.id,
"id": guest.id,
"im_status": "offline",
"im_status_access_token": guest._get_im_status_access_token(),
"name": "Visitor",
"offline_since": False,
"write_date": fields.Datetime.to_string(guest.write_date),
},
],
)
self.assertEqual(
data["res.partner"],
self._filter_partners_fields(
{
"active": False,
"avatar_128_access_token": self.partner_root._get_avatar_128_access_token(),
"id": self.user_root.partner_id.id,
"im_status": "bot",
"im_status_access_token": self.partner_root._get_im_status_access_token(),
"is_company": False,
"main_user_id": self.user_root.id,
"name": "OdooBot",
"write_date": fields.Datetime.to_string(self.user_root.partner_id.write_date),
},
{
"active": True,
"avatar_128_access_token": operator.partner_id._get_avatar_128_access_token(),
"country_id": False,
"id": operator.partner_id.id,
"im_status": "offline",
"im_status_access_token": operator.partner_id._get_im_status_access_token(),
"is_public": False,
"mention_token": operator.partner_id._get_mention_token(),
"user_livechat_username": "Michel Operator",
"write_date": fields.Datetime.to_string(operator.write_date),
},
),
)
self.assertEqual(
data["res.users"],
self._filter_users_fields(
{
"id": self.user_root.id,
"partner_id": self.partner_root.id,
"share": False,
},
),
)
# ensure visitor info are correct with real user
self.authenticate(test_user.login, self.password)
data = self.make_jsonrpc_request('/im_livechat/get_session', {
'previous_operator_id': operator.partner_id.id,
'channel_id': self.livechat_channel.id,
})["store_data"]
channel_info = data["discuss.channel"][0]
self.assertEqual(channel_info["name"], "Roger Michel Operator")
self.assertEqual(channel_info["country_id"], belgium.id)
self.assertEqual(data["res.country"], [{"code": "BE", "id": belgium.id, "name": "Belgium"}])
operator_member_domain = [
('channel_id', '=', channel_info['id']),
('partner_id', '=', operator.partner_id.id),
]
operator_member = self.env['discuss.channel.member'].search(operator_member_domain)
visitor_member_domain = [
('channel_id', '=', channel_info['id']),
('partner_id', '=', test_user.partner_id.id),
]
visitor_member = self.env['discuss.channel.member'].search(visitor_member_domain)
self.assertEqual(
data["res.partner"],
self._filter_partners_fields(
{
"active": False,
"avatar_128_access_token": self.partner_root._get_avatar_128_access_token(),
"email": "odoobot@example.com",
"id": self.user_root.partner_id.id,
"im_status": "bot",
"im_status_access_token": self.partner_root._get_im_status_access_token(),
"is_company": False,
"main_user_id": self.user_root.id,
"name": "OdooBot",
"write_date": fields.Datetime.to_string(self.user_root.partner_id.write_date),
},
{
"active": True,
"avatar_128_access_token": test_user.partner_id._get_avatar_128_access_token(),
"country_id": belgium.id,
"id": test_user.partner_id.id,
"im_status": "offline",
"im_status_access_token": test_user.partner_id._get_im_status_access_token(),
"is_public": False,
"main_user_id": test_user.id,
"mention_token": test_user.partner_id._get_mention_token(),
"name": "Roger",
"email": test_user.partner_id.email,
"offline_since": False,
"user_livechat_username": False,
"write_date": fields.Datetime.to_string(test_user.write_date),
},
{
"active": True,
"avatar_128_access_token": operator.partner_id._get_avatar_128_access_token(),
"country_id": False,
"id": operator.partner_id.id,
"im_status": "offline",
"im_status_access_token": operator.partner_id._get_im_status_access_token(),
"is_public": False,
"mention_token": operator.partner_id._get_mention_token(),
"user_livechat_username": "Michel Operator",
"write_date": fields.Datetime.to_string(operator.write_date),
},
),
)
self.assertEqual(
data["res.users"],
self._filter_users_fields(
{
"id": self.user_root.id,
"employee_ids": [],
"partner_id": self.partner_root.id,
"share": False,
},
{
"id": test_user.id,
"is_admin": False,
"is_livechat_manager": False,
"notification_type": "email",
"partner_id": test_user.partner_id.id,
"signature": ["markup", str(test_user.signature)],
"share": False,
},
),
)
self.assertEqual(
data["discuss.channel.member"],
[
{
"create_date": fields.Datetime.to_string(operator_member.create_date),
"fetched_message_id": False,
"id": operator_member.id,
"livechat_member_type": "agent",
"last_seen_dt": False,
"partner_id": operator.partner_id.id,
"seen_message_id": False,
"channel_id": {"id": channel_info["id"], "model": "discuss.channel"},
},
{
"create_date": fields.Datetime.to_string(visitor_member.create_date),
"custom_channel_name": False,
"custom_notifications": False,
"fetched_message_id": False,
"id": visitor_member.id,
"livechat_member_type": "visitor",
"last_interest_dt": fields.Datetime.to_string(visitor_member.last_interest_dt),
"last_seen_dt": False,
"message_unread_counter": 0,
"message_unread_counter_bus_id": self.env["bus.bus"]._bus_last_id() - 2,
"mute_until_dt": False,
"new_message_separator": 0,
"partner_id": test_user.partner_id.id,
"rtc_inviting_session_id": False,
"seen_message_id": False,
"unpin_dt": False,
"channel_id": {"id": channel_info["id"], "model": "discuss.channel"},
},
],
)
self.assertEqual(data["res.country"], [{"code": "BE", "id": belgium.id, "name": "Belgium"}])
# ensure visitor info are correct when operator is testing themselves
operator = self.operators[0]
self.authenticate(operator.login, self.password)
data = self.make_jsonrpc_request('/im_livechat/get_session', {
'previous_operator_id': operator.partner_id.id,
'channel_id': self.livechat_channel.id,
})["store_data"]
channel_info = data["discuss.channel"][0]
operator_member_domain = [
('channel_id', '=', channel_info['id']),
('partner_id', '=', operator.partner_id.id),
]
operator_member = self.env['discuss.channel.member'].search(operator_member_domain)
self.assertEqual(channel_info['livechat_operator_id'], operator.partner_id.id)
self.assertEqual(channel_info["name"], "Michel Michel Operator")
self.assertEqual(channel_info['country_id'], False)
self.assertEqual(
data["res.partner"],
self._filter_partners_fields(
{
"active": False,
"avatar_128_access_token": self.partner_root._get_avatar_128_access_token(),
"email": "odoobot@example.com",
"id": self.user_root.partner_id.id,
"im_status": "bot",
"im_status_access_token": self.partner_root._get_im_status_access_token(),
"is_company": False,
"main_user_id": self.user_root.id,
"name": "OdooBot",
"write_date": fields.Datetime.to_string(self.user_root.partner_id.write_date),
},
{
"active": True,
"avatar_128_access_token": operator.partner_id._get_avatar_128_access_token(),
"country_id": False,
"id": operator.partner_id.id,
"im_status": "offline",
"im_status_access_token": operator.partner_id._get_im_status_access_token(),
"is_public": False,
"main_user_id": operator.id,
"mention_token": operator.partner_id._get_mention_token(),
"name": "Michel",
"email": operator.email,
"user_livechat_username": "Michel Operator",
"write_date": fields.Datetime.to_string(operator.partner_id.write_date),
},
),
)
self.assertEqual(
data["discuss.channel.member"],
[
{
"create_date": fields.Datetime.to_string(operator_member.create_date),
"custom_channel_name": False,
"custom_notifications": False,
"fetched_message_id": False,
"id": operator_member.id,
"livechat_member_type": "agent",
"last_interest_dt": fields.Datetime.to_string(operator_member.last_interest_dt),
"last_seen_dt": False,
"message_unread_counter": 0,
"message_unread_counter_bus_id": self.env["bus.bus"]._bus_last_id() - 2,
"mute_until_dt": False,
"new_message_separator": 0,
"partner_id": operator.partner_id.id,
"rtc_inviting_session_id": False,
"seen_message_id": False,
"unpin_dt": fields.Datetime.to_string(operator_member.unpin_dt),
"channel_id": {"id": channel_info["id"], "model": "discuss.channel"},
},
],
)
self.assertEqual(
data["res.users"],
self._filter_users_fields(
{
"id": self.user_root.id,
"employee_ids": [],
"partner_id": self.partner_root.id,
"share": False,
},
{
"id": operator.id,
"is_admin": False,
"is_livechat_manager": False,
"notification_type": "email",
"partner_id": operator.partner_id.id,
"share": False,
"signature": ["markup", str(operator.signature)],
},
),
)
def _open_livechat_discuss_channel(self):
discuss_channels = []
for _i in range(5):
data = self.make_jsonrpc_request(
"/im_livechat/get_session", {"channel_id": self.livechat_channel.id}
)
discuss_channels.append(
next(
filter(
lambda c: c["id"] == data["channel_id"],
data["store_data"]["discuss.channel"],
)
)
)
# send a message to mark this channel as 'active'
self.env["discuss.channel"].browse(data["channel_id"]).message_post(body="cc")
return discuss_channels
def test_channel_not_pinned_for_operator_before_first_message(self):
operator = self.operators[0]
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"channel_id": self.livechat_channel.id,
"previous_operator_id": operator.partner_id.id,
},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
member = channel.with_user(operator).self_member_id
self.assertEqual(member.partner_id, operator.partner_id, "operator should be member of channel")
self.assertFalse(member.is_pinned, "channel should not be pinned for operator initially")
channel.message_post(body="cc", message_type="comment")
self.assertTrue(member.is_pinned, "channel should be pinned for operator after visitor sent a message")
self.authenticate(operator.login, self.password)
data = self.make_jsonrpc_request("/mail/data", {"fetch_params": ["channels_as_member"]})
channel_ids = [channel["id"] for channel in data["discuss.channel"]]
self.assertIn(channel.id, channel_ids, "channel should be fetched by operator on new page")
def test_read_channel_unpined_for_operator_after_one_day(self):
data = self.make_jsonrpc_request(
"/im_livechat/get_session", {"channel_id": self.livechat_channel.id}
)
member_of_operator = self.env["discuss.channel.member"].search(
[
("channel_id", "=", data["channel_id"]),
("partner_id", "in", self.operators.partner_id.ids),
]
)
message = self.env["discuss.channel"].browse(data["channel_id"]).message_post(body="cc", message_type="comment")
member_of_operator._mark_as_read(message.id)
with freeze_time(fields.Datetime.to_string(fields.Datetime.now() + timedelta(days=1))):
member_of_operator._gc_unpin_livechat_sessions()
self.assertFalse(member_of_operator.is_pinned, "read channel should be unpinned after one day")
self.assertTrue(member_of_operator.channel_id.livechat_end_dt)
def test_unread_channel_not_unpined_for_operator_after_autovacuum(self):
data = self.make_jsonrpc_request(
"/im_livechat/get_session", {"channel_id": self.livechat_channel.id}
)
member_of_operator = self.env["discuss.channel.member"].search(
[
("channel_id", "=", data["channel_id"]),
("partner_id", "in", self.operators.partner_id.ids),
]
)
self.env["discuss.channel"].browse(data["channel_id"]).message_post(body="cc", message_type="comment")
with freeze_time(fields.Datetime.to_string(fields.Datetime.now() + timedelta(days=1))):
member_of_operator._gc_unpin_livechat_sessions()
self.assertTrue(member_of_operator.is_pinned, "unread channel should not be unpinned after autovacuum")
self.assertFalse(member_of_operator.channel_id.livechat_end_dt)
def test_livechat_manager_can_invite_anyone(self):
channel = self.env["discuss.channel"].create(
{
"channel_type": "livechat",
"livechat_operator_id": self.operators[2].partner_id.id,
"name": "test",
}
)
other_member = channel.with_user(self.operators[0])._add_members(users=self.operators[1])
self.assertEqual(other_member.partner_id, self.operators[1].partner_id)
self_member = channel.with_user(self.operators[0])._add_members(users=self.operators[0])
self.assertEqual(self_member.partner_id, self.operators[0].partner_id)
def test_livechat_operator_can_see_all_livechat_conversations_and_members(self):
bob_user = new_test_user(
self.env, "bob_user", groups="base.group_user,im_livechat.im_livechat_group_user"
)
livechat_session = self.env["discuss.channel"].create(
{
"channel_type": "livechat",
"livechat_operator_id": self.operators[0].partner_id.id,
"name": "test",
}
)
livechat_session.with_user(self.operators[0])._add_members(users=self.operators[1])
self.assertEqual(
self.env["discuss.channel"].with_user(bob_user).search([("id", "=", livechat_session.id)]),
livechat_session
)
self.assertEqual(
self.env["discuss.channel.member"].with_user(bob_user).search([("channel_id", "=", livechat_session.id)]),
livechat_session.channel_member_ids
)
def test_user_prevails_over_guest_when_creating_member(self):
test_user = new_test_user(self.env, "meow_user")
guest = self.env["mail.guest"].create({"name": "Guest"})
self.authenticate(test_user.login, test_user.password)
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": self.livechat_channel.id},
cookies={guest._cookie_name: guest._format_auth_cookie()},
)
channel_members = self.env["discuss.channel"].browse(data["channel_id"]).channel_member_ids
agent = channel_members.filtered(lambda member: member.livechat_member_type == "agent")
visitor = channel_members.filtered(lambda member: member.livechat_member_type == "visitor")
self.assertEqual(len(agent), 1)
self.assertEqual(len(visitor), 1)
self.assertEqual(visitor.partner_id, test_user.partner_id)

165
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_get_mail_channel.py
View file

@ -1,165 +0,0 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from datetime import timedelta
from freezegun import freeze_time
from odoo import fields
from odoo.addons.im_livechat.tests.common import TestImLivechatCommon
class TestGetMailChannel(TestImLivechatCommon):
def test_get_mail_channel(self):
"""For a livechat with 5 available operators, we open 5 channels 5 times (25 channels total).
For every 5 channels opening, we check that all operators were assigned.
"""
for i in range(5):
mail_channels = self._open_livechat_mail_channel()
channel_operators = [channel_info['operator_pid'] for channel_info in mail_channels]
channel_operator_ids = [channel_operator[0] for channel_operator in channel_operators]
self.assertTrue(all(partner_id in channel_operator_ids for partner_id in self.operators.mapped('partner_id').ids))
def test_channel_get_livechat_visitor_info(self):
belgium = self.env.ref('base.be')
public_user = self.env.ref('base.public_user')
test_user = self.env['res.users'].create({'name': 'Roger', 'login': 'roger', 'country_id': belgium.id})
# ensure visitor info are correct with anonymous
operator = self.operators[0]
channel_info = self.livechat_channel.with_user(public_user)._open_livechat_mail_channel(anonymous_name='Visitor 22', previous_operator_id=operator.partner_id.id, country_id=belgium.id)
self.assertEqual(channel_info['channel']['anonymous_name'], "Visitor 22")
self.assertEqual(channel_info['channel']['anonymous_country'], {'code': 'BE', 'id': belgium.id, 'name': 'Belgium'})
# ensure member info are hidden (in particular email and real name when livechat username is present)
# shape of channelMembers is [('insert', data...)], [0][1] accesses the data
self.assertEqual(sorted(map(lambda m: m['persona']['partner'], channel_info['channel']['channelMembers'][0][1]), key=lambda m: m['id']), sorted([{
'active': True,
'country': [('clear',)],
'id': operator.partner_id.id,
'is_public': False,
'user_livechat_username': 'Michel Operator',
}, {
'active': False,
'id': public_user.partner_id.id,
'is_public': True,
'name': 'Public user',
}], key=lambda m: m['id']))
# ensure visitor info are correct with real user
channel_info = self.livechat_channel.with_user(test_user)._open_livechat_mail_channel(anonymous_name='whatever', previous_operator_id=operator.partner_id.id, user_id=test_user.id)
self.assertFalse(channel_info['channel']['anonymous_name'])
self.assertEqual(channel_info['channel']['anonymous_country'], [('clear',)])
self.assertEqual(channel_info['channel']['channelMembers'], [('insert', [
{
'channel': {'id': channel_info['id']},
'id': self.env['mail.channel.member'].search([('channel_id', '=', channel_info['id']), ('partner_id', '=', operator.partner_id.id)]).id,
'persona': {
'partner': {
'active': True,
'country': [('clear',)],
'id': operator.partner_id.id,
'is_public': False,
'user_livechat_username': 'Michel Operator',
},
},
},
{
'channel': {'id': channel_info['id']},
'id': self.env['mail.channel.member'].search([('channel_id', '=', channel_info['id']), ('partner_id', '=', test_user.partner_id.id)]).id,
'persona': {
'partner': {
'active': True,
'country': {
'code': 'BE',
'id': belgium.id,
'name': 'Belgium',
},
'id': test_user.partner_id.id,
'is_public': False,
'name': 'Roger',
},
},
},
])])
# ensure visitor info are correct when operator is testing themselves
operator = self.operators[0]
channel_info = self.livechat_channel.with_user(operator)._open_livechat_mail_channel(anonymous_name='whatever', previous_operator_id=operator.partner_id.id, user_id=operator.id)
self.assertEqual(channel_info['operator_pid'], (operator.partner_id.id, "Michel Operator"))
self.assertFalse(channel_info['channel']['anonymous_name'])
self.assertEqual(channel_info['channel']['anonymous_country'], [('clear',)])
self.assertEqual(channel_info['channel']['channelMembers'], [('insert', [
{
'channel': {'id': channel_info['id']},
'id': self.env['mail.channel.member'].search([('channel_id', '=', channel_info['id']), ('partner_id', '=', operator.partner_id.id)]).id,
'persona': {
'partner': {
'active': True,
'country': [('clear',)],
'id': operator.partner_id.id,
'is_public': False,
'user_livechat_username': 'Michel Operator',
},
},
},
])])
def _open_livechat_mail_channel(self):
mail_channels = []
for i in range(5):
mail_channel = self.livechat_channel._open_livechat_mail_channel('Anonymous')
mail_channels.append(mail_channel)
# send a message to mark this channel as 'active'
self.env['mail.channel'].browse(mail_channel['id']).message_post(body='cc')
return mail_channels
def test_channel_not_pinned_for_operator_before_first_message(self):
public_user = self.env.ref('base.public_user')
channel_info = self.livechat_channel.with_user(public_user)._open_livechat_mail_channel(anonymous_name='whatever')
operator_channel_member = self.env['mail.channel.member'].search([('channel_id', '=', channel_info['id']), ('partner_id', 'in', self.operators.partner_id.ids)])
self.assertEqual(len(operator_channel_member), 1, "operator should be member of channel")
self.assertFalse(operator_channel_member.is_pinned, "channel should not be pinned for operator initially")
self.env['mail.channel'].browse(channel_info['id']).message_post(body='cc')
self.assertTrue(operator_channel_member.is_pinned, "channel should be pinned for operator after visitor sent a message")
self.assertIn(channel_info['id'], operator_channel_member.partner_id._get_channels_as_member().ids, "channel should be fetched by operator on new page")
def test_operator_livechat_username(self):
"""Ensures the operator livechat_username is returned by `_channel_fetch_message`, which is
the method called by the public route displaying chat history."""
public_user = self.env.ref('base.public_user')
operator = self.operators[0]
operator.write({
'email': 'michel@example.com',
'livechat_username': 'Michel at your service',
})
channel_info = self.livechat_channel.with_user(public_user).sudo()._open_livechat_mail_channel(anonymous_name='whatever')
channel = self.env['mail.channel'].browse(channel_info['id'])
channel.with_user(operator).message_post(body='Hello', message_type='comment', subtype_xmlid='mail.mt_comment')
message_formats = channel.with_user(public_user).sudo()._channel_fetch_message()
self.assertEqual(len(message_formats), 1)
self.assertEqual(message_formats[0]['author']['id'], operator.partner_id.id)
self.assertEqual(message_formats[0]['author']['user_livechat_username'], operator.livechat_username)
self.assertFalse(message_formats[0].get('email_from'), "should not send email_from to livechat user")
def test_read_channel_unpined_for_operator_after_one_day(self):
public_user = self.env.ref('base.public_user')
channel_info = self.livechat_channel.with_user(public_user)._open_livechat_mail_channel(anonymous_name='visitor')
member_of_operator = self.env['mail.channel.member'].search([('channel_id', '=', channel_info['id']), ('partner_id', 'in', self.operators.partner_id.ids)])
message = self.env['mail.channel'].browse(channel_info['id']).message_post(body='cc')
member_of_operator.channel_id.with_user(self.operators.filtered(
lambda operator: operator.partner_id == member_of_operator.partner_id
))._channel_seen(message.id)
with freeze_time(fields.Datetime.to_string(fields.datetime.now() + timedelta(days=1))):
member_of_operator._gc_unpin_livechat_sessions()
self.assertFalse(member_of_operator.is_pinned, "read channel should be unpinned after one day")
def test_unread_channel_not_unpined_for_operator_after_autovacuum(self):
public_user = self.env.ref('base.public_user')
channel_info = self.livechat_channel.with_user(public_user)._open_livechat_mail_channel(anonymous_name='visitor')
member_of_operator = self.env['mail.channel.member'].search([('channel_id', '=', channel_info['id']), ('partner_id', 'in', self.operators.partner_id.ids)])
self.env['mail.channel'].browse(channel_info['id']).message_post(body='cc')
with freeze_time(fields.Datetime.to_string(fields.datetime.now() + timedelta(days=1))):
member_of_operator._gc_unpin_livechat_sessions()
self.assertTrue(member_of_operator.is_pinned, "unread channel should not be unpinned after autovacuum")

484
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_get_operator.py Normal file
View file

@ -0,0 +1,484 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from datetime import timedelta
from unittest.mock import patch
import odoo
from odoo import Command, fields
from odoo.addons.im_livechat.tests.common import TestGetOperatorCommon
from odoo.addons.mail.tests.common import MailCommon, freeze_all_time
from odoo.tests.common import users
@odoo.tests.tagged("-at_install", "post_install")
class TestGetOperator(MailCommon, TestGetOperatorCommon):
def setUp(self):
super().setUp()
random_choice_patch = patch("random.choice", lambda arr: arr[0])
self.startPatcher(random_choice_patch)
def test_get_by_lang(self):
fr_operator = self._create_operator("fr_FR")
en_operator = self._create_operator("en_US")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [fr_operator.id, en_operator.id],
}
)
self.assertEqual(fr_operator, livechat_channel._get_operator(lang="fr_FR"))
self.assertEqual(en_operator, livechat_channel._get_operator(lang="en_US"))
def test_get_by_lang_both_operator_active(self):
fr_operator = self._create_operator("fr_FR")
en_operator = self._create_operator("en_US")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [fr_operator.id, en_operator.id],
}
)
self._create_conversation(livechat_channel, fr_operator)
self._create_conversation(livechat_channel, en_operator)
self._create_conversation(livechat_channel, en_operator)
self.assertEqual(en_operator, livechat_channel._get_operator(lang="en_US"))
def test_get_by_lang_no_operator_matching_lang(self):
fr_operator = self._create_operator("fr_FR")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [fr_operator.id],
}
)
self.assertEqual(fr_operator, livechat_channel._get_operator(lang="en_US"))
def test_get_by_country(self):
fr_operator = self._create_operator(country_code="FR")
en_operator = self._create_operator(country_code="US")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [fr_operator.id, en_operator.id],
}
)
self.assertEqual(
fr_operator,
livechat_channel._get_operator(country_id=self.env["res.country"].search([("code", "=", "FR")]).id),
)
self.assertEqual(
en_operator,
livechat_channel._get_operator(country_id=self.env["res.country"].search([("code", "=", "US")]).id),
)
def test_get_by_country_no_operator_matching_country(self):
fr_operator = self._create_operator(country_code="FR")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [fr_operator.id],
}
)
self.assertEqual(
fr_operator,
livechat_channel._get_operator(country_id=self.env["res.country"].search([("code", "=", "US")]).id),
)
def test_get_by_lang_and_country_prioritize_lang(self):
fr_operator = self._create_operator("fr_FR", "FR")
en_operator = self._create_operator("en_US", "US")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [fr_operator.id, en_operator.id],
}
)
self.assertEqual(
fr_operator,
livechat_channel._get_operator(
lang="fr_FR", country_id=self.env["res.country"].search([("code", "=", "US")]).id
),
)
self.assertEqual(
en_operator,
livechat_channel._get_operator(
lang="en_US", country_id=self.env["res.country"].search([("code", "=", "FR")]).id
),
)
def test_operator_in_call_no_more_than_two_chats(self):
first_operator = self._create_operator("fr_FR", "FR")
second_operator = self._create_operator("fr_FR", "FR")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [first_operator.id, second_operator.id],
}
)
with freeze_all_time():
self._create_conversation(livechat_channel, first_operator)
self._create_conversation(livechat_channel, first_operator)
# Previous operator is not in a call so it should be available, even if
# he already has two ongoing chats.
self.assertEqual(
first_operator, livechat_channel._get_operator(previous_operator_id=first_operator.partner_id.id)
)
self._create_conversation(livechat_channel, first_operator, in_call=True)
# Previous operator is in a call so it should not be available anymore.
self.assertEqual(
second_operator, livechat_channel._get_operator(previous_operator_id=first_operator.partner_id.id)
)
def test_priority_by_number_of_chat(self):
first_operator = self._create_operator()
second_operator = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [first_operator.id, second_operator.id],
}
)
with freeze_all_time():
self._create_conversation(livechat_channel, first_operator)
self._create_conversation(livechat_channel, second_operator)
self._create_conversation(livechat_channel, second_operator)
self.assertEqual(first_operator, livechat_channel._get_operator())
def test_in_call_operator_not_prioritized(self):
first_operator = self._create_operator()
second_operator = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [first_operator.id, second_operator.id],
}
)
self._create_conversation(livechat_channel, first_operator, in_call=True)
self._create_conversation(livechat_channel, second_operator)
self.assertEqual(second_operator, livechat_channel._get_operator())
def test_priority_by_number_of_chat_with_call_limit_not_exceeded(self):
first_operator = self._create_operator()
second_operator = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [first_operator.id, second_operator.id],
}
)
with freeze_all_time():
self._create_conversation(livechat_channel, first_operator, in_call=True)
self._create_conversation(livechat_channel, second_operator)
self._create_conversation(livechat_channel, second_operator)
self.assertEqual(first_operator, livechat_channel._get_operator())
def test_priority_by_number_of_chat_all_operators_exceed_limit(self):
first_operator = self._create_operator()
second_operator = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [first_operator.id, second_operator.id],
}
)
with freeze_all_time():
self._create_conversation(livechat_channel, first_operator, in_call=True)
self._create_conversation(livechat_channel, first_operator)
self._create_conversation(livechat_channel, second_operator, in_call=True)
self._create_conversation(livechat_channel, second_operator)
self._create_conversation(livechat_channel, second_operator)
self.assertEqual(first_operator, livechat_channel._get_operator())
def test_get_by_expertise(self):
dog_expert = self.env["im_livechat.expertise"].create({"name": "dog"})
cat_expert = self.env["im_livechat.expertise"].create({"name": "cat"})
operator_dog = self._create_operator(expertises=dog_expert)
operator_car = self._create_operator(expertises=cat_expert)
all_operators = operator_dog + operator_car
pets_support = self.env["im_livechat.channel"].create(
{"name": "Pets", "user_ids": all_operators.ids}
)
self.assertEqual(operator_dog, pets_support._get_operator(expertises=dog_expert))
self.assertEqual(operator_car, pets_support._get_operator(expertises=cat_expert))
def test_get_by_expertise_amongst_same_language(self):
dog_expert = self.env["im_livechat.expertise"].create({"name": "dog"})
cat_expert = self.env["im_livechat.expertise"].create({"name": "cat"})
operator_fr_dog = self._create_operator("fr_FR", expertises=dog_expert)
operator_fr_cat = self._create_operator("fr_FR", expertises=cat_expert)
operator_fr_dog_cat = self._create_operator("fr_FR", expertises=dog_expert + cat_expert)
operator_en_dog = self._create_operator("en_US", expertises=dog_expert)
operator_en_cat = self._create_operator("en_US", expertises=cat_expert)
all_operators = (
operator_fr_dog
+ operator_fr_cat
+ operator_fr_dog_cat
+ operator_en_dog
+ operator_en_cat
)
pets_support = self.env["im_livechat.channel"].create(
{"name": "Pets", "user_ids": all_operators.ids}
)
self.assertEqual(
operator_fr_dog, pets_support._get_operator(lang="fr_FR", expertises=dog_expert)
)
self.assertEqual(
operator_en_cat, pets_support._get_operator(lang="en_US", expertises=cat_expert)
)
self.assertEqual(
operator_fr_dog_cat, pets_support._get_operator(lang="fr_FR", expertises=dog_expert + cat_expert)
)
self.assertEqual(
operator_en_dog, pets_support._get_operator(lang="en_US", expertises=dog_expert + cat_expert)
)
@users("employee")
def test_max_sessions_mode_limited(self):
"""Test operator is not available when they reached the livechat channel limit."""
operator = self._create_operator()
livechat_channel_data = {
"name": "Livechat Channel",
"user_ids": operator,
"max_sessions_mode": "limited",
"max_sessions": 2,
}
livechat_channel = self.env["im_livechat.channel"].sudo().create(livechat_channel_data)
self.assertEqual(livechat_channel.available_operator_ids, operator)
self._create_conversation(livechat_channel, operator)
self.assertEqual(livechat_channel.available_operator_ids, operator)
self._create_conversation(livechat_channel, operator)
self.assertFalse(livechat_channel.available_operator_ids)
@users("employee")
def test_max_sessions_mode_limited_multi_operators(self):
"""Test second operator is available when first operator reached the livechat channel
limit."""
operator_1 = self._create_operator()
operator_2 = self._create_operator()
livechat_channel_data = {
"name": "Livechat Channel",
"user_ids": operator_1 + operator_2,
"max_sessions_mode": "limited",
"max_sessions": 2,
}
livechat_channel = self.env["im_livechat.channel"].sudo().create(livechat_channel_data)
self._create_conversation(livechat_channel, operator_1)
self.assertEqual(livechat_channel.available_operator_ids, operator_1 + operator_2)
self._create_conversation(livechat_channel, operator_1)
self.assertEqual(livechat_channel.available_operator_ids, operator_2)
self._create_conversation(livechat_channel, operator_2)
self.assertEqual(livechat_channel.available_operator_ids, operator_2)
@users("employee")
def test_block_assignment_during_call(self):
"""Test operator is not available when they are in call, even below the livechat channel
limit."""
operator = self._create_operator()
livechat_channel_data = {
"name": "Livechat Channel",
"user_ids": operator,
"block_assignment_during_call": True,
}
livechat_channel = self.env["im_livechat.channel"].sudo().create(livechat_channel_data)
with freeze_all_time():
self._create_conversation(livechat_channel, operator, in_call=True)
self.assertFalse(livechat_channel.available_operator_ids)
@users("employee")
def test_max_sessions_mode_multi_channel(self):
"""Test operator is available in second channel even when they reached the livechat channel
limit on the first channel."""
operator = self._create_operator()
livechat_channels_data = [
{
"name": "Livechat Channel",
"user_ids": [operator.id],
"max_sessions_mode": "limited",
"max_sessions": 2,
},
{
"name": "Livechat Channel",
"user_ids": [operator.id],
},
]
livechat_channels = self.env["im_livechat.channel"].sudo().create(livechat_channels_data)
self._create_conversation(livechat_channels[0], operator)
self._create_conversation(livechat_channels[0], operator)
self.assertFalse(livechat_channels[0].available_operator_ids)
self.assertEqual(livechat_channels[1].available_operator_ids, operator)
@users("employee")
def test_operator_max(self):
operator = self._create_operator()
livechat_channels_data = [
{
"name": "Livechat Channel",
"user_ids": [operator.id],
"max_sessions_mode": "limited",
"max_sessions": 2,
},
{
"name": "Livechat Channel",
"user_ids": [operator.id],
},
]
livechat_channels = self.env["im_livechat.channel"].sudo().create(livechat_channels_data)
self._create_conversation(livechat_channels[1], operator)
self._create_conversation(livechat_channels[1], operator)
self.assertEqual(livechat_channels[0].available_operator_ids, operator)
@users("employee")
def test_operator_expired_channel(self):
operator = self._create_operator()
livechat_channel_data = {
"name": "Livechat Channel",
"user_ids": [operator.id],
"max_sessions_mode": "limited",
"max_sessions": 1,
}
livechat_channel = self.env["im_livechat.channel"].sudo().create(livechat_channel_data)
channel_data = {
"name": "Visitor 1",
"channel_type": "livechat",
"livechat_channel_id": livechat_channel.id,
"livechat_operator_id": operator.partner_id.id,
"channel_member_ids": [Command.create({"partner_id": operator.partner_id.id})],
"last_interest_dt": fields.Datetime.now() - timedelta(minutes=4),
}
channel = self.env["discuss.channel"].create(channel_data)
self.assertFalse(livechat_channel.available_operator_ids)
channel.write({"last_interest_dt": fields.Datetime.now() - timedelta(minutes=20)})
self.assertEqual(livechat_channel.available_operator_ids, operator)
@users("employee")
def test_non_member_operator_availability(self):
"""Test the availability of an operator not member of any livechat channel is properly
computed when explicitly passing them to _get_available_operators_by_livechat_channel."""
operator = self._create_operator()
livechat_channels_data = [
{
"name": "Livechat Channel 1",
"max_sessions_mode": "limited",
"max_sessions": 2,
},
{
"name": "Livechat Channel 2",
},
]
livechat_channels = self.env["im_livechat.channel"].sudo().create(livechat_channels_data)
self.assertFalse(livechat_channels[0].available_operator_ids)
self.assertFalse(livechat_channels[1].available_operator_ids)
self.assertEqual(
livechat_channels._get_available_operators_by_livechat_channel(operator),
{
livechat_channels[0]: operator,
livechat_channels[1]: operator,
},
)
self._create_conversation(livechat_channels[0], operator)
self._create_conversation(livechat_channels[0], operator)
self.assertEqual(
livechat_channels._get_available_operators_by_livechat_channel(operator),
{
livechat_channels[0]: self.env["res.users"],
livechat_channels[1]: operator,
},
)
operator.presence_ids.status = "offline"
self.assertEqual(
livechat_channels._get_available_operators_by_livechat_channel(operator),
{
livechat_channels[0]: self.env["res.users"],
livechat_channels[1]: self.env["res.users"],
},
)
@users("employee")
def test_get_non_member_operator(self):
"""Test _get_operator works with a given list of operators that are not members of the
livechat channel"""
operator_1 = self._create_operator(lang_code="fr_FR")
operator_2 = self._create_operator(lang_code="en_US")
all_operators = operator_1 + operator_2
livechat_channel_data = {"name": "Livechat Channel 2"}
livechat_channel = self.env["im_livechat.channel"].sudo().create(livechat_channel_data)
self.assertFalse(livechat_channel._get_operator())
self.assertFalse(
livechat_channel._get_operator(previous_operator_id=operator_1.partner_id.id)
)
self.assertEqual(livechat_channel._get_operator(users=all_operators), operator_1)
self.assertEqual(
livechat_channel._get_operator(previous_operator_id=operator_2.partner_id.id, users=all_operators),
operator_2,
)
self.assertEqual(
livechat_channel._get_operator(lang="en_US", users=all_operators), operator_2
)
def test_buffer_time_multi_operator(self):
first_operator = self._create_operator()
second_operator = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [first_operator.id, second_operator.id],
}
)
now = fields.Datetime.now()
with freeze_all_time(now + timedelta(minutes=-3)):
self._create_conversation(livechat_channel, second_operator)
with freeze_all_time(now):
self._create_conversation(livechat_channel, first_operator)
self.assertEqual(second_operator, livechat_channel._get_operator())
with freeze_all_time(now + timedelta(seconds=121)):
self.assertEqual(first_operator, livechat_channel._get_operator())
def test_bypass_buffer_time_when_impossible_selection(self):
first_operator = self._create_operator()
second_operator = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [first_operator.id, second_operator.id],
}
)
now = fields.Datetime.now()
with freeze_all_time(now):
self._create_conversation(livechat_channel, first_operator)
self._create_conversation(livechat_channel, second_operator)
self.assertEqual(first_operator, livechat_channel._get_operator())
def test_operator_freed_after_chat_ends(self):
first_operator = self._create_operator()
second_operator = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [first_operator.id, second_operator.id],
}
)
self.assertEqual(first_operator, livechat_channel._get_operator())
chat = self._create_conversation(livechat_channel, first_operator)
self.assertEqual(second_operator, livechat_channel._get_operator())
chat.livechat_end_dt = fields.Datetime.now()
chat.flush_recordset(["livechat_end_dt"])
self.assertEqual(first_operator, livechat_channel._get_operator())
def test_agent_availability_not_affected_by_custom_im_status(self):
agent = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [agent.id],
}
)
self.assertEqual(agent.presence_ids.status, "online")
self.assertEqual(agent, livechat_channel._get_operator())
agent.manual_im_status = "busy"
self.assertEqual(agent, livechat_channel._get_operator())
agent.manual_im_status = "away"
self.assertEqual(agent, livechat_channel._get_operator())
agent.manual_im_status = "offline"
self.assertEqual(agent, livechat_channel._get_operator())
agent.presence_ids.status = "away"
self.assertEqual(self.env["res.users"], livechat_channel._get_operator())

29
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_im_livechat_calls.py Normal file
View file

@ -0,0 +1,29 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from functools import wraps
from unittest.mock import patch
from odoo.tests.common import tagged
from odoo.addons.im_livechat.controllers.main import LivechatController
from odoo.addons.im_livechat.tests.common import TestImLivechatCommon
@tagged("post_install", "-at_install")
class TestImLivechatCalls(TestImLivechatCommon):
def test_meeting_view(self):
og_get_session = LivechatController.get_session
def _patched_get_session(*args, **kwargs):
result = og_get_session(*args, **kwargs)
if kwargs["persisted"]:
self.env.flush_all()
channel = self.env["discuss.channel"].search([("id", "=", result["channel_id"])])
agent = channel.channel_member_ids.filtered(lambda m: m.partner_id)
agent.sudo()._rtc_join_call()
return result
with patch.object(LivechatController, "get_session", wraps(og_get_session)(_patched_get_session)):
self.start_tour(
f"/im_livechat/support/{self.livechat_channel.id}",
"im_livechat.meeting_view_tour",
)

116
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_im_livechat_channel.py Normal file
View file

@ -0,0 +1,116 @@
from datetime import timedelta
from odoo.tests import new_test_user, tagged
from odoo.exceptions import AccessError, ValidationError
from odoo.addons.im_livechat.tests.common import TestImLivechatCommon
from odoo.addons.im_livechat.tests.test_get_operator import TestGetOperator
from odoo.fields import Command, Datetime
@tagged("-at_install", "post_install")
class TestImLivechatChannel(TestImLivechatCommon, TestGetOperator):
def test_user_cant_join_livechat_channel(self):
bob_user = new_test_user(self.env, "bob_user", groups="base.group_user")
with self.assertRaises(AccessError):
self.livechat_channel.with_user(bob_user).action_join()
def test_operator_join_leave_livechat_channel(self):
bob_operator = new_test_user(
self.env, "bob_user", groups="base.group_user,im_livechat.im_livechat_group_user"
)
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
self.livechat_channel.with_user(bob_operator).action_join()
self.assertIn(bob_operator, self.livechat_channel.user_ids)
self.livechat_channel.with_user(bob_operator).action_quit()
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
def test_leave_livechat_channels_when_operator_access_revoked(self):
bob_operator = new_test_user(
self.env, "bob_user", groups="im_livechat.im_livechat_group_user"
)
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
self.livechat_channel.with_user(bob_operator).action_join()
self.assertIn(bob_operator, self.livechat_channel.user_ids)
livechat_operator_group = self.env.ref("im_livechat.im_livechat_group_user")
bob_operator.write({
"group_ids": [Command.unlink(livechat_operator_group.id)],
})
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
bob_operator.write({
"group_ids": [Command.link(livechat_operator_group.id)],
})
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
def test_leave_livechat_channels_when_manager_access_revoked(self):
bob_operator = new_test_user(
self.env, "bob_user", groups="im_livechat.im_livechat_group_manager"
)
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
self.livechat_channel.with_user(bob_operator).action_join()
self.assertIn(bob_operator, self.livechat_channel.user_ids)
livechat_manager_group = self.env.ref("im_livechat.im_livechat_group_manager")
bob_operator.write({
"group_ids": [Command.unlink(livechat_manager_group.id)],
})
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
bob_operator.write({
"group_ids": [Command.link(livechat_manager_group.id)],
})
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
def test_leave_livechat_channels_when_operator_removed_from_group(self):
bob_operator = new_test_user(
self.env, "bob_user", groups="im_livechat.im_livechat_group_user"
)
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
self.livechat_channel.with_user(bob_operator).action_join()
self.assertIn(bob_operator, self.livechat_channel.user_ids)
livechat_operator_group = self.env.ref("im_livechat.im_livechat_group_user")
livechat_operator_group.write({
"user_ids": [Command.unlink(bob_operator.id)],
})
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
livechat_operator_group.write({
"user_ids": [Command.link(bob_operator.id)],
})
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
def test_leave_livechat_channels_when_manager_removed_from_group(self):
bob_operator = new_test_user(
self.env, "bob_user", groups="im_livechat.im_livechat_group_manager"
)
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
self.livechat_channel.with_user(bob_operator).action_join()
self.assertIn(bob_operator, self.livechat_channel.user_ids)
livechat_manager_group = self.env.ref("im_livechat.im_livechat_group_manager")
livechat_manager_group.write({
"user_ids": [Command.unlink(bob_operator.id)],
})
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
livechat_manager_group.write({
"user_ids": [Command.link(bob_operator.id)],
})
self.assertNotIn(bob_operator, self.livechat_channel.user_ids)
def test_review_link(self):
with self.assertRaises(ValidationError):
self.livechat_channel.review_link = "javascript:alert('hello')"
with self.assertRaises(ValidationError):
self.livechat_channel.review_link = "https://"
self.livechat_channel.review_link = "https://www.odoo.com"
self.assertEqual(self.livechat_channel.review_link, "https://www.odoo.com")
def test_ongoing_session_count(self):
self.authenticate(None, None)
john = self._create_operator("fr_FR")
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Livechat Channel", "user_ids": [john.id]},
)
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": livechat_channel.id},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
self.assertEqual(livechat_channel.ongoing_session_count, 1)
channel.livechat_end_dt = Datetime.now() - timedelta(minutes=2)
self.assertEqual(livechat_channel.ongoing_session_count, 0)

166
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_im_livechat_report.py
View file

@ -1,55 +1,177 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from datetime import datetime, timedelta
from freezegun import freeze_time
from unittest.mock import patch
from odoo import Command
from odoo.addons.im_livechat.tests.common import TestImLivechatCommon
from odoo.tests.common import new_test_user, tagged
@tagged("post_install", "-at_install")
class TestImLivechatReport(TestImLivechatCommon):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.env['mail.channel'].search([('livechat_channel_id', '!=', False)]).unlink()
def setUp(self):
super().setUp()
self.env['discuss.channel'].search([('livechat_channel_id', '!=', False)]).unlink()
with patch.object(type(cls.env['im_livechat.channel']), '_get_available_users', lambda _: cls.operators):
channel_id = cls.livechat_channel._open_livechat_mail_channel('Anonymous')['id']
def _compute_available_operator_ids(channel_self):
for record in channel_self:
record.available_operator_ids = self.operators
with (
patch.object(
type(self.env["im_livechat.channel"]),
"_compute_available_operator_ids",
_compute_available_operator_ids,
),
freeze_time("2023-03-17 06:05:54"),
):
channel_id = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": self.livechat_channel.id},
)["channel_id"]
channel = cls.env['mail.channel'].browse(channel_id)
cls.operator = channel.livechat_operator_id
channel = self.env['discuss.channel'].browse(channel_id)
self.operator = channel.livechat_operator_id
cls._create_message(channel, cls.visitor_user.partner_id, '2023-03-17 06:05:54')
cls._create_message(channel, cls.operator, '2023-03-17 08:15:54')
cls._create_message(channel, cls.operator, '2023-03-17 08:45:54')
self._create_message(channel, self.visitor_user.partner_id, '2023-03-17 06:06:59')
self._create_message(channel, self.operator, '2023-03-17 08:15:54')
self._create_message(channel, self.operator, '2023-03-17 08:45:54')
# message with the same record id, but with a different model
# should not be taken into account for statistics
partner_message = cls._create_message(channel, cls.operator, '2023-03-17 05:05:54')
partner_message |= cls._create_message(channel, cls.operator, '2023-03-17 09:15:54')
partner_message = self._create_message(channel, self.operator, '2023-03-17 05:05:54')
partner_message |= self._create_message(channel, self.operator, '2023-03-17 09:15:54')
partner_message.model = 'res.partner'
cls.env['mail.message'].flush_model()
with freeze_time("2023-03-17 09:20:54"):
self.make_jsonrpc_request(
"/im_livechat/visitor_leave_session",
{"channel_id": channel_id}
)
self.env['mail.message'].flush_model()
def test_im_livechat_report_channel(self):
report = self.env['im_livechat.report.channel'].search([('livechat_channel_id', '=', self.livechat_channel.id)])
self.assertEqual(len(report), 1, 'Should have one channel report for this live channel')
# We have those messages, ordered by creation;
# 05:05:54: wrong model
# 06:05:54: visitor message
# 06:05:54: session create
# 06:06:59: visitor message
# 08:15:54: operator first answer
# 08:45:54: operator second answer
# 09:15:54: wrong model
# So the duration of the session is: (08:45:54 - 06:05:54) = 2h40 = 9600 seconds
# So the duration of the session is: (09:20:54 - 06:05:54) = 3h15 = 195 minutes
# The time to answer of this session is: (08:15:54 - 06:05:54) = 2h10 = 7800 seconds
self.assertEqual(int(report.time_to_answer), 7800)
self.assertEqual(int(report.duration), 9600)
self.assertEqual(report.time_to_answer, 7800 / 3600)
self.assertEqual(int(report.duration), 195)
def test_im_livechat_report_operator(self):
result = self.env['im_livechat.report.operator'].read_group([], ['time_to_answer:avg', 'duration:avg'], [])
result = self.env["im_livechat.report.channel"].formatted_read_group([], aggregates=["time_to_answer:avg", "duration:avg"])
self.assertEqual(len(result), 1)
self.assertEqual(int(result[0]['time_to_answer']), 7800)
self.assertEqual(int(result[0]['duration']), 9600)
self.assertEqual(result[0]["time_to_answer:avg"], 7800 / 3600)
self.assertEqual(int(result[0]['duration:avg']), 195)
channel = self.env["discuss.channel"].search([("livechat_channel_id", "=", self.livechat_channel.id)])
rated_channel = channel.copy({"rating_last_value": 5})
self._create_message(rated_channel, self.operator, "2023-03-18 11:00:00")
result = self.env["im_livechat.report.channel"].formatted_read_group([], aggregates=["rating:avg"])
self.assertEqual(result[0]["rating:avg"], 5, "Rating average should be 5, excluding unrated sessions")
@classmethod
def _create_message(cls, channel, author, date):
with patch.object(cls.env.cr, 'now', lambda: date):
return channel.message_post(author_id=author.id, body=f'Message {date}')
def test_redirect_to_form_from_pivot(self):
operator_1 = new_test_user(self.env, login="operator_1", groups="im_livechat.im_livechat_group_manager")
operator_2 = new_test_user(self.env, login="operator_2")
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Support", "user_ids": [operator_1.id, operator_2.id]}
)
[partner_1, partner_2] = self.env["res.partner"].create([{"name": "test 1"}, {"name": "test 2"}])
[channel_1, channel_2, channel_3] = self.env["discuss.channel"].create(
[{
"name": "test 1",
"channel_type": "livechat",
"livechat_channel_id": livechat_channel.id,
"livechat_operator_id": operator_1.partner_id.id,
"channel_member_ids": [Command.create({"partner_id": partner_1.id})],
},
{
"name": "test 2",
"channel_type": "livechat",
"livechat_channel_id": livechat_channel.id,
"livechat_operator_id": operator_2.partner_id.id,
"channel_member_ids": [Command.create({"partner_id": partner_2.id})],
},
{
"name": "test 3",
"channel_type": "livechat",
"livechat_channel_id": livechat_channel.id,
"livechat_operator_id": operator_2.partner_id.id,
"channel_member_ids": [Command.create({"partner_id": partner_2.id})],
}]
)
self._create_message(channel_1, operator_1.partner_id, "2025-06-26 10:05:00")
self._create_message(channel_2, operator_2.partner_id, "2025-06-26 10:15:00")
self._create_message(channel_3, operator_2.partner_id, "2025-06-26 10:25:00")
agent_report_action = self.env.ref("im_livechat.im_livechat_agent_history_action")
session_report_action = self.env.ref("im_livechat.im_livechat_report_channel_action")
self.start_tour(
f"/odoo/action-{agent_report_action.id}?view_type=pivot",
"im_livechat_agents_report_pivot_redirect_tour",
login="operator_1",
)
self.start_tour(
f"/odoo/action-{session_report_action.id}?view_type=pivot",
"im_livechat_sessions_report_pivot_redirect_tour",
login="operator_1",
)
def test_day_of_week_ordered_by_lang_week_start(self):
agent = new_test_user(self.env, login="test_agent", groups="im_livechat.im_livechat_group_manager")
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Test Support", "user_ids": [agent.id]}
)
today_dt = datetime.now()
for i in range(1, 8):
date = today_dt + timedelta(days=i)
with freeze_time(date):
channel_id = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"anonymous_name": f"Visitor_{i}", "channel_id": livechat_channel.id},
)["channel_id"]
channel = self.env["discuss.channel"].browse(channel_id)
self._create_message(channel, channel.livechat_operator_id, date)
en_lang = self.env["res.lang"]._activate_lang("en_US")
report_model = self.env["im_livechat.report.channel"].with_user(agent)
result = report_model.formatted_read_group(domain=[], groupby=["day_number"])
expected_order = ["0", "1", "2", "3", "4", "5", "6"]
actual_order = [group["day_number"] for group in result]
self.assertEqual(actual_order, expected_order)
en_lang.week_start = "6"
result = report_model.formatted_read_group(domain=[], groupby=["day_number"])
expected_order = ["6", "0", "1", "2", "3", "4", "5"]
actual_order = [group["day_number"] for group in result]
self.assertEqual(
actual_order,
expected_order,
f"Days should be ordered starting from Saturday. Got {actual_order}, expected {expected_order}"
)
def test_time_to_answer_does_not_count_messages_after_close(self):
with freeze_time("2025-05-20 06:00:00") as frozen_time:
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": self.livechat_channel.id},
)
chat = self.env["discuss.channel"].browse(data["channel_id"])
frozen_time.tick(delta=timedelta(minutes=1))
chat.message_post(body="Question", message_type="comment")
self._create_message(chat, self.env.user.partner_id, datetime.now())
frozen_time.tick(delta=timedelta(minutes=5))
self.make_jsonrpc_request("/im_livechat/visitor_leave_session", {"channel_id": chat.id})
frozen_time.tick(delta=timedelta(minutes=5))
self._create_message(chat, chat.livechat_operator_id, datetime.now())
report = self.env["im_livechat.report.channel"].search([("channel_id", "=", chat.id)])
self.assertEqual(report.time_to_answer, 0.0)

41
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_im_livechat_support_page.py
View file

@ -1,30 +1,25 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from unittest.mock import patch
import odoo
from odoo.tests import HttpCase
from odoo.addons.im_livechat.controllers.main import LivechatController
from odoo.addons.im_livechat.tests.common import TestGetOperatorCommon
@odoo.tests.tagged('-at_install', 'post_install')
class TestImLivechatSupportPage(HttpCase):
@odoo.tests.tagged("-at_install", "post_install")
class TestImLivechatSupportPage(TestGetOperatorCommon):
def test_load_modules(self):
"""Checks that all javascript modules load correctly on the livechat support page"""
operator = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Support Channel", "user_ids": [operator.id]}
)
self.start_tour(f"/im_livechat/support/{livechat_channel.id}", "im_livechat.basic_tour")
# Give some time to the assets to load to prevent fetch
# interrupt errors then ensures all the assets are loaded.
check_js_modules = """
setTimeout(() => {
const { missing, failed, unloaded } = odoo.__DEBUG__.jsModules;
if ([missing, failed, unloaded].some(arr => arr.length)) {
console.error("Couldn't load all JS modules.", JSON.stringify({ missing, failed, unloaded }));
} else {
console.log("test successful");
}
Object.assign(console, {
log: () => {},
error: () => {},
warn: () => {},
});
}, 1000);
"""
self.browser_js("/im_livechat/support/1", code=check_js_modules, ready="odoo.__DEBUG__.didLogInfo")
def test_load_modules_cors(self):
operator = self._create_operator()
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Support Channel", "user_ids": [operator.id]}
)
with patch.object(LivechatController, "_is_cors_request", return_value=True):
self.start_tour(f"/im_livechat/support/{livechat_channel.id}", "im_livechat.basic_tour")

17
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_js.py Normal file
View file

@ -0,0 +1,17 @@
import odoo
from odoo.addons.web.tests.test_js import unit_test_error_checker
@odoo.tests.tagged("post_install", "-at_install")
class ExternalTestSuite(odoo.tests.HttpCase):
def test_external_livechat(self):
# webclient external test suite
self.browser_js(
"/web/tests/livechat?headless&loglevel=2&preset=desktop",
"",
"",
login='admin',
timeout=1800,
success_signal="[HOOT] Test suite succeeded",
error_checker=unit_test_error_checker,
)

154
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_member_history.py Normal file
View file

@ -0,0 +1,154 @@
from odoo.addons.im_livechat.tests import chatbot_common
from odoo.exceptions import ValidationError
from odoo.tests.common import tagged, new_test_user
from odoo.addons.im_livechat.tests.common import TestGetOperatorCommon
@tagged("post_install", "-at_install")
class TestLivechatMemberHistory(TestGetOperatorCommon, chatbot_common.ChatbotCase):
def test_get_session_create_history(self):
john = self._create_operator("fr_FR")
bob = self._create_operator("fr_FR")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [bob.id],
},
)
data = self.make_jsonrpc_request(
"/im_livechat/get_session", {"channel_id": livechat_channel.id}
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
self.assertEqual(len(channel.channel_member_ids.livechat_member_history_ids), 2)
self.assertEqual(
channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id == bob.partner_id
).livechat_member_type,
"agent",
)
self.assertEqual(
channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id != bob.partner_id
).livechat_member_type,
"visitor",
)
channel.add_members(partner_ids=john.partner_id.ids)
self.assertEqual(len(channel.channel_member_ids.livechat_member_history_ids), 3)
self.assertEqual(
channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id == john.partner_id
).livechat_member_type,
"agent",
)
def test_get_session_create_history_with_bot(self):
john = self._create_operator("fr_FR")
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"chatbot_script_id": self.chatbot_script.id,
"channel_id": self.livechat_channel.id,
},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
self.assertEqual(len(channel.channel_member_ids.livechat_member_history_ids), 2)
self.assertEqual(
channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id == self.chatbot_script.operator_partner_id
).livechat_member_type,
"bot",
)
self.assertEqual(
channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id != self.chatbot_script.operator_partner_id
).livechat_member_type,
"visitor",
)
channel.add_members(partner_ids=john.partner_id.ids)
self.assertEqual(len(channel.channel_member_ids.livechat_member_history_ids), 3)
self.assertEqual(
channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id == john.partner_id
).livechat_member_type,
"agent",
)
def test_marked_as_visitor_when_joining_after_log_in(self):
self.authenticate(None, None)
john = self._create_operator("fr_FR")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": john.ids,
},
)
data = self.make_jsonrpc_request(
"/im_livechat/get_session", {"channel_id": livechat_channel.id}
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
self.assertEqual(len(channel.channel_member_ids.livechat_member_history_ids), 2)
self.assertEqual(
channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id == john.partner_id,
).livechat_member_type,
"agent",
)
guest_visitor_history = channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.guest_id
)
self.assertEqual(guest_visitor_history.livechat_member_type, "visitor")
visitor_user = new_test_user(
self.env, login="visitor_user", groups="im_livechat.im_livechat_group_manager"
)
self.authenticate("visitor_user", "visitor_user")
data = self.make_jsonrpc_request(
"/discuss/channel/notify_typing",
{"channel_id": channel.id, "is_typing": True},
cookies={
guest_visitor_history.guest_id._cookie_name: guest_visitor_history.guest_id._format_auth_cookie()
},
)
self.assertEqual(
channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id == visitor_user.partner_id,
).livechat_member_type,
"visitor",
)
def test_can_only_create_history_for_livechats(self):
john = self._create_operator("fr_FR")
channel = self.env["discuss.channel"]._create_channel(name="General", group_id=None)
member = channel.add_members(partner_ids=john.partner_id.ids)
with self.assertRaises(ValidationError):
self.env["im_livechat.channel.member.history"].create({"member_id": member.id}).channel_id
def test_update_history_on_second_join(self):
john = self._create_operator("fr_FR")
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Livechat Channel", "user_ids": [john.id]},
)
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": livechat_channel.id},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
og_history = channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id == john.partner_id
)
john_member = channel.channel_member_ids.filtered(lambda m: m.partner_id == john.partner_id)
self.assertEqual(og_history.livechat_member_type, "agent")
self.assertEqual(og_history.member_id, john_member)
channel.with_user(john).action_unfollow()
john_history = channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id == john.partner_id
)
self.assertFalse(john_history.member_id)
self.assertNotIn(john.partner_id, channel.channel_member_ids.partner_id)
channel._add_members(users=john)
self.assertIn(john.partner_id, channel.channel_member_ids.partner_id)
john_member = channel.channel_member_ids.filtered(lambda m: m.partner_id == john.partner_id)
john_history = channel.channel_member_ids.livechat_member_history_ids.filtered(
lambda m: m.partner_id == john.partner_id
)
self.assertEqual(og_history, john_history)
self.assertEqual(john_member, john_history.member_id)

398
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_message.py
View file

@ -1,84 +1,139 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo import Command
from freezegun import freeze_time
from markupsafe import Markup
from odoo import Command, fields
from odoo.exceptions import AccessError
from odoo.tests.common import users, tagged
from odoo.addons.mail.tests.common import MailCommon
from odoo.addons.mail.tools.discuss import Store
from odoo.addons.im_livechat.tests.chatbot_common import ChatbotCase
@tagged('post_install', '-at_install')
class TestImLivechatMessage(ChatbotCase):
class TestImLivechatMessage(ChatbotCase, MailCommon):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls._create_portal_user()
def setUp(self):
super().setUp()
self.password = 'Pl1bhD@2!kXZ'
self.users = self.env['res.users'].create([
{
'email': 'e.e@example.com',
'groups_id': [Command.link(self.env.ref('base.group_user').id)],
'group_ids': [Command.link(self.env.ref('base.group_user').id)],
'login': 'emp',
'password': self.password,
'name': 'Ernest Employee',
'notification_type': 'inbox',
'odoobot_state': 'disabled',
'signature': '--\nErnest',
},
{'name': 'test1', 'login': 'test1', 'email': 'test1@example.com'},
{
"email": "test1@example.com",
"login": "test1",
"name": "test1",
"password": self.password,
},
])
settings = self.env["res.users.settings"]._find_or_create_for_user(self.users[1])
settings.livechat_username = "chuck"
self.maxDiff = None
def test_update_username(self):
user = self.env['res.users'].create({
'name': 'User',
'login': 'User',
'password': self.password,
'email': 'user@example.com',
'livechat_username': 'edit me'
})
with self.assertRaises(AccessError):
self.env['res.users'].with_user(user).check_access('write')
user.with_user(user).livechat_username = 'New username'
self.assertEqual(user.livechat_username, 'New username')
@users('emp')
def test_chatbot_message_format(self):
user = self.env.user
channel_info = self.livechat_channel.with_user(user)._open_livechat_mail_channel(
anonymous_name='Test Chatbot',
previous_operator_id=self.chatbot_script.operator_partner_id.id,
chatbot_script=self.chatbot_script,
user_id=user.id
self.authenticate(self.users[0].login, self.password)
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"channel_id": self.livechat_channel.id,
"chatbot_script_id": self.chatbot_script.id,
"persisted": True,
},
)
mail_channel = self.env['mail.channel'].browse(channel_info['id'])
discuss_channel = self.env['discuss.channel'].browse(data["channel_id"])
self._post_answer_and_trigger_next_step(
mail_channel,
discuss_channel,
self.step_dispatch_buy_software.name,
chatbot_script_answer=self.step_dispatch_buy_software
)
chatbot_message = mail_channel.chatbot_message_ids.mail_message_id[-1:]
self.assertEqual(chatbot_message.message_format(), [{
'id': chatbot_message.id,
'body': '<p>Can you give us your email please?</p>',
'date': chatbot_message.date,
'message_type': 'comment',
'subtype_id': (self.env.ref('mail.mt_comment').id, 'Discussions'),
'subject': False,
'model': 'mail.channel',
'res_id': mail_channel.id,
'record_name': 'Testing Bot',
'starred_partner_ids': [],
'author': {
'id': self.chatbot_script.operator_partner_id.id,
'name': 'Testing Bot'
},
'guestAuthor': [('clear',)],
'notifications': [],
'attachment_ids': [],
'trackingValues': [],
'linkPreviews': [],
'messageReactionGroups': [],
'chatbot_script_step_id': self.step_email.id,
'needaction_partner_ids': [],
'history_partner_ids': [],
'is_note': False,
'is_discussion': True,
'subtype_description': False,
'is_notification': False,
'recipients': [],
'module_icon': '/mail/static/description/icon.png',
'sms_ids': []
}])
chatbot_message = discuss_channel.chatbot_message_ids.mail_message_id[:1]
self.assertEqual(
Store().add(chatbot_message).get_result()["mail.message"],
[
{
"attachment_ids": [],
"author_guest_id": False,
"author_id": self.chatbot_script.operator_partner_id.id,
"body": ["markup", "<p>Can you give us your email please?</p>"],
"chatbotStep": {
"message": chatbot_message.id,
"scriptStep": self.step_email.id,
},
"create_date": fields.Datetime.to_string(chatbot_message.create_date),
"date": fields.Datetime.to_string(chatbot_message.date),
"default_subject": "Testing Bot",
"email_from": False,
"id": chatbot_message.id,
"incoming_email_cc": False,
"incoming_email_to": False,
"message_link_preview_ids": [],
"message_type": "comment",
"model": "discuss.channel",
"needaction": False,
"notification_ids": [],
"parent_id": False,
"partner_ids": [],
"pinned_at": False,
"rating_id": False,
"reactions": [],
"record_name": "Testing Bot",
"res_id": discuss_channel.id,
"scheduledDatetime": False,
"starred": False,
"thread": {
"id": discuss_channel.id,
"model": "discuss.channel",
},
"subject": False,
"subtype_id": self.env.ref("mail.mt_comment").id,
"trackingValues": [],
"write_date": fields.Datetime.to_string(chatbot_message.write_date),
}
],
)
@users('emp')
def test_message_format(self):
def test_message_to_store(self):
im_livechat_channel = self.env['im_livechat.channel'].sudo().create({'name': 'support', 'user_ids': [Command.link(self.users[0].id)]})
self.users[0].im_status = 'online' # make available for livechat (ignore leave)
channel_livechat_1 = self.env['mail.channel'].browse(im_livechat_channel._open_livechat_mail_channel(anonymous_name='anon 1', previous_operator_id=self.users[0].partner_id.id, user_id=self.users[1].id, country_id=self.env.ref('base.in').id)['id'])
self.env["mail.presence"]._update_presence(self.users[0])
self.authenticate(self.users[1].login, self.password)
channel_livechat_1 = self.env["discuss.channel"].browse(
self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"previous_operator_id": self.users[0].partner_id.id,
"channel_id": im_livechat_channel.id,
},
)["channel_id"]
)
record_rating = self.env['rating.rating'].create({
'res_model_id': self.env['ir.model']._get('mail.channel').id,
'res_model_id': self.env['ir.model']._get('discuss.channel').id,
'res_id': channel_livechat_1.id,
'parent_res_model_id': self.env['ir.model']._get('im_livechat.channel').id,
'parent_res_id': im_livechat_channel.id,
@ -89,43 +144,212 @@ class TestImLivechatMessage(ChatbotCase):
})
message = channel_livechat_1.message_post(
author_id=record_rating.partner_id.id,
body="<img src='%s' alt=':%s/5' style='width:18px;height:18px;float:left;margin-right: 5px;'/>%s"
body=Markup("<img src='%s' alt=':%s/5' style='width:18px;height:18px;float:left;margin-right: 5px;'/>%s")
% (record_rating.rating_image_url, record_rating.rating, record_rating.feedback),
rating_id=record_rating.id,
)
self.assertEqual(message.message_format(), [{
'attachment_ids': [],
'author': {
'id': self.users[1].partner_id.id,
'name': "test1",
self.assertEqual(
Store().add(message).get_result(),
{
"mail.message": self._filter_messages_fields(
{
"attachment_ids": [],
"author_guest_id": False,
"author_id": self.users[1].partner_id.id,
"body": ["markup", message.body],
"date": fields.Datetime.to_string(message.date),
"write_date": fields.Datetime.to_string(message.write_date),
"create_date": fields.Datetime.to_string(message.create_date),
"id": message.id,
"default_subject": "test1 Ernest Employee",
"email_from": '"test1" <test1@example.com>',
"incoming_email_cc": False,
"incoming_email_to": False,
"message_link_preview_ids": [],
"message_type": "notification",
"reactions": [],
"model": "discuss.channel",
"needaction": False,
"notification_ids": [],
"thread": {"id": channel_livechat_1.id, "model": "discuss.channel"},
"parent_id": False,
"partner_ids": [],
"pinned_at": False,
"rating_id": record_rating.id,
"record_name": "test1 Ernest Employee",
"res_id": channel_livechat_1.id,
"scheduledDatetime": False,
"starred": False,
"subject": False,
"subtype_id": self.env.ref("mail.mt_note").id,
"trackingValues": [],
},
),
"mail.message.subtype": [
{"description": False, "id": self.env.ref("mail.mt_note").id}
],
"mail.thread": self._filter_threads_fields(
{
"display_name": "test1 Ernest Employee",
"id": channel_livechat_1.id,
"model": "discuss.channel",
"module_icon": "/mail/static/description/icon.png",
"rating_avg": 5.0,
"rating_count": 1,
},
),
"rating.rating": [
{
"id": record_rating.id,
"rating": 5.0,
"rating_image_url": record_rating.rating_image_url,
"rating_text": "top",
},
],
"res.partner": self._filter_partners_fields(
{
"avatar_128_access_token": self.users[
1
].partner_id._get_avatar_128_access_token(),
"id": self.users[1].partner_id.id,
"is_company": False,
"main_user_id": self.users[1].id,
"user_livechat_username": "chuck",
"write_date": fields.Datetime.to_string(self.users[1].write_date),
},
),
"res.users": self._filter_users_fields(
{
"id": self.users[1].id,
"partner_id": self.users[1].partner_id.id,
"share": False,
},
),
},
'body': message.body,
'date': message.date,
'guestAuthor': [('clear',)],
'history_partner_ids': [],
'id': message.id,
'is_discussion': False,
'is_note': True,
'is_notification': False,
'linkPreviews': [],
'message_type': 'notification',
'messageReactionGroups': [],
'model': 'mail.channel',
'module_icon': '/mail/static/description/icon.png',
'needaction_partner_ids': [],
'notifications': [],
'rating': {
'id': record_rating.id,
'ratingImageUrl': record_rating.rating_image_url,
'ratingText': record_rating.rating_text,
},
'recipients': [],
'record_name': "test1 Ernest Employee",
'res_id': channel_livechat_1.id,
'sms_ids': [],
'starred_partner_ids': [],
'subject': False,
'subtype_description': False,
'subtype_id': (self.env.ref('mail.mt_note').id, 'Note'),
'trackingValues': [],
}])
)
@users("portal_test")
@freeze_time("2020-03-22 10:42:06")
def test_feedback_message(self):
"""Test posting a feedback message as a portal user, and ensure the proper bus
notifications are sent."""
livechat_channel_vals = {"name": "support", "user_ids": [Command.link(self.users[0].id)]}
im_livechat_channel = self.env["im_livechat.channel"].sudo().create(livechat_channel_vals)
self.env["mail.presence"]._update_presence(self.users[0])
self.authenticate(self.env.user.login, self.env.user.login)
channel = self.env["discuss.channel"].browse(
self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"previous_operator_id": self.users[0].partner_id.id,
"channel_id": im_livechat_channel.id,
},
)["channel_id"]
)
def _get_feedback_bus():
message = self.env["mail.message"].sudo().search([], order="id desc", limit=1)
rating = self.env["rating.rating"].sudo().search([], order="id desc", limit=1)
return (
[
# unread counter/new message separator (not asserted below)
(self.env.cr.dbname, "res.partner", self.env.user.partner_id.id),
# new_message
(self.env.cr.dbname, "discuss.channel", channel.id),
],
[
{
"type": "discuss.channel/new_message",
"payload": {
"data": {
"mail.message": self._filter_messages_fields(
{
"attachment_ids": [],
"author_guest_id": False,
"author_id": self.env.user.partner_id.id,
"body": [
"markup",
'<div class="o_mail_notification o_hide_author">Rating: <img class="o_livechat_emoji_rating" src="/rating/static/src/img/rating_5.png" alt="rating"><br>\nGood service</div>',
],
"create_date": fields.Datetime.to_string(
message.create_date
),
"date": fields.Datetime.to_string(message.date),
"default_subject": "Chell Gladys Ernest Employee",
"id": message.id,
"incoming_email_cc": False,
"incoming_email_to": False,
"message_link_preview_ids": [],
"message_type": "notification",
"model": "discuss.channel",
"parent_id": False,
"partner_ids": [],
"pinned_at": False,
"rating_id": rating.id,
"reactions": [],
"record_name": "Chell Gladys Ernest Employee",
"res_id": channel.id,
"scheduledDatetime": False,
"subject": False,
"subtype_id": self.env.ref("mail.mt_comment").id,
"thread": {"id": channel.id, "model": "discuss.channel"},
"write_date": fields.Datetime.to_string(message.write_date),
},
),
"mail.message.subtype": [
{"description": False, "id": self.env.ref("mail.mt_comment").id}
],
"mail.thread": self._filter_threads_fields(
{
"display_name": "Chell Gladys Ernest Employee",
"id": channel.id,
"model": "discuss.channel",
"module_icon": "/mail/static/description/icon.png",
"rating_avg": 5.0,
"rating_count": 1,
},
),
"rating.rating": [
{
"id": rating.id,
"rating": 5.0,
"rating_image_url": rating.rating_image_url,
"rating_text": "top",
},
],
"res.partner": self._filter_partners_fields(
{
"avatar_128_access_token": self.env.user.partner_id._get_avatar_128_access_token(),
"id": self.env.user.partner_id.id,
"is_company": False,
"main_user_id": self.env.user.id,
"name": "Chell Gladys",
"user_livechat_username": False,
"write_date": fields.Datetime.to_string(
self.env.user.write_date
),
},
),
"res.users": self._filter_users_fields(
{
"id": self.env.user.id,
"partner_id": self.env.user.partner_id.id,
"share": True,
},
),
},
"id": channel.id,
},
},
],
)
with self.assertBus(get_params=_get_feedback_bus):
self.make_jsonrpc_request(
"/im_livechat/feedback",
{
"channel_id": channel.id,
"rate": 5,
"reason": "Good service",
},
)

20
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_res_users.py Normal file
View file

@ -0,0 +1,20 @@
from odoo.tests import new_test_user
from odoo.tests.common import TransactionCase, tagged
@tagged("post_install", "-at_install")
class TestLiveChatResUsers(TransactionCase):
def test_livechat_create_res_users(self):
access_user = new_test_user(
self.env,
login="admin_access",
name="admin_access",
groups="base.group_erp_manager,base.group_partner_manager",
)
access_user.with_user(access_user.id).create({
"login": "test_can_be_created",
"name": "test_can_be_created",
"livechat_username": False,
"livechat_lang_ids": [],
})

156
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_session_views.py Normal file
View file

@ -0,0 +1,156 @@
from odoo import Command
from odoo.tests import new_test_user
from odoo.addons.im_livechat.tests.common import TestImLivechatCommon
from odoo.tests.common import users, tagged
@tagged("-at_install", "post_install")
class TestImLivechatSessionViews(TestImLivechatCommon):
def test_session_history_navigation_back_and_forth(self):
operator = new_test_user(
self.env,
login="operator",
groups="base.group_user,im_livechat.im_livechat_group_manager",
)
self.env["mail.presence"]._update_presence(operator)
self.livechat_channel.user_ids |= operator
self.authenticate(None, None)
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"channel_id": self.livechat_channel.id,
"previous_operator_id": operator.partner_id.id,
},
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
channel.with_user(operator).message_post(body="Hello, how can I help you?")
self._reset_bus()
action = self.env.ref("im_livechat.discuss_channel_action_from_livechat_channel")
self.start_tour(
f"/odoo/livechat/{self.livechat_channel.id}/action-{action.id}",
"im_livechat_history_back_and_forth_tour",
login="operator",
)
@users("admin")
def test_form_view_embed_thread(self):
operator = new_test_user(
self.env,
login="operator",
groups="base.group_user,im_livechat.im_livechat_group_manager",
)
[user_1, user_2] = self.env["res.partner"].create([{"name": "test 1"}, {"name": "test 2"}])
[channel1, channel2] = self.env["discuss.channel"].create(
[
{
"name": "test 1",
"channel_type": "livechat",
"livechat_channel_id": self.livechat_channel.id,
"livechat_operator_id": operator.partner_id.id,
"channel_member_ids": [Command.create({"partner_id": user_1.id})],
},
{
"name": "test 2",
"channel_type": "livechat",
"livechat_channel_id": self.livechat_channel.id,
"livechat_operator_id": operator.partner_id.id,
"channel_member_ids": [Command.create({"partner_id": user_2.id})],
},
]
)
channel1.message_post(
body="Test Channel 1 Msg", message_type="comment", subtype_xmlid="mail.mt_comment"
)
channel2.message_post(
body="Test Channel 2 Msg", message_type="comment", subtype_xmlid="mail.mt_comment"
)
action = self.env.ref("im_livechat.discuss_channel_action_from_livechat_channel")
self.start_tour(
f"/odoo/livechat/{self.livechat_channel.id}/action-{action.id}",
"im_livechat_session_history_open",
login="operator",
)
def test_partner_display_name(self):
user = new_test_user(self.env, login="agent", name="john")
company = self.env["res.partner"].create({"name": "TestCompany", "is_company": True})
user.partner_id.parent_id = company.id
self.assertEqual(
user.with_context(im_livechat_hide_partner_company=True).partner_id.display_name,
"john",
)
self.assertEqual(user.partner_id.display_name, "TestCompany, john")
class TestImLivechatLookingForHelpViews(TestImLivechatSessionViews):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.bob = new_test_user(
cls.env,
login="bob_looking_for_help",
groups="base.group_user,im_livechat.im_livechat_group_user",
)
cls.livechat_channel.user_ids |= cls.bob
cls.looking_for_help_action = cls.env.ref(
"im_livechat.discuss_channel_looking_for_help_action"
)
def start_needhelp_session(self, guest_name=None):
self.authenticate(None, None)
cookies = {}
if guest_name:
guest = self.env["mail.guest"].create({"name": guest_name})
cookies = {guest._cookie_name: guest._format_auth_cookie()}
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"channel_id": self.livechat_channel.id,
"previous_operator_id": self.bob.partner_id.id,
},
cookies=cookies,
)
chat = self.env["discuss.channel"].browse(data["channel_id"])
chat.livechat_status = "need_help"
return chat
def test_looking_for_help_list_real_time_update(self):
self.start_needhelp_session()
self.start_tour(
f"/odoo/action-{self.looking_for_help_action.id}",
"im_livechat.looking_for_help_list_real_time_update_tour",
login="bob_looking_for_help",
)
def test_looking_for_help_kanban_real_time_update(self):
self.start_needhelp_session()
self.start_tour(
f"/odoo/action-{self.looking_for_help_action.id}?view_type=kanban",
"im_livechat.looking_for_help_kanban_real_time_update_tour",
login="bob_looking_for_help",
)
def test_looking_for_help_tags_real_time_update(self):
self.start_needhelp_session()
self.start_tour(
f"/odoo/action-{self.looking_for_help_action.id}",
"im_livechat.looking_for_help_tags_real_time_update_tour",
login="bob_looking_for_help",
)
def test_looking_for_help_discuss_category(self):
self.env["discuss.channel"].search([("livechat_status", "=", "need_help")]).unlink()
agent = new_test_user(self.env, "agent", groups="im_livechat.im_livechat_group_user")
accounting_expertise, sales_expertise = self.env["im_livechat.expertise"].create(
[{"name": "Accounting"}, {"name": "Sales"}],
)
agent.livechat_expertise_ids = sales_expertise
accounting_chat = self.start_needhelp_session(guest_name="Visitor Accounting")
accounting_chat.livechat_expertise_ids = accounting_expertise
sales_chat = self.start_needhelp_session(guest_name="Visitor Sales")
sales_chat.livechat_expertise_ids = sales_expertise
self._reset_bus()
self.start_tour(
"/odoo/discuss", "im_livechat.looking_for_help_discuss_category_tour", login="agent"
)

41
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_transcript.py Normal file
View file

@ -0,0 +1,41 @@
from odoo.tests.common import JsonRpcException, tagged
from odoo.tools import mute_logger
from odoo.addons.base.tests.common import HttpCaseWithUserDemo, HttpCaseWithUserPortal
from odoo.addons.im_livechat.tests.common import TestImLivechatCommon
@tagged("-at_install", "post_install")
class TestImLivechatTranscript(TestImLivechatCommon, HttpCaseWithUserDemo, HttpCaseWithUserPortal):
def test_download_transcript(self):
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": self.livechat_channel.id},
)
res = self.url_open(f"/im_livechat/download_transcript/{data['channel_id']}")
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers["Content-Type"], "application/pdf")
def test_download_transcript_non_member(self):
self.authenticate("demo", "demo")
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": self.livechat_channel.id},
)
chat = self.env["discuss.channel"].browse(data["channel_id"])
self.authenticate(None, None)
self.assertFalse(chat.is_member)
with mute_logger("odoo.http"):
res = self.url_open(f"/im_livechat/download_transcript/{data['channel_id']}")
self.assertEqual(res.status_code, 404)
def test_email_transcript_portal_user(self):
self.authenticate(self.user_portal.login, self.user_portal.login)
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{"channel_id": self.livechat_channel.id},
)
with self.assertRaises(JsonRpcException, msg="werkzeug.exceptions.NotFound"):
self.make_jsonrpc_request(
"/im_livechat/email_livechat_transcript",
{"channel_id": data["channel_id"], "email": self.partner_portal.email},
)

37
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_upload_attachment.py Normal file
View file

@ -0,0 +1,37 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo import http
from odoo.tests.common import tagged, HttpCase
from odoo.tools import mute_logger, file_open
@tagged("post_install", "-at_install")
class TestUploadAttachment(HttpCase):
def test_visitor_cannot_upload_on_closed_livechat(self):
self.authenticate(None, None)
operator = self.env["res.users"].create({"name": "Operator", "login": "operator"})
self.env["mail.presence"]._update_presence(operator)
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Test Livechat Channel", "user_ids": [operator.id]}
)
data = self.make_jsonrpc_request(
"/im_livechat/get_session",
{
"channel_id": livechat_channel.id,
"persisted": True,
},
)
self.make_jsonrpc_request(
"/im_livechat/visitor_leave_session", {"channel_id": data["channel_id"]}
)
with mute_logger("odoo.http"), file_open("addons/web/__init__.py") as file:
response = self.url_open(
"/mail/attachment/upload",
{
"csrf_token": http.Request.csrf_token(self),
"thread_id": data["channel_id"],
"thread_model": "discuss.channel",
},
files={"ufile": file},
)
self.assertEqual(response.status_code, 403)

81
odoo-bringout-oca-ocb-im_livechat/im_livechat/tests/test_user_livechat_username.py Normal file
View file

@ -0,0 +1,81 @@
from odoo import fields
from odoo.tests.common import tagged
from odoo.addons.im_livechat.tests.common import TestGetOperatorCommon
@tagged("post_install", "-at_install")
class TestUserLivechatUsername(TestGetOperatorCommon):
def test_user_livechat_username_channel_invite_notification(self):
john = self._create_operator("fr_FR")
bob = self._create_operator("fr_FR")
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "Livechat Channel",
"user_ids": [bob.id],
}
)
data = self.make_jsonrpc_request(
"/im_livechat/get_session", {"channel_id": livechat_channel.id, "persisted": True}
)
john.partner_id.user_livechat_username = "ELOPERADOR"
channel = self.env["discuss.channel"].browse(data["channel_id"])
channel._add_members(users=john)
self.assertEqual(
channel.message_ids[-1].body,
f'<div class="o_mail_notification" data-oe-type="channel-joined">invited <a href="#" data-oe-model="res.partner" data-oe-id="{john.partner_id.id}">@ELOPERADOR</a> to the channel</div>',
)
def test_user_livechat_username_reactions(self):
john = self._create_operator("fr_FR")
john.livechat_username = "ELOPERADOR"
livechat_channel = self.env["im_livechat.channel"].create(
{"name": "Livechat Channel", "user_ids": [john.id]}
)
data = self.make_jsonrpc_request(
"/im_livechat/get_session", {"channel_id": livechat_channel.id}
)
channel = self.env["discuss.channel"].browse(data["channel_id"])
message = channel.message_post(body="Hello, How can I help you?")
session = self.authenticate(john.login, john.login)
data = self.make_jsonrpc_request(
"/mail/message/reaction",
{"action": "add", "content": "👍", "message_id": message.id},
cookies={"session_id": session.sid},
)
self.assertEqual(
data["res.partner"][0],
{
"avatar_128_access_token": john.partner_id._get_avatar_128_access_token(),
"id": john.partner_id.id,
"user_livechat_username": "ELOPERADOR",
"write_date": fields.Datetime.to_string(john.partner_id.write_date),
},
)
def test_user_livechat_username_in_store(self):
john = self._create_operator("fr_FR")
operator = self._create_operator("en_US")
john.livechat_username = "ELOPERADOR"
operator.group_ids = [
fields.Command.link(self.env.ref("im_livechat.im_livechat_group_user").id),
]
livechat_channel = self.env["im_livechat.channel"].create(
{
"name": "The channel",
"user_ids": [fields.Command.link(john.id), fields.Command.link(operator.id)],
}
)
channel = self.env["discuss.channel"].with_user(operator).create(
{
"name": "Livechat session",
"channel_type": "livechat",
"livechat_operator_id": operator.partner_id.id,
"livechat_channel_id": livechat_channel.id,
}
)
data = operator.partner_id.with_user(operator).search_for_channel_invite("fr_FR", channel.id)["store_data"]
john_data = next(filter(lambda partner: partner["id"] == john.partner_id.id, data["res.partner"]))
self.assertEqual(
john_data["user_livechat_username"],
"ELOPERADOR",
)