~canonical-platform-qa/ubuntu-system-tests/use_keyboard_reset_lp1481770

« back to all changes in this revision

Viewing changes to ubuntu_system_tests/tests/test_mtp.py

This branch has: 
- Refactoring 
- Test implementation of mtp test test_copy_from_device
.

Approved by PS Jenkins bot, Richard Huddie.

Show diffs side-by-side

added added

removed removed

Lines of Context:
32
32
 
33
33
TEMP_DIR = '/tmp'
34
34
DEFAULT_XDG_RUNTIME_DIR = '/run/user/1000'
 
35
 
35
36
MUSIC = "Music"
36
37
MUSIC_DIR = os.path.join(os.path.expanduser("~"), MUSIC)
37
38
DOCS = "Documents"
38
39
DOCS_DIR = os.path.join(os.path.expanduser("~"), DOCS)
39
 
FILE_SIZE = 10000
 
40
PICTURES = "Pictures"
 
41
PICTURES_DIR = os.path.join(os.path.expanduser("~"), PICTURES)
 
42
VIDEOS = "Videos"
 
43
VIDEOS_DIR = os.path.join(os.path.expanduser("~"), VIDEOS)
40
44
 
41
45
FileMusicDesc = namedtuple('FileMusicDesc', ['file', 'album'])
42
46
logger = logging.getLogger(__name__)
53
57
}
54
58
 
55
59
 
 
60
class MTPFile:
 
61
    """ This class is a data model which represents a file to be copied
 
62
    through mtp
 
63
    """
 
64
    def __init__(self, remote_helper):
 
65
        self.remote_helper = remote_helper
 
66
        self.local_path = ""
 
67
        self.remote_src_path = ""
 
68
        self.remote_dst_path = ""
 
69
        self.sha1 = ""
 
70
 
 
71
    @property
 
72
    def name(self):
 
73
        return os.path.basename(self.local_path) or ""
 
74
 
 
75
    def clean_file(self):
 
76
        if self.local_path:
 
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)
 
82
 
 
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,
 
91
                                     self.remote_dst_path)
 
92
 
 
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)
 
97
 
 
98
 
56
99
class MTPTest(base.BaseUbuntuSystemTestCase):
57
100
    """This Class tests the mtp protocol"""
58
101
 
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()
 
113
 
 
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)
 
117
        if sha1:
 
118
            file.sha1 = fs.calculate_file_sha1(file.local_path)
 
119
        self.addCleanup(file.clean_file)
 
120
        return file
69
121
 
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)
92
144
        return None
93
145
 
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.
101
149
        """
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)
105
154
 
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
 
159
        """
 
160
        dir = os.path.join(self.mtp_dir, *paths)
 
161
        return '"' + dir + '"'
109
162
 
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
125
178
        """
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)
129
182
 
130
183
        return dir_name, dir_path
131
184
 
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)
154
207
        return dst_file
155
208
 
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)
218
271
 
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)
273
326
 
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)
277
 
 
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,
281
 
                                                 albums_name)
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)
283
331
 
284
332
        # Verify the albums have ben copied to the music folder
285
333
        copied_albums = self.verify_dir_copied(MUSIC_DIR, albums_name)
303
351
 
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)
307
355
 
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)
314
362
 
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)
318
 
 
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)
335
379
 
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()
339
 
 
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()
343
383
 
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)
351
391
 
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
379
419
        """
380
420
        self._test_copy_big_file(gb=4)
 
421
 
 
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
 
425
        """
 
426
 
 
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)
 
430
 
 
431
        video = self.create_random_file(VIDEOS_DIR, ext='.avi')
 
432
        video.remote_src_path = self.get_device_mtp_path(VIDEOS, video.name)
 
433
 
 
434
        picture = self.create_random_file(PICTURES_DIR, ext='.jpg')
 
435
        picture.remote_src_path = self.get_device_mtp_path(PICTURES,
 
436
                                                           picture.name)
 
437
 
 
438
        files = [song, video, picture]
 
439
 
 
440
        # Set destination path to files
 
441
        for file in files:
 
442
            file.remote_dst_path = os.path.join(TEMP_DIR, file.name)
 
443
 
 
444
        # Copy the files from mtp to tmp directory and verify the sha1
 
445
        for file in files:
 
446
            file.remote_copy()
 
447
            copied_sha1 = file.get_remote_dst_sha1()
 
448
            self.assertEqual(file.sha1, copied_sha1)
 
449
 
 
450
        # Remove remote files and verify the files were deleted in the device
 
451
        for file in files:
 
452
            self.remote_helper.remove_file(file.remote_src_path)
 
453
            self.assertFalse(os.path.isfile(file.local_path))