aboutsummaryrefslogtreecommitdiff
path: root/catapult/devil/devil/android/crash_handler_devicetest.py
blob: 6365104deb62340fc1325e034a7dd7b8753448e5 (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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
#!/usr/bin/env python
# Copyright 2017 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import os
import sys
import unittest

if __name__ == '__main__':
  sys.path.append(
      os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', )))

from devil.android import crash_handler
from devil.android import device_errors
from devil.android import device_utils
from devil.android import device_temp_file
from devil.android import device_test_case
from devil.utils import cmd_helper
from devil.utils import reraiser_thread
from devil.utils import timeout_retry


class DeviceCrashTest(device_test_case.DeviceTestCase):

  def setUp(self):
    super(DeviceCrashTest, self).setUp()
    self.device = device_utils.DeviceUtils(self.serial)

  def testCrashDuringCommand(self):
    self.device.EnableRoot()
    with device_temp_file.DeviceTempFile(self.device.adb) as trigger_file:

      trigger_text = 'hello world'

      def victim():
        trigger_cmd = 'echo -n %s > %s; sleep 20' % (
            cmd_helper.SingleQuote(trigger_text),
            cmd_helper.SingleQuote(trigger_file.name))
        crash_handler.RetryOnSystemCrash(
            lambda d: d.RunShellCommand(
                trigger_cmd, shell=True, check_return=True, retries=1,
                as_root=True, timeout=180),
            device=self.device)
        self.assertEquals(
            trigger_text,
            self.device.ReadFile(trigger_file.name, retries=0).strip())
        return True

      def crasher():
        def ready_to_crash():
          try:
            return trigger_text == self.device.ReadFile(
                trigger_file.name, retries=0).strip()
          except device_errors.CommandFailedError:
            return False

        timeout_retry.WaitFor(ready_to_crash, wait_period=2, max_tries=10)
        if not ready_to_crash():
          return False
        self.device.adb.Shell(
            'echo c > /proc/sysrq-trigger',
            expect_status=None, timeout=60, retries=0)
        return True

    self.assertEquals([True, True],
                      reraiser_thread.RunAsync([crasher, victim]))


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