Есть отличная удаленная работа для php+codeception+jenkins+allure+docker спецов. 100% remote! Присоединиться к проекту

[Рецепт] Создание и юнит тестирование простого, но универсального механизма ожидания на Python


(Mykhailo Poliarush) #1

Очень частой проблемой в автоматизации является синхронизация и ожидание данных. Недавно на моей персональной консультации :two_men_holding_hands: по автоматизации возникла проблема ожидания данных и мы с учащимся создали небольшой, простой, но правильный пример реализации механизма ожидания. А заодно и поучились писать код по TDD.

Вам же выдаю готовый рецепт, который можно спокойно использовать в ваших проектах. Долго описывать код не буду, кому надо сами разберетесь. Код нетривиальный, потому если что-то непонятно задавайте вопросы ну и обязательно лайкайте пост :heart: , чтобы я знал что делаю что-то полезное :smile:

Все исходники выложены в общий репозиторий примеров на github https://github.com/atinfo/at.info-knowledge-base (кстати присоединяйтесь ко мне и присылай pull request с полезными примерами)

А кому интересно изучить python вместе со мной приходите на :books: http://lessons2.ru/ где я обучаю питонировать и автоматизировать.

Механизм ожидания реализован через декоратор

def wait(function, expected_condition=None, timeout=None, frequency=None):
    """simple implementation of wait mechanism"""
    if not timeout:
        timeout = 6

    if not frequency:
        frequency = 2

    if not expected_condition:
        def expected_condition(results):
            return results


    @wraps(function)
    def wrapper(*args, **kwargs):
        exception = None
        results = None
        for i in xrange(timeout / frequency):
            try:
                results = function(*args, **kwargs)
            except Exception, e:
                exception = e.message
            finally:
                if results:
                    if expected_condition:
                        if expected_condition(results):
                            break
                if timeout / frequency - i < 2:
                    break
                time.sleep(frequency)

        if exception:
            #todo: make your custom exception
            msg = "wrapped function exception {}".format(exception)
            raise Exception(msg)

        if not results:
            #todo: make your custom exception
            msg = "not retrieved results exception"
            raise Exception(msg)

        if results:
            if expected_condition:
                if not expected_condition(results):
                    #todo: make your custom exception
                    msg = "expected condition exception"
                    raise Exception(msg)
        return results

    return wrapper

Реализация проверки времени через контекстный менеджер

@contextmanager
def assert_timeout_manager(expected_to_not_exceed_in_seconds=1):
    start = time.time()
    yield
    end = time.time()
    msg = "elapsed time is {}, but expected {}".format(end - start, expected_to_not_exceed_in_seconds)
    assert (end - start) <= expected_to_not_exceed_in_seconds, msg

Проверка тестового метода через декоратор

def assert_timeout(expected_to_not_exceed_in_seconds=1):
    def method_decorator(func):
        @wraps(func)
        def wrapper(self, *argv, **kwargv):
            with assert_timeout_manager(expected_to_not_exceed_in_seconds):
                results = func(self, *argv, **kwargv)
            return results

        return wrapper

    return method_decorator

Ну и сами тесты

class TestWaiter(unittest.TestCase):
    default_timeout = 6
    default_frequency = 2

    @assert_timeout
    def test_wait_success(self):
        assert wait(lambda: True)()

    def method(self):
        time.sleep(1)
        return True

    def test_wait_success_call_method(self):
        class Some(object):
            def method(self):
                time.sleep(2)
                return True

        assert wait(self.method)()
        assert wait(Some().method)()

    @assert_timeout(default_timeout)
    def test_wait_with_delayed_success_result(self):
        def func():
            time.sleep(5)
            return True

        assert wait(func)()

#.....

if __name__ == "__main__":
    unittest.main(verbosity=2)

Весь код целиком можно посмотреть здесь, а также без проблем его можно скачать и запустить:


Дайджест полезных ссылок для тестировщиков-автоматизаторов #030
Готовые рецепты или активизация сообщества автоматизаторов на #atinfo
(Dmytro Makhno) #2

Всегда интересовало как подобные вещи отлаживать… :smile:
Поставил точку отладки, зазевался, упало по таймауту.

Для себя нашел более подходящим сочетание:
https://wiki.python.org/moin/PythonDecoratorLibrary#Retry (нравится что точно количество)
https://github.com/SeleniumHQ/selenium/blob/master/py/selenium/webdriver/support/wait.py (нравится что exceptions как аргумент).

В результате получил:

def retry(tries=5, delay=3, backoff=2, retry_exception=None):
    """
    Retry "tries" times, with initial "delay", increasing delay "delay*backoff" each time.
    Without exception success means when function returns valid object.
    With exception success when no exceptions
    """
    assert tries > 0, "tries must be 1 or greater"
    catching_mode = bool(retry_exception)

    def deco_retry(f):
        def f_retry(*args, **kwargs):
            mtries, mdelay = tries, delay

            while mtries > 0:
                try:
                    rv = f(*args, **kwargs)
                    if not catching_mode and rv:
                        return rv
                except retry_exception:
                    pass
                else:
                    if catching_mode:
                        return rv
                mtries -= 1
                if mtries is 0 and not catching_mode:
                    return False
                if mtries is 0 and catching_mode:
                    return f(*args, **kwargs)  # extra try, to avoid except-raise syntax
                log.debug("...{0} sleeping for {1} s in retry".format(f.__name__, mdelay))
                time.sleep(mdelay)
                mdelay *= backoff
            raise Exception("unreachable code")
        return f_retry
    return deco_retry

