"""Test issues with threading and signals for http://bugs.python.org/issue1975.
"""

import os
import time
import threading
import select
import signal
import subprocess
import unittest

class _PopenWrapper(object):

    # Simple wrapper around subprocess.Popen.
    #
    # We use this so we can get the PID and/or subprocess objects back when we
    # create the subprocess in a thread.

    def __init__(self, *args, **kwargs):
        self._args = args
        self._kwargs = kwargs
        self.pid = None
        self.sub = None

    def create(self):
        """Start the subprocess."""
        self.sub = subprocess.Popen(*self._args, **self._kwargs)
        self.pid = self.sub.pid


class TestThreadSubprocess(unittest.TestCase):

    """Tests kill-ability of a subprocess created in and not in a thread."""

    def setUp(self):
        self._sub_wrapper = _PopenWrapper(
            args=('sleep', '60'), cwd=None, env=None,
            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            close_fds=True)

    def _verify(self):
        time.sleep(0.1)
        self.assertTrue(self._sub_wrapper.sub.poll() is None)
        os.kill(self._sub_wrapper.pid, signal.SIGTERM)
        time.sleep(0.1)
        self.assertTrue(
            self._sub_wrapper.sub.poll() is not None,
            'Process %d could not be killed.' % (self._sub_wrapper.pid))

    def test_thr(self):
        """Subprocess created in a thread."""
        thr = threading.Thread(target=self._sub_wrapper.create)
        thr.start()
        thr.join()
        self._verify()

    def test_no_thr(self):
        """Subprocess created in main."""
        self._sub_wrapper.create()
        self._verify()


class TestSignal(unittest.TestCase):

    """Tests that signals are properly received by the main thread.

    The signal should be received by the main thread, interrupting the current
    operation where applicable.
    """

    def _thread(self):
        # Kill the current process with SIGINT.
        time.sleep(0.05)
        os.kill(os.getpid(), signal.SIGINT)
        time.sleep(0.05)

    def test_signal(self):
        t = threading.Thread(target=self._thread)
        t.start()
        now = time.time()
        try:
            # Will block for 1 second, unless interrupted by a signal.
            select.select((), (), (), 1)
        except KeyboardInterrupt:
            pass

        self.assertLess(time.time() - now, 0.8)
        t.join()


if __name__ == '__main__':
    unittest.main()