Gooey/gooey/tests/processor/test_processor.py

import re
import signal
import unittest

import time

from gooey.gui.processor import ProcessController


class TestProcessor(unittest.TestCase):

    def test_extract_progress(self):
        # should pull out a number based on the supplied
        # regex and expression
        processor = ProcessController(r"^progress: (\d+)%$", None, False, 'utf-8')
        self.assertEqual(processor._extract_progress(b'progress: 50%'), 50)

        processor = ProcessController(r"total: (\d+)%$", None, False, 'utf-8')
        self.assertEqual(processor._extract_progress(b'my cool total: 100%'), 100)

    def test_extract_progress_returns_none_if_no_regex_supplied(self):
        processor = ProcessController(None, None, False, 'utf-8')
        self.assertIsNone(processor._extract_progress(b'Total progress: 100%'))


    def test_extract_progress_returns_none_if_no_match_found(self):
        processor = ProcessController(r'(\d+)%$', None, False, 'utf-8')
        self.assertIsNone(processor._extract_progress(b'No match in dis string'))


    def test_eval_progress(self):
        # given a match in the string, should eval the result
        regex = r'(\d+)/(\d+)$'
        processor = ProcessController(regex, r'x[0] / x[1]', False,False, 'utf-8')
        match = re.search(regex, '50/50')
        self.assertEqual(processor._eval_progress(match), 1.0)
    def test_eval_progress_returns_none_on_failure(self):
        # given a match in the string, should eval the result
        regex = r'(\d+)/(\d+)$'
        processor = ProcessController(regex, r'x[0] *^/* x[1]', False, False,'utf-8')
        match = re.search(regex, '50/50')
        self.assertIsNone(processor._eval_progress(match))


    def test_all_interrupts_halt_process(self):
        """
        TODO: These tests are hella flaky. I'm confident that the feature works. However, getting
        signals, subprocesses and unittest to all play together reliably is proving tricky. It
        primarily seems to come down to how long the time.sleep() is before sending the shutdown
        signal.
        """
        import os
        cmd = 'python ' + os.path.join(os.getcwd(), 'files', 'infinite_loop.py')

        windows_signals = [signal.SIGTERM, signal.CTRL_BREAK_EVENT, signal.CTRL_C_EVENT]

        try:
            for sig in windows_signals:
                print('sig', sig)
                processor = ProcessController(None, None, False, 'utf-8', True, shutdown_signal=sig)

                import subprocess

                processor.run(cmd)
                self.assertTrue(processor.running())

                # super-duper important sleep so that the
                # signal is actually received by the child process
                # see: https://stackoverflow.com/questions/32023719/how-to-simulate-a-terminal-ctrl-c-event-from-a-unittest
                time.sleep(1)
                processor.stop()
                max_wait = time.time() + 4
                while processor.running() and time.time() < max_wait:
                    time.sleep(0.1)
                self.assertFalse(processor.running())
        except KeyboardInterrupt:
            pass