summaryrefslogtreecommitdiff
path: root/functional_tests/test_multiprocessing/test_process_timeout.py
blob: 535ecdba9e3744f4eeef90649bb384b3af878850 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import os
import unittest

from nose.plugins import PluginTester
from nose.plugins.skip import SkipTest
from nose.plugins.multiprocess import MultiProcess

support = os.path.join(os.path.dirname(__file__), 'support')


def setup():
    try:
        import multiprocessing
        if 'active' in MultiProcess.status:
            raise SkipTest("Multiprocess plugin is active. Skipping tests of "
                           "plugin itself.")
    except ImportError:
        raise SkipTest("multiprocessing module not available")



class TestMPTimeout(PluginTester, unittest.TestCase):
    activate = '--processes=2'
    args = ['--process-timeout=1']
    plugins = [MultiProcess()]
    suitepath = os.path.join(support, 'timeout.py')

    def runTest(self):
        assert "TimedOutException: 'timeout.test_timeout'" in self.output


class TestMPTimeoutPass(TestMPTimeout):
    args = ['--process-timeout=3']

    def runTest(self):
        assert "TimedOutException: 'timeout.test_timeout'" not in self.output
        assert str(self.output).strip().endswith('OK')