18 ноября 2015 г.

Парсим бесконечный вывод logcat с таймаутом (Python 3)

    Столкнулся с типичной ситуацией - нужно читать вывод logcat в Python 3. Вообще это не проблема, когда мы точно знаем, что нужная строка там появится. Но бывают ситуации, когда нужно выходить по таймауту:
  • Нужно сделать отказоустойчивость автотеста. Ведь в будущих сборках ожидаемой строки может не быть вообще или ключевые слова могут быть изменены
  • Операция не должна длиться больше заданного времени
    Весь текст ниже рассчитан на таких же нубов, как я. Если кто-то придёт и пояснит по хардкору, как же правильно сделать (кодом, а не словами, словами и я могу), я буду очень благодарен. А пока же мой автотест работает (на самом деле ещё в процессе написания) так, как описано ниже.
    Первое, что нужно понять. Речь идёт только о "бесконечных" процессах. В данном случае вывод лога. Если процесс всё-таки должен завершаться, но вы желаете ограничить время его выполнение (например, у вас обработка большого файла занимает часы), используйте subprocess.run(*popenargs, timeout=X), обёрнутый в try … except subprocess.TimeoutExpired.
    Итак, первая проблема в том, что subprocess.Popen() не имеет своего параметра таймаута, а subprocess.run() хоть и имеет его, но предусматривает, что операция завершится и вернёт код; в случае таймаута будет выброшено исключение, которое нужно будет обработать, но при этом ваш proc.stdout будет пуст. Вторая проблема в том, что лог нужно обрабатывать в while True, а это затрудняет использование subprocess.communicate().
    Моё нубское решение, которое я предлагаю вам, такое:
  1. Готовим переменную с нужным нам объектом subprocess.Popen
  2. Получаем текущее время
  3. Уходим в "бесконечный" цикл:
    • Если с получения текущего времени прошло секунд больше, чем мы желаем:
      • Записываем отладочную информацию, чтобы на утро сказать "ну вот, опять автотест завалился", а затем понять, почему же это произошло
      • Убиваем процесс
      • Уходим из цикла
    • Обрабатываем вывод adb logcat
    Думаю, тут всё понятно. Пишем код:
timeout = 60
proc = subprocess.Popen(['adb','logcat', '-v', 'threadtime', '*:V'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
before = time.time()
while proc.poll() is None: # While proc is alive
    if time.time() - before < timeout:
        # Your code here
        print(proc.stdout.readline().decode()) # Print logcat
        if 'your string' in proc.stdout.readline().decode():
            # Your code here
            proc.kill()
            # return Your success return
    else:
        print('Timeout error')
        proc.kill()
        # return Your timeout return

    Здесь несколько пояснений:
  • Это код в отрыве от контекста. Само-собою, его нужно засунуть в функцию и передавать ему нужные данные с вызовом функции
  • timeout - желаемый таймаут в секундах. Эта переменная отлично ляжет в код функции:
    def spam(timeout=60):
        proc = #...code...
        while proc.poll() is None:
            #...code...
            if time.time() - before < timeout:
                #...code...
    
    
  • subprocess.poll() возвращает None, если дочерний процесс жив и всякое другое, если он уже мёртв. Другими словами, у нас получается выполнение кода, пока процесс жив. Если внутри цикла происходит убийства процесса (proc.kill()), то вместо None прилетает нечто иное.
  • Нельзя уходить из функции, не убив процесс. Ведь logcat выводится бесконечно (пока подключено устройство) и процесс продолжит свою работу, тогда как ваш скрипт продолжит своё выполнение
  • Не стоит использовать universal_newlines=True, потому что в логе гарантированно попадутся символы, которые будут не в той кодировке, которую ожидает Python. Придётся ловить исключения и обрабатывать их. Вместо этого лучше правильно использовать .decode()
    Этот код вполне рабочий, но здесь есть проблема. Типичная ошибка тестировщиков - не воткнуть кабель в телефон или воткнуть не в тот. И казалось бы, что такого. Добавь сравнение if 'waiting for device' in proc.stdout.readline().decode(), но нет. Проблема в том, что в stdout уходит единственная строка, которая не закрывается. То есть .readline() не возвращает результат, пока не воткнёшь кабель в телефон (не включишь отладку, не включишь телефон и прочее). Имеем зависание в строке.

    К счастью, adb имеет минимум три команды, которыми можно разрулить эту ситуацию:
  1. adb getserialno возвращает серийный номер физического устройства и эмулятора. В случае эмулятора будет явное указание на это: emulator-5554
  2. adb get-state возвращает либо unknown, если устройство не готово (не подключено, не включена отладка, не авторизован компьютер), либо device, если всё готово, либо offline, если есть проблемы с драйверами.
  3. adb wait-for-device будет бесконечно ожидать устройства и передаст управление, как только оно появится. Либо передаст сразу, если оно уже доступно. При этом можно сразу после wait-for-device указать дополнительные команды, которые будут выполнены, как только устройство будет доступно. Или же сразу, если оно изначально было доступно:
    C:\Windows\System32>adb wait-for-device get-state
    device
    
  4. adb devices возвращает и серийный номер, и состояние по всем подключенным устройствам.
    В общем, есть из чего выбрать. Например, выполнять команды, как только устройство доступно. Можно и разово проверить доступность, чтобы приступить к остальной части сценария, либо сразу свалить его:

try:
    subprocess.run(['adb', 'wait-for-device'], timeout=5)
    print('Device ready')
    # return True
except subprocess.TimeoutExpired:
    print('Device not ready')
    # return False

    Просто добавляете эту проверку заранее и наслаждаетесь.
if __name__ == '__main__':
    if not device_ready(timeout=5):
        print('Stop!')
    else:
        # your code

    Всем спасибо. Всем замечаниям, если они по делу, да с реальным кодом, буду рад.