PR #12471 from Eran: add run-unit-tests --device <>

Author: maloel

unit-tests/py/rspy/devices.py

@@ -333,16 +333,18 @@ def by_name( name, ignored_products ):
                 result.add(device.serial_number)
     return result
 
-def _get_sns_from_spec( spec, ignored_products ):
+def by_spec( spec, ignored_products ):
     """
     Helper function for by_configuration. Yields all serial-numbers matching the given spec
-    :param spec: A product name/line (as a string) we want to get serial number of
+    :param spec: A product name/line (as a string) we want to get serial number of, or an actual s/n
     :param ignored_products: List of products we want to ignore. e.g. ['D455', 'D457', etc.]
     :return: A set of device serial-numbers
     """
     if spec.endswith( '*' ):
         for sn in by_product_line( spec[:-1], ignored_products ):
             yield sn
+    elif get( spec ):
+        yield spec   # the device serial number
     else:
         for sn in by_name( spec, ignored_products ):
             yield sn
@@ -357,7 +359,7 @@ def expand_specs( specs ):
     """
     expanded = set()
     for spec in specs:
-        sns = {sn for sn in _get_sns_from_spec( spec )}
+        sns = {sn for sn in by_spec( spec )}
         if sns:
             expanded.update( sns )
         else:
@@ -387,16 +389,17 @@ def load_specs_from_file( filename ):
     return exceptions
 
 
-def by_configuration( config, exceptions = None ):
+def by_configuration( config, exceptions=None, inclusions=None ):
     """
     Yields the serial numbers fitting the given configuration. If configuration includes an 'each' directive
     will yield all fitting serial numbers one at a time. Otherwise yields one set of serial numbers fitting the configuration
 
-    :param config: A test:device line collection of arguments (e.g., [L515 D400*])
+    :param config: A test:device line collection of arguments (e.g., [L515 D400*]) or serial numbers
     :param exceptions: A collection of serial-numbers that serve as exceptions that will never get matched
+    :param inclusions: A collection of serial-numbers from which to match - nothing else will get matched
 
-    If no device matches the configuration devices specified, a RuntimeError will be
-    raised!
+    If no device matches the configuration devices specified, a RuntimeError will be raised unless
+    'inclusions' is provided and the configuration is simple, and an empty set yielded to signify.
     """
     exceptions = exceptions or set()
     # split the current config to two lists:
@@ -411,33 +414,49 @@ def by_configuration( config, exceptions = None ):
         else:
             new_config.append(p)
 
+    nothing_matched = True
     if len( new_config ) > 0 and re.fullmatch( r'each\(.+\)', new_config[0], re.IGNORECASE ):
         spec = new_config[0][5:-1]
-        for sn in _get_sns_from_spec( spec, ignored_products ):
-            if sn not in exceptions:
-                yield { sn }
+        for sn in by_spec( spec, ignored_products ):
+            if sn in exceptions:
+                continue
+            if inclusions and sn not in inclusions:
+                continue
+            nothing_matched = False
+            yield { sn }
     else:
         sns = set()
         for spec in new_config:
             old_len = len(sns)
-            for sn in _get_sns_from_spec( spec, ignored_products ):
+            for sn in by_spec( spec, ignored_products ):
                 if sn in exceptions:
                     continue
+                if inclusions and sn not in inclusions:
+                    continue
                 if sn not in sns:
                     sns.add( sn )
                     break
             new_len = len(sns)
             if new_len == old_len:
-                error = 'no device matches configuration "' + spec + '"'
-                if old_len:
-                    error += ' (after already matching ' + str(sns) + ')'
-                if ignored_products:
-                    error += ' (!' + str(ignored_products) + ')'
-                if exceptions:
-                    error += ' (-' + str(exceptions) + ')'
-                raise RuntimeError( error )
+                # No new device matches the spec:
+                #   - if no inclusions were specified, this is always an error
+                #   - with inclusions, it's not an error only if it's the only spec
+                if not inclusions or len(new_config) > 1:
+                    error = 'no device matches configuration "' + spec + '"'
+                    if old_len:
+                        error += ' (after already matching ' + str(sns) + ')'
+                    if ignored_products:
+                        error += ' (!' + str(ignored_products) + ')'
+                    if exceptions:
+                        error += ' (-' + str(exceptions) + ')'
+                    if inclusions:
+                        error += ' (+' + str(inclusions) + ')'
+                    raise RuntimeError( error )
         if sns:
+            nothing_matched = False
             yield sns
+    if nothing_matched and inclusions:
+        yield set()  # let the caller decide how to deal with it
 
 
 def get_first( sns ):

unit-tests/run-unit-tests.py

@@ -47,6 +47,7 @@ def usage():
     print( '        --context <>         The context to use for test configuration' )
     print( '        --repeat <#>         Repeat each test <#> times' )
     print( '        --config <>          Ignore test configurations; use the one provided' )
+    print( '        --device <>          Run only on the specified devices; ignore any test that does not match (implies --live)' )
     print( '        --no-reset           Do not try to reset any devices, with or without Acroname' )
     print( '        --acroname-reset     If acroname is available, reset the acroname itself' )
     print( '        --rslog              Enable LibRS logging (LOG_DEBUG etc.) to console in each test' )
@@ -77,7 +78,7 @@ def usage():
     opts, args = getopt.getopt( sys.argv[1:], 'hvqr:st:',
                                 longopts=['help', 'verbose', 'debug', 'quiet', 'regex=', 'stdout', 'tag=', 'list-tags',
                                           'list-tests', 'no-exceptions', 'context=', 'repeat=', 'config=', 'no-reset', 'acroname-reset',
-                                          'rslog', 'skip-disconnected', 'live', 'not-live'] )
+                                          'rslog', 'skip-disconnected', 'live', 'not-live', 'device='] )
 except getopt.GetoptError as err:
     log.e( err )  # something like "option -a not recognized"
     usage()
@@ -90,6 +91,7 @@ def usage():
 context = []
 repeat = 1
 forced_configurations = None
+device_set = None
 no_reset = False
 acroname_reset = False
 skip_disconnected = False
@@ -124,6 +126,12 @@ def usage():
         repeat = int(arg)
     elif opt == '--config':
         forced_configurations = [[arg]]
+    elif opt == '--device':
+        if only_not_live:
+            log.e( "--device and --not-live are mutually exclusive" )
+            usage()
+        only_live = True
+        device_set = arg.split()
     elif opt == '--no-reset':
         no_reset = True
     elif opt == '--acroname-reset':
@@ -237,16 +245,20 @@ def find_build_dir( dir ):
     os.environ["PYTHONPATH"] += os.pathsep + dir
 
 
-def configuration_str( configuration, repetition=1, retry=0, sns=None, prefix='', suffix='' ):
+def serial_numbers_to_string( sns ):
+    return ' '.join( [f'{devices.get(sn).name}_{sn}' for sn in sns] )
+
+
+def configuration_str( configuration, repetition=0, retry=0, sns=None, prefix='', suffix='' ):
     """ Return a string repr (with a prefix and/or suffix) of the configuration or '' if it's None """
     s = ''
     if configuration is not None:
         s += '[' + ' '.join( configuration )
         if sns is not None:
