@@ 1,203 1,267 @@
-# Copyright (c) 2017-2021, Mudita Sp. z.o.o. All rights reserved.
+# Copyright (c) 2017-2022, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
-
import pytest
-from harness.interface.defs import status
-
-MESSAGES_PAGE_SIZE = 4
-
-
-@pytest.mark.service_desktop_test
-@pytest.mark.usefixtures("phone_unlocked")
-def test_get_messages_without_pagination(harness):
- limit = MESSAGES_PAGE_SIZE - 1
- get_body = {"category": "message", "limit": limit, "offset": 0}
- ret = harness.endpoint_request("messages", "get", get_body)
- assert ret["status"] == status["OK"]
-
- messages = ret["body"]["entries"]
- messages_count = len(messages)
- assert messages_count == limit
- message_types = [1, 2, 4, 8, 16, 18, 255]
- for message in messages:
- assert type(message["contactID"]) == int
- assert type(message["messageBody"]) == str
- assert type(message["messageID"]) == int
- assert type(message["messageType"]) == int
- assert type(message["createdAt"]) == int
- assert type(message["threadID"]) == int
- assert message["messageType"] in message_types
+from harness.request import TransactionError
+from harness.api.messages import GetMessagesCount, GetMessagesWithOffsetAndLimit, GetMessageById, \
+ GetMessagesByThreadIdWithOffsetAndLimit, AddMessage, DeleteMessageById
+
+
+class MessagesTester:
+ def __init__(self, harness):
+ self.harness = harness
+
+ def get_messages_count(self):
+ try:
+ count = GetMessagesCount().run(self.harness).count
+ except TransactionError:
+ return False
+ else:
+ return True, count
+
+ def get_messages_with_offset_and_limit(self, offset, limit):
+ try:
+ result = GetMessagesWithOffsetAndLimit(offset, limit).run(self.harness)
+ except TransactionError:
+ return False
+ else:
+ return True, result
+
+ def get_message_by_id(self, message_record_id):
+ try:
+ result = GetMessageById(message_record_id).run(self.harness)
+ except TransactionError:
+ return False
+ else:
+ return True, result.message
+
+ def get_messages_by_thread_id(self, thread_record_id, offset, limit):
+ try:
+ result = GetMessagesByThreadIdWithOffsetAndLimit(thread_record_id, offset, limit).run(self.harness)
+ except TransactionError:
+ return False
+ else:
+ return True, result
+
+ def add_message(self, message_number, message_body):
+ try:
+ message = AddMessage(message_number, message_body).run(self.harness).message
+ except TransactionError:
+ return False
+ else:
+ return True, message
+
+ def delete_message_by_id(self, message_record_id):
+ try:
+ DeleteMessageById(message_record_id).run(self.harness)
+ except TransactionError:
+ return False
+ else:
+ return True
@pytest.mark.service_desktop_test
@pytest.mark.usefixtures("phone_unlocked")
-def test_get_messages_with_pagination(harness):
- limit = MESSAGES_PAGE_SIZE + 1
- body = {"category": "message", "limit": limit, "offset": 0}
- ret = harness.endpoint_request("messages", "get", body)
- assert ret["status"] == status["OK"]
-
- messages = ret["body"]["entries"]
- messages_count = len(messages)
- assert messages_count == MESSAGES_PAGE_SIZE
-
+def test_add_and_delete_message(harness):
+ messages_tester = MessagesTester(harness)
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ initial_number_of_messages_records = received_messages_records_count
-@pytest.mark.service_desktop_test
-@pytest.mark.usefixtures("phone_unlocked")
-def test_get_all_messages(harness):
- # get messages count
- body = {"category": "message", "count": True}
- ret = harness.endpoint_request("messages", "get", body)
- assert ret["status"] == status["OK"]
+ message_number = "123456789"
+ message_body = "Hello, how are you?"
- count = ret["body"]["count"]
- assert count > 0
+ result, message_record = messages_tester.add_message(message_number, message_body)
+ assert result, "Failed to add message!"
+ assert message_record["messageBody"] == message_body, "Message body corrupted!"
- # get first messages page
- body = {"category": "message", "limit": count, "offset": 0}
- ret = harness.endpoint_request("messages", "get", body)
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ assert received_messages_records_count == initial_number_of_messages_records + 1, "Wrong number of messages!"
- assert ret["status"] == status["OK"]
- assert count == ret["body"]["totalCount"]
+ assert messages_tester.delete_message_by_id(message_record["messageID"]), "Failed to delete a message!"
- all_messages = ret["body"]["entries"]
- if count > MESSAGES_PAGE_SIZE:
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ assert received_messages_records_count == initial_number_of_messages_records, "Wrong number of messages!"
- # get rest of the messages
- while MESSAGES_PAGE_SIZE == len(ret["body"]["entries"]):
- if "nextPage" in ret["body"]:
-
- offset = ret["body"]["nextPage"]["offset"]
- messages_left_count = count - len(all_messages)
-
- body = {"category": "message", "limit": messages_left_count, "offset": offset}
- ret = harness.endpoint_request("messages", "get", body)
-
- assert ret["status"] == status["OK"]
-
- all_messages += ret["body"]["entries"]
+@pytest.mark.service_desktop_test
+@pytest.mark.usefixtures("phone_unlocked")
+def test_get_message_by_id(harness):
+ messages_tester = MessagesTester(harness)
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ initial_number_of_messages_records = received_messages_records_count
- else:
- break
+ message_number = "123456789"
+ message_body = "Hello, how are you?"
- assert count == len(all_messages)
+ result, message_record = messages_tester.add_message(message_number, message_body)
+ assert result, "Failed to add message!"
+ result, received_message_record = messages_tester.get_message_by_id(message_record["messageID"])
+ assert result, "Failed to get message by id!"
+ assert received_message_record["messageID"] == message_record["messageID"], "Wrong message id!"
+ assert received_message_record["messageBody"] == message_body, "Wrong message body!"
-@pytest.mark.service_desktop_test
-@pytest.mark.usefixtures("phone_unlocked")
-def test_get_messages_by_thread_id_without_pagination(harness):
- thread_id = 1
- limit = MESSAGES_PAGE_SIZE - 1
- body = {"category": "message", "threadID": thread_id, "limit": limit, "offset": 0}
- ret = harness.endpoint_request("messages", "get", body)
- assert ret["status"] == status["OK"]
+ assert messages_tester.delete_message_by_id(message_record["messageID"]), "Failed to delete a message!"
- messages = ret["body"]["entries"]
- messages_count = len(messages)
- assert messages_count == limit
- for message in ret["body"]["entries"]:
- assert message["threadID"] == thread_id
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ assert received_messages_records_count == initial_number_of_messages_records, "Wrong number of messages!"
@pytest.mark.service_desktop_test
@pytest.mark.usefixtures("phone_unlocked")
-def test_get_messages_by_thread_id_with_pagination(harness):
- thread_id = 1
- limit = MESSAGES_PAGE_SIZE + 1
- body = {"category": "message", "threadID": thread_id, "limit": limit, "offset": 0}
- ret = harness.endpoint_request("messages", "get", body)
- assert ret["status"] == status["OK"]
-
- messages = ret["body"]["entries"]
- messages_count = len(messages)
- assert messages_count == MESSAGES_PAGE_SIZE
- for message in ret["body"]["entries"]:
- assert message["threadID"] == thread_id
+def test_get_messages_by_thread_id(harness):
+ messages_tester = MessagesTester(harness)
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ initial_number_of_messages_records = received_messages_records_count
+
+ message_number = "123456789"
+ message_1_body = "Hello, how are you?"
+ message_2_body = "Are you there?"
+
+ result, message_1_record = messages_tester.add_message(message_number, message_1_body)
+ assert result, "Failed to add message!"
+ result, message_2_record = messages_tester.add_message(message_number, message_2_body)
+ assert result, "Failed to add message!"
+
+ result, received_message_1_record = messages_tester.get_message_by_id(message_1_record["messageID"])
+ assert result, "Failed to get message by id!"
+
+ result, response = messages_tester.get_messages_by_thread_id(
+ received_message_1_record["threadID"], 0, 0)
+ assert result, "Failed to get messages by thread id!"
+ found_message_1_id = False
+ for record in response.messages:
+ if message_1_record["messageID"] == record["messageID"]:
+ found_message_1_id = True
+ break
+ found_message_2_id = False
+ for record in response.messages:
+ if message_2_record["messageID"] == record["messageID"]:
+ found_message_2_id = True
+ break
+ assert found_message_1_id and found_message_2_id
+
+ assert messages_tester.delete_message_by_id(message_1_record["messageID"]), "Failed to delete a message!"
+ assert messages_tester.delete_message_by_id(message_2_record["messageID"]), "Failed to delete a message!"
+
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ assert received_messages_records_count == initial_number_of_messages_records, "Wrong number of messages!"
@pytest.mark.service_desktop_test
@pytest.mark.usefixtures("phone_unlocked")
-def test_get_all_messages_by_thread_id(harness):
- thread_id = 1
- # get messages count
- body = {"category": "message", "count": True}
- ret = harness.endpoint_request("messages", "get", body)
- assert ret["status"] == status["OK"]
-
- messages_count = ret["body"]["count"]
- assert messages_count > 0
-
- # get first messages page
- body = {"category": "message", "threadID": thread_id, "limit": messages_count, "offset": 0}
- ret = harness.endpoint_request("messages", "get", body)
-
- assert ret["status"] == status["OK"]
- messages_by_thread_id_count = ret["body"]["totalCount"]
-
- all_messages = ret["body"]["entries"]
- if messages_by_thread_id_count > MESSAGES_PAGE_SIZE:
-
- # get rest of the messages
- while MESSAGES_PAGE_SIZE == len(ret["body"]["entries"]):
-
- if "nextPage" in ret["body"]:
-
- offset = ret["body"]["nextPage"]["offset"]
- messages_left_count = messages_by_thread_id_count - len(all_messages)
-
- body = {"category": "message", "threadID": thread_id, "limit": messages_left_count, "offset": offset}
- ret = harness.endpoint_request("messages", "get", body)
-
- assert ret["status"] == status["OK"]
-
- all_messages += ret["body"]["entries"]
-
- else:
+def test_get_multiple_messages(harness):
+ messages_tester = MessagesTester(harness)
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ initial_number_of_messages_records = received_messages_records_count
+
+ messages_list = ["Hello, how are you?", "Are you there?", "What's up?", "Did you checked your email recently?"]
+ message_phone_number = "123456789"
+ message_records = []
+
+ for message_body in messages_list:
+ if len(message_records) == 2:
+ message_phone_number = 223456789 # changing number to create second thread
+ result, message_record = messages_tester.add_message(message_phone_number, message_body)
+ assert result, "Failed to add message!"
+ message_records.append(message_record)
+
+ result, response = messages_tester.get_messages_with_offset_and_limit(
+ initial_number_of_messages_records, 4)
+ assert result, "Failed to get messages with offset and limit!"
+
+ for message_record in message_records:
+ found_message_id = False
+ for record in response.messages:
+ if message_record["messageID"] == record["messageID"]:
+ found_message_id = True
break
+ assert found_message_id
- assert messages_by_thread_id_count == len(all_messages)
+ for message_record in message_records:
+ assert messages_tester.delete_message_by_id(message_record["messageID"]), "Failed to delete a message!"
+
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ assert received_messages_records_count == initial_number_of_messages_records, "Wrong number of messages!"
@pytest.mark.service_desktop_test
@pytest.mark.usefixtures("phone_unlocked")
-def test_remove_message(harness):
- # get messages count
- body = {"category": "message", "count": True}
- ret = harness.endpoint_request("messages", "get", body)
- assert ret["status"] == status["OK"]
-
- count = ret["body"]["count"]
- assert count > 0
-
- # get message
- number_of_requested_messages = 1
- body = {"category": "message", "limit": number_of_requested_messages, "offset": 0}
- ret = harness.endpoint_request("messages", "get", body)
- assert ret["status"] == status["OK"]
-
- messages = ret["body"]["entries"]
- messages_count = len(messages)
- assert messages_count == number_of_requested_messages
-
- # remove message
- sms_to_remove = messages[0]
- body = {"category": "message", "messageID": sms_to_remove["messageID"]}
- ret = harness.endpoint_request("messages", "del", body)
- assert ret["status"] == status["NoContent"]
-
- # get messages count again
- body = {"category": "message", "count": True}
- ret = harness.endpoint_request("messages", "get", body)
- assert ret["status"] == status["OK"]
- assert ret["body"]["count"] == count - 1
+def test_pagination(harness):
+ messages_tester = MessagesTester(harness)
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ initial_number_of_messages_records = received_messages_records_count
+
+ message_number = "123456789"
+ message_body = "Test message"
+ message_records = []
+
+ num_of_messages = 6
+ while num_of_messages > 0:
+ result, message_record = messages_tester.add_message(message_number, message_body)
+ assert result, "Failed to add message!"
+ message_records.append(message_record)
+ num_of_messages -= 1
+
+ result, response = messages_tester.get_messages_with_offset_and_limit(
+ initial_number_of_messages_records, 6)
+ assert result, "Failed to get messages with offset and limit!"
+ MESSAGES_PAGE_SIZE = 4
+ assert len(response.messages) == MESSAGES_PAGE_SIZE
+ assert response.totalCount == initial_number_of_messages_records + 6
+ assert response.nextPage["limit"] == 2
+ assert response.nextPage["offset"] == initial_number_of_messages_records + 4
+ received_messages = response.messages
+ result, response = messages_tester.get_messages_with_offset_and_limit(
+ response.nextPage["offset"], response.nextPage["limit"])
+ assert result, "Failed to get messages with offset and limit!"
+ assert len(response.messages) == 2
+ received_messages += response.messages
+
+ for message_record in message_records:
+ found_message_id = False
+ for record in received_messages:
+ if message_record["messageID"] == record["messageID"]:
+ found_message_id = True
+ break
+ assert found_message_id
+
+ thread_id = received_messages[0]["threadID"]
+ result, response = messages_tester.get_messages_by_thread_id(thread_id,
+ initial_number_of_messages_records, 6)
+ assert result, "Failed to get messages by thread id!"
+ assert len(response.messages) == MESSAGES_PAGE_SIZE
+ assert response.totalCount == initial_number_of_messages_records + 6
+ assert response.nextPage["limit"] == 2
+ assert response.nextPage["offset"] == initial_number_of_messages_records + 4
+ received_messages = response.messages
+ result, response = messages_tester.get_messages_by_thread_id(thread_id,
+ response.nextPage["offset"],
+ response.nextPage["limit"])
+ assert result, "Failed to get messages with offset and limit!"
+ assert len(response.messages) == 2
+ received_messages += response.messages
+
+ for message_record in message_records:
+ found_message_id = False
+ for record in received_messages:
+ if message_record["messageID"] == record["messageID"]:
+ found_message_id = True
+ break
+ assert found_message_id
+ for message_record in message_records:
+ assert messages_tester.delete_message_by_id(message_record["messageID"]), "Failed to delete a message!"
-@pytest.mark.service_desktop_test
-@pytest.mark.usefixtures("phone_unlocked")
-def test_get_message_by_id(harness):
- message_id = 1
- get_body = {"category": "message", "messageID": message_id}
- ret = harness.endpoint_request("messages", "get", get_body)
- assert ret["status"] == status["OK"]
- assert ret["body"]["messageID"] == message_id
+ result, received_messages_records_count = messages_tester.get_messages_count()
+ assert result, "Failed to get messages count!"
+ assert received_messages_records_count == initial_number_of_messages_records, "Wrong number of messages!"