1 #!/usr/bin/env python 2 # Copyright 2015 The Chromium Authors. All rights reserved. 3 # Use of this source code is governed by a BSD-style license that can be 4 # found in the LICENSE file. 5 6 """ 7 Unit tests for the contents of device_utils.py (mostly DeviceUtils). 8 The test will invoke real devices 9 """ 10 11 import os 12 import posixpath 13 import sys 14 import tempfile 15 import unittest 16 17 if __name__ == '__main__': 18 sys.path.append( 19 os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', ))) 20 21 from devil.android import device_test_case 22 from devil.android import device_utils 23 from devil.android.sdk import adb_wrapper 24 from devil.utils import cmd_helper 25 26 _OLD_CONTENTS = "foo" 27 _NEW_CONTENTS = "bar" 28 _DEVICE_DIR = "/data/local/tmp/device_utils_test" 29 _SUB_DIR = "sub" 30 _SUB_DIR1 = "sub1" 31 _SUB_DIR2 = "sub2" 32 33 34 class DeviceUtilsPushDeleteFilesTest(device_test_case.DeviceTestCase): 35 36 def setUp(self): 37 super(DeviceUtilsPushDeleteFilesTest, self).setUp() 38 self.adb = adb_wrapper.AdbWrapper(self.serial) 39 self.adb.WaitForDevice() 40 self.device = device_utils.DeviceUtils( 41 self.adb, default_timeout=10, default_retries=0) 42 43 @staticmethod 44 def _MakeTempFile(contents): 45 """Make a temporary file with the given contents. 46 47 Args: 48 contents: string to write to the temporary file. 49 50 Returns: 51 the tuple contains the absolute path to the file and the file name 52 """ 53 fi, path = tempfile.mkstemp(text=True) 54 with os.fdopen(fi, 'w') as f: 55 f.write(contents) 56 file_name = os.path.basename(path) 57 return (path, file_name) 58 59 @staticmethod 60 def _MakeTempFileGivenDir(directory, contents): 61 """Make a temporary file under the given directory 62 with the given contents 63 64 Args: 65 directory: the temp directory to create the file 66 contents: string to write to the temp file 67 68 Returns: 69 the list contains the absolute path to the file and the file name 70 """ 71 fi, path = tempfile.mkstemp(dir=directory, text=True) 72 with os.fdopen(fi, 'w') as f: 73 f.write(contents) 74 file_name = os.path.basename(path) 75 return (path, file_name) 76 77 @staticmethod 78 def _ChangeTempFile(path, contents): 79 with os.open(path, 'w') as f: 80 f.write(contents) 81 82 @staticmethod 83 def _DeleteTempFile(path): 84 os.remove(path) 85 86 def testPushChangedFiles_noFileChange(self): 87 (host_file_path, file_name) = self._MakeTempFile(_OLD_CONTENTS) 88 device_file_path = "%s/%s" % (_DEVICE_DIR, file_name) 89 self.adb.Push(host_file_path, device_file_path) 90 self.device.PushChangedFiles([(host_file_path, device_file_path)]) 91 result = self.device.RunShellCommand( 92 ['cat', device_file_path], check_return=True, single_line=True) 93 self.assertEqual(_OLD_CONTENTS, result) 94 95 cmd_helper.RunCmd(['rm', host_file_path]) 96 self.device.RemovePath(_DEVICE_DIR, recursive=True, force=True) 97 98 def testPushChangedFiles_singleFileChange(self): 99 (host_file_path, file_name) = self._MakeTempFile(_OLD_CONTENTS) 100 device_file_path = "%s/%s" % (_DEVICE_DIR, file_name) 101 self.adb.Push(host_file_path, device_file_path) 102 103 with open(host_file_path, 'w') as f: 104 f.write(_NEW_CONTENTS) 105 self.device.PushChangedFiles([(host_file_path, device_file_path)]) 106 result = self.device.RunShellCommand( 107 ['cat', device_file_path], check_return=True, single_line=True) 108 self.assertEqual(_NEW_CONTENTS, result) 109 110 cmd_helper.RunCmd(['rm', host_file_path]) 111 self.device.RemovePath(_DEVICE_DIR, recursive=True, force=True) 112 113 def testDeleteFiles(self): 114 host_tmp_dir = tempfile.mkdtemp() 115 (host_file_path, file_name) = self._MakeTempFileGivenDir( 116 host_tmp_dir, _OLD_CONTENTS) 117 118 device_file_path = "%s/%s" % (_DEVICE_DIR, file_name) 119 self.adb.Push(host_file_path, device_file_path) 120 121 cmd_helper.RunCmd(['rm', host_file_path]) 122 self.device.PushChangedFiles([(host_tmp_dir, _DEVICE_DIR)], 123 delete_device_stale=True) 124 filenames = self.device.ListDirectory(_DEVICE_DIR) 125 self.assertEqual([], filenames) 126 127 cmd_helper.RunCmd(['rm', '-rf', host_tmp_dir]) 128 self.device.RemovePath(_DEVICE_DIR, recursive=True, force=True) 129 130 def testPushAndDeleteFiles_noSubDir(self): 131 host_tmp_dir = tempfile.mkdtemp() 132 (host_file_path1, file_name1) = self._MakeTempFileGivenDir( 133 host_tmp_dir, _OLD_CONTENTS) 134 (host_file_path2, file_name2) = self._MakeTempFileGivenDir( 135 host_tmp_dir, _OLD_CONTENTS) 136 137 device_file_path1 = "%s/%s" % (_DEVICE_DIR, file_name1) 138 device_file_path2 = "%s/%s" % (_DEVICE_DIR, file_name2) 139 self.adb.Push(host_file_path1, device_file_path1) 140 self.adb.Push(host_file_path2, device_file_path2) 141 142 with open(host_file_path1, 'w') as f: 143 f.write(_NEW_CONTENTS) 144 cmd_helper.RunCmd(['rm', host_file_path2]) 145 146 self.device.PushChangedFiles([(host_tmp_dir, _DEVICE_DIR)], 147 delete_device_stale=True) 148 result = self.device.RunShellCommand( 149 ['cat', device_file_path1], check_return=True, single_line=True) 150 self.assertEqual(_NEW_CONTENTS, result) 151 152 filenames = self.device.ListDirectory(_DEVICE_DIR) 153 self.assertEqual([file_name1], filenames) 154 155 cmd_helper.RunCmd(['rm', '-rf', host_tmp_dir]) 156 self.device.RemovePath(_DEVICE_DIR, recursive=True, force=True) 157 158 def testPushAndDeleteFiles_SubDir(self): 159 host_tmp_dir = tempfile.mkdtemp() 160 host_sub_dir1 = "%s/%s" % (host_tmp_dir, _SUB_DIR1) 161 host_sub_dir2 = "%s/%s/%s" % (host_tmp_dir, _SUB_DIR, _SUB_DIR2) 162 cmd_helper.RunCmd(['mkdir', '-p', host_sub_dir1]) 163 cmd_helper.RunCmd(['mkdir', '-p', host_sub_dir2]) 164 165 (host_file_path1, file_name1) = self._MakeTempFileGivenDir( 166 host_tmp_dir, _OLD_CONTENTS) 167 (host_file_path2, file_name2) = self._MakeTempFileGivenDir( 168 host_tmp_dir, _OLD_CONTENTS) 169 (host_file_path3, file_name3) = self._MakeTempFileGivenDir( 170 host_sub_dir1, _OLD_CONTENTS) 171 (host_file_path4, file_name4) = self._MakeTempFileGivenDir( 172 host_sub_dir2, _OLD_CONTENTS) 173 174 device_file_path1 = "%s/%s" % (_DEVICE_DIR, file_name1) 175 device_file_path2 = "%s/%s" % (_DEVICE_DIR, file_name2) 176 device_file_path3 = "%s/%s/%s" % (_DEVICE_DIR, _SUB_DIR1, file_name3) 177 device_file_path4 = "%s/%s/%s/%s" % (_DEVICE_DIR, _SUB_DIR, 178 _SUB_DIR2, file_name4) 179 180 self.adb.Push(host_file_path1, device_file_path1) 181 self.adb.Push(host_file_path2, device_file_path2) 182 self.adb.Push(host_file_path3, device_file_path3) 183 self.adb.Push(host_file_path4, device_file_path4) 184 185 with open(host_file_path1, 'w') as f: 186 f.write(_NEW_CONTENTS) 187 cmd_helper.RunCmd(['rm', host_file_path2]) 188 cmd_helper.RunCmd(['rm', host_file_path4]) 189 190 self.device.PushChangedFiles([(host_tmp_dir, _DEVICE_DIR)], 191 delete_device_stale=True) 192 result = self.device.RunShellCommand( 193 ['cat', device_file_path1], check_return=True, single_line=True) 194 self.assertEqual(_NEW_CONTENTS, result) 195 196 filenames = self.device.ListDirectory(_DEVICE_DIR) 197 self.assertIn(file_name1, filenames) 198 self.assertIn(_SUB_DIR1, filenames) 199 self.assertIn(_SUB_DIR, filenames) 200 self.assertEqual(3, len(filenames)) 201 202 result = self.device.RunShellCommand( 203 ['cat', device_file_path3], check_return=True, single_line=True) 204 self.assertEqual(_OLD_CONTENTS, result) 205 206 filenames = self.device.ListDirectory( 207 posixpath.join(_DEVICE_DIR, _SUB_DIR, _SUB_DIR2)) 208 self.assertEqual([], filenames) 209 210 cmd_helper.RunCmd(['rm', '-rf', host_tmp_dir]) 211 self.device.RemovePath(_DEVICE_DIR, recursive=True, force=True) 212 213 def testRestartAdbd(self): 214 def get_adbd_pid(): 215 # TODO(catapult:#3215): Migrate to device.GetPids(). 216 ps_output = self.device.RunShellCommand(['ps'], check_return=True) 217 for ps_line in ps_output: 218 if 'adbd' in ps_line: 219 return ps_line.split()[1] 220 self.fail('Unable to find adbd') 221 222 old_adbd_pid = get_adbd_pid() 223 self.device.RestartAdbd() 224 new_adbd_pid = get_adbd_pid() 225 self.assertNotEqual(old_adbd_pid, new_adbd_pid) 226 227 228 if __name__ == '__main__': 229 unittest.main() 230
