unittests/framework/test/test_base.py

   1 # coding=utf-8
   2 # Copyright (c) 2014, 2016, 2019 Intel Corporation
   3
   4 # Permission is hereby granted, free of charge, to any person obtaining a copy
   5 # of this software and associated documentation files (the "Software"), to deal
   6 # in the Software without restriction, including without limitation the rights
   7 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   8 # copies of the Software, and to permit persons to whom the Software is
   9 # furnished to do so, subject to the following conditions:
  10
  11 # The above copyright notice and this permission notice shall be included in
  12 # all copies or substantial portions of the Software.
  13
  14 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  19 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  20 # SOFTWARE.
  21
  22 """ Tests for the exectest module """
  23
  24 import os
  25 import textwrap
  26 try:
  27     import subprocess32 as subprocess
  28 except ImportError:
  29     import subprocess
  30
  31 import pytest
  32
  33 from framework import dmesg
  34 from framework import log
  35 from framework import monitoring
  36 from framework import status
  37 from framework.options import _Options as Options
  38 from framework.test import base
  39
  40 from ..test_status import PROBLEMS
  41 from .. import skip
  42
  43 # pylint: disable=invalid-name,no-self-use,protected-access
  44
  45
  46 class _Test(base.Test):
  47     """Helper with stubbed interpret_results method."""
  48     def interpret_result(self):
  49         super(_Test, self).interpret_result()
  50
  51
  52 class TestTest(object):
  53     """Tests for the Test class."""
  54
  55     class TestRun(object):
  56         """Tests for Test.run."""
  57
  58         def test_return_early_when_run_command_excepts(self, mocker):
  59             """Test.run exits early when Test._run_command raises an exception.
  60             """
  61             t = _Test(['foo'])
  62             mocker.patch.object(t, '_run_command',
  63                                 side_effect=base.TestRunError('foo', 'pass'))
  64             mocker.patch.object(t, 'interpret_result',
  65                                 side_effect=Exception('Test failure'))
  66
  67             t.run()
  68
  69     @skip.posix
  70     class TestRunCommand(object):
  71         """Tests for Test._run_command."""
  72
  73         @classmethod
  74         def setup_class(cls):
  75             if os.name == 'posix':
  76                 cls.command = ['sleep', '60']
  77             else:
  78                 cls.command = ['timeout', '/t', '/nobreak', '60']
  79
  80         @pytest.mark.timeout(6)
  81         def test_timeout_kill_children(self, mocker, tmpdir):
  82             """test.base.Test: kill children if terminate fails.
  83
  84             This creates a process that forks multiple times, and then checks
  85             that the children have been killed.
  86
  87             This test could leave processes running if it fails.
  88             """
  89             # This function is the only user of psutil, and by putting it in
  90             # the function itself we avoid needed to try/except it, and then
  91             # skip if we don't have it.
  92             psutil = pytest.importorskip('psutil')
  93
  94             class PopenProxy(object):
  95                 """An object that proxies Popen, and saves the Popen instance
  96                 as an attribute.
  97
  98                 This is useful for testing the Popen instance.
  99                 """
 100
 101                 def __init__(self):
 102                     self.popen = None
 103
 104                 def __call__(self, *args, **kwargs):
 105                     self.popen = subprocess.Popen(*args, **kwargs)
 106
 107                     # if communicate is called successfully then the proc will
 108                     # be reset to None, which will make the test fail.
 109                     self.popen.communicate = mocker.Mock(
 110                         return_value=('out', 'err'))
 111
 112                     return self.popen
 113
 114             # localpath doesn't have a seek option
 115             with tmpdir.join('test').open(mode='w', ensure=True) as f:
 116                 # Create a file that will be executed as a python script Create
 117                 # a process with two subproccesses (not threads) that will run
 118                 # for a long time.
 119                 f.write(textwrap.dedent("""\
 120                     import time
 121                     from multiprocessing import Process
 122
 123                     def p():
 124                         for _ in range(100):
 125                             time.sleep(1)
 126
 127                     if __name__ == "__main__":
 128                         a = Process(target=p)
 129                         b = Process(target=p)
 130                         a.start()
 131                         b.start()
 132                         a.join()
 133                         b.join()
 134                 """))
 135                 f.seek(0)  # we'll need to read the file back
 136
 137                 # Create an object that will return a popen object, but also
 138                 # store it so we can access it later
 139                 proxy = PopenProxy()
 140
 141                 test = _Test(['python3', str(f)])
 142                 test.timeout = 1
 143
 144                 # mock out subprocess.Popen with our proxy object
 145                 mock_subp = mocker.patch('framework.test.base.subprocess')
 146                 mock_subp.Popen = proxy
 147                 mock_subp.TimeoutExpired = subprocess.TimeoutExpired
 148                 test.run()
 149
 150                 # Check to see if the Popen has children, even after it should
 151                 # have received a TimeoutExpired.
 152                 proc = psutil.Process(os.getsid(proxy.popen.pid))
 153                 children = proc.children(recursive=True)
 154
 155             if children:
 156                 # If there are still running children attempt to clean them up,
 157                 # starting with the final generation and working back to the
 158                 # first
 159                 for child in reversed(children):
 160                     child.kill()
 161
 162                 raise Exception('Test process had children when it should not')
 163
 164         @pytest.mark.slow
 165         @pytest.mark.timeout(6)
 166         def test_timeout(self):
 167             """test.base.Test: Stops running test after timeout expires.
 168
 169             This is a little bit of extra time here, but without a sleep of 60
 170             seconds if the test runs 6 seconds it's run too long, we do need to
 171             allow a bit of time for teardown to happen.
 172             """
 173             test = _Test(self.command)
 174             test.timeout = 1
 175             test.run()
 176
 177         @pytest.mark.slow
 178         @pytest.mark.timeout(6)
 179         def test_timeout_status(self):
 180             """test.base.Test: Setst the status to 'timeout' when then timeout
 181             is exceeded
 182
 183             This is a little bit of extra time here, but without a sleep of 60
 184             seconds if the test runs 6 seconds it's run too long, we do need to
 185             allow a bit of time for teardown to happen.
 186             """
 187             test = _Test(self.command)
 188             test.timeout = 1
 189             test.run()
 190             assert test.result.result is status.TIMEOUT
 191
 192     class TestExecuteTraceback(object):
 193         """Test.execute tests for Traceback handling."""
 194
 195         class Sentinel(Exception):
 196             pass
 197
 198         @pytest.fixture
 199         def shared_test(self, mocker):
 200             """Do some setup."""
 201             test = _Test(['foo'])
 202             test.run = mocker.Mock(side_effect=self.Sentinel)
 203
 204             test.execute(mocker.Mock(spec=str),
 205                          mocker.Mock(spec=log.BaseLog),
 206                          {'dmesg': mocker.Mock(spec=dmesg.BaseDmesg),
 207                           'monitor': mocker.Mock(spec=monitoring.Monitoring)})
 208             return test.result
 209
 210         def test_result(self, shared_test):
 211             """Test.execute (exception): Sets the result to fail."""
 212             assert shared_test.result is status.FAIL
 213
 214         def test_traceback(self, shared_test):
 215             """Test.execute (exception): Sets the traceback.
 216
 217             It's fragile to record the actual traceback, and it's unlikely that
 218             it can easily be implemented differently than the way the original
 219             code is implemented, so this doesn't do that, it just verifies
 220             there is a value.
 221             """
 222             assert shared_test.traceback != ''
 223             assert isinstance(shared_test.traceback, str)
 224
 225         def test_exception(self, shared_test):
 226             """Test.execute (exception): Sets the exception.
 227
 228             This is much like the traceback, it's difficult to get the correct
 229             value, so just make sure it's being set.
 230             """
 231             assert shared_test.exception != ''
 232             assert isinstance(shared_test.exception, str)
 233
 234     class TestCommand(object):
 235         """Tests for Test.command."""
 236
 237         def test_string_for_command(self):
 238             """Asserts if it is passed a string instead of a list."""
 239             with pytest.raises(AssertionError):
 240                 _Test('foo')
 241
 242         def test_mutation(self):
 243             """test.base.Test.command: does not mutate the value it was
 244             provided.
 245
 246             There was a very subtle bug in all.py that causes the key values to
 247             be changed before they are assigned in some cases. This is because
 248             the right side of an assignment is evalated before the left side,
 249             so:
 250
 251             >>> profile = {}
 252             >>> args = ['a', 'b']
 253             >>> profile[' '.join(args)] = PiglitGLTest(args)
 254             >>> list(profile.keys())
 255             ['bin/a b']
 256             """
 257             class MyTest(_Test):
 258                 def __init__(self, *args, **kwargs):
 259                     super(MyTest, self).__init__(*args, **kwargs)
 260                     self._command[0] = 'bin/' + self._command[0]
 261
 262             args = ['a', 'b']
 263             _Test(args)
 264
 265             assert args == ['a', 'b']
 266
 267     class TestInterpretResult(object):
 268         """Tests for Test.interpret_result."""
 269
 270         def test_returncode_greater_zero(self):
 271             """A test with status > 0 is fail."""
 272             test = _Test(['foobar'])
 273             test.result.returncode = 1
 274             test.result.out = 'this is some\nstdout'
 275             test.result.err = 'this is some\nerrors'
 276             test.interpret_result()
 277
 278             assert test.result.result is status.FAIL
 279
 280         def test_crash_subtest_before_start(self):
 281             """A test for a test with a subtest, that crashes at the start
 282             of the run.
 283             """
 284             test = _Test(['foobar'])
 285             test.result.returncode = -1
 286             for x in (str(y) for y in range(5)):
 287                 test.result.subtests[x] = status.NOTRUN
 288             test.interpret_result()
 289
 290             assert test.result.result is status.CRASH
 291             assert test.result.subtests['0'] is status.CRASH
 292             for x in (str(y) for y in range(1, 5)):
 293                 assert test.result.subtests[x] is status.NOTRUN
 294
 295         def test_crash_subtest_mid(self):
 296             """A test for a test with a subtest, that crashes in the middle
 297             of the run.
 298             """
 299             test = _Test(['foobar'])
 300             test.result.returncode = -1
 301             for x in (str(y) for y in range(2)):
 302                 test.result.subtests[x] = status.PASS
 303             for x in (str(y) for y in range(2, 5)):
 304                 test.result.subtests[x] = status.NOTRUN
 305             test.interpret_result()
 306
 307             assert test.result.result is status.CRASH
 308             for x in (str(y) for y in range(2)):
 309                 assert test.result.subtests[x] is status.PASS
 310             assert test.result.subtests['2'] is status.CRASH
 311             for x in (str(y) for y in range(3, 5)):
 312                 assert test.result.subtests[x] is status.NOTRUN
 313
 314
 315 class TestWindowResizeMixin(object):
 316     """Tests for the WindowResizeMixin class."""
 317
 318     def test_rerun(self):
 319         """test.base.WindowResizeMixin: runs multiple when spurious resize
 320         detected.
 321         """
 322         # Because of Python's inheritance order we need another mixin.
 323         class Mixin(object):
 324             def __init__(self, *args, **kwargs):
 325                 super(Mixin, self).__init__(*args, **kwargs)
 326                 self.__return_spurious = True
 327
 328             def _run_command(self, *args, **kwargs):  # pylint: disable=unused-argument
 329                 self.result.returncode = None
 330
 331                 # IF this is run only once we'll have "got spurious window resize"
 332                 # in result.out, if it runs multiple times we'll get 'all good'
 333                 if self.__return_spurious:
 334                     self.result.out = "Got spurious window resize"
 335                     self.__return_spurious = False
 336                 else:
 337                     self.result.out = 'all good'
 338
 339         class Test_(base.WindowResizeMixin, Mixin, _Test):
 340             pass
 341
 342         test = Test_(['foo'])
 343         test.run()
 344         assert test.result.out == 'all good'
 345
 346
 347 class TestValgrindMixin(object):
 348     """Tests for the ValgrindMixin class."""
 349
 350     def test_command(self, mocker):
 351         """test.base.ValgrindMixin.command: self.command doesn't change."""
 352         opts = mocker.patch('framework.test.base.OPTIONS',
 353                             new_callable=Options)
 354
 355         class Test(base.ValgrindMixin, _Test):
 356             pass
 357
 358         opts.valgrind = True
 359
 360         test = Test(['foo'])
 361         assert test.command == ['foo']
 362
 363     def test_keys(self, mocker):
 364         """test.base.ValgrindMixin.keys: return 'valgrind' with its keys."""
 365         opts = mocker.patch('framework.test.base.OPTIONS',
 366                             new_callable=Options)
 367
 368         class Test(base.ValgrindMixin, _Test):
 369             pass
 370
 371         opts.valgrind = True
 372
 373         test = Test(['foo'])
 374         assert test.keys() == ['valgrind', '--quiet', '--error-exitcode=1',
 375                     '--tool=memcheck']
 376
 377     class TestRun(object):
 378         """Tests for the run method."""
 379
 380         @classmethod
 381         def setup_class(cls):
 382             class _NoRunTest(_Test):
 383                 def run(self):
 384                     self.interpret_result()
 385
 386             class Test(base.ValgrindMixin, _NoRunTest):
 387                 pass
 388
 389             cls.test = Test
 390
 391         # The ids function here is a bit of a hack to work around the
 392         # pytest-timeout plugin, which is broken. when 'timeout' is passed as a
 393         # string using str it tries to grab that value and a flaming pile
 394         # ensues
 395         @pytest.mark.parametrize("starting", PROBLEMS,
 396                                  ids=lambda x: str(x).upper())
 397         def test_problem_status_changes_valgrind_enabled(self, starting, mocker):
 398             """When running with valgrind mode we're actually testing the test
 399             binary itself, so any status other than pass is irrelevant, and
 400             should be marked as skip.
 401             """
 402             mock_opts = mocker.patch('framework.test.base.OPTIONS',
 403                                      new_callable=Options)
 404             mock_opts.valgrind = True
 405
 406             test = self.test(['foo'])
 407             test.result.result = starting
 408             test.run()
 409             assert test.result.result is status.SKIP
 410
 411         @pytest.mark.parametrize("starting", PROBLEMS,
 412                                  ids=lambda x: str(x).upper())
 413         def test_problems_with_valgrind_disabled(self, starting, mocker):
 414             """When valgrind is disabled nothing should change
 415             """
 416             mock_opts = mocker.patch('framework.test.base.OPTIONS',
 417                                      new_callable=Options)
 418             mock_opts.valgrind = False
 419
 420             test = self.test(['foo'])
 421             test.result.result = starting
 422             test.result.returncode = 0
 423             test.run()
 424             assert test.result.result is starting
 425
 426         def test_passed_valgrind(self, mocker):
 427             """test.base.ValgrindMixin.run: when test is 'pass' and returncode
 428             is '0' result is pass.
 429             """
 430             mock_opts = mocker.patch('framework.test.base.OPTIONS',
 431                                      new_callable=Options)
 432             test = self.test(['foo'])
 433             mock_opts.valgrind = True
 434             test.result.result = status.PASS
 435             test.result.returncode = 0
 436             test.run()
 437             assert test.result.result is status.PASS
 438
 439         def test_failed_valgrind(self, mocker):
 440             """test.base.ValgrindMixin.run: when a test is 'pass' but
 441             returncode is not 0 it's 'fail'.
 442             """
 443             mock_opts = mocker.patch('framework.test.base.OPTIONS',
 444                                      new_callable=Options)
 445             mock_opts.valgrind = True
 446             test = self.test(['foo'])
 447             test.result.result = status.PASS
 448             test.result.returncode = 1
 449             test.interpret_result()
 450             assert test.result.result is status.FAIL
 451
 452
 453 class TestReducedProcessMixin(object):
 454     """Tests for the ReducedProcessMixin class."""
 455
 456     class MPTest(base.ReducedProcessMixin, _Test):
 457         def _resume(self, current):
 458             return [self.command[0]] + self._expected[current:]
 459
 460         def _is_subtest(self, line):
 461             return line.startswith('TEST')
 462
 463     def test_populate_subtests(self):
 464         test = self.MPTest(['foobar'], subtests=['a', 'b', 'c'])
 465         assert set(test.result.subtests.keys()) == {'a', 'b', 'c'}
 466
 467     class TestRunCommand(object):
 468         """Tests for the _run_command method."""
 469
 470         @pytest.fixture(scope='module')
 471         def test_class(self):
 472             """Defines a test class that uses generators to ease testing."""
 473             class _Shim(object):
 474                 """This shim goes between the Mixin and the Test class and
 475                 provides a way to set the output of the test.
 476                 """
 477
 478                 def __init__(self, *args, **kwargs):
 479                     super(_Shim, self).__init__(*args, **kwargs)
 480                     self.gen_rcode = None
 481                     self.gen_out = None
 482                     self.get_err = None
 483
 484                 def _run_command(self, *args, **kwargs):  # pylint: disable=unused-argument
 485                     # pylint: disable=no-member
 486                     self.result.returncode = next(self.gen_rcode)
 487                     self.result.out = next(self.gen_out)
 488                     self.result.err = next(self.gen_err)
 489
 490             class Test(base.ReducedProcessMixin, _Shim, _Test):
 491                 """The actual Class returned by the fixture.
 492
 493                 This class implements the abstract bits from
 494                 ReducedProcessMixin, and inserts the _Shim class. The
 495                 _is_subtest method is implemented such that any line starting
 496                 with SUBTEST is a subtest.
 497                 """
 498
 499                 def _is_subtest(self, line):
 500                     return line.startswith('SUBTEST')
 501
 502                 def _resume(self, cur, **kwargs):  # pylint: disable=unused-argument
 503                     return self._expected[cur:]
 504
 505                 def interpret_result(self):
 506                     name = None
 507
 508                     for line in self.result.out.split('\n'):
 509                         if self._is_subtest(line):
 510                             name = line[len('SUBTEST: '):]
 511                         elif line.startswith('RESULT: '):
 512                             self.result.subtests[name] = line[len('RESULT: '):]
 513                             name = None
 514
 515             return Test
 516
 517         def test_result(self, test_class):
 518             """Test result attributes."""
 519             test = test_class(['foobar'], ['a', 'b'])
 520             test.gen_out = iter(['SUBTEST: a', 'SUBTEST: b'])
 521             test.gen_err = iter(['err output', 'err output'])
 522             test.gen_rcode = iter([2, 0])
 523             test._run_command()
 524             assert test.result.out == \
 525                 'SUBTEST: a\n\n====RESUME====\n\nSUBTEST: b'
 526             assert test.result.err == \
 527                 'err output\n\n====RESUME====\n\nerr output'
 528             assert test.result.returncode == 2
 529
 530         @pytest.mark.timeout(5)
 531         def test_infinite_loop(self, test_class):
 532             """Test that we don't get into an infinite loop."""
 533             test = test_class(['foobar'], ['a', 'b'])
 534             test.gen_out = iter(['a', 'a'])
 535             test.gen_err = iter(['a', 'a'])
 536             test.gen_rcode = iter([1, 1])
 537             test._run_command()
 538
 539         def test_crash_first(self, test_class):
 540             """Handles the first test crashing."""
 541             test = test_class(['foo'], ['a', 'b'])
 542             test.gen_out = iter(['', 'SUBTEST: a'])
 543             test.gen_err = iter(['', ''])
 544             test.gen_rcode = iter([1, 0])
 545
 546             # Since interpret_result isn't called this would normally be left
 547             # as NOTRUN, but we want to ensure that _run_command isn't mucking
 548             # with it, so we set it to this PASS, which acts as a sentinel
 549             test.result.subtests['b'] = status.PASS
 550             test._run_command()
 551
 552             assert test.result.subtests['a'] is status.CRASH
 553             assert test.result.subtests['b'] is status.PASS
 554
 555         def test_middle_crash(self, test_class):
 556             """handle the final subtest crashing."""
 557             test = test_class(['foo'], ['a', 'b', 'c'])
 558             test.gen_out = iter(['SUBTEST: a\nRESULT: pass\nSUBTEST: b\n',
 559                                  'SUBTEST: c\nRESULT: pass\n'])
 560             test.gen_err = iter(['', ''])
 561             test.gen_rcode = iter([1, 0])
 562
 563             test._run_command()
 564             test.interpret_result()
 565
 566             assert test.result.subtests['a'] == status.PASS
 567             assert test.result.subtests['b'] == status.CRASH
 568             assert test.result.subtests['c'] == status.PASS
 569
 570         def test_final_crash(self, test_class):
 571             """handle the final subtest crashing."""
 572             test = test_class(['foo'], ['a', 'b', 'c'])
 573             test.gen_out = iter(['SUBTEST: a\nRESULT: pass\n'
 574                                  'SUBTEST: b\nRESULT: pass\n'
 575                                  'SUBTEST: c\n'])
 576             test.gen_err = iter([''])
 577             test.gen_rcode = iter([1])
 578
 579             test._run_command()
 580             test.interpret_result()
 581
 582             assert test.result.subtests['a'] == status.PASS
 583             assert test.result.subtests['b'] == status.PASS
 584             assert test.result.subtests['c'] == status.CRASH