Очень частой проблемой в автоматизации является синхронизация и ожидание данных. Недавно на моей персональной консультации по автоматизации возникла проблема ожидания данных и мы с учащимся создали небольшой, простой, но правильный пример реализации механизма ожидания. А заодно и поучились писать код по TDD.
Вам же выдаю готовый рецепт, который можно спокойно использовать в ваших проектах. Долго описывать код не буду, кому надо сами разберетесь. Код нетривиальный, потому если что-то непонятно задавайте вопросы ну и обязательно лайкайте пост , чтобы я знал что делаю что-то полезное
Все исходники выложены в общий репозиторий примеров на github GitHub - atinfo/at.info-knowledge-base: http://automated-testing.info knowledge base on test automation examples (кстати присоединяйтесь ко мне и присылай pull request с полезными примерами)
А кому интересно изучить python вместе со мной приходите на 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)
Весь код целиком можно посмотреть здесь, а также без проблем его можно скачать и запустить: