test/py: ums: add filesystem-based testing

Enhance the UMS test to optionally mount a partition and read/write a file
to it, validating that the content written and read back are identical.

This enhancement is backwards-compatible; old boardenv contents that don't
define the new configuration data will cause the test code to perform as
before.

test/ums/ is deleted since the Python test now performs the same testing
that it did.

The code is also re-written to make use of the recently added utility
module, and split it up into nested functions so the overall logic of
the test process can be followed more easily without the details
cluttering the code.

Signed-off-by: Stephen Warren <swarren@nvidia.com>
Acked-by: Lukasz Majewski <l.majewski@samsung.com>
Acked-by: Simon Glass <sjg@chromium.org>

1 files changed, 188 insertions, 48 deletions

diff --git a/test/py/tests/test_ums.py b/test/py/tests/test_ums.py
index a137221..f482cfe 100644
--- a/test/py/tests/test_ums.py
+++ b/test/py/tests/test_ums.py
@@ -2,13 +2,17 @@
 #
 # SPDX-License-Identifier: GPL-2.0
 
-# Test U-Boot's "ums" command. At present, this test only ensures that a UMS
-# device can be enumerated by the host/test machine. In the future, this test
-# should be enhanced to validate disk IO.
+# Test U-Boot's "ums" command. The test starts UMS in U-Boot, waits for USB
+# device enumeration on the host, reads a small block of data from the UMS
+# block device, optionally mounts a partition and performs filesystem-based
+# read/write tests, and finally aborts the "ums" command in U-Boot.
 
 import os
+import os.path
 import pytest
+import re
 import time
+import u_boot_utils
 
 '''
 Note: This test relies on:
@@ -17,13 +21,36 @@ a) boardenv_* to contain configuration values to define which USB ports are
 available for testing. Without this, this test will be automatically skipped.
 For example:
 
+# Leave this list empty if you have no block_devs below with writable
+# partitions defined.
+env__mount_points = (
+    "/mnt/ubtest-mnt-p2371-2180-na",
+)
+
 env__usb_dev_ports = (
-    {'tgt_usb_ctlr': '0', 'host_ums_dev_node': '/dev/disk/by-path/pci-0000:00:14.0-usb-0:13:1.0-scsi-0:0:0:0'},
+    {
+        "tgt_usb_ctlr": "0",
+        "host_ums_dev_node": "/dev/disk/by-path/pci-0000:00:14.0-usb-0:13:1.0-scsi-0:0:0:0",
+    },
 )
 
 env__block_devs = (
-    {'type': 'mmc', 'id': '0'}, # eMMC; always present
-    {'type': 'mmc', 'id': '1'}, # SD card; present since I plugged one in
+    # eMMC; always present
+    {
+        "type": "mmc",
+        "id": "0",
+        # The following two properties are optional.
+        # If present, the partition will be mounted and a file written-to and
+        # read-from it. If missing, only a simple block read test will be
+        # performed.
+        "writable_fs_partition": 1,
+        "writable_fs_subdir": "tmp/",
+    },
+    # SD card; present since I plugged one in
+    {
+        "type": "mmc",
+        "id": "1"
+    },
 )
 
 b) udev rules to set permissions on devices nodes, so that sudo is not
@@ -34,47 +61,42 @@ ACTION=="add", SUBSYSTEM=="block", SUBSYSTEMS=="usb", KERNELS=="3-13", MODE:="66
 (You may wish to change the group ID instead of setting the permissions wide
 open. All that matters is that the user ID running the test can access the
 device.)
-'''
 
-def open_ums_device(host_ums_dev_node):
-    '''Attempt to open a device node, returning either the opened file handle,
-    or None on any error.'''
+c) /etc/fstab entries to allow the block device to be mounted without requiring
+root permissions. For example:
 
