"""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 TestThreadSubprocess(unittest.TestCase):

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

    def _create(self):
        self.sub = subprocess.Popen(
            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.poll() is None)
        os.kill(self.sub.pid, signal.SIGTERM)
        time.sleep(0.1)
        self.assertTrue(
            self.sub.poll() is not None,
            'Process %d could not be killed.' % (self.sub.pid))

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

    def test_no_thr(self):
        """Subprocess created in main."""
        self._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.assertTrue(time.time() - now < 0.8)
        t.join()


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