-            s += ' -> ' + ' '.join( [f'{devices.get(sn).name}_{sn}' for sn in sns] )
+            s += ' -> ' + serial_numbers_to_string( sns )
         s += ']'
     elif sns is not None:
-        s += '[' + ' '.join( [f'{devices.get(sn).name}_{sn}' for sn in sns] ) + ']'
+        s += '[' + serial_numbers_to_string( sns ) + ']'
     if repetition:
         s += '[' + str(repetition+1) + ']'
     if retry:
@@ -379,11 +391,14 @@ def devices_by_test_config( test, exceptions ):
 
     :param test: The test (of class type Test) we're interested in
     """
-    global forced_configurations
+    global forced_configurations, device_set
     for configuration in ( forced_configurations  or  test.config.configurations ):
         try:
-            for serial_numbers in devices.by_configuration( configuration, exceptions ):
-                yield configuration, serial_numbers
+            for serial_numbers in devices.by_configuration( configuration, exceptions, device_set ):
+                if not serial_numbers:
+                    log.d( 'configuration:', configuration_str( configuration ), 'has no matching device; ignoring' )
+                else:
+                    yield configuration, serial_numbers
         except RuntimeError as e:
             if devices.acroname:
                 log.e( log.red + test.name + log.reset + ': ' + str( e ) )
@@ -462,6 +477,17 @@ def test_wrapper( test, configuration=None, repetition=1, sns=None ):
                 finally:
                     log.debug_unindent()
         #
+        if device_set is not None:
+            sns = set()  # convert the list of specs to a list of serial numbers
+            ignored_list = list()
+            for spec in device_set:
+                included_devices = [sn for sn in devices.by_spec( spec, ignored_list )]
+                if not included_devices:
+                    log.f( f'No match for --device "{spec}"' )
+                sns.update( included_devices )
+            device_set = sns
+            log.d( f'ignoring devices other than: {serial_numbers_to_string( device_set )}' )
+        #
         log.progress()
     #
     # Automatically detect github actions based on environment variable
@@ -543,7 +569,7 @@ def test_wrapper( test, configuration=None, repetition=1, sns=None ):
             for configuration, serial_numbers in devices_by_test_config( test, exceptions ):
                 for repetition in range(repeat):
                     try:
-                        log.d( 'configuration:', configuration )
+                        log.d( 'configuration:', configuration_str( configuration, repetition, sns=serial_numbers ) )
                         log.debug_indent()
                         if not no_reset:
                             devices.enable_only( serial_numbers, recycle=True )