OLD | NEW |
| (Empty) |
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 tempfile | |
13 import unittest | |
14 | |
15 from pylib import cmd_helper | |
16 from pylib import constants | |
17 from pylib.device import adb_wrapper | |
18 from pylib.device import device_utils | |
19 from pylib.utils import md5sum | |
20 | |
21 _OLD_CONTENTS = "foo" | |
22 _NEW_CONTENTS = "bar" | |
23 _DEVICE_DIR = "/data/local/tmp/device_utils_test" | |
24 _SUB_DIR = "sub" | |
25 _SUB_DIR1 = "sub1" | |
26 _SUB_DIR2 = "sub2" | |
27 | |
28 class DeviceUtilsPushDeleteFilesTest(unittest.TestCase): | |
29 | |
30 def setUp(self): | |
31 devices = adb_wrapper.AdbWrapper.Devices() | |
32 assert devices, 'A device must be attached' | |
33 self.adb = devices[0] | |
34 self.adb.WaitForDevice() | |
35 self.device = device_utils.DeviceUtils( | |
36 self.adb, default_timeout=10, default_retries=0) | |
37 default_build_type = os.environ.get('BUILDTYPE', 'Debug') | |
38 constants.SetBuildType(default_build_type) | |
39 | |
40 @staticmethod | |
41 def _MakeTempFile(contents): | |
42 """Make a temporary file with the given contents. | |
43 | |
44 Args: | |
45 contents: string to write to the temporary file. | |
46 | |
47 Returns: | |
48 the tuple contains the absolute path to the file and the file name | |
49 """ | |
50 fi, path = tempfile.mkstemp(text=True) | |
51 with os.fdopen(fi, 'w') as f: | |
52 f.write(contents) | |
53 file_name = os.path.basename(path) | |
54 return (path, file_name) | |
55 | |
56 @staticmethod | |
57 def _MakeTempFileGivenDir(directory, contents): | |
58 """Make a temporary file under the given directory | |
59 with the given contents | |
60 | |
61 Args: | |
62 directory: the temp directory to create the file | |
63 contents: string to write to the temp file | |
64 | |
65 Returns: | |
66 the list contains the absolute path to the file and the file name | |
67 """ | |
68 fi, path = tempfile.mkstemp(dir=directory, text=True) | |
69 with os.fdopen(fi, 'w') as f: | |
70 f.write(contents) | |
71 file_name = os.path.basename(path) | |
72 return (path, file_name) | |
73 | |
74 @staticmethod | |
75 def _ChangeTempFile(path, contents): | |
76 with os.open(path, 'w') as f: | |
77 f.write(contents) | |
78 | |
79 @staticmethod | |
80 def _DeleteTempFile(path): | |
81 os.remove(path) | |
82 | |
83 def testPushChangedFiles_noFileChange(self): | |
84 (host_file_path, file_name) = self._MakeTempFile(_OLD_CONTENTS) | |
85 device_file_path = "%s/%s" % (_DEVICE_DIR, file_name) | |
86 self.adb.Push(host_file_path, device_file_path) | |
87 self.device.PushChangedFiles([(host_file_path, device_file_path)]) | |
88 result = self.device.RunShellCommand(['cat', device_file_path], | |
89 single_line=True) | |
90 self.assertEqual(_OLD_CONTENTS, result) | |
91 | |
92 cmd_helper.RunCmd(['rm', host_file_path]) | |
93 self.device.RunShellCommand(['rm', '-rf', _DEVICE_DIR]) | |
94 | |
95 def testPushChangedFiles_singleFileChange(self): | |
96 (host_file_path, file_name) = self._MakeTempFile(_OLD_CONTENTS) | |
97 device_file_path = "%s/%s" % (_DEVICE_DIR, file_name) | |
98 self.adb.Push(host_file_path, device_file_path) | |
99 | |
100 with open(host_file_path, 'w') as f: | |
101 f.write(_NEW_CONTENTS) | |
102 self.device.PushChangedFiles([(host_file_path, device_file_path)]) | |
103 result = self.device.RunShellCommand(['cat', device_file_path], | |
104 single_line=True) | |
105 self.assertEqual(_NEW_CONTENTS, result) | |
106 | |
107 cmd_helper.RunCmd(['rm', host_file_path]) | |
108 self.device.RunShellCommand(['rm', '-rf', _DEVICE_DIR]) | |
109 | |
110 def testDeleteFiles(self): | |
111 host_tmp_dir = tempfile.mkdtemp() | |
112 (host_file_path, file_name) = self._MakeTempFileGivenDir( | |
113 host_tmp_dir, _OLD_CONTENTS) | |
114 | |
115 device_file_path = "%s/%s" % (_DEVICE_DIR, file_name) | |
116 self.adb.Push(host_file_path, device_file_path) | |
117 | |
118 cmd_helper.RunCmd(['rm', host_file_path]) | |
119 self.device.PushChangedFiles([(host_tmp_dir, _DEVICE_DIR)], | |
120 delete_device_stale=True) | |
121 result = self.device.RunShellCommand(['ls', _DEVICE_DIR], single_line=True) | |
122 self.assertEqual('', result) | |
123 | |
124 cmd_helper.RunCmd(['rm', '-rf', host_tmp_dir]) | |
125 self.device.RunShellCommand(['rm', '-rf', _DEVICE_DIR]) | |
126 | |
127 def testPushAndDeleteFiles_noSubDir(self): | |
128 host_tmp_dir = tempfile.mkdtemp() | |
129 (host_file_path1, file_name1) = self._MakeTempFileGivenDir( | |
130 host_tmp_dir, _OLD_CONTENTS) | |
131 (host_file_path2, file_name2) = self._MakeTempFileGivenDir( | |
132 host_tmp_dir, _OLD_CONTENTS) | |
133 | |
134 device_file_path1 = "%s/%s" % (_DEVICE_DIR, file_name1) | |
135 device_file_path2 = "%s/%s" % (_DEVICE_DIR, file_name2) | |
136 self.adb.Push(host_file_path1, device_file_path1) | |
137 self.adb.Push(host_file_path2, device_file_path2) | |
138 | |
139 with open(host_file_path1, 'w') as f: | |
140 f.write(_NEW_CONTENTS) | |
141 cmd_helper.RunCmd(['rm', host_file_path2]) | |
142 | |
143 self.device.PushChangedFiles([(host_tmp_dir, _DEVICE_DIR)], | |
144 delete_device_stale=True) | |
145 result = self.device.RunShellCommand(['cat', device_file_path1], | |
146 single_line=True) | |
147 self.assertEqual(_NEW_CONTENTS, result) | |
148 result = self.device.RunShellCommand(['ls', _DEVICE_DIR], single_line=True) | |
149 self.assertEqual(file_name1, result) | |
150 | |
151 self.device.RunShellCommand(['rm', '-rf', _DEVICE_DIR]) | |
152 cmd_helper.RunCmd(['rm', '-rf', host_tmp_dir]) | |
153 | |
154 def testPushAndDeleteFiles_SubDir(self): | |
155 host_tmp_dir = tempfile.mkdtemp() | |
156 host_sub_dir1 = "%s/%s" % (host_tmp_dir, _SUB_DIR1) | |
157 host_sub_dir2 = "%s/%s/%s" % (host_tmp_dir, _SUB_DIR, _SUB_DIR2) | |
158 cmd_helper.RunCmd(['mkdir', '-p', host_sub_dir1]) | |
159 cmd_helper.RunCmd(['mkdir', '-p', host_sub_dir2]) | |
160 | |
161 (host_file_path1, file_name1) = self._MakeTempFileGivenDir( | |
162 host_tmp_dir, _OLD_CONTENTS) | |
163 (host_file_path2, file_name2) = self._MakeTempFileGivenDir( | |
164 host_tmp_dir, _OLD_CONTENTS) | |
165 (host_file_path3, file_name3) = self._MakeTempFileGivenDir( | |
166 host_sub_dir1, _OLD_CONTENTS) | |
167 (host_file_path4, file_name4) = self._MakeTempFileGivenDir( | |
168 host_sub_dir2, _OLD_CONTENTS) | |
169 | |
170 device_file_path1 = "%s/%s" % (_DEVICE_DIR, file_name1) | |
171 device_file_path2 = "%s/%s" % (_DEVICE_DIR, file_name2) | |
172 device_file_path3 = "%s/%s/%s" % (_DEVICE_DIR, _SUB_DIR1, file_name3) | |
173 device_file_path4 = "%s/%s/%s/%s" % (_DEVICE_DIR, _SUB_DIR, | |
174 _SUB_DIR2, file_name4) | |
175 | |
176 self.adb.Push(host_file_path1, device_file_path1) | |
177 self.adb.Push(host_file_path2, device_file_path2) | |
178 self.adb.Push(host_file_path3, device_file_path3) | |
179 self.adb.Push(host_file_path4, device_file_path4) | |
180 | |
181 with open(host_file_path1, 'w') as f: | |
182 f.write(_NEW_CONTENTS) | |
183 cmd_helper.RunCmd(['rm', host_file_path2]) | |
184 cmd_helper.RunCmd(['rm', host_file_path4]) | |
185 | |
186 self.device.PushChangedFiles([(host_tmp_dir, _DEVICE_DIR)], | |
187 delete_device_stale=True) | |
188 result = self.device.RunShellCommand(['cat', device_file_path1], | |
189 single_line=True) | |
190 self.assertEqual(_NEW_CONTENTS, result) | |
191 | |
192 result = self.device.RunShellCommand(['ls', _DEVICE_DIR]) | |
193 self.assertIn(file_name1, result) | |
194 self.assertIn(_SUB_DIR1, result) | |
195 self.assertIn(_SUB_DIR, result) | |
196 self.assertEqual(3, len(result)) | |
197 | |
198 result = self.device.RunShellCommand(['cat', device_file_path3], | |
199 single_line=True) | |
200 self.assertEqual(_OLD_CONTENTS, result) | |
201 | |
202 result = self.device.RunShellCommand(["ls", "%s/%s/%s" | |
203 % (_DEVICE_DIR, _SUB_DIR, _SUB_DIR2)], | |
204 single_line=True) | |
205 self.assertEqual('', result) | |
206 | |
207 self.device.RunShellCommand(['rm', '-rf', _DEVICE_DIR]) | |
208 cmd_helper.RunCmd(['rm', '-rf', host_tmp_dir]) | |
209 | |
210 def testRestartAdbd(self): | |
211 old_adbd_pid = self.device.RunShellCommand( | |
212 ['ps', '|', 'grep', 'adbd'])[1].split()[1] | |
213 self.device.RestartAdbd() | |
214 new_adbd_pid = self.device.RunShellCommand( | |
215 ['ps', '|', 'grep', 'adbd'])[1].split()[1] | |
216 self.assertNotEqual(old_adbd_pid, new_adbd_pid) | |
217 | |
218 | |
219 if __name__ == '__main__': | |
220 unittest.main() | |
OLD | NEW |