# Copyright (c) 2010-2022 Belledonne Communications SARL. # # This file is part of Liblinphone # (see https://gitlab.linphone.org/BC/public/liblinphone). # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . from nose.tools import assert_equals from copy import deepcopy import filecmp import linphone from linphonetester import * import os import os.path import time class TestMessage: @classmethod def msg_state_changed(cls, msg, state): stats = msg.chat_room.core.user_data().stats linphonetester_logger.info("[TESTER] Message [{text}] [{state}]".format(text=msg.text, state=linphone.ChatMessageState.string(state))) if state == linphone.ChatMessageState.Delivered: stats.number_of_LinphoneMessageDelivered += 1 elif state == linphone.ChatMessageState.NotDelivered: stats.number_of_LinphoneMessageNotDelivered += 1 elif state == linphone.ChatMessageState.InProgress: stats.number_of_LinphoneMessageInProgress += 1 elif state == linphone.ChatMessageState.FileTransferError: stats.number_of_LinphoneMessageNotDelivered += 1 elif state == linphone.ChatMessageState.FileTransferDone: stats.number_of_LinphoneMessageFileTransferDone += 1 elif state == linphone.ChatMessageState.DeliveredToUser: stats.number_of_LinphoneMessageDeliveredToUser += 1 elif state == linphone.ChatMessageState.Displayed: stats.number_of_LinphoneMessageDisplayed += 1 else: linphonetester_logger.error("[TESTER] Unexpected state [{state}] for message [{msg}]".format(msg=msg, state=linphone.ChatMessageState.string(state))) @classmethod def file_transfer_progress_indication(cls, msg, content, offset, total): stats = msg.chat_room.core.user_data().stats progress = int((offset * 100) / total) direction = 'received' tofrom = 'from' address = msg.from_address if msg.is_outgoing: direction = 'sent' tofrom = 'to' address = msg.to_address linphonetester_logger.info("[TESTER] File transfer [{progress}%] {direction} of type [{type}/{subtype}] {tofrom} {address}".format( progress=progress, direction=direction, type=content.type, subtype=content.subtype, tofrom=tofrom, address=address.as_string())); stats.progress_of_LinphoneFileTransfer = progress @classmethod def file_transfer_send(cls, msg, content, offset, size): send_filepath = msg.user_data send_filesize = os.path.getsize(send_filepath) if offset >= send_filesize: return linphone.Buffer.new() # end of file f = open(send_filepath, 'rb') f.seek(offset, 0) if (send_filesize - offset) < size: size = send_filesize - offset lb = linphone.Buffer.new_from_data(bytearray(f.read(size))) f.close() return lb @classmethod def file_transfer_recv(cls, msg, content, buf): receive_filepath = msg.user_data stats = msg.chat_room.core.user_data().stats if buf.is_empty: # Transfer complete stats.number_of_LinphoneFileTransferDownloadSuccessful += 1 else: # Store content f = open(receive_filepath, 'ab') f.write(buf.content) f.close() def create_message_from_sintel_trailer(self, chat_room): send_filepath = os.path.join(tester_resources_path, 'sounds', 'sintel_trailer_opus_h264.mkv') content = chat_room.core.create_content() content.type = 'video' content.subtype = 'mkv' linphonetester_logger.info("[TESTER] os.path.getsize(send_filepath)={}".format(os.path.getsize(send_filepath))) content.size = os.path.getsize(send_filepath) # total size to be transfered linphonetester_logger.info("[TESTER] content.size={}".format(content.size)) content.name = 'sintel_trailer_opus_h264.mkv' message = chat_room.create_file_transfer_message(content) message.callbacks.msg_state_changed = TestMessage.msg_state_changed message.callbacks.file_transfer_send = TestMessage.file_transfer_send message.callbacks.file_transfer_progress_indication = TestMessage.file_transfer_progress_indication message.user_data = send_filepath return message def wait_for_server_to_purge_messages(self, manager1, manager2): # Wait a little bit just to have time to purge message stored in the server CoreManager.wait_for_until(manager1, manager2, lambda manager1, manager2: False, 100) manager1.stats.reset() manager2.stats.reset() def teardown(self): linphone.Factory.clean() def test_text_message(self): marie = CoreManager('marie_rc') pauline = CoreManager('pauline_tcp_rc') chat_room = pauline.lc.get_chat_room(marie.identity) self.wait_for_server_to_purge_messages(marie, pauline) msg = chat_room.create_message("hello") msg.callbacks.msg_state_changed = TestMessage.msg_state_changed chat_room.send_chat_message_2(msg) assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneMessageReceived == 1), True) assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneMessageDelivered == 1), True) assert marie.lc.get_chat_room(pauline.identity) is not None def test_text_message_within_call_dialog(self): marie = CoreManager('marie_rc') pauline = CoreManager('pauline_tcp_rc') pauline.lc.config.set_int('sip', 'chat_use_call_dialogs', 1) self.wait_for_server_to_purge_messages(marie, pauline) assert_equals(CoreManager.call(marie, pauline), True) chat_room = pauline.lc.get_chat_room(marie.identity) msg = chat_room.create_message("Bla bla bla bla") msg.callbacks.msg_state_changed = TestMessage.msg_state_changed chat_room.send_chat_message_2(msg) assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneMessageReceived == 1), True) # When using call dialogs, we will never receive delivered status assert_equals(pauline.stats.number_of_LinphoneMessageDelivered, 0) assert marie.lc.get_chat_room(pauline.identity) is not None CoreManager.end_call(marie, pauline) def test_transfer_message(self): marie = CoreManager('marie_rc') pauline = CoreManager('pauline_tcp_rc') send_filepath = os.path.join(tester_resources_path, 'sounds', 'sintel_trailer_opus_h264.mkv') receive_filepath = 'receive_file.dump' if os.path.exists(receive_filepath): os.remove(receive_filepath) pauline.lc.file_transfer_server = "https://www.linphone.org:444/lft.php" self.wait_for_server_to_purge_messages(marie, pauline) chat_room = pauline.lc.get_chat_room(marie.identity) message = self.create_message_from_sintel_trailer(chat_room) chat_room.send_chat_message_2(message) assert_equals(CoreManager.wait_for_until(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneMessageReceivedWithFile == 1, 60000), True) if marie.stats.last_received_chat_message is not None: cbs = marie.stats.last_received_chat_message.callbacks cbs.msg_state_changed = TestMessage.msg_state_changed cbs.file_transfer_recv = TestMessage.file_transfer_recv cbs.file_transfer_progress_indication = TestMessage.file_transfer_progress_indication marie.stats.last_received_chat_message.user_data = receive_filepath marie.stats.last_received_chat_message.download_file() assert_equals(CoreManager.wait_for_until(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneFileTransferDownloadSuccessful == 1, 55000), True) assert_equals(pauline.stats.number_of_LinphoneMessageInProgress, 2) assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneMessageDelivered == 1), True) assert_equals(filecmp.cmp(send_filepath, receive_filepath, shallow=False), True) if os.path.exists(receive_filepath): os.remove(receive_filepath) def test_transfer_message_upload_cancelled(self): marie = CoreManager('marie_rc') pauline = CoreManager('pauline_tcp_rc') pauline.lc.file_transfer_server = "https://www.linphone.org:444/lft.php" self.wait_for_server_to_purge_messages(marie, pauline) chat_room = pauline.lc.get_chat_room(marie.identity) message = self.create_message_from_sintel_trailer(chat_room) chat_room.send_chat_message_2(message) # Wait for file to be at least 25% uploaded and cancel the transfer assert_equals(CoreManager.wait_for_until(pauline, marie, lambda pauline, marie: pauline.stats.progress_of_LinphoneFileTransfer >= 25, 60000), True) message.cancel_file_transfer() assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneMessageNotDelivered == 1), True) assert_equals(pauline.stats.number_of_LinphoneMessageNotDelivered, 1) assert_equals(marie.stats.number_of_LinphoneFileTransferDownloadSuccessful, 0)