1# Copyright 2018 The Chromium OS Authors. All rights reserved. 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5import logging 6import os 7import sys 8 9from dbus.mainloop.glib import DBusGMainLoop 10 11from autotest_lib.client.bin import test 12from autotest_lib.client.common_lib import error 13from autotest_lib.client.common_lib import utils 14from autotest_lib.client.common_lib.cros import smbprovider 15 16class enterprise_SmbProviderDaemon(test.test): 17 """ 18 Test for SmbProvider Daemon. 19 20 """ 21 22 version = 1 23 24 WORKGROUP = '' 25 USERNAME = '' 26 PASSWORD = '' 27 28 def setup(self): 29 """ 30 Compiles protobufs for error type and input/output parameters. 31 32 """ 33 34 os.chdir(self.srcdir) 35 utils.make('OUT_DIR=.') 36 37 def initialize(self): 38 """ 39 Initializes the D-Bus loop and creates Python wrapper. 40 41 """ 42 43 bus_loop = DBusGMainLoop(set_as_default=True) 44 self._smbprovider = smbprovider.SmbProvider(bus_loop, self.srcdir) 45 46 # Append path for directory_entry_pb2 imports. 47 sys.path.append(self.srcdir) 48 49 def run_once(self, mount_path): 50 """ 51 Runs smbproviderd D-Bus commands. 52 53 @param mount_path: Address of the SMB share. 54 """ 55 56 self.sanity_test(mount_path) 57 58 def _generate_random_id(self, size): 59 """ 60 Generates a random string of size N. 61 62 @param size: Size of the generated string. 63 64 @return: Returns a random alphanumeric string of size N. 65 66 """ 67 68 import string 69 import random 70 71 return ''.join(random.choice(string.ascii_uppercase + 72 string.digits) for i in range(size)) 73 74 def sanity_test(self, mount_path): 75 """ 76 Sanity test that runs through all filesystem operations 77 on the SmbProvider Daemon. 78 79 @param mount_path: Address of the SMB share. 80 81 """ 82 83 from directory_entry_pb2 import ERROR_EXISTS 84 85 # Mount the SMB share. 86 mount_id = self._check_mount(mount_path) 87 88 # Generate random directory. 89 rand_dir_id = self._generate_random_id(10) 90 test_dir = '/autotest_' + rand_dir_id + '/' 91 self._check_create_directory(mount_id, test_dir, False) 92 93 # Get metadata of a directory. 94 metadata = self._check_get_metadata(mount_id, test_dir) 95 96 # Check that GetMetadata has correct values of a directory. 97 self._check_metadata(test_dir[1:-1], 0, True, metadata) 98 99 # Create file inside directory. 100 test_file = test_dir + '1.txt' 101 self._check_create_file(mount_id, test_file) 102 103 # Open file with Read-Only privileges. 104 file_id = self._check_open_file(mount_id, test_file, False) 105 self._check_close_file(mount_id, file_id) 106 107 # Open + Close file with Read-Only privileges. 108 file_id = self._check_open_file(mount_id, test_file, False) 109 self._check_close_file(mount_id, file_id) 110 111 # Open file for writing. 112 file_id = self._check_open_file(mount_id, test_file, True) 113 114 # Write data to file. 115 data = 'Hello World!' 116 self._check_write_file(mount_id, file_id, 0, data) 117 118 # Read data from file. 119 read_data = self._check_read_file(mount_id, file_id, 0, len(data)) 120 121 # Close file. 122 self._check_close_file(mount_id, file_id) 123 124 # Verify data is written to file correctly. 125 self._check_contents(data, read_data) 126 127 # Get the metadata of the file. 128 metadata = self._check_get_metadata(mount_id, test_file) 129 130 # Check that GetMetadeta has correct values of a file. 131 # TODO(jimmyxgong): len() only works properly for UTF-8. Find way to 132 # get size universally. 133 self._check_metadata('1.txt', len(data), False, metadata) 134 135 # Delete file. 136 self._check_delete_entry(mount_id, test_file, False) 137 138 # Create recursive directories. 139 recursive_dir = test_dir + 'test1/test2/' 140 self._check_create_directory(mount_id, recursive_dir, True) 141 142 # Create file within the new directory. 143 test_file2 = recursive_dir + '2.txt' 144 self._check_create_file(mount_id, test_file2) 145 146 # Check moving to existing entry is handled. 147 self._check_move_entry(mount_id, test_file2, test_dir, ERROR_EXISTS) 148 149 # Move file up to root test directory. 150 self._check_move_entry(mount_id, test_file2, test_dir + 'moved.txt') 151 152 # Move back down to original location. 153 self._check_move_entry(mount_id, test_dir + 'moved.txt', test_file2) 154 155 # TODO(jimmyxgong): Delete contents of autotest directory recursively. 156 self._check_delete_entry(mount_id, test_file2, False) 157 self._check_delete_entry(mount_id, test_dir + 'test1/test2/', False) 158 self._check_delete_entry(mount_id, test_dir + 'test1/', False) 159 160 # Delete autotest directory. 161 self._check_delete_entry(mount_id, test_dir, False) 162 163 # Unmount the SMB share. 164 self._check_unmount(mount_id) 165 166 def _check_mount(self, mount_path): 167 """ 168 Checks that mount is working. 169 170 @param mount_path: Address of the SMB share. 171 172 @return mount_id: Unique identifier of the mount. 173 174 """ 175 176 from directory_entry_pb2 import ERROR_OK 177 178 error, mount_id = self._smbprovider.mount(mount_path, 179 self.WORKGROUP, 180 self.USERNAME, 181 self.PASSWORD) 182 183 if mount_id < 0 : 184 raise error.TestFail('Unexpected failure with mount id.') 185 186 self._check_result('Mount', error) 187 return mount_id 188 189 def _check_unmount(self, mount_id): 190 """ 191 Checks that unmount is working. 192 193 @param mount_id: Unique identifier of the mount. 194 195 """ 196 197 error = self._smbprovider.unmount(mount_id) 198 199 self._check_result('Unmount', error) 200 201 def _check_get_metadata(self, mount_id, entry_path): 202 """ 203 Checks that get metadata is working. 204 205 @param mount_id: Unique identifier of the mount. 206 @param entry_path: Path of the entry. 207 208 @return: GetMetaDataEntryOptionsProto blob string returned by the D-Bus 209 call. 210 211 """ 212 213 error, metadata_blob = self._smbprovider.get_metadata(mount_id, 214 entry_path) 215 216 self._check_result('Get Metadata', error) 217 218 return metadata_blob 219 220 def _check_metadata(self, entry_path, size, is_dir, metadata_blob): 221 """ 222 Checks that metadata_blob has the correct values. 223 224 @param entry_path: File path of the entry we are checking. 225 @param size: Size of the entry in bytes. 226 @param is_dir: Boolean that indicates whether the entry is a directory. 227 @param metadata_blob: Blob that contains metadata of the entry. 228 229 """ 230 231 if entry_path != metadata_blob.name or \ 232 size != metadata_blob.size or \ 233 is_dir != metadata_blob.is_directory: 234 logging.error('Failed: Metadata is incorrect') 235 raise error.TestFail('Unexpected error with metadata') 236 237 def _check_create_file(self, mount_id, file_path): 238 """ 239 Checks that create file is working. 240 241 @param mount_id: Unique identifier of the mount. 242 @param file_path: Path of where the new file will be created. 243 244 """ 245 246 error = self._smbprovider.create_file(mount_id, file_path) 247 248 self._check_result('Create File', error) 249 250 def _check_open_file(self, mount_id, file_path, writeable): 251 """ 252 Checks that open file is working. 253 254 @param mount_id: Unique identifier of the mount. 255 @param file_path: Path of where the file is located. 256 @param writeable: Boolean to indicated whether the file should 257 be opened with write access. 258 259 """ 260 261 error, file_id = self._smbprovider.open_file(mount_id, 262 file_path, 263 writeable) 264 if file_id < 0: 265 raise error.TestFail('Unexpected file id failure.') 266 267 self._check_result('Open File', error) 268 269 return file_id 270 271 def _check_close_file(self, mount_id, file_id): 272 """ 273 Checks that close file is working. 274 275 @param mount_id: Unique identifier of the mount. 276 @param file_id: Unique identifier of the file. 277 278 """ 279 280 error = self._smbprovider.close_file(mount_id, file_id) 281 282 self._check_result('Close File', error) 283 284 def _check_write_file(self, mount_id, file_id, offset, data): 285 """ 286 Checks that write file is working. 287 288 @param mount_id: Unique identifier of the mount. 289 @param file_id: Unique identifier of the file. 290 @param offset: Offset of the file to start writing to. 291 @param data: Data to be written. 292 293 """ 294 295 error = self._smbprovider.write_file(mount_id, file_id, offset, data) 296 297 self._check_result('Write File', error) 298 299 def _check_read_file(self, mount_id, file_id, offset, length): 300 """ 301 Checks that read file is working. 302 303 @param mount_id: Unique identifier of the mount. 304 @param file_id: Unique identifier of the file. 305 @param offset: Offset of the file to start reading from. 306 @param length: Length of data to read in bytes. 307 308 @return A buffer containing the data read. 309 310 """ 311 312 error, fd = self._smbprovider.read_file(mount_id, file_id, offset, 313 length) 314 315 self._check_result('Read File', error) 316 317 return fd 318 319 def _check_contents(self, data, read_data): 320 """ 321 Checks that read_data is equal to data. 322 323 @param data: Original data to be compared to. 324 @param read_data: Data to be compared to the original data. 325 326 """ 327 328 if data != read_data: 329 logging.error('Failed: Written data does not match Read data') 330 raise error.TestFail( 331 'Unexpected mismatch of written data and read data.\ 332 Expected: %s , but got: %s' % (data, read_data)) 333 334 def _check_create_directory(self, mount_id, 335 directory_path, 336 recursive): 337 """ 338 Checks that create directory is working. 339 340 @param mount_id: Unique identifier of the mount. 341 @param directory_path: Path for the test directory. 342 @param recursive: Boolean to indicate whether directories should be 343 created recursively. 344 345 """ 346 347 error = self._smbprovider.create_directory(mount_id, 348 directory_path, 349 recursive) 350 351 self._check_result('Create Directory', error) 352 353 def _check_delete_entry(self, mount_id, entry_path, recursive): 354 """ 355 Checks that delete an entry works. 356 357 @param mount_id: Unique identifier of the mount. 358 @param entry_path: Path to the file/directory to delete. 359 @param recursive: Boolean to indicate recursive deletes. 360 361 """ 362 363 error = self._smbprovider.delete_entry(mount_id, 364 entry_path, 365 recursive) 366 367 self._check_result('Delete Entry', error) 368 369 def _check_move_entry(self, mount_id, source_path, target_path, 370 expected=None): 371 """ 372 Checks that move entry is working. 373 374 @param mount_id: Unique identifier of the mount. 375 @param source_path: Path of the entry to be moved. 376 @param target_path: Path of the destination for the entry. 377 @param expected: Expected ErrorType. Default: None (ERROR_OK) 378 379 """ 380 381 error = self._smbprovider.move_entry(mount_id, 382 source_path, 383 target_path) 384 385 self._check_result('Move Entry', error, expected) 386 387 def _check_result(self, method_name, result, expected=None): 388 """ 389 Helper to check error codes and throw on mismatch. 390 391 Checks whether the returned ErrorType from a D-Bus call to smbproviderd 392 matches the expected ErrorType. In case of a mismatch, throws a 393 TestError. 394 395 @param method_name: Name of the D-Bus method that was called. 396 @param result: ErrorType returned from the D-Bus call. 397 @param expected: Expected ErrorType. Default: ErrorType.ERROR_OK. 398 399 """ 400 401 from directory_entry_pb2 import ErrorType 402 from directory_entry_pb2 import ERROR_OK 403 404 if not expected: 405 expected = ERROR_OK 406 407 if result != expected: 408 logging.error('Failed to run %s', method_name) 409 raise error.TestFail( 410 '%s failed with error %s (%s), expected %s (%s)' % ( 411 method_name, result, ErrorType.Name(result), expected, 412 ErrorType.Name(expected))) 413