class RetryTest(unittest.TestCase):
    def setUp(self):
        class Counter(object):
            i = 0

        self.counter = Counter()

    def test_bypass_return_with_only_try(self):
        @retry(1, 0, 0)
        def return_smth():
            self.counter.i += 1
            return "something"

        self.assertEqual(return_smth(), "something")
        self.assertEqual(self.counter.i, 1)

    def test_bypass_return(self):
        @retry(3, 0, 0, Exception)
        def return_smth():
            self.counter.i += 1
            return "something"

        self.assertEqual(return_smth(), "something")
        self.assertEqual(self.counter.i, 1)

    def test_false_without_exceptions_is_retry_indicator(self):
        @retry(3, 0, 0)
        def return_false():
            self.counter.i += 1
            return False

        self.assertEqual(return_false(), False)
        self.assertEqual(self.counter.i, 3)

    def test_true_without_exceptions_is_stop_indicator(self):
        @retry(5, 0, 0)
        def return_false():
            self.counter.i += 1
            if self.counter.i < 3:
                return False
            return True

        self.assertEqual(return_false(), True)
        self.assertEqual(self.counter.i, 3)

    def test_retry_on_unknown_exception(self):
        @retry(5, 0, 0, AssertionError)
        def return_good():
            self.counter.i += 1
            if self.counter.i < 3:
                assert False
            return "good"

        self.assertEqual(return_good(), "good")
        self.assertEqual(self.counter.i, 3)

    def test_fail_on_unexpected_exception(self):
        @retry(5, 0, 0, TypeError)
        def return_smth():
            self.counter.i += 1
            if self.counter.i < 3:
                raise ArithmeticError("boom")
            return "good"

        with self.assertRaises(ArithmeticError):
            return_smth()
        self.assertEqual(self.counter.i, 1)

    def test_bypass_return_after_catching(self):
        @retry(5, 0, 0, AssertionError)
        def return_good():
            self.counter.i += 1
            if self.counter.i < 3:
                assert False
            return "good"

        self.assertEqual(return_good(), "good")

    def test_fail_on_exception_for_non_cathching_mode(self):
        @retry(5, 0, 0)
        def return_exception():
            self.counter.i += 1
            raise ArithmeticError("boom")

        with self.assertRaises(ArithmeticError):
            return_exception()
        self.assertEqual(self.counter.i, 1)

    def test_propagate_known_exception_on_retries_out(self):
        @retry(5, 0, 0, ArithmeticError)
        def return_smth():
            self.counter.i += 1
            raise ArithmeticError("boom")

        with self.assertRaises(ArithmeticError):
            return_smth()
        self.assertEqual(self.counter.i, 6)

    def test_propagate_known_exceptions_on_retries_out(self):

        @retry(5, 0, 0, (ArithmeticError, TypeError))
        def return_smth():
            self.counter.i += 1
            if self.counter.i % 2 == 0:
                raise TypeError("boom_even")
            raise ArithmeticError("boom_odd")

        with self.assertRaises(TypeError):
            return_smth()
        self.assertEqual(self.counter.i, 6)

И его потомки декораторы

def webdriver_retry(catch_all=False):
    """
    Method decorator, that retries action if WebDriver exception appeared
    Note: 'sum([delay*backoff**i for i in range(tries)])' ~= 11 seconds
    """

    return retry(tries=60, delay=0.1, backoff=1.02, retry_exception=exceptions.WebDriverException if catch_all else (
        exceptions.NoSuchElementException, exceptions.StaleElementReferenceException))

def eventually(*exceptions):
    """
    Method decorator, that waits when something inside eventually happens
    Note: 'sum([delay*backoff**i for i in range(tries)])' ~= 580 seconds ~= 10 minutes
    """
    return retry(tries=50, delay=0.5, backoff=1.1, retry_exception=exceptions)

в тестах где агент, что-то долго делает

def test_something_should_happen_but_do_not_know_when(self):
        @eventually(AssertionError)
        def eventually_assert():
            assert agent.work_completed()
        eventually_assert()

def test_some_collection_should_get_specific_key_and_value(self)
        @eventually(AssertionError, KeyError)
        def eventually_assert():
            #     possible KeyError -v
            assert agent.some_map["hello"] == "at.info"
            #     possible AssertionError -^
        eventually_assert()

Note: вдохновил опыт с http://doc.scalatest.org/2.1.3/index.html#org.scalatest.concurrent.Eventually

P.S. Модераторы, почему-то когда в последнем снипете маркдаун пишу явно - трибэктикаpython, подсветка пропадает.


(Mykhailo Poliarush) #3

@dmakhno было бы неплохо добавить эти примеры в наш мега-склад примеров на :inbox_tray: GitHub https://github.com/atinfo/at.info-knowledge-base/tree/master/programming/python. Буду рад увидеть pull request. Спасибо!


(Mykhailo Poliarush) #4

Хочется еще добавить одну уже написанную небольшую библиотеку, которая поможет в этом вопросе.