У меня есть много модульных тестов для отдельных компонентов в моем стеке AWS, но я хотел бы провести функциональные тесты в нескольких асинхронных системах.
Эти системы асинхронны из-за очередей, в которых они работают. Одно приложение может принять событие и сгенерировать несколько элементов очереди в разных очередях для последующей обработки. В конечном счете, я хотел бы поместить событие на одном конце моей системы и увидеть результаты на другом конце системы. Я уже делаю это вручную, но есть много вариантов, которые я хотел бы протестировать, и полное системное регрессионное тестирование для каждого выпуска функции становится непростым делом.
За единицу работы часто требуется несколько минут, чтобы пройти через все очереди, потоки и процессы и расположиться на другой стороне.
![function testing](https://i.stack.imgur.com/XAFfP.png)
Я знаю , как Я мог бы проверить что-то подобное. Но это не очень красиво.
def test_event():
event = { 'user_id': 1, 'action': 'like_post', 'data': {} }
put_event_to_queue(event)
retry_count = 0
success = False
while retry_count < 3 and not success:
time.sleep(60)
if (is_action_in_rds()
and is_action_in_elasticsearch()
and is_action_in_athena()):
success = True
retry_count += 1
self.assertTrue(success)
Мой подход заключается в том, чтобы поместить что-то в очередь для обработки и проверить, существует ли результат через какое-то произвольное количество времени.
Это будет стандартный подход? Существует ли структура тестирования, которая лучше всего подходит для такого рода испытаний?