Files
MuditaOS/test/pytest/test_send_message.py
Pawel Olejniczak f535268624 [EGD-5586] Clean up messages endpoint API
Introducing changes according to new messages endpoint api
proposal doc. All request have been updated, and some new were added.
Harness tests were updated too. These changes will be followed up with
pagination implementation.
2021-02-16 09:19:50 +01:00

362 lines
13 KiB
Python

# Copyright (c) 2017-2021, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import time
import pytest
from harness.interface.defs import key_codes, SMSType, status
from harness.interface.CDCSerial import Keytype
def erase_all_templates(harness):
# getting the templates count
body = {"category": "template", "count": True}
ret = harness.endpoint_request("messages", "get", body)
assert ret["status"] == status["OK"]
count = ret["body"]["count"]
# getting all templates
body = {"category": "template", "limit": count}
ret = harness.endpoint_request("messages", "get", body)
assert ret["status"] == status["OK"]
assert len(ret["body"][1][1]) == count
for template in ret["body"][1][1]:
body = {"category": "template", "templateID": template["templateID"]}
del_res = harness.endpoint_request("messages", "del", body)
assert del_res["status"] == status["OK"]
def add_new_template(harness, template_text: str):
# adding new template
body = {"category": "template", "templateBody": template_text}
ret = harness.endpoint_request("messages", "post", body)
assert ret["status"] == status["OK"]
def erase_contacts_by_name(harness, name):
body = {"count": True}
ret = harness.endpoint_request("contacts", "get", body)
assert ret["status"] == status["OK"]
count = ret["body"]["count"]
if count == 0:
return 0
# try to get more than available
batch_size = 30
divider = int((count + 10) / batch_size)
reminder = (count + 10) % batch_size
contacts = []
for i in range(divider):
body = {"count": batch_size, "offset": batch_size * i}
ret = harness.endpoint_request("contacts", "get", body)
assert ret["status"] == status["OK"]
contacts = contacts + ret["body"]
body = {"count": reminder, "offset": (count + 10) - reminder}
ret = harness.endpoint_request("contacts", "get", body)
assert ret["status"] == status["OK"]
contacts = contacts + ret["body"]
ids = []
for contact in contacts:
if name in contact["priName"]:
ids.append(contact["id"])
# erase all contacts by id
for identifier in ids:
# removing added contact
body = {"id": identifier}
ret = harness.endpoint_request("contacts", "del", body)
assert ret["status"] == status["OK"]
def get_message_by_text(harness, message: str, phone_number: str):
body = {"category": "message", "messageBody": message, "phoneNumber": phone_number}
return harness.endpoint_request("messages", "get", body)["body"]
# default sms type is draft
def prepare_sms(harness, message: str, phone_number: str, sms_type: int = 1):
body = {"smsCommand": "smsAdd", "messageBody": message, "phoneNumber": phone_number, "type": sms_type}
return harness.endpoint_request("developerMode", "post", body)
def prepare_sms_template(harness, message: str, phone_number: str):
body = {"template": True, "messageBody": message, "phoneNumber": phone_number}
return harness.endpoint_request("developerMode", "post", body)
def compare_messages(old_messages, new_messages, sms_type: SMSType = SMSType.OUTBOX):
diff_messages = []
for message in new_messages:
if message not in old_messages:
diff_messages.append(message)
assert len(diff_messages) == 1
assert SMSType(diff_messages[0]["type"]) == sms_type
def enter_messages_menu(harness):
harness.connection.send_key_code(key_codes["enter"])
harness.open_application("messages")
if harness.get_application_name() != "ApplicationMessages":
time.sleep(2)
assert harness.get_application_name() == "ApplicationMessages"
def enter_contacts_menu(harness):
harness.connection.send_key_code(key_codes["enter"])
harness.open_application("contacts")
if harness.get_application_name() != "ApplicationPhonebook":
time.sleep(2)
assert harness.get_application_name() == "ApplicationPhonebook"
@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
def test_send_message(harness, phone_number, sms_text):
old_messages = get_message_by_text(harness, sms_text, str(phone_number))
# enter menu
enter_messages_menu(harness)
# create new message
harness.connection.send_key_code(key_codes["left"])
# enter phone number
harness.send_number(str(phone_number))
# move down to message body
harness.connection.send_key_code(key_codes["down"])
# write a message
harness.send_text(sms_text)
# send
harness.connection.send_key_code(key_codes["enter"])
# go back to main screen
for _ in range(3):
time.sleep(1.2) # it take horrendous amount of time to go back to thread view
harness.connection.send_key_code(key_codes["fnRight"])
time.sleep(3)
# check if we back to ApplicationDesktop
assert harness.get_application_name() == "ApplicationDesktop"
new_messages = get_message_by_text(harness, sms_text, str(phone_number))
compare_messages(old_messages, new_messages)
@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
def test_send_prepared_message(harness, phone_number, sms_text):
old_messages = get_message_by_text(harness, sms_text, str(phone_number))
# prepare sms and send it (add sms to sending queue)
prepare_sms(harness, sms_text, str(phone_number), SMSType.QUEUED.value)
# time to send message
time.sleep(3)
# check whether message was sent
new_messages = get_message_by_text(harness, sms_text, str(phone_number))
compare_messages(old_messages, new_messages)
testdata = [
"Ala1Ma śżżń 😚",
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA😚"
]
@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
@pytest.mark.parametrize("sms_text", testdata, ids=["special_characters", "long_text"])
def test_send_prepared_draft_message(harness, phone_number, sms_text):
old_messages = get_message_by_text(harness, sms_text, str(phone_number))
# prepare sms and send it (add sms to sending queue)
prepare_sms(harness, sms_text, str(phone_number), SMSType.DRAFT.value)
# enter menu
enter_messages_menu(harness)
# enter thread with draft
harness.connection.send_key_code(key_codes["enter"])
# send
harness.connection.send_key_code(key_codes["enter"])
# go back to main screen
harness.connection.send_key_code(key_codes["fnRight"], Keytype.long_press)
# wait for sms to send
time.sleep(3)
assert harness.get_application_name() == "ApplicationDesktop"
# check whether message was sent
new_messages = get_message_by_text(harness, sms_text, str(phone_number))
compare_messages(old_messages, new_messages)
@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
def test_send_message_from_template(harness, phone_number, sms_text):
old_messages = get_message_by_text(harness, sms_text, str(phone_number))
# erasing all templates
erase_all_templates(harness)
# add own (well-known) template
add_new_template(harness, sms_text)
# try to send message based on template
# enter menu
enter_messages_menu(harness)
# create new message
harness.connection.send_key_code(key_codes["left"])
# enter phone number
harness.send_number(str(phone_number))
# move down to message body
harness.connection.send_key_code(key_codes["down"])
# move to options
harness.connection.send_key_code(key_codes["fnLeft"])
# move to templates
harness.connection.send_key_code(key_codes["enter"])
# select template
harness.connection.send_key_code(key_codes["enter"])
# send message
harness.connection.send_key_code(key_codes["enter"])
# go back to main screen
harness.connection.send_key_code(key_codes["fnRight"], Keytype.long_press)
# give some time to back to ApplicationDesktop and send message
time.sleep(3)
assert harness.get_application_name() == "ApplicationDesktop"
# check whether message was sent
new_messages = get_message_by_text(harness, sms_text, str(phone_number))
compare_messages(old_messages, new_messages)
@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
def test_forward_message(harness, phone_number, sms_text):
# send first message in order to forward it
test_send_prepared_message(harness, phone_number, sms_text)
old_messages = get_message_by_text(harness, sms_text, str(phone_number))
# enter menu
enter_messages_menu(harness)
# enter thread with sent message
harness.connection.send_key_code(key_codes["enter"])
# select message
harness.connection.send_key_code(key_codes["up"])
# move to options
harness.connection.send_key_code(key_codes["fnLeft"])
# move down to forward message option (2 times down)
for _ in range(2):
harness.connection.send_key_code(key_codes["down"])
# enter thread with sent message
harness.connection.send_key_code(key_codes["enter"])
# go to number field
harness.connection.send_key_code(key_codes["up"])
# enter phone number
harness.send_number(str(phone_number))
# go to sms body
harness.connection.send_key_code(key_codes["down"])
# send message
harness.connection.send_key_code(key_codes["enter"])
# go back to main screen
harness.connection.send_key_code(key_codes["fnRight"], Keytype.long_press)
# give some time to back to ApplicationDesktop and send message
time.sleep(3)
assert harness.get_application_name() == "ApplicationDesktop"
# check whether message was sent
new_messages = get_message_by_text(harness, sms_text, str(phone_number))
compare_messages(old_messages, new_messages)
@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
def test_resend_message(harness, phone_number, sms_text):
# send first message in order to resend it
prepare_sms(harness, sms_text, str(phone_number), SMSType.FAILED.value)
old_messages = get_message_by_text(harness, sms_text, str(phone_number))
# enter menu
enter_messages_menu(harness)
# enter thread with sent message
harness.connection.send_key_code(key_codes["enter"])
# select message
harness.connection.send_key_code(key_codes["up"])
# move to options
harness.connection.send_key_code(key_codes["fnLeft"])
# resending message
harness.connection.send_key_code(key_codes["enter"])
# go back to main screen
harness.connection.send_key_code(key_codes["fnRight"], Keytype.long_press)
# give some time to back to ApplicationDesktop and send message
time.sleep(3)
# check if we back to ApplicationDesktop
assert harness.get_application_name() == "ApplicationDesktop"
# check whether message was sent
new_messages = get_message_by_text(harness, sms_text, str(phone_number))
compare_messages(old_messages, new_messages)
@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
def test_send_message_from_phonebook(harness, phone_number, sms_text):
# erase Test contacts
erase_contacts_by_name(harness, "Test")
old_messages = get_message_by_text(harness, sms_text, str(phone_number))
# adding new test contact
body = {"address": "6 Czeczota St.\n02600 Warsaw",
"altName": "Testowy",
"blocked": False,
"favourite": True,
"numbers": [str(phone_number)],
"priName": "Test"}
ret = harness.endpoint_request("contacts", "put", body)
assert ret["status"] == status["OK"]
added_contact_id = ret["body"]["id"]
# enter contacts (phonebook) menu
enter_contacts_menu(harness)
# search test contact
# select message
harness.connection.send_key_code(key_codes["right"])
# write search text
harness.send_text("Test")
# search for added contact
harness.connection.send_key_code(key_codes["enter"])
# enter contact
harness.connection.send_key_code(key_codes["enter"])
# go to messages in contact
harness.connection.send_key_code(key_codes["right"])
# enter new message window
harness.connection.send_key_code(key_codes["enter"])
assert harness.get_application_name() == "ApplicationMessages"
# go to text in new message windows
harness.connection.send_key_code(key_codes["down"])
# write a message
harness.send_text(sms_text)
# send
harness.connection.send_key_code(key_codes["enter"])
# go back to main screen
harness.connection.send_key_code(key_codes["fnRight"], Keytype.long_press)
# give some time to back to ApplicationDesktop and send message
time.sleep(3)
# check if we back to ApplicationDesktop
assert harness.get_application_name() == "ApplicationDesktop"
new_messages = get_message_by_text(harness, sms_text, str(phone_number))
compare_messages(old_messages, new_messages)
# removing added contact
body = {"id": added_contact_id}
ret = harness.endpoint_request("contacts", "del", body)
assert ret["status"] == status["OK"]