status-mobile/test/appium/tests/test_message_reliability.py

import random
import string
import time

from itertools import cycle
from timeit import timeit

from appium.webdriver.common.mobileby import MobileBy
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait

from tests import info, marks
from tests.base_test_case import MessageReliabilityTestCase
from views.base_element import BaseButton
from views.sign_in_view import SignInView


def wrapper(func, *args, **kwargs):
    def wrapped():
        return func(*args, **kwargs)

    return wrapped


@marks.message_reliability
class TestMessageReliability(MessageReliabilityTestCase):

    def test_message_reliability_1_1_chat(self, messages_number, message_wait_time):
        user_a_sent_messages = 0
        user_a_received_messages = 0
        user_b_sent_messages = 0
        user_b_received_messages = 0
        user_a_message_time = dict()
        user_b_message_time = dict()
        try:
            self.create_drivers(2, max_duration=10800, custom_implicitly_wait=2)
            device_1, device_2 = SignInView(self.drivers[0]), SignInView(self.drivers[1])
            device_1_home, device_2_home = device_1.create_user(username='user_a'), device_2.create_user(
                username='user_b')
            device_2_public_key = device_2_home.get_public_key()
            device_2_home.home_button.click()
            device_1_chat = device_1_home.add_contact(device_2_public_key)
            device_1_chat.chat_message_input.send_keys('hello')
            device_1_chat.send_message_button.click()
            device_2_home.element_by_text('hello').click()
            device_2_chat = device_2_home.get_chat_view()
            device_2_chat.add_to_contacts.click()

            start_time = time.time()
            for i in range(int(messages_number / 2)):
                message_1 = ''.join(random.sample(string.ascii_lowercase, k=10))
                device_1_chat.chat_message_input.send_keys(message_1)
                device_1_chat.send_message_button.click()
                user_a_sent_messages += 1
                try:
                    user_b_receive_time = timeit(wrapper(device_2_chat.wait_for_element_starts_with_text,
                                                         message_1, message_wait_time),
                                                 number=1)
                    duration_time = round(time.time() - start_time, ndigits=2)
                    user_b_message_time[duration_time] = user_b_receive_time
                    user_b_received_messages += 1
                except TimeoutException:
                    info("Message with text '%s' was not received by user_b" % message_1)
                message_2 = ''.join(random.sample(string.ascii_lowercase, k=10))
                device_2_chat.chat_message_input.send_keys(message_2)
                device_2_chat.send_message_button.click()
                user_b_sent_messages += 1
                try:
                    user_a_receive_time = timeit(wrapper(device_1_chat.wait_for_element_starts_with_text,
                                                         message_2, message_wait_time),
                                                 number=1)
                    duration_time = round(time.time() - start_time, ndigits=2)
                    user_a_message_time[duration_time] = user_a_receive_time
                    user_a_received_messages += 1
                except TimeoutException:
                    info("Message with text '%s' was not received by user_a" % message_2)
        finally:
            self.one_to_one_chat_data['user_a'] = {'sent_messages': user_a_sent_messages,
                                                   'message_time': user_a_message_time}
            self.one_to_one_chat_data['user_b'] = {'sent_messages': user_b_sent_messages,
                                                   'message_time': user_b_message_time}

    def test_message_reliability_1_1_chat_with_predefined_user(self, messages_number, user_public_key):
        self.create_drivers(1, max_duration=10800, custom_implicitly_wait=2)
        sign_in_view = SignInView(self.drivers[0])
        home_view = sign_in_view.create_user(username='user_a')
        home_view.add_contact(user_public_key)
        chat_view = home_view.get_chat_view()
        for i in range(messages_number):
            message_text = '%s %s' % (i + 1, ''.join(random.sample(string.ascii_lowercase, k=10)))
            chat_view.chat_message_input.send_keys(message_text)
            chat_view.send_message_button.click()

    def test_message_reliability_public_chat(self, messages_number, message_wait_time, participants_number, chat_name):
        self.public_chat_data['sent_messages'] = int()
        self.public_chat_data['message_time'] = dict()

        self.create_drivers(participants_number, max_duration=10800, custom_implicitly_wait=2)
        users = ['user_%s' % i for i in range(participants_number)]
        chat_views = list()
        chat_name = chat_name if chat_name else ''.join(random.choice(string.ascii_lowercase) for _ in range(7))
        for i in range(participants_number):
            device = SignInView(self.drivers[i])
            home_view = device.create_user(username=users[i])
            home_view.join_public_chat(chat_name)
            chat_views.append(home_view.get_chat_view())

        start_time = time.time()
        repeat = cycle(range(participants_number))
        next_user = next(repeat)
        while self.public_chat_data['sent_messages'] <= messages_number:
            this_user, next_user = next_user, next(repeat)
            message_text = '%s %s' % (self.public_chat_data['sent_messages'] + 1,
                                      ''.join(random.sample(string.ascii_lowercase, k=10)))
            chat_views[this_user].chat_message_input.send_keys(message_text)
            chat_views[this_user].send_message_button.click()
            self.public_chat_data['sent_messages'] += 1
            try:
                receive_time = timeit(wrapper(chat_views[next_user].wait_for_element_starts_with_text,
                                              message_text, message_wait_time),
                                      number=1)
                duration_time = round(time.time() - start_time, ndigits=2)
                self.public_chat_data['message_time'][duration_time] = receive_time
            except TimeoutException:
                pass
            if self.public_chat_data['sent_messages'] == messages_number:
                break

    def test_message_reliability_offline_public_chat(self, messages_number, message_wait_time, chat_name):
        self.public_chat_data['sent_messages'] = int()
        self.public_chat_data['message_time'] = dict()

        self.create_drivers(1, max_duration=10800, custom_implicitly_wait=2, offline_mode=True)
        driver = self.drivers[0]
        sign_in_view = SignInView(driver)
        home_view = sign_in_view.create_user()
        chat_name = chat_name if chat_name else home_view.get_public_chat_name()
        home_view.join_public_chat(chat_name)

        start_time = time.time()
        iterations = int(messages_number / 10 if messages_number > 10 else messages_number)
        for _ in range(iterations):
            home_view.get_back_to_home_view()
            driver.set_network_connection(1)  # airplane mode

            sent_messages_texts = self.network_api.start_chat_bot(chat_name=chat_name, messages_number=10)
            self.public_chat_data['sent_messages'] += 10

            driver.set_network_connection(2)  # turning on WiFi connection

            home_view.get_chat_with_user('#' + chat_name).click()
            chat_view = home_view.get_chat_view()
            for message in sent_messages_texts:
                try:
                    receive_time = timeit(wrapper(chat_view.wait_for_element_starts_with_text,
                                                  message, message_wait_time),
                                          number=1)
                    duration_time = round(time.time() - start_time, ndigits=2)
                    self.public_chat_data['message_time'][duration_time] = receive_time
                except TimeoutException:
                    pass

    def test_message_reliability_offline_1_1_chat(self, messages_number, message_wait_time):
        user_a_sent_messages = 0
        user_a_received_messages = 0
        user_b_sent_messages = 0
        user_b_received_messages = 0
        user_a_message_time = dict()
        user_b_message_time = dict()
        try:
            self.create_drivers(2, max_duration=10800, custom_implicitly_wait=2, offline_mode=True)
            device_1, device_2 = self.drivers[0], self.drivers[1]
            sign_in_1, sign_in_2 = SignInView(device_1), SignInView(device_2)
            sign_in_1.create_user(username='user_a')
            sign_in_2.create_user(username='user_b')
            device_1_home, device_2_home = sign_in_1.get_home_view(), sign_in_2.get_home_view()
            device_2_public_key = device_2_home.get_public_key()
            device_2_home.home_button.click()
            device_1_home.add_contact(device_2_public_key)
            device_1_chat = device_1_home.get_chat_view()
            device_1_chat.chat_message_input.send_keys('hello')
            device_1_chat.send_message_button.click()
            device_2_home.element_by_text('hello').click()
            device_2_chat = device_2_home.get_chat_view()
            device_2_chat.add_to_contacts.click()

            iterations = int((messages_number / 10 if messages_number > 10 else messages_number) / 2)
            start_time = time.time()
            for i in range(iterations):
                device_2.set_network_connection(1)  # airplane mode

                messages_1 = list()
                for _ in range(10):
                    message_1 = '%s %s' % (user_a_sent_messages + 1, ''.join(random.sample(string.ascii_lowercase,
                                                                                           k=10)))
                    device_1_chat.chat_message_input.send_keys(message_1)
                    device_1_chat.send_message_button.click()
                    messages_1.append(messages_1)
                    user_a_sent_messages += 1

                device_2.set_network_connection(2)  # turning on WiFi connection

                for message in messages_1:
                    try:
                        user_b_receive_time = timeit(wrapper(device_2_chat.wait_for_element_starts_with_text,
                                                             message, message_wait_time),
                                                     number=1)
                        duration_time = round(time.time() - start_time, ndigits=2)
                        user_b_message_time[duration_time] = user_b_receive_time
                        user_b_received_messages += 1
                    except TimeoutException:
                        info("Message with text '%s' was not received by user_b" % message)

                messages_2 = list()
                for _ in range(10):
                    message_2 = '%s %s' % (user_b_sent_messages + 1, ''.join(random.sample(string.ascii_lowercase,
                                                                                           k=10)))
                    device_2_chat.chat_message_input.send_keys(message_2)
                    device_2_chat.send_message_button.click()
                    messages_2.append(message_2)
                    user_b_sent_messages += 1
                    for message in messages_2:
                        try:
                            user_a_receive_time = timeit(wrapper(device_1_chat.wait_for_element_starts_with_text,
                                                                 message, message_wait_time),
                                                         number=1)
                            duration_time = round(time.time() - start_time, ndigits=2)
                            user_a_message_time[duration_time] = user_a_receive_time
                            user_a_received_messages += 1
                        except TimeoutException:
                            info("Message with text '%s' was not received by user_a" % message)
        finally:
            self.one_to_one_chat_data['user_a'] = {'sent_messages': user_a_sent_messages,
                                                   'message_time': user_a_message_time}
            self.one_to_one_chat_data['user_b'] = {'sent_messages': user_b_sent_messages,
                                                   'message_time': user_b_message_time}

    def test_message_reliability_push_notifications(self, message_wait_time):
        self.create_drivers(2, max_duration=10800, custom_implicitly_wait=2)
        device_1, device_2 = self.drivers[0], self.drivers[1]
        sign_in_1, sign_in_2 = SignInView(device_1), SignInView(device_2)
        sign_in_1.create_user(username='user_a')
        sign_in_2.create_user(username='user_b')
        device_1_home, device_2_home = sign_in_1.get_home_view(), sign_in_2.get_home_view()
        device_2_public_key = device_2_home.get_public_key()
        device_2_home.home_button.click()

        device_2.close_app()

        device_1_home.add_contact(device_2_public_key)
        device_1_chat = device_1_home.get_chat_view()
        device_1_chat.chat_message_input.send_keys('hello')
        device_1_chat.send_message_button.click()

        device_2.open_notifications()
        try:
            WebDriverWait(device_2, message_wait_time) \
                .until(
                expected_conditions.presence_of_element_located((MobileBy.XPATH, '//*[contains(@text, "Status")]')))
            element = BaseButton(device_2)
            element.locator = element.Locator.xpath_selector("//*[contains(@text,'Status')]")
            element.click()
        except TimeoutException as exception:
            exception.msg = "Push notification is not received during '%s' seconds" % message_wait_time
            raise exception

        sign_in_2.sign_in()
        device_2_home.element_by_text('hello').click()
        device_2_chat = device_2_home.get_chat_view()
        device_2_chat.wait_for_element_starts_with_text('hello')