Python tools improvements

docs/opentitan/pymod.md

@@ -6,8 +6,9 @@ using a TCP socket).
 
 * `python/qemu/jtag`: JTAG / TAP controller client, using the _Remote BitBang Protocol_
 * `python/qemu/ot`: OpenTitan tools
-  * `dtm`: Debug Transport Module support,
+  * `devproxy`: implementation of the communication channel with the QEMU devproxy device.
   * `dm`: RISC-V Debug Module support,
+  * `dtm`: Debug Transport Module support,
   * `eflash`: Embedded Flash support,
   * `gpio`: GPIO support,
   * `km`: Key Manager support,
@@ -22,7 +23,7 @@ using a TCP socket).
   * `spi`: support SPI device communication, _i.e._ acts as a SPI master connected to QEMU SPI
      device port,
   * `util`: miscellaneous utililies such as ELF format tools and logging utilities,
-  * `devproxy`: implementation of the communication channel with the QEMU devproxy device.
+     scrambler/descrambler used in OTP image files for HW digest verification.
 
 Please check the [Python tools](tools.md) documentation for details and scripts that rely
 on these APIs.

docs/opentitan/tools.md

@@ -53,8 +53,6 @@ of options and the available features.
   using the same parameters as the KeyManager DPE. It is dedicated to unit test purposes.
 * `ot-format.sh` is a simple shell wrapper to run clang-format (code formatter) on OpenTitan files
 * `ot-tidy.sh` is a simple shell wrapper to run clang-tidy (C linter) on OpenTitan files
-* `present.py` implements the Present 128-bit scrambler/descrambler used in OTP image files for
-  HW digest verification.
 * [spidevice.py](spidevice.md) is a tiny script to upload a binary using the SPI device.
 * `treillis/` directory contains the test application to test the [GPIO](gpio.md) device.
 * [`uartmux.py`](uartmux.md) is a tiny stream wrapper to help dealing with multiple QEMU output

python/qemu/ot/.pylintrc

@@ -12,5 +12,6 @@ disable=
     too-many-nested-blocks,
     too-many-positional-arguments,
     too-many-public-methods,
+    too-many-return-statements,
     too-many-statements,
     unspecified-encoding

python/qemu/ot/devproxy.py

@@ -366,7 +366,8 @@ def enumerate_interrupts(self, out: bool) -> Iterator[InterruptGroup]:
             if group.out == out:
                 yield name, group
 
-    def signal_interrupt(self, group: str, irq: int, level: int | bool) -> None:
+    def signal_interrupt(self, group: str, irq: int, level: Union[int, bool]) \
+            -> None:
         """Set the level of an input interrupt line.
 
            :param group: the name of the group

python/qemu/ot/otp/partition.py

@@ -16,7 +16,7 @@
 
 try:
     # try to load Present if available
-    from present import Present
+    from ot.util.present import Present
 except ImportError:
     Present = None
 

python/qemu/ot/pyot/executer.py

@@ -40,6 +40,8 @@ class Executer:
        :param args: parsed arguments
     """
 
