Как написать модульный тест для проверки правильности поведения в условиях временных ограничений системных ресурсов?

Question

У меня есть приложение на Python 3, которое использует многопроцессорный модуль для параллельного распределения больших, ресурсоемких научных расчетов по многим процессорам. Внутри вычисления есть несколько шагов, которые требуют от процесса выделения умеренно больших массивов. Приложение работает очень хорошо, когда я запускаю его на своем ноутбуке с OS X, однако в долгосрочной перспективе это, как правило, будет выполняться из док-контейнера, который, в свою очередь, будет работать на экземпляре Amazon Web Services EC2 или другом сопоставимом облаке. виртуальная машина на основе - эффективно вкладывает приложение в два уровня технологии виртуализации машин.

Я ожидаю, что в будущем какой-то другой пользователь может попытаться запустить приложение с ресурсами виртуальной машины (память, пространство подкачки и т. Д.), Настроенными на значения, которые были установлены слишком агрессивно малыми. (Существуют явные финансовые стимулы для этого, потому что вы, как правило, платите меньше за услуги облачных вычислений, тем меньше ресурсов вы используете.)

Это приводит к тому, что при сценарии с жесткими ограничениями ресурсов один из процессов может попытаться выделить память для массива в момент, когда достаточно большой блок памяти временно недоступен, что вызывает запуск Python. MemoryError исключение.

Решение этой проблемы может выглядеть примерно так: фрагмент кода ниже: попробуйте выделить память, перехватить исключение, если оно возникнет, затем подождать некоторое время и повторить попытку:

import numpy as np
import time
import datetime
import os
from warnings import warn

def getHugeArray(n, retry=1, warnuser=5):

    # Don't retry too often
    if retry < 0.1:
        retry = 1
    # Don't send a useless flood of warning messages
    if warnuser < retry:
        warnuser = retry

    success, outarray = False, None
    startwait, lastcount = datetime.datetime.now(), 0
    # Keep re-asking the OS for memory allocation until the OS gives it to us
    while success is False:
        try:
            outarray = np.zeros(n)
            success = True
        except MemoryError:
            time.sleep(retry)
            wait = (datetime.datetime.now()-startwait).total_seconds()
            newcount = int(wait/warnuser)
            if newcount > lastcount:
                msg = 'PID {0}: waiting for memory allocation for {1} seconds'
                warn(msg.format(os.getpid(), wait))
                lastcount = newcount

    return outarray

Мой вопрос: как мне создать модульный тест для проверки поведения в режиме ожидания? Или это вообще возможно? Похоже, мне нужно написать тест, который настраивает себя, сначала загружая большую часть ресурсов памяти в моей системе, а затем некоторое время ждет, пока getHugeArray() функция начинает выполняться, а затем освобождает ресурсы, чтобы getHugeArray() может захватить их, и в конечном итоге проверяет, если getHugeArray() функция возвращается с правильным возвращаемым значением.

Является ли это чем-то, что было бы лучше попробовать сделать в рамках интеграционного тестирования, а не в качестве модульного теста? Или в любом случае, какой подходящий инструмент или инфраструктуру тестирования я мог бы использовать для тестирования своего кода в такой ситуации и как мне его настроить?