-    try:
-        return open(host_ums_dev_node, 'rb')
-    except:
-        return None
-
-def wait_for_ums_device(host_ums_dev_node):
-    '''Continually attempt to open the device node exported by the "ums"
-    command, and either return the opened file handle, or raise an exception
-    after a timeout.'''
-
-    for i in xrange(100):
-        fh = open_ums_device(host_ums_dev_node)
-        if fh:
-            return fh
-        time.sleep(0.1)
-    raise Exception('UMS device did not appear')
-
-def wait_for_ums_device_gone(host_ums_dev_node):
-    '''Continually attempt to open the device node exported by the "ums"
-    command, and either return once the device has disappeared, or raise an
-    exception if it does not before a timeout occurs.'''
-
-    for i in xrange(100):
-        fh = open_ums_device(host_ums_dev_node)
-        if not fh:
-            return
-        fh.close()
-        time.sleep(0.1)
-    raise Exception('UMS device did not disappear')
+/dev/disk/by-path/pci-0000:00:14.0-usb-0:13:1.0-scsi-0:0:0:0-part1 /mnt/ubtest-mnt-p2371-2180-na ext4 noauto,user,nosuid,nodev
+
+This entry is only needed if any block_devs above contain a
+writable_fs_partition value.
+'''
 
 @pytest.mark.buildconfigspec('cmd_usb_mass_storage')
 def test_ums(u_boot_console, env__usb_dev_port, env__block_devs):
     '''Test the "ums" command; the host system must be able to enumerate a UMS
-    device when "ums" is running, and this device must disappear when "ums" is
-    aborted.'''
+    device when "ums" is running, block and optionally file I/O are tested,
+    and this device must disappear when "ums" is aborted.
+
+    Args:
+        u_boot_console: A U-Boot console connection.
+        env__usb_dev_port: The single USB device-mode port specification on
+            which to run the test. See the file-level comment above for
+            details of the format.
+        env__block_devs: The list of block devices that the target U-Boot
+            device has attached. See the file-level comment above for details
+            of the format.
+
+    Returns:
+        Nothing.
+    '''
+
+    have_writable_fs_partition = 'writable_fs_partition' in env__block_devs[0]
+    if not have_writable_fs_partition:
+        # If 'writable_fs_subdir' is missing, we'll skip all parts of the
+        # testing which mount filesystems.
+        u_boot_console.log.warning(
+            'boardenv missing "writable_fs_partition"; ' +
+            'UMS testing will be limited.')
 
     tgt_usb_ctlr = env__usb_dev_port['tgt_usb_ctlr']
     host_ums_dev_node = env__usb_dev_port['host_ums_dev_node']
@@ -84,11 +106,129 @@ def test_ums(u_boot_console, env__usb_dev_port, env__block_devs):
     # device list here. We'll test each block device somewhere else.
     tgt_dev_type = env__block_devs[0]['type']
     tgt_dev_id = env__block_devs[0]['id']
+    if have_writable_fs_partition:
+        mount_point = u_boot_console.config.env['env__mount_points'][0]
+        mount_subdir = env__block_devs[0]['writable_fs_subdir']
+        part_num = env__block_devs[0]['writable_fs_partition']
+        host_ums_part_node = '%s-part%d' % (host_ums_dev_node, part_num)
+    else:
+        host_ums_part_node = host_ums_dev_node
+
+    test_f = u_boot_utils.PersistentRandomFile(u_boot_console, 'ums.bin',
+        1024 * 1024);
+    if have_writable_fs_partition:
+        mounted_test_fn = mount_point + '/' + mount_subdir + test_f.fn
+
+    def start_ums():
+        '''Start U-Boot's ums shell command.
+
+        This also waits for the host-side USB enumeration process to complete.
+
+        Args:
+            None.
+
+        Returns:
+            Nothing.
+        '''
+
+        u_boot_console.log.action(
+            'Starting long-running U-Boot ums shell command')
+        cmd = 'ums %s %s %s' % (tgt_usb_ctlr, tgt_dev_type, tgt_dev_id)
+        u_boot_console.run_command(cmd, wait_for_prompt=False)
+        u_boot_console.wait_for(re.compile('UMS: LUN.*[\r\n]'))
+        fh = u_boot_utils.wait_until_open_succeeds(host_ums_part_node)
+        u_boot_console.log.action('Reading raw data from UMS device')
+        fh.read(4096)
+        fh.close()
+
+    def mount():
+        '''Mount the block device that U-Boot exports.
+
+        Args:
+            None.
+
+        Returns:
+            Nothing.
+        '''
+
+        u_boot_console.log.action('Mounting exported UMS device')
+        cmd = ('/bin/mount', host_ums_part_node)
+        u_boot_utils.run_and_log(u_boot_console, cmd)
 
