34
34
DEFAULT_XDG_RUNTIME_DIR = '/run/user/1000'
36
37
MUSIC_DIR = os.path.join(os.path.expanduser("~"), MUSIC)
38
39
DOCS_DIR = os.path.join(os.path.expanduser("~"), DOCS)
41
PICTURES_DIR = os.path.join(os.path.expanduser("~"), PICTURES)
43
VIDEOS_DIR = os.path.join(os.path.expanduser("~"), VIDEOS)
41
45
FileMusicDesc = namedtuple('FileMusicDesc', ['file', 'album'])
42
46
logger = logging.getLogger(__name__)
61
""" This class is a data model which represents a file to be copied
64
def __init__(self, remote_helper):
65
self.remote_helper = remote_helper
67
self.remote_src_path = ""
68
self.remote_dst_path = ""
73
return os.path.basename(self.local_path) or ""
77
fs.delete_file(self.local_path)
78
if self.remote_src_path:
79
self.remote_helper.remove_file(self.remote_src_path)
80
if self.remote_dst_path:
81
self.remote_helper.remove_file(self.remote_dst_path)
83
def remote_copy(self):
84
if not self.remote_src_path or not self.remote_dst_path:
85
raise RuntimeError("Remote copy cannot be done, it is required to "
86
"define src and dst paths,")
87
logger.info("Copying remote file from {src_path} to {dst_path}".
88
format(src_path=self.remote_src_path,
89
dst_path=self.remote_dst_path))
90
self.remote_helper.copy_file(self.remote_src_path,
93
def get_remote_dst_sha1(self):
94
if not self.remote_dst_path:
95
raise RuntimeError("Attribute remote_dst_path has to de defined")
96
return self.remote_helper.get_sha1(self.remote_dst_path)
56
99
class MTPTest(base.BaseUbuntuSystemTestCase):
57
100
"""This Class tests the mtp protocol"""
66
109
self.config_stack.get('ssh_user'),
67
110
self.config_stack.get('ssh_password'))
68
111
self.addCleanup(self.remote_helper.close)
112
self.mtp_dir = self.calculate_device_mtp_dir()
114
def create_random_file(self, dir, ext='', sha1=True):
115
file = MTPFile(self.remote_helper)
116
file.local_path = fs.create_random_file(dir, ext)
118
file.sha1 = fs.calculate_file_sha1(file.local_path)
119
self.addCleanup(file.clean_file)
70
122
def get_first_elem_in_dir_command(self, path):
71
123
"""Command to retrieve the first element in a dir"""
91
143
return os.path.dirname(file_path)
94
def calculate_device_mtp_dir(self, dir_path, dir_name):
95
""" Get the destination path to copy the file in the ubuntu host
96
machine which maps to the dir_name dir in the device. Use a random
97
file in order to identify the right destination.
98
:param dir_path: path in the device to create the tmp file
99
:param dir_name: name of dir with maps in the host the dir_path
100
:return: The mtp path which maps to the dir_name dir in the device
146
def calculate_device_mtp_dir(self):
147
""" Get the path in the ubuntu host machine which maps to the mtp dir.
148
Use a random file to identify the device internal storage destination.
102
file = fs.create_random_file(dir_path, ext='.tmp')
150
file = fs.create_random_file(DOCS_DIR, ext='.tmp')
103
151
self.addCleanup(os.remove, file)
104
return self.get_device_mtp_dir(dir_name, os.path.basename(file))
152
docs_mtp_dir = self.get_device_mtp_dir(DOCS, os.path.basename(file))
153
return os.path.dirname(docs_mtp_dir)
106
def escape_dir(self, base_dir, sub_dir=''):
107
""" Escape a dir and subdir to be used through commandline"""
108
return '"' + os.path.join(base_dir, sub_dir) + '"'
155
def get_device_mtp_path(self, *paths):
156
""" Retrieve the mtp path following the desired paths
157
:param paths: A list of subdir to the destination path
158
:return: The mtp path already escaped to the destination path
160
dir = os.path.join(self.mtp_dir, *paths)
161
return '"' + dir + '"'
110
163
def verify_music_album_is_shown(self, album):
111
164
""" Determines if there is an album with the desired name in the music
126
179
dir_name = fs.get_random_string()
127
180
dir_path = os.path.join(parent_dir, dir_name)
128
self.remote_helper.create_remote_dir(dir_path)
181
self.remote_helper.create_dir(dir_path)
130
183
return dir_name, dir_path
150
203
"{dst_file}".format(file_path=file_path,
151
204
dst_file=dst_file))
152
205
self.remote_helper.copy(file_path, dst_file)
153
self.addCleanup(self.remote_helper.remove_remote_file, dst_file)
206
self.addCleanup(self.remote_helper.remove_file, dst_file)
156
209
def verify_same_files(self, src_files, dst_files):
214
267
(albums_name, albums_dir) = self.create_remote_dir(TEMP_DIR)
215
268
(album_mp3_name, album_mp3_dir) = self.create_remote_dir(albums_dir)
216
269
(album_ogg_name, album_ogg_dir) = self.create_remote_dir(albums_dir)
217
self.addCleanup(self.remote_helper.remove_remote_dir, albums_dir)
270
self.addCleanup(self.remote_helper.remove_dir, albums_dir)
219
272
# Copy the music files to the ubuntu host machine
220
273
dst_tmp_mp3_file = \
271
324
albums_name, albums_dir, album_mp3_name, album_ogg_name = \
272
325
self.prepare_music_albums(src_mp3_file, src_ogg_file)
274
# Get the destination path to copy the file in the ubuntu host
275
# machine which maps to the Music dir in the device.
276
device_mtp_music_path = self.calculate_device_mtp_dir(MUSIC_DIR, MUSIC)
278
327
# Copy the albums directory from the temporal dir in the ubuntu host
279
328
# machine to the dir which maps the music dir in the device
280
device_mtp_albuns_path = self.escape_dir(device_mtp_music_path,
282
self.remote_helper.copy_dir(albums_dir, device_mtp_albuns_path)
329
device_mtp_albums_path = self.get_device_mtp_path(MUSIC, albums_name)
330
self.remote_helper.copy_dir(albums_dir, device_mtp_albums_path)
284
332
# Verify the albums have ben copied to the music folder
285
333
copied_albums = self.verify_dir_copied(MUSIC_DIR, albums_name)
304
352
# Create the dir to store the files in the host machine
305
353
(dir_name, dir_path) = self.create_remote_dir(TEMP_DIR)
306
self.addCleanup(self.remote_helper.remove_remote_dir, dir_path)
354
self.addCleanup(self.remote_helper.remove_dir, dir_path)
308
356
# Copy the src file to the ubuntu host machine
309
357
dst_tmp_file = self.copy_file_to_remote_host(src_file, TEMP_DIR)
312
360
original_files = [src_file_name + "_" + str(i) for i in range(amount)]
313
361
self.replicate_file(dst_tmp_file, dir_path, original_files)
315
# Generate the destination path to copy the file in the ubuntu host
316
# machine which maps to the Documents dir in the device.
317
device_mtp_doc_path = self.calculate_device_mtp_dir(DOCS_DIR, DOCS)
319
363
# Copy the directory from the temporal dir in the ubuntu host
320
364
# machine to the dir which maps the documents dir in the device
321
device_mtp_dir_path = self.escape_dir(device_mtp_doc_path, dir_name)
365
device_mtp_dir_path = self.get_device_mtp_path(DOCS, dir_name)
322
366
logger.info("Copying remote files to mtp directory from {src_path} to "
323
367
"{mtp_path}".format(src_path=dir_path,
324
368
mtp_path=device_mtp_dir_path))
334
378
file_path = os.path.join(TEMP_DIR, file_name)
336
380
self.remote_helper.create_random_file(file_path, kb=kb, mb=mb, gb=gb)
337
self.addCleanup(self.remote_helper.remove_remote_file, file_path)
338
file_sha1 = self.remote_helper.calculate_file_sha1(file_path).strip()
340
# Generate the destination path to copy the file in the ubuntu host
341
# machine which maps to the Documents dir in the device.
342
device_mtp_doc_path = self.calculate_device_mtp_dir(DOCS_DIR, DOCS)
381
self.addCleanup(self.remote_helper.remove_file, file_path)
382
file_sha1 = self.remote_helper.get_sha1(file_path).strip()
344
384
# Copy the directory from the temporal dir in the ubuntu host
345
385
# machine to the dir which maps the documents dir in the device
346
device_mtp_dest_path = self.escape_dir(device_mtp_doc_path, file_name)
386
device_mtp_dst_path = self.get_device_mtp_path(DOCS, file_name)
347
387
logger.info("Copying remote files to mtp directory from {src_path} to "
348
388
"{mtp_path}".format(src_path=file_path,
349
mtp_path=device_mtp_dest_path))
350
self.remote_helper.copy_file(file_path, device_mtp_dest_path)
389
mtp_path=device_mtp_dst_path))
390
self.remote_helper.copy_file(file_path, device_mtp_dst_path)
352
392
# Verify the files have ben copied correctly to the destiation
353
393
copied_file = self.verify_file_copied(DOCS_DIR, file_name)
378
418
https://bugs.launchpad.net/ubuntu/+source/mtp/+bug/1472789
380
420
self._test_copy_big_file(gb=4)
422
def test_copy_from_device(self):
423
""" Copy Music, Videos and Pictures from the device to the ubuntu host
424
machine and delete the files in the device from the host
427
# Create random source data and calculate the sha1 digest
428
song = self.create_random_file(MUSIC_DIR, ext='.mp3')
429
song.remote_src_path = self.get_device_mtp_path(MUSIC, song.name)
431
video = self.create_random_file(VIDEOS_DIR, ext='.avi')
432
video.remote_src_path = self.get_device_mtp_path(VIDEOS, video.name)
434
picture = self.create_random_file(PICTURES_DIR, ext='.jpg')
435
picture.remote_src_path = self.get_device_mtp_path(PICTURES,
438
files = [song, video, picture]
440
# Set destination path to files
442
file.remote_dst_path = os.path.join(TEMP_DIR, file.name)
444
# Copy the files from mtp to tmp directory and verify the sha1
447
copied_sha1 = file.get_remote_dst_sha1()
448
self.assertEqual(file.sha1, copied_sha1)
450
# Remove remote files and verify the files were deleted in the device
452
self.remote_helper.remove_file(file.remote_src_path)
453
self.assertFalse(os.path.isfile(file.local_path))