+    CONTEXT_ERROR = 99
+
     RESULT_MAP = {
         -1: 'SKIPPED',
         0: 'PASS',
@@ -49,7 +51,7 @@ class Executer:
         Wrapper.GUEST_ERROR_OFFSET - 1: 'GUEST_ESC',
         Wrapper.GUEST_ERROR_OFFSET + 1: 'FAIL',
         98: 'UNEXP_SUCCESS',
-        99: 'CONTEXT',
+        CONTEXT_ERROR: 'CONTEXT',
         # convention: exit code in [100..115] range report uncaught exceptions
         100: 'INST_ADDR_MISALIGN',
         101: 'INST_ACCESS_FAULT',
@@ -81,6 +83,9 @@ class Executer:
     DEFAULT_SERIAL_PORT = 'serial0'
     """Default VCP name."""
 
+    WRAPPER = Wrapper
+    """Default wrapper."""
+
     def __init__(self, tfm: FileManager, config: dict[str, any],
                  args: Namespace):
         self._log = getLogger('pyot.exec')
@@ -129,7 +134,7 @@ def run(self, debug: bool, allow_no_test: bool) -> int:
            :return: success or the code of the first encountered error
         """
         log_classifiers = self._config.get('logclass', {})
-        qot = Wrapper(log_classifiers, debug)
+        qot = self.WRAPPER(log_classifiers, debug)
         ret = 0
         results = defaultdict(int)
         result_file = self._argdict.get('result')
@@ -190,7 +195,9 @@ def run(self, debug: bool, allow_no_test: bool) -> int:
                     exec_info = self._build_test_command(test)
                     exec_info.test_name = test_name
                     vcplogfile = self._log_vcp_streams(exec_info)
-                    exec_info.context.execute('pre')
+                    context = 'pre'
+                    exec_info.context.execute(context)
+                    context = 'with'
                     tret, xtime, err = qot.run(exec_info)
                     cret = exec_info.context.finalize()
                     if exec_info.expect_result != 0:
@@ -203,16 +210,18 @@ def run(self, debug: bool, allow_no_test: bool) -> int:
                                               'error %d, assume error', tret)
                             tret = 98
                     if tret == 0 and cret != 0:
-                        tret = 99
+                        tret = self.CONTEXT_ERROR
                     if tret and not err:
                         err = exec_info.context.first_error
-                    exec_info.context.execute('post', tret)
+                    if not tret:
+                        context = 'post'
+                    exec_info.context.execute(context, 'post')
                 # pylint: disable=broad-except
                 except Exception as exc:
                     self._log.critical('%s', str(exc))
                     if debug:
                         print(format_exc(chain=False), file=sys.stderr)
-                    tret = 99
+                    tret = self.CONTEXT_ERROR
                     xtime = ExecTime(0.0)
                     err = str(exc)
                 finally:
@@ -223,7 +232,9 @@ def run(self, debug: bool, allow_no_test: bool) -> int:
                     flush_memory_loggers(['pyot', 'pyot.vcp', 'pyot.ctx',
                                           'pyot.file'], LOG_INFO)
                 results[tret] += 1
-                sret = self.RESULT_MAP.get(tret, tret)
+                sret = self.RESULT_MAP.get(tret) or str(tret)
+                if tret == self.CONTEXT_ERROR:
+                    sret = f'{sret} ({context.upper()})'
                 try:
                     targs = exec_info.args
                     icount = self.get_namespace_arg(targs, 'icount')

python/qemu/ot/pyot/qemu/executer.py

@@ -17,6 +17,7 @@
 from ot.util.misc import EasyDict
 
 from ..executer import Executer
+from .wrapper import Wrapper
 
 
 class QEMUExecuter(Executer):
@@ -35,6 +36,9 @@ class QEMUExecuter(Executer):
     }
     """Shortcut names for QEMU log sources."""
 
+    WRAPPER = Wrapper
+    """QEMU wrapper."""
+
     def _build_fw_args(self, args: Namespace) \
             -> tuple[str, Optional[str], list[str], Optional[str]]:
         rom_exec = bool(args.rom_exec)

python/qemu/ot/pyot/util.py

@@ -67,8 +67,8 @@ def show(self, spacing: bool = False, result: Optional[str] = None) -> None:
         if spacing:
             print('')
         # third row is time, always defined as ms, see ExecTime
-        total_time = sum(float(r[2].strip().split(' ')[0])
-                               for r in self._results[1:])
+        exec_times = [r[2].strip().split(' ')[0] for r in self._results[1:]]
+        total_time = sum(float(t) for t in exec_times if t)
         tt_str = f'{total_time / 1000:.1f} s'
         last_row = ['TEST SESSION', result or '?', tt_str]
         last_row.extend([''] * (len(self._results) - len(last_row)))

python/qemu/ot/util/misc.py

@@ -7,7 +7,10 @@
 """
 
 from io import BytesIO
+from os.path import abspath, dirname, exists, isdir, isfile, join as joinpath
+from subprocess import check_output
 from sys import stdout
+from textwrap import dedent, indent
 from typing import Any, Iterable, Optional, TextIO, Union
 import re
 
@@ -193,10 +196,83 @@ def to_bool(value, permissive=True, prohibit_int=False):
         return False
     raise ValueError(f"Invalid boolean value: '{value}")
 
+
 def alphanum_key(text: str) -> list[Union[int, str]]:
     """Alphanumerical sorting key.
 
        :param text: the text to generate a sorting key from
        :return: a list of alternating str and integer values
     """
     return [int(t) if t.isdigit() else t for t in re.split(r'(\d+)', text)]
+
+
+def redent(text: str, spc: int = 0, strip_end: bool = False) -> str:
+    """Utility function to re-indent code string.
+
+        :param text: the text to re-indent
+        :param spc: the number of leading empty space chars to prefix lines
+        :param strip_end: whether to strip trailing whitespace and newline
+    """
+    text = dedent(text.lstrip('\n'))
+    text = indent(text, ' ' * spc)
+    if strip_end:
+        text = text.rstrip(' ').rstrip('\n')
+    return text
+
+
+def retrieve_git_version(path: str, max_tag_dist: int = 100) \
+        -> Optional[str]:
+    """Return a Git identifier whenever possible.
+
+        :param path: the configuration file or directory to track the repository
+                     version identifier. Note that the returned Git identifier
+                     is not the commit version of this file / directory, but the
+                     one of the repo top-level directory.
+        :param max_tag_dist: maximum distance (in commit number) to the closest
+                             tag. If distance is greater, only emit the commit
+                             identifier
+        :return: Git version of the top-level directory
+    """
+    cfgdir: Optional[str] = None
+    path = abspath(path)
+    if isfile(path):
+        cfgdir = dirname(path)
+    elif isdir(path):
+        cfgdir = path
+    else:
+        return None
+    while cfgdir and isdir(cfgdir):
+        gitdir = joinpath(cfgdir, '.git')
+        if exists(gitdir):  # either a dir or a file for worktree
+            break
+        cfgdir = dirname(cfgdir)
+    else:
+        return None
+    assert cfgdir is not None
+    try:
+        descstr = check_output(['git', 'describe', '--long', '--dirty'],
+                               text=True, cwd=cfgdir).strip()
+        gmo = re.match(r'^(?P<tag>.*)-(?P<dist>\d+)-g(?P<commit>[0-9a-f]+)'
+                       r'(?:-(?P<dirty>dirty))?$', descstr)
+        if not gmo:
+            raise ValueError('Unknown Git describe format')
+        distance = int(gmo.group('dist'))
+        dirty = gmo.group('dirty')
+        if distance == 0:
+            return '-'.join(filter(None, (gmo.group('tag'), dirty)))
+        if distance <= max_tag_dist:
+            return descstr
+        return  '-'.join(filter(None, (gmo.group('commit'), dirty)))
+    except (OSError, ValueError):
+        pass
+    try:
+        change = check_output(['git', 'status', '--porcelain'],
+                              text=True, cwd=cfgdir).strip()
+        descstr = check_output(['git', 'rev-parse', '--short', 'HEAD'],
+                               text=True, cwd=cfgdir).strip()
+        if len(change) > 1:
+            descstr = f'{descstr}-dirty'
+        return descstr
+    except OSError:
+        pass
+    return None

scripts/opentitan/present.py → python/qemu/ot/util/present.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python3
-
 # Python PRESENT implementation
 #
 # Copyright (c) 2008
@@ -54,6 +52,8 @@ class Present:
        :param rounds: the number of rounds as an integer
     """
 
+    BLOCK_BIT_SIZE = 64
+
     SBOX = (12, 5, 6, 11, 9, 0, 10, 13, 3, 14, 15, 8, 4, 7, 1, 2)
 
     SBOX_INV = _tinvert(SBOX)

scripts/opentitan/.pylintrc

@@ -13,6 +13,7 @@ disable=
     too-many-nested-blocks,
     too-many-positional-arguments,
     too-many-public-methods,
+    too-many-return-statements,
     too-many-statements,
     unspecified-encoding,
     wrong-import-order,

scripts/opentitan/verilate.py

@@ -36,7 +36,7 @@
 
 getLogger = logging.getLogger
 
-from ot.util.argparse import ArgumentParser, FileType  # noqa: E402
+from ot.util.arg import ArgumentParser, FileType  # noqa: E402
 from ot.util.file import guess_file_type, make_vmem_from_elf
 from ot.util.log import ColorLogFormatter, configure_loggers