-    cmd = 'ums %s %s %s' % (tgt_usb_ctlr, tgt_dev_type, tgt_dev_id)
-    u_boot_console.run_command('ums 0 mmc 0', wait_for_prompt=False)
-    fh = wait_for_ums_device(host_ums_dev_node)
-    fh.read(4096)
-    fh.close()
-    u_boot_console.ctrlc()
-    wait_for_ums_device_gone(host_ums_dev_node)
+    def umount(ignore_errors):
+        '''Unmount the block device that U-Boot exports.
+
+        Args:
+            ignore_errors: Ignore any errors. This is useful if an error has
+                already been detected, and the code is performing best-effort
+                cleanup. In this case, we do not want to mask the original
+                error by "honoring" any new errors.
+
+        Returns:
+            Nothing.
+        '''
+
+        u_boot_console.log.action('Unmounting UMS device')
+        cmd = ('/bin/umount', host_ums_part_node)
+        u_boot_utils.run_and_log(u_boot_console, cmd, ignore_errors)
+
+    def stop_ums(ignore_errors):
+        '''Stop U-Boot's ums shell command from executing.
+
+        This also waits for the host-side USB de-enumeration process to
+        complete.
+
+        Args:
+            ignore_errors: Ignore any errors. This is useful if an error has
+                already been detected, and the code is performing best-effort
+                cleanup. In this case, we do not want to mask the original
+                error by "honoring" any new errors.
+
+        Returns:
+            Nothing.
+        '''
+
+        u_boot_console.log.action(
+            'Stopping long-running U-Boot ums shell command')
+        u_boot_console.ctrlc()
+        u_boot_utils.wait_until_file_open_fails(host_ums_part_node,
+            ignore_errors)
+
+    ignore_cleanup_errors = True
+    try:
+        start_ums()
+        if not have_writable_fs_partition:
+            # Skip filesystem-based testing if not configured
+            return
+        try:
+            mount()
+            u_boot_console.log.action('Writing test file via UMS')
+            cmd = ('rm', '-f', mounted_test_fn)
+            u_boot_utils.run_and_log(u_boot_console, cmd)
+            if os.path.exists(mounted_test_fn):
+                raise Exception('Could not rm target UMS test file')
+            cmd = ('cp', test_f.abs_fn, mounted_test_fn)
+            u_boot_utils.run_and_log(u_boot_console, cmd)
+            ignore_cleanup_errors = False
+        finally:
+            umount(ignore_errors=ignore_cleanup_errors)
+    finally:
+        stop_ums(ignore_errors=ignore_cleanup_errors)
+
+    ignore_cleanup_errors = True
+    try:
+        start_ums()
+        try:
+            mount()
+            u_boot_console.log.action('Reading test file back via UMS')
+            read_back_hash = u_boot_utils.md5sum_file(mounted_test_fn)
+            cmd = ('rm', '-f', mounted_test_fn)
+            u_boot_utils.run_and_log(u_boot_console, cmd)
+            ignore_cleanup_errors = False
+        finally:
+            umount(ignore_errors=ignore_cleanup_errors)
+    finally:
+        stop_ums(ignore_errors=ignore_cleanup_errors)
+
+    written_hash = test_f.content_hash
+    assert(written_hash == read_back_hash)