Skip to content

Utilities

Helper modules used across sts-libs.

Command Line

sts.utils.cmdline

Command execution: run(), run_argv(), and BackgroundCommand.

BackgroundCommand

A background subprocess with managed lifecycle.

Created via run_background() or run_background_argv(). Always reap with stop() or wait() to avoid zombie processes.

Output is captured to temporary files so the child never blocks on a full pipe buffer.

Source code in sts_libs/src/sts/utils/cmdline.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
class BackgroundCommand:
    """A background subprocess with managed lifecycle.

    Created via ``run_background()`` or ``run_background_argv()``.
    Always reap with ``stop()`` or ``wait()`` to avoid zombie processes.

    Output is captured to temporary files so the child never blocks on a
    full pipe buffer.
    """

    _STOP_TIMEOUT: float = 5

    def __init__(
        self,
        process: subprocess.Popen[bytes],
        command: str,
        stdout_file: BinaryIO,
        stderr_file: BinaryIO,
    ) -> None:
        self._process = process
        self.command = command
        self._stdout_file = stdout_file
        self._stderr_file = stderr_file
        if process.poll() is not None:
            stderr = self._read_file(self._stderr_file)
            self._close_files()
            raise STSError(f"Process exited immediately: '{command}' (rc={process.returncode})\n{stderr}")
        logger.info(f"Started background process (pid={process.pid}): '{command}'")

    @property
    def running(self) -> bool:
        """True if the process has not yet exited."""
        return self._process.poll() is None

    @property
    def pid(self) -> int:
        """OS process ID."""
        return self._process.pid

    def stop(self, timeout: float = 30) -> CommandResult:
        """Terminate the process and return its result."""
        self._signal(signal.SIGTERM)
        return self._reap(timeout)

    def wait(self, timeout: float | None = None) -> CommandResult:
        """Wait for the process to exit on its own, then return its result."""
        try:
            self._process.wait(timeout=timeout)
            return self._result()
        except subprocess.TimeoutExpired:
            logger.warning(f"Background process timed out after {timeout}s: '{self.command}'")
            result = self.stop()
            return result.model_copy(update={'timed_out': True})

    def _signal(self, sig: int) -> None:
        """Signal the process group (covers shell-spawned children)."""
        with suppress(ProcessLookupError, PermissionError):
            os.killpg(self._process.pid, sig)

    def _reap(self, timeout: float) -> CommandResult:
        """Wait for exit, escalating from terminate to kill."""
        try:
            self._process.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            self._signal(signal.SIGKILL)
            with suppress(subprocess.TimeoutExpired):
                self._process.wait(timeout=self._STOP_TIMEOUT)
        if self._process.returncode is None:
            logger.error(f"Process did not exit after SIGKILL (pid={self._process.pid}): '{self.command}'")
        return self._result()

    @staticmethod
    def _read_file(f: BinaryIO) -> str:
        f.seek(0)
        return f.read().decode(errors='replace')

    def _close_files(self) -> None:
        self._stdout_file.close()
        self._stderr_file.close()

    def _result(self) -> CommandResult:
        stdout = self._read_file(self._stdout_file)
        stderr = self._read_file(self._stderr_file)
        self._close_files()
        rc = self._process.returncode if self._process.returncode is not None else -1
        level = logging.DEBUG if rc < 0 else (logging.WARNING if rc != 0 else logging.DEBUG)
        logger.log(level, f"Background process finished (rc={rc}): '{self.command}'")
        return CommandResult(command=self.command, rc=rc, stdout=stdout, stderr=stderr)

pid property

OS process ID.

running property

True if the process has not yet exited.

stop(timeout=30)

Terminate the process and return its result.

Source code in sts_libs/src/sts/utils/cmdline.py 
205
206
207
208
def stop(self, timeout: float = 30) -> CommandResult:
    """Terminate the process and return its result."""
    self._signal(signal.SIGTERM)
    return self._reap(timeout)

wait(timeout=None)

Wait for the process to exit on its own, then return its result.

Source code in sts_libs/src/sts/utils/cmdline.py 
210
211
212
213
214
215
216
217
218
def wait(self, timeout: float | None = None) -> CommandResult:
    """Wait for the process to exit on its own, then return its result."""
    try:
        self._process.wait(timeout=timeout)
        return self._result()
    except subprocess.TimeoutExpired:
        logger.warning(f"Background process timed out after {timeout}s: '{self.command}'")
        result = self.stop()
        return result.model_copy(update={'timed_out': True})

CommandResult pydantic-model

Bases: ReportModel

Result of a command execution.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Result of a command execution.",
  "properties": {
    "command": {
      "title": "Command",
      "type": "string"
    },
    "rc": {
      "title": "Rc",
      "type": "integer"
    },
    "stdout": {
      "default": "",
      "title": "Stdout",
      "type": "string"
    },
    "stderr": {
      "default": "",
      "title": "Stderr",
      "type": "string"
    },
    "timed_out": {
      "default": false,
      "title": "Timed Out",
      "type": "boolean"
    }
  },
  "required": [
    "command",
    "rc"
  ],
  "title": "CommandResult",
  "type": "object"
}

Config:

  • extra: forbid

Fields:

  • command (str)
  • rc (int)
  • stdout (str)
  • stderr (str)
  • timed_out (bool)
Source code in sts_libs/src/sts/utils/cmdline.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class CommandResult(ReportModel):
    """Result of a command execution."""

    # Constructed internally by run()/run_argv(), not from external JSON —
    # forbid unknown fields so misspelled kwargs raise immediately.
    model_config = ConfigDict(extra='forbid')

    command: str
    rc: int
    stdout: str = ''
    stderr: str = ''
    timed_out: bool = False

    @property
    def succeeded(self) -> bool:  # noqa: D102
        return self.rc == 0

    @property
    def failed(self) -> bool:  # noqa: D102
        return not self.succeeded

    def assert_ok(self, msg: str | None = None) -> CommandResult:
        """Raise if the command failed; return self for chaining.

        Args:
            msg: Optional context message prepended to the error.

        Raises:
            STSError: If the command failed, with command, rc, and stderr.
        """
        if self.succeeded:
            return self
        parts: list[str] = []
        if msg:
            parts.append(msg)
        if self.timed_out:
            parts.append(f"Command timed out: '{self.command}'")
        else:
            parts.append(f"Command failed (rc={self.rc}): '{self.command}'")
        if self.stderr:
            parts.append(self.stderr)
        raise STSError('\n'.join(parts))

assert_ok(msg=None)

Raise if the command failed; return self for chaining.

Parameters:

Name Type Description Default
msg str | None

Optional context message prepended to the error.

 None

Raises:

Type Description
STSError

If the command failed, with command, rc, and stderr.
Source code in sts_libs/src/sts/utils/cmdline.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def assert_ok(self, msg: str | None = None) -> CommandResult:
    """Raise if the command failed; return self for chaining.

    Args:
        msg: Optional context message prepended to the error.

    Raises:
        STSError: If the command failed, with command, rc, and stderr.
    """
    if self.succeeded:
        return self
    parts: list[str] = []
    if msg:
        parts.append(msg)
    if self.timed_out:
        parts.append(f"Command timed out: '{self.command}'")
    else:
        parts.append(f"Command failed (rc={self.rc}): '{self.command}'")
    if self.stderr:
        parts.append(self.stderr)
    raise STSError('\n'.join(parts))

build_options(**options)

Build CLI option list from keyword arguments.

Parameters:

Name Type Description Default
**options str | int | bool | None

Keyword arguments where underscores become hyphens. True → --flag, str → --key=value, False/None/empty string → omitted.

 {}

Returns:

Type Description
list[str]

List of CLI option strings.
Example 
build_options(report_format='json', yes=True)
# ['--report-format=json', '--yes']
Source code in sts_libs/src/sts/utils/cmdline.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def build_options(**options: str | int | bool | None) -> list[str]:
    """Build CLI option list from keyword arguments.

    Args:
        **options: Keyword arguments where underscores become hyphens.
            True → ``--flag``, str → ``--key=value``,
            False/None/empty string → omitted.

    Returns:
        List of CLI option strings.

    Example:
        ```python
        build_options(report_format='json', yes=True)
        # ['--report-format=json', '--yes']
        ```
    """
    argv: list[str] = []
    for key, value in options.items():
        flag = f'-{key}' if len(key) == 1 else f'--{key.replace("_", "-")}'
        if value is True:
            argv.append(flag)
        elif value not in (False, None, ''):
            if len(key) == 1:
                argv.extend([flag, str(value)])
            else:
                argv.append(f'{flag}={value}')
    return argv

exists(cmd)

Check if command exists in PATH.

Source code in sts_libs/src/sts/utils/cmdline.py 
161
162
163
def exists(cmd: str) -> bool:
    """Check if command exists in PATH."""
    return shutil.which(cmd) is not None

run(cmd, msg=None, timeout=600, log_level=logging.INFO)

Run a shell command and return a CommandResult.

Source code in sts_libs/src/sts/utils/cmdline.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def run(cmd: str, msg: str | None = None, timeout: int | None = 600, log_level: int = logging.INFO) -> CommandResult:
    """Run a shell command and return a ``CommandResult``."""
    msg = msg or 'Running'
    logger.log(log_level, f"{msg}: '{cmd}'")
    try:
        result = subprocess.run(  # noqa: S602
            cmd, shell=True, capture_output=True, check=False, timeout=timeout, encoding='utf-8', errors='replace'
        )
    except subprocess.TimeoutExpired as e:
        logger.exception(f"Command timed out after {timeout}s: '{cmd}'")
        return CommandResult(
            command=cmd,
            rc=-1,
            stdout=str(e.stdout or ''),
            stderr=str(e.stderr or ''),
            timed_out=True,
        )
    fail_level = logging.WARNING if log_level >= logging.INFO else log_level
    if result.returncode != 0:
        logger.log(fail_level, f"Command failed (rc={result.returncode}): '{cmd}'")
        if result.stderr:
            logger.log(fail_level, f'stderr: {result.stderr.rstrip()}')
        if result.stdout:
            logger.debug(f'stdout: {result.stdout.rstrip()}')
    elif result.stdout:
        logger.debug(f'stdout: {result.stdout.rstrip()}')
    return CommandResult(command=cmd, rc=result.returncode, stdout=result.stdout, stderr=result.stderr)

run_argv(argv, msg=None, timeout=600, log_level=logging.INFO)

Run a command from an argument vector (no shell) and return a CommandResult.

Source code in sts_libs/src/sts/utils/cmdline.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def run_argv(
    argv: Sequence[str], msg: str | None = None, timeout: int | None = 600, log_level: int = logging.INFO
) -> CommandResult:
    """Run a command from an argument vector (no shell) and return a ``CommandResult``."""
    cmd_str = shlex.join(argv)
    logger.log(log_level, f"{msg or 'Running'}: '{cmd_str}'")
    try:
        result = subprocess.run(  # noqa: S603
            list(argv), capture_output=True, check=False, timeout=timeout, encoding='utf-8', errors='replace'
        )
    except subprocess.TimeoutExpired as e:
        logger.exception(f"Command timed out after {timeout}s: '{cmd_str}'")
        return CommandResult(
            command=cmd_str, rc=-1, timed_out=True, stdout=str(e.stdout or ''), stderr=str(e.stderr or '')
        )
    fail_level = logging.WARNING if log_level >= logging.INFO else log_level
    if result.returncode != 0:
        logger.log(fail_level, f"Command failed (rc={result.returncode}): '{cmd_str}'")
        if result.stderr:
            logger.log(fail_level, f'stderr: {result.stderr.rstrip()}')
        if result.stdout:
            logger.debug(f'stdout: {result.stdout.rstrip()}')
    elif result.stdout:
        logger.debug(f'stdout: {result.stdout.rstrip()}')
    return CommandResult(command=cmd_str, rc=result.returncode, stdout=result.stdout, stderr=result.stderr)

run_background(cmd)

Start a shell command in the background.

Source code in sts_libs/src/sts/utils/cmdline.py 
260
261
262
263
264
265
266
267
268
269
270
271
def run_background(cmd: str) -> BackgroundCommand:
    """Start a shell command in the background."""
    stdout_file, stderr_file = _open_output(), _open_output()
    process = subprocess.Popen(  # noqa: S602
        cmd,
        shell=True,
        stdin=subprocess.DEVNULL,
        stdout=stdout_file,
        stderr=stderr_file,
        start_new_session=True,
    )
    return BackgroundCommand(process, cmd, stdout_file, stderr_file)

run_background_argv(argv)

Start a command from an argument vector in the background.

Source code in sts_libs/src/sts/utils/cmdline.py 
274
275
276
277
278
279
280
281
282
283
284
285
def run_background_argv(argv: Sequence[str]) -> BackgroundCommand:
    """Start a command from an argument vector in the background."""
    cmd_str = shlex.join(argv)
    stdout_file, stderr_file = _open_output(), _open_output()
    process = subprocess.Popen(  # noqa: S603
        list(argv),
        stdin=subprocess.DEVNULL,
        stdout=stdout_file,
        stderr=stderr_file,
        start_new_session=True,
    )
    return BackgroundCommand(process, cmd_str, stdout_file, stderr_file)

wait_for_device_io(device, *, timeout=10.0, poll_interval=0.1)

Block until at least one new I/O operation completes on a block device.

Returns True if I/O was observed, False on timeout.

Source code in sts_libs/src/sts/utils/cmdline.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
def wait_for_device_io(device: str | Path, *, timeout: float = 10.0, poll_interval: float = 0.1) -> bool:
    """Block until at least one new I/O operation completes on a block device.

    Returns True if I/O was observed, False on timeout.
    """
    stat_path = Path(f'/sys/block/{Path(device).name}/stat')
    fields = stat_path.read_text().split()
    baseline = int(fields[0]) + int(fields[4])
    deadline = monotonic() + timeout
    while monotonic() < deadline:
        fields = stat_path.read_text().split()
        if int(fields[0]) + int(fields[4]) > baseline:
            return True
        sleep(poll_interval)
    logger.warning(f'No I/O observed on {device} within {timeout}s')
    return False

System Management

sts.utils.system

System information, service management, and time control.

LogFormat

Bases: Enum

Log format options.

Source code in sts_libs/src/sts/utils/system.py 
63
64
65
66
67
68
class LogFormat(Enum):
    """Log format options."""

    DEFAULT = auto()
    KERNEL = auto()
    REVERSE = auto()

LogOptions pydantic-model

Bases: StsBaseModel

Options for SystemManager.get_logs().

Show JSON schema: 
{
  "$defs": {
    "LogFormat": {
      "description": "Log format options.",
      "enum": [
        1,
        2,
        3
      ],
      "title": "LogFormat",
      "type": "integer"
    }
  },
  "additionalProperties": false,
  "description": "Options for ``SystemManager.get_logs()``.",
  "properties": {
    "format": {
      "$ref": "#/$defs/LogFormat",
      "default": 1
    },
    "length": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Length"
    },
    "since": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Since"
    },
    "grep": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Grep"
    },
    "options": {
      "items": {
        "type": "string"
      },
      "title": "Options",
      "type": "array"
    }
  },
  "title": "LogOptions",
  "type": "object"
}

Fields:

  • format (LogFormat)
  • length (int | None)
  • since (str | None)
  • grep (str | None)
  • options (list[str])
Source code in sts_libs/src/sts/utils/system.py 
71
72
73
74
75
76
77
78
class LogOptions(StsBaseModel):
    """Options for ``SystemManager.get_logs()``."""

    format: LogFormat = LogFormat.DEFAULT
    length: int | None = None
    since: str | None = None
    grep: str | None = None
    options: list[str] = Field(default_factory=list)

SystemInfo pydantic-model

Bases: StsBaseModel

Lazily-discovered system information (hostname, kernel, arch, distro).

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Lazily-discovered system information (hostname, kernel, arch, distro).",
  "properties": {},
  "title": "SystemInfo",
  "type": "object"
}
Source code in sts_libs/src/sts/utils/system.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class SystemInfo(StsBaseModel):
    """Lazily-discovered system information (hostname, kernel, arch, distro)."""

    _hostname: str | None = PrivateAttr(default=None)
    _kernel: str | None = PrivateAttr(default=None)
    _arch: str | None = PrivateAttr(default=None)
    _distribution: str | None = PrivateAttr(default=None)
    _release: str | None = PrivateAttr(default=None)
    _codename: str | None = PrivateAttr(default=None)

    @property
    def hostname(self) -> str | None:
        """System hostname (cached)."""
        if self._hostname is None:
            result = run('hostname', log_level=logging.DEBUG)
            if result.succeeded:
                self._hostname = result.stdout.strip()
        return self._hostname

    @property
    def kernel(self) -> str | None:
        """Kernel version string (cached)."""
        if self._kernel is None:
            try:
                self._kernel = Path('/proc/sys/kernel/osrelease').read_text().strip()
            except OSError:
                self._kernel = run('uname -r').stdout.strip()

        return self._kernel

    @property
    def arch(self) -> str | None:
        """System architecture (cached)."""
        if self._arch is None:
            self._arch = platform.machine()
        return self._arch

    @property
    def distribution(self) -> str | None:
        """Distribution ID from os-release (cached)."""
        if self._distribution is None:
            self._distribution = _get_os_release().get('ID')
        return self._distribution

    @property
    def release(self) -> str | None:
        """Distribution VERSION_ID from os-release (cached)."""
        if self._release is None:
            self._release = _get_os_release().get('VERSION_ID')
        return self._release

    @property
    def version(self) -> VersionInfo:
        """Distribution release parsed as VersionInfo."""
        if self.release:
            return VersionInfo.from_string(self.release)
        return VersionInfo(major=0)

    @property
    def codename(self) -> str | None:
        """Distribution codename from os-release (cached)."""
        if self._codename is None:
            self._codename = _get_os_release().get('VERSION_CODENAME')
        return self._codename

    @classmethod
    def get_current(cls) -> SystemInfo:
        """Create a SystemInfo (properties are discovered lazily)."""
        return cls()  # Values will be discovered when needed

    @property
    def is_debug(self) -> bool:
        """True if running a +debug kernel."""
        return bool(self.kernel and '+debug' in self.kernel)

    @property
    def in_container(self) -> bool:
        """True if running inside a container."""
        try:
            proc_current = Path('/proc/1/attr/current').read_text()
            if 'container_t' in proc_current or 'unconfined' in proc_current:
                return True
            if 'docker' in Path('/proc/self/cgroup').read_text():
                return True
        except PermissionError:
            logger.debug('Assuming containerized environment')
            return True
        return False

    def log_all(self) -> None:
        """Log all system information at debug level."""
        logger.debug(f'Hostname: {self.hostname}')
        logger.debug(f'Kernel: {self.kernel}')
        logger.debug(f'Architecture: {self.arch}')
        logger.debug(f'Distribution: {self.distribution}')
        logger.debug(f'Release: {self.release}')
        logger.debug(f'Codename: {self.codename}')

arch property

System architecture (cached).

codename property

Distribution codename from os-release (cached).

distribution property

Distribution ID from os-release (cached).

hostname property

System hostname (cached).

in_container property

True if running inside a container.

is_debug property

True if running a +debug kernel.

kernel property

Kernel version string (cached).

release property

Distribution VERSION_ID from os-release (cached).

version property

Distribution release parsed as VersionInfo.

get_current() classmethod

Create a SystemInfo (properties are discovered lazily).

Source code in sts_libs/src/sts/utils/system.py 
146
147
148
149
@classmethod
def get_current(cls) -> SystemInfo:
    """Create a SystemInfo (properties are discovered lazily)."""
    return cls()  # Values will be discovered when needed

log_all()

Log all system information at debug level.

Source code in sts_libs/src/sts/utils/system.py 
170
171
172
173
174
175
176
177
def log_all(self) -> None:
    """Log all system information at debug level."""
    logger.debug(f'Hostname: {self.hostname}')
    logger.debug(f'Kernel: {self.kernel}')
    logger.debug(f'Architecture: {self.arch}')
    logger.debug(f'Distribution: {self.distribution}')
    logger.debug(f'Release: {self.release}')
    logger.debug(f'Codename: {self.codename}')

SystemManager

System logs, sosreport generation, and systemd service management.

Source code in sts_libs/src/sts/utils/system.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
class SystemManager:
    """System logs, sosreport generation, and systemd service management."""

    def __init__(self) -> None:
        self.info = SystemInfo.get_current()
        self.package_manager = RpmOstree() if check_rpm_ostree_status() else Dnf()

    @staticmethod
    def get_logs(options: LogOptions | None = None) -> str | None:
        """Get system logs via journalctl, formatted to match /var/log/messages."""
        options = options or LogOptions()

        # Build command
        cmd = ['journalctl']
        if options.format == LogFormat.KERNEL:
            cmd.append('-k')
        if options.length:
            cmd.extend(['-n', str(options.length)])
        if options.format == LogFormat.REVERSE:
            cmd.append('-r')
        if options.since:
            cmd.extend(['-S', options.since])
        if options.options:
            cmd.extend(options.options)
        if options.grep:
            cmd.extend(['|', 'grep', f"'{options.grep}'"])

        result = run(' '.join(cmd))
        if result.failed:
            logger.error('Failed to get system logs')
            return None

        # Format output to match /var/log/messages
        output: list[str] = []
        for line in result.stdout.splitlines():
            parts = line.split()
            if len(parts) < MIN_LOG_PARTS:
                continue
            parts[3] = parts[3].split('.')[0]
            output.append(' '.join(parts))

        return '\n'.join(output)

    @staticmethod
    def generate_sosreport(skip_plugins: str | None = None, plugin_timeout: int = 300) -> str | None:
        """Generate a sosreport, returning the archive path or None on failure."""
        ensure_installed('sos')

        cmd = ['sos', 'report', '--batch', f'--plugin-timeout={plugin_timeout}']
        if skip_plugins:
            cmd.extend(['--skip-plugins', skip_plugins])

        result = run(' '.join(cmd))
        if result.failed:
            logger.error('Failed to generate sosreport')
            return None

        # Find sosreport path in output
        for line in result.stdout.splitlines():
            if '/tmp/sosreport' in line:
                return line.strip()

        return None

    @staticmethod
    def get_timestamp(timezone_: Literal['utc', 'local'] = 'local') -> str:
        """Current timestamp as 'YYYYMMDDhhmmss'."""
        return datetime.now(tz=UTC if timezone_ == 'utc' else None).strftime('%Y%m%d%H%M%S')

    @staticmethod
    def clear_logs() -> None:
        """Clear dmesg."""
        run('dmesg -c')

    @staticmethod
    def is_service_enabled(service: str) -> bool:
        """Check if a systemd service is enabled."""
        result = run(f'systemctl is-enabled {service}')
        return result.succeeded

    @staticmethod
    def is_service_running(service: str) -> bool:
        """Check if a systemd service is active."""
        result = run(f'systemctl is-active {service}')
        return result.succeeded

    def service_exists(self, service: str) -> bool:
        """Check if a systemd unit file exists."""
        result = run(f'systemctl cat {service}')
        return result.succeeded

    def service_enable(self, service: str) -> bool:
        """Enable a systemd service."""
        result = run(f'systemctl enable {service}')
        return result.succeeded

    @staticmethod
    def service_disable(service: str) -> bool:
        """Disable a systemd service."""
        result = run(f'systemctl disable {service}')
        return result.succeeded

    @staticmethod
    def service_start(service: str) -> bool:
        """Start a systemd service."""
        result = run(f'systemctl start {service}')
        return result.succeeded

    @staticmethod
    def service_stop(service: str) -> bool:
        """Stop a systemd service."""
        result = run(f'systemctl stop {service}')
        return result.succeeded

    @staticmethod
    def service_restart(service: str) -> bool:
        """Restart a systemd service."""
        result = run(f'systemctl restart {service}')
        return result.succeeded

    @staticmethod
    def daemon_reload() -> bool:
        """Run ``systemctl daemon-reload`` (needed after fstab or generator changes)."""
        result = run('systemctl daemon-reload')
        return result.succeeded

    @staticmethod
    def get_unit_property(unit: str, prop: str) -> str | None:
        """Get a systemd unit property via ``systemctl show``."""
        result = run(f'systemctl show {unit} -p {prop} --value')
        if result.failed:
            logger.error(f'Failed to get property {prop} for unit {unit}: {result.stderr}')
            return None
        return result.stdout.strip()

    @staticmethod
    def escape_path_to_unit(path: str | Path, unit_type: str | None = None) -> str | None:
        """Escape a path to a systemd unit name via ``systemd-escape``."""
        cmd = f'systemd-escape --path {path!s}'
        if unit_type:
            cmd += f' --suffix={unit_type}'

        result = run(cmd)
        if result.failed:
            logger.error(f'systemd-escape failed for {path}: {result.stderr}')
            return None
        return result.stdout.strip()

    def _test_service_enable_cycle(self, service: str) -> bool:
        """Toggle enable/disable and verify it round-trips."""
        if self.is_service_enabled(service):
            # Test disable -> enable
            if not self.service_disable(service):
                return False
            time.sleep(SERVICE_WAIT_TIME)
            return self.service_enable(service)
        # Test enable -> disable
        if not self.service_enable(service):
            return False
        time.sleep(SERVICE_WAIT_TIME)
        return self.service_disable(service)

    def _test_service_start_cycle(self, service: str) -> bool:
        """Toggle start/stop and verify it round-trips."""
        if SystemManager.is_service_running(service):
            # Test stop -> start
            if not SystemManager.service_stop(service):
                return False
            time.sleep(SERVICE_WAIT_TIME)
            return SystemManager.service_start(service)
        # Test start -> stop
        if not SystemManager.service_start(service):
            return False
        time.sleep(SERVICE_WAIT_TIME)
        return self.service_stop(service)

    @staticmethod
    def _test_service_restart(service: str) -> bool:
        """Restart and verify the service is running."""
        if not SystemManager.service_restart(service):
            return False
        time.sleep(SERVICE_WAIT_TIME)
        return SystemManager.is_service_running(service)

    def test_service(self, service: str) -> bool:
        """Run enable/disable, start/stop, and restart cycles on a service."""
        # Test enable/disable cycle
        if not self._test_service_enable_cycle(service):
            return False

        # Test start/stop cycle
        if not self._test_service_start_cycle(service):
            return False

        # Test restart
        return self._test_service_restart(service)

clear_logs() staticmethod

Clear dmesg.

Source code in sts_libs/src/sts/utils/system.py 
249
250
251
252
@staticmethod
def clear_logs() -> None:
    """Clear dmesg."""
    run('dmesg -c')

daemon_reload() staticmethod

Run systemctl daemon-reload (needed after fstab or generator changes).

Source code in sts_libs/src/sts/utils/system.py 
300
301
302
303
304
@staticmethod
def daemon_reload() -> bool:
    """Run ``systemctl daemon-reload`` (needed after fstab or generator changes)."""
    result = run('systemctl daemon-reload')
    return result.succeeded

escape_path_to_unit(path, unit_type=None) staticmethod

Escape a path to a systemd unit name via systemd-escape.

Source code in sts_libs/src/sts/utils/system.py 
315
316
317
318
319
320
321
322
323
324
325
326
@staticmethod
def escape_path_to_unit(path: str | Path, unit_type: str | None = None) -> str | None:
    """Escape a path to a systemd unit name via ``systemd-escape``."""
    cmd = f'systemd-escape --path {path!s}'
    if unit_type:
        cmd += f' --suffix={unit_type}'

    result = run(cmd)
    if result.failed:
        logger.error(f'systemd-escape failed for {path}: {result.stderr}')
        return None
    return result.stdout.strip()

generate_sosreport(skip_plugins=None, plugin_timeout=300) staticmethod

Generate a sosreport, returning the archive path or None on failure.

Source code in sts_libs/src/sts/utils/system.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
@staticmethod
def generate_sosreport(skip_plugins: str | None = None, plugin_timeout: int = 300) -> str | None:
    """Generate a sosreport, returning the archive path or None on failure."""
    ensure_installed('sos')

    cmd = ['sos', 'report', '--batch', f'--plugin-timeout={plugin_timeout}']
    if skip_plugins:
        cmd.extend(['--skip-plugins', skip_plugins])

    result = run(' '.join(cmd))
    if result.failed:
        logger.error('Failed to generate sosreport')
        return None

    # Find sosreport path in output
    for line in result.stdout.splitlines():
        if '/tmp/sosreport' in line:
            return line.strip()

    return None

get_logs(options=None) staticmethod

Get system logs via journalctl, formatted to match /var/log/messages.

Source code in sts_libs/src/sts/utils/system.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
@staticmethod
def get_logs(options: LogOptions | None = None) -> str | None:
    """Get system logs via journalctl, formatted to match /var/log/messages."""
    options = options or LogOptions()

    # Build command
    cmd = ['journalctl']
    if options.format == LogFormat.KERNEL:
        cmd.append('-k')
    if options.length:
        cmd.extend(['-n', str(options.length)])
    if options.format == LogFormat.REVERSE:
        cmd.append('-r')
    if options.since:
        cmd.extend(['-S', options.since])
    if options.options:
        cmd.extend(options.options)
    if options.grep:
        cmd.extend(['|', 'grep', f"'{options.grep}'"])

    result = run(' '.join(cmd))
    if result.failed:
        logger.error('Failed to get system logs')
        return None

    # Format output to match /var/log/messages
    output: list[str] = []
    for line in result.stdout.splitlines():
        parts = line.split()
        if len(parts) < MIN_LOG_PARTS:
            continue
        parts[3] = parts[3].split('.')[0]
        output.append(' '.join(parts))

    return '\n'.join(output)

get_timestamp(timezone_='local') staticmethod

Current timestamp as 'YYYYMMDDhhmmss'.

Source code in sts_libs/src/sts/utils/system.py 
244
245
246
247
@staticmethod
def get_timestamp(timezone_: Literal['utc', 'local'] = 'local') -> str:
    """Current timestamp as 'YYYYMMDDhhmmss'."""
    return datetime.now(tz=UTC if timezone_ == 'utc' else None).strftime('%Y%m%d%H%M%S')

get_unit_property(unit, prop) staticmethod

Get a systemd unit property via systemctl show.

Source code in sts_libs/src/sts/utils/system.py 
306
307
308
309
310
311
312
313
@staticmethod
def get_unit_property(unit: str, prop: str) -> str | None:
    """Get a systemd unit property via ``systemctl show``."""
    result = run(f'systemctl show {unit} -p {prop} --value')
    if result.failed:
        logger.error(f'Failed to get property {prop} for unit {unit}: {result.stderr}')
        return None
    return result.stdout.strip()

is_service_enabled(service) staticmethod

Check if a systemd service is enabled.

Source code in sts_libs/src/sts/utils/system.py 
254
255
256
257
258
@staticmethod
def is_service_enabled(service: str) -> bool:
    """Check if a systemd service is enabled."""
    result = run(f'systemctl is-enabled {service}')
    return result.succeeded

is_service_running(service) staticmethod

Check if a systemd service is active.

Source code in sts_libs/src/sts/utils/system.py 
260
261
262
263
264
@staticmethod
def is_service_running(service: str) -> bool:
    """Check if a systemd service is active."""
    result = run(f'systemctl is-active {service}')
    return result.succeeded

service_disable(service) staticmethod

Disable a systemd service.

Source code in sts_libs/src/sts/utils/system.py 
276
277
278
279
280
@staticmethod
def service_disable(service: str) -> bool:
    """Disable a systemd service."""
    result = run(f'systemctl disable {service}')
    return result.succeeded

service_enable(service)

Enable a systemd service.

Source code in sts_libs/src/sts/utils/system.py 
271
272
273
274
def service_enable(self, service: str) -> bool:
    """Enable a systemd service."""
    result = run(f'systemctl enable {service}')
    return result.succeeded

service_exists(service)

Check if a systemd unit file exists.

Source code in sts_libs/src/sts/utils/system.py 
266
267
268
269
def service_exists(self, service: str) -> bool:
    """Check if a systemd unit file exists."""
    result = run(f'systemctl cat {service}')
    return result.succeeded

service_restart(service) staticmethod

Restart a systemd service.

Source code in sts_libs/src/sts/utils/system.py 
294
295
296
297
298
@staticmethod
def service_restart(service: str) -> bool:
    """Restart a systemd service."""
    result = run(f'systemctl restart {service}')
    return result.succeeded

service_start(service) staticmethod

Start a systemd service.

Source code in sts_libs/src/sts/utils/system.py 
282
283
284
285
286
@staticmethod
def service_start(service: str) -> bool:
    """Start a systemd service."""
    result = run(f'systemctl start {service}')
    return result.succeeded

service_stop(service) staticmethod

Stop a systemd service.

Source code in sts_libs/src/sts/utils/system.py 
288
289
290
291
292
@staticmethod
def service_stop(service: str) -> bool:
    """Stop a systemd service."""
    result = run(f'systemctl stop {service}')
    return result.succeeded

test_service(service)

Run enable/disable, start/stop, and restart cycles on a service.

Source code in sts_libs/src/sts/utils/system.py 
364
365
366
367
368
369
370
371
372
373
374
375
def test_service(self, service: str) -> bool:
    """Run enable/disable, start/stop, and restart cycles on a service."""
    # Test enable/disable cycle
    if not self._test_service_enable_cycle(service):
        return False

    # Test start/stop cycle
    if not self._test_service_start_cycle(service):
        return False

    # Test restart
    return self._test_service_restart(service)

TimeController

Temporarily shift system time for testing (requires root).

Example 
tc = TimeController()
tc.disable_ntp()
try:
    with tc.time_offset(hours=-5):
        create_something_timestamped()
finally:
    tc.enable_ntp()
Source code in sts_libs/src/sts/utils/system.py 
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
class TimeController:
    """Temporarily shift system time for testing (requires root).

    Example:
        ```python
        tc = TimeController()
        tc.disable_ntp()
        try:
            with tc.time_offset(hours=-5):
                create_something_timestamped()
        finally:
            tc.enable_ntp()
        ```
    """

    def __init__(self) -> None:
        self._ntp_was_enabled: bool | None = None

    def is_ntp_enabled(self) -> bool:
        """Check if NTP synchronization is enabled."""
        result = run('timedatectl show --property=NTP --value')
        return result.succeeded and result.stdout.strip().lower() == 'yes'

    def disable_ntp(self) -> bool:
        """Disable NTP (saves prior state for ``restore_ntp()``)."""
        self._ntp_was_enabled = self.is_ntp_enabled()
        result = run('timedatectl set-ntp false')
        if result.failed:
            logger.error(f'Failed to disable NTP: {result.stderr}')
            return False
        return True

    def enable_ntp(self) -> bool:
        """Enable NTP synchronization (triggers time sync)."""
        result = run('timedatectl set-ntp true')
        if result.failed:
            logger.error(f'Failed to enable NTP: {result.stderr}')
            return False
        return True

    def restore_ntp(self) -> bool:
        """Restore NTP to its state before ``disable_ntp()`` was called."""
        if self._ntp_was_enabled is None:
            # Never saved state, enable NTP as safe default
            return self.enable_ntp()
        if self._ntp_was_enabled:
            return self.enable_ntp()
        return True  # NTP was disabled, leave it disabled

    def set_time(self, time_str: str) -> bool:
        """Set system time (format: 'YYYY-MM-DD HH:MM:SS')."""
        result = run(f"timedatectl set-time '{time_str}'")
        if result.failed:
            logger.error(f'Failed to set time: {result.stderr}')
            return False
        logger.info(f'System time set to: {time_str}')
        return True

    def set_time_offset(self, hours: int = 0, days: int = 0) -> bool:
        """Set system time to an offset from now (negative = past)."""
        offset = timedelta(hours=hours, days=days)
        target_time = datetime.now(tz=UTC).astimezone() + offset
        time_str = target_time.strftime('%Y-%m-%d %H:%M:%S')
        return self.set_time(time_str)

    @contextlib.contextmanager
    def time_offset(self, hours: int = 0, days: int = 0) -> Generator[None, None, None]:
        """Context manager that shifts system time for the duration of the block.

        NTP must be disabled before entering. Time is NOT auto-restored on exit --
        call ``enable_ntp()`` in a finally block to re-sync.
        """
        if not self.set_time_offset(hours=hours, days=days):
            msg = f'Failed to set time offset (hours={hours}, days={days})'
            raise RuntimeError(msg)
        yield

disable_ntp()

Disable NTP (saves prior state for restore_ntp()).

Source code in sts_libs/src/sts/utils/system.py 
401
402
403
404
405
406
407
408
def disable_ntp(self) -> bool:
    """Disable NTP (saves prior state for ``restore_ntp()``)."""
    self._ntp_was_enabled = self.is_ntp_enabled()
    result = run('timedatectl set-ntp false')
    if result.failed:
        logger.error(f'Failed to disable NTP: {result.stderr}')
        return False
    return True

enable_ntp()

Enable NTP synchronization (triggers time sync).

Source code in sts_libs/src/sts/utils/system.py 
410
411
412
413
414
415
416
def enable_ntp(self) -> bool:
    """Enable NTP synchronization (triggers time sync)."""
    result = run('timedatectl set-ntp true')
    if result.failed:
        logger.error(f'Failed to enable NTP: {result.stderr}')
        return False
    return True

is_ntp_enabled()

Check if NTP synchronization is enabled.

Source code in sts_libs/src/sts/utils/system.py 
396
397
398
399
def is_ntp_enabled(self) -> bool:
    """Check if NTP synchronization is enabled."""
    result = run('timedatectl show --property=NTP --value')
    return result.succeeded and result.stdout.strip().lower() == 'yes'

restore_ntp()

Restore NTP to its state before disable_ntp() was called.

Source code in sts_libs/src/sts/utils/system.py 
418
419
420
421
422
423
424
425
def restore_ntp(self) -> bool:
    """Restore NTP to its state before ``disable_ntp()`` was called."""
    if self._ntp_was_enabled is None:
        # Never saved state, enable NTP as safe default
        return self.enable_ntp()
    if self._ntp_was_enabled:
        return self.enable_ntp()
    return True  # NTP was disabled, leave it disabled

set_time(time_str)

Set system time (format: 'YYYY-MM-DD HH:MM:SS').

Source code in sts_libs/src/sts/utils/system.py 
427
428
429
430
431
432
433
434
def set_time(self, time_str: str) -> bool:
    """Set system time (format: 'YYYY-MM-DD HH:MM:SS')."""
    result = run(f"timedatectl set-time '{time_str}'")
    if result.failed:
        logger.error(f'Failed to set time: {result.stderr}')
        return False
    logger.info(f'System time set to: {time_str}')
    return True

set_time_offset(hours=0, days=0)

Set system time to an offset from now (negative = past).

Source code in sts_libs/src/sts/utils/system.py 
436
437
438
439
440
441
def set_time_offset(self, hours: int = 0, days: int = 0) -> bool:
    """Set system time to an offset from now (negative = past)."""
    offset = timedelta(hours=hours, days=days)
    target_time = datetime.now(tz=UTC).astimezone() + offset
    time_str = target_time.strftime('%Y-%m-%d %H:%M:%S')
    return self.set_time(time_str)

time_offset(hours=0, days=0)

Context manager that shifts system time for the duration of the block.

NTP must be disabled before entering. Time is NOT auto-restored on exit -- call enable_ntp() in a finally block to re-sync.

Source code in sts_libs/src/sts/utils/system.py 
443
444
445
446
447
448
449
450
451
452
453
@contextlib.contextmanager
def time_offset(self, hours: int = 0, days: int = 0) -> Generator[None, None, None]:
    """Context manager that shifts system time for the duration of the block.

    NTP must be disabled before entering. Time is NOT auto-restored on exit --
    call ``enable_ntp()`` in a finally block to re-sync.
    """
    if not self.set_time_offset(hours=hours, days=days):
        msg = f'Failed to set time offset (hours={hours}, days={days})'
        raise RuntimeError(msg)
    yield

Package Management

sts.utils.packages

RPM package management (dnf, rpm-ostree) and repository configuration.

Dnf

DNF package manager operations.

Source code in sts_libs/src/sts/utils/packages.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class Dnf:
    """DNF package manager operations."""

    def __init__(self) -> None:
        self.repo_path = Path('/etc/yum.repos.d')

    @staticmethod
    def install(package: str) -> bool:
        """Install package (no-op if already installed)."""
        if get_package(package).is_installed:
            return True

        result = run(f'dnf install -y {package}')
        if result.failed:
            logger.error(f'Failed to install {package}:\n{result.stderr}')
            return False

        logger.debug(f'Successfully installed {package}')
        return True

    @staticmethod
    def remove(package: str) -> bool:
        """Remove package (no-op if not installed)."""
        if not get_package(package).is_installed:
            return True

        result = run(f'dnf remove -y {package}')
        if result.failed:
            logger.error(f'Failed to remove {package}:\n{result.stderr}')
            return False

        logger.debug(f'Successfully removed {package}')
        return True

    def add_repo(self, config: RepoConfig) -> bool:
        """Write a .repo file for the given configuration (no-op if it already exists)."""
        if not config.baseurl and not config.metalink:
            logger.error('Either baseurl or metalink required')
            return False

        repo_file = self.repo_path / f'{config.name.lower()}.repo'
        if repo_file.exists():
            logger.debug(f'Repository {config.name} already exists')
            return True

        # Write repo file
        try:
            content = [f'[{config.name}]']
            content.extend(f'{k}={v}' for k, v in config.to_config().items())
            repo_file.write_text('\n'.join(content))
        except OSError:
            logger.exception(f'Failed to write repository file {repo_file}')
            return False

        return True

    def remove_repo(self, name: str) -> bool:
        """Delete the .repo file for a repository."""
        repo_file = self.repo_path / f'{name.lower()}.repo'
        try:
            repo_file.unlink(missing_ok=True)
        except OSError:
            logger.exception(f'Failed to remove repository file {repo_file}')
            return False

        return True

    @staticmethod
    def repo_exists(name: str) -> bool:
        """Check if repository exists via ``dnf repoinfo``."""
        result = run(f'dnf repoinfo {name}')
        if result.failed:
            logger.error(f'Repository {name} not found:\n{result.stderr}')
            return False
        return True

    def repo_enabled(self, name: str) -> bool:
        """Check if repository exists and is enabled."""
        if not self.repo_exists(name):
            return False

        result = run(f'dnf repoinfo {name}')
        if 'enabled' not in result.stdout:
            logger.error(f'Repository {name} not enabled:\n{result.stderr}')
            return False

        return True

    def download_repo(self, url: str, name: str | None = None, *, overwrite: bool = True) -> bool:
        """Download a .repo file from *url* into /etc/yum.repos.d/."""
        if not self.install('curl'):
            return False

        if not name:
            name = url.rsplit('/', maxsplit=1)[-1]
        if not name.endswith('.repo'):
            name = f'{name}.repo'

        repo_file = self.repo_path / name
        if repo_file.exists() and not overwrite:
            logger.debug(f'Repository file {repo_file} already exists')
            return True

        result = run(f'curl {url} --output {repo_file}')
        if result.failed:
            logger.error(f'Failed to download repository file from {url}:\n{result.stderr}')
            return False

        return True

add_repo(config)

Write a .repo file for the given configuration (no-op if it already exists).

Source code in sts_libs/src/sts/utils/packages.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def add_repo(self, config: RepoConfig) -> bool:
    """Write a .repo file for the given configuration (no-op if it already exists)."""
    if not config.baseurl and not config.metalink:
        logger.error('Either baseurl or metalink required')
        return False

    repo_file = self.repo_path / f'{config.name.lower()}.repo'
    if repo_file.exists():
        logger.debug(f'Repository {config.name} already exists')
        return True

    # Write repo file
    try:
        content = [f'[{config.name}]']
        content.extend(f'{k}={v}' for k, v in config.to_config().items())
        repo_file.write_text('\n'.join(content))
    except OSError:
        logger.exception(f'Failed to write repository file {repo_file}')
        return False

    return True

download_repo(url, name=None, *, overwrite=True)

Download a .repo file from url into /etc/yum.repos.d/.

Source code in sts_libs/src/sts/utils/packages.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def download_repo(self, url: str, name: str | None = None, *, overwrite: bool = True) -> bool:
    """Download a .repo file from *url* into /etc/yum.repos.d/."""
    if not self.install('curl'):
        return False

    if not name:
        name = url.rsplit('/', maxsplit=1)[-1]
    if not name.endswith('.repo'):
        name = f'{name}.repo'

    repo_file = self.repo_path / name
    if repo_file.exists() and not overwrite:
        logger.debug(f'Repository file {repo_file} already exists')
        return True

    result = run(f'curl {url} --output {repo_file}')
    if result.failed:
        logger.error(f'Failed to download repository file from {url}:\n{result.stderr}')
        return False

    return True

install(package) staticmethod

Install package (no-op if already installed).

Source code in sts_libs/src/sts/utils/packages.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
@staticmethod
def install(package: str) -> bool:
    """Install package (no-op if already installed)."""
    if get_package(package).is_installed:
        return True

    result = run(f'dnf install -y {package}')
    if result.failed:
        logger.error(f'Failed to install {package}:\n{result.stderr}')
        return False

    logger.debug(f'Successfully installed {package}')
    return True

remove(package) staticmethod

Remove package (no-op if not installed).

Source code in sts_libs/src/sts/utils/packages.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
@staticmethod
def remove(package: str) -> bool:
    """Remove package (no-op if not installed)."""
    if not get_package(package).is_installed:
        return True

    result = run(f'dnf remove -y {package}')
    if result.failed:
        logger.error(f'Failed to remove {package}:\n{result.stderr}')
        return False

    logger.debug(f'Successfully removed {package}')
    return True

remove_repo(name)

Delete the .repo file for a repository.

Source code in sts_libs/src/sts/utils/packages.py 
122
123
124
125
126
127
128
129
130
131
def remove_repo(self, name: str) -> bool:
    """Delete the .repo file for a repository."""
    repo_file = self.repo_path / f'{name.lower()}.repo'
    try:
        repo_file.unlink(missing_ok=True)
    except OSError:
        logger.exception(f'Failed to remove repository file {repo_file}')
        return False

    return True

repo_enabled(name)

Check if repository exists and is enabled.

Source code in sts_libs/src/sts/utils/packages.py 
142
143
144
145
146
147
148
149
150
151
152
def repo_enabled(self, name: str) -> bool:
    """Check if repository exists and is enabled."""
    if not self.repo_exists(name):
        return False

    result = run(f'dnf repoinfo {name}')
    if 'enabled' not in result.stdout:
        logger.error(f'Repository {name} not enabled:\n{result.stderr}')
        return False

    return True

repo_exists(name) staticmethod

Check if repository exists via dnf repoinfo.

Source code in sts_libs/src/sts/utils/packages.py 
133
134
135
136
137
138
139
140
@staticmethod
def repo_exists(name: str) -> bool:
    """Check if repository exists via ``dnf repoinfo``."""
    result = run(f'dnf repoinfo {name}')
    if result.failed:
        logger.error(f'Repository {name} not found:\n{result.stderr}')
        return False
    return True

PackageInfo pydantic-model

Bases: ReportModel

RPM package information.

Show JSON schema: 
{
  "description": "RPM package information.",
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "is_installed": {
      "title": "Is Installed",
      "type": "boolean"
    },
    "version": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Version"
    },
    "release": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Release"
    }
  },
  "required": [
    "name",
    "is_installed"
  ],
  "title": "PackageInfo",
  "type": "object"
}

Fields:

  • name (str)
  • is_installed (bool)
  • version (str | None)
  • release (str | None)
Source code in sts_libs/src/sts/utils/packages.py 
18
19
20
21
22
23
24
class PackageInfo(ReportModel):
    """RPM package information."""

    name: str
    is_installed: bool
    version: str | None = None
    release: str | None = None

RepoConfig pydantic-model

Bases: StsBaseModel

DNF repository configuration (baseurl or metalink).

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "DNF repository configuration (baseurl or metalink).",
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "baseurl": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Baseurl"
    },
    "metalink": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Metalink"
    },
    "enabled": {
      "default": true,
      "title": "Enabled",
      "type": "boolean"
    },
    "gpgcheck": {
      "default": false,
      "title": "Gpgcheck",
      "type": "boolean"
    },
    "skip_if_unavailable": {
      "default": true,
      "title": "Skip If Unavailable",
      "type": "boolean"
    }
  },
  "required": [
    "name"
  ],
  "title": "RepoConfig",
  "type": "object"
}

Fields:

  • name (str)
  • baseurl (str | None)
  • metalink (str | None)
  • enabled (bool)
  • gpgcheck (bool)
  • skip_if_unavailable (bool)
Source code in sts_libs/src/sts/utils/packages.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class RepoConfig(StsBaseModel):
    """DNF repository configuration (baseurl or metalink)."""

    name: str
    baseurl: str | None = None
    metalink: str | None = None
    enabled: bool = True
    gpgcheck: bool = False
    skip_if_unavailable: bool = True

    def to_config(self) -> dict[str, str]:
        """Convert to .repo file key-value pairs."""
        config = {
            'name': self.name,
            'enabled': '1' if self.enabled else '0',
            'gpgcheck': '1' if self.gpgcheck else '0',
            'skip_if_unavailable': '1' if self.skip_if_unavailable else '0',
        }
        if self.baseurl:
            config['baseurl'] = self.baseurl
        if self.metalink:
            config['metalink'] = self.metalink
        return config

to_config()

Convert to .repo file key-value pairs.

Source code in sts_libs/src/sts/utils/packages.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
def to_config(self) -> dict[str, str]:
    """Convert to .repo file key-value pairs."""
    config = {
        'name': self.name,
        'enabled': '1' if self.enabled else '0',
        'gpgcheck': '1' if self.gpgcheck else '0',
        'skip_if_unavailable': '1' if self.skip_if_unavailable else '0',
    }
    if self.baseurl:
        config['baseurl'] = self.baseurl
    if self.metalink:
        config['metalink'] = self.metalink
    return config

RpmOstree

RPM-OSTree package manager operations.

Source code in sts_libs/src/sts/utils/packages.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class RpmOstree:
    """RPM-OSTree package manager operations."""

    @staticmethod
    def install(package: str) -> bool:
        """Install package via rpm-ostree (no-op if already installed)."""
        if get_package(package).is_installed:
            return True

        result = run(f'rpm-ostree install --apply-live --idempotent --allow-inactive --assumeyes {package}')
        if result.failed:
            logger.error(f'Failed to install {package}:\n{result.stderr}')
            return False

        logger.debug(f'Successfully installed {package}')
        return True

    @staticmethod
    def remove(package: str) -> bool:
        """Remove package via rpm-ostree (no-op if not installed)."""
        if not get_package(package).is_installed:
            return True

        result = run(f'rpm-ostree uninstall --assumeyes {package}')
        if result.failed:
            logger.error(f'Failed to remove {package}:\n{result.stderr}')
            return False

        logger.debug(f'Successfully removed {package}')
        return True

install(package) staticmethod

Install package via rpm-ostree (no-op if already installed).

Source code in sts_libs/src/sts/utils/packages.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
@staticmethod
def install(package: str) -> bool:
    """Install package via rpm-ostree (no-op if already installed)."""
    if get_package(package).is_installed:
        return True

    result = run(f'rpm-ostree install --apply-live --idempotent --allow-inactive --assumeyes {package}')
    if result.failed:
        logger.error(f'Failed to install {package}:\n{result.stderr}')
        return False

    logger.debug(f'Successfully installed {package}')
    return True

remove(package) staticmethod

Remove package via rpm-ostree (no-op if not installed).

Source code in sts_libs/src/sts/utils/packages.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
@staticmethod
def remove(package: str) -> bool:
    """Remove package via rpm-ostree (no-op if not installed)."""
    if not get_package(package).is_installed:
        return True

    result = run(f'rpm-ostree uninstall --assumeyes {package}')
    if result.failed:
        logger.error(f'Failed to remove {package}:\n{result.stderr}')
        return False

    logger.debug(f'Successfully removed {package}')
    return True

check_rpm_ostree_status()

Check if the system is managed by rpm-ostree (cached).

Source code in sts_libs/src/sts/utils/packages.py 
212
213
214
215
216
217
def check_rpm_ostree_status() -> bool:
    """Check if the system is managed by rpm-ostree (cached)."""
    global _rpm_ostree_cached  # noqa: PLW0603
    if _rpm_ostree_cached is None:
        _rpm_ostree_cached = Path('/run/ostree-booted').exists()
    return _rpm_ostree_cached

ensure_installed(*packages)

Install any missing packages, auto-detecting dnf vs rpm-ostree.

Source code in sts_libs/src/sts/utils/packages.py 
220
221
222
223
224
def ensure_installed(*packages: str) -> bool:
    """Install any missing packages, auto-detecting dnf vs rpm-ostree."""
    pm = RpmOstree() if check_rpm_ostree_status() else Dnf()

    return all(pm.install(package) for package in packages)

get_package(name)

Query RPM for package installation status and version.

Source code in sts_libs/src/sts/utils/packages.py 
27
28
29
30
31
32
33
34
35
36
37
38
def get_package(name: str) -> PackageInfo:
    """Query RPM for package installation status and version."""
    result = run(f'rpm -q --queryformat "%{{VERSION}} %{{RELEASE}}\\n" {name}', log_level=logging.DEBUG)
    if result.failed:
        return PackageInfo(name=name, is_installed=False)
    parts = result.stdout.strip().splitlines()[0].split()
    return PackageInfo(
        name=name,
        is_installed=True,
        version=parts[0] if parts else None,
        release=parts[1] if len(parts) > 1 else None,
    )

get_package_version(package_name)

Get installed package version as VersionInfo, or None if not installed.

Source code in sts_libs/src/sts/utils/packages.py 
227
228
229
230
231
232
def get_package_version(package_name: str) -> VersionInfo | None:
    """Get installed package version as VersionInfo, or None if not installed."""
    pkg = get_package(package_name)
    if not pkg.is_installed or not pkg.version:
        return None
    return VersionInfo.from_string(pkg.version)

log_package_versions(*package_names)

Log installed version (or 'not installed') for each package.

Source code in sts_libs/src/sts/utils/packages.py 
235
236
237
238
239
240
241
242
243
def log_package_versions(*package_names: str) -> None:
    """Log installed version (or 'not installed') for each package."""
    for package_name in package_names:
        pkg = get_package(package_name)
        if not pkg.is_installed:
            logger.debug(f'Package {package_name} version: not installed')
        else:
            version = f'{pkg.version}-{pkg.release}'
            logger.debug(f'Package {package_name} version: {version}')

Module Management

sts.utils.modules

Kernel module loading, unloading, and introspection.

ManagedModule

Acquire a kernel module and release only a load owned by this object.

Parameters:

Name Type Description Default
name str

Kernel module name.

 required
parameters Mapping[str, str | int | bool] | None

Parameters verified after loading and passed to modprobe.

 None
load_only_parameters Mapping[str, str | int | bool] | None

Extra parameters passed to modprobe but not verified via sysfs (e.g. trigger parameters like add_host).

 None
exclusive bool

Refuse acquisition when the module was already loaded.

 False
unreadable_parameters_postcondition Callable[[], bool] | None

Explicit fallback used only after this object loads a module and none of the requested parameters can be read back.

 None
Source code in sts_libs/src/sts/utils/modules.py 
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
class ManagedModule:
    """Acquire a kernel module and release only a load owned by this object.

    Args:
        name: Kernel module name.
        parameters: Parameters verified after loading and passed to modprobe.
        load_only_parameters: Extra parameters passed to modprobe but not
            verified via sysfs (e.g. trigger parameters like ``add_host``).
        exclusive: Refuse acquisition when the module was already loaded.
        unreadable_parameters_postcondition: Explicit fallback used only after
            this object loads a module and none of the requested parameters can
            be read back.
    """

    def __init__(
        self,
        name: str,
        *,
        parameters: Mapping[str, str | int | bool] | None = None,
        load_only_parameters: Mapping[str, str | int | bool] | None = None,
        exclusive: bool = False,
        unreadable_parameters_postcondition: Callable[[], bool] | None = None,
    ) -> None:
        self.name = name
        self.parameters = dict(parameters or {})
        self._load_parameters = {**self.parameters, **(load_only_parameters or {})}
        self.exclusive = exclusive
        self.unreadable_parameters_postcondition = unreadable_parameters_postcondition
        self.info: ModuleInfo | None = None
        self._owned = False

    @property
    def owns_load(self) -> bool:
        """Whether this lifecycle currently owns the module load."""
        return self._owned

    def acquire(self) -> ModuleInfo:
        """Make the module available or raise an explicit acquisition outcome.

        Missing modules and exclusive conflicts are typed pre-mutation outcomes.
        Load and post-load verification failures are typed setup errors. Any
        failed rollback retains ownership for a later :meth:`release` retry.
        """
        info = ModuleInfo.from_name(self.name)
        if info is None:
            raise ModuleUnavailableError(f'Module {self.name} is unavailable')
        if info.loaded:
            if self.exclusive:
                raise ModuleAlreadyLoadedError(f'Module {self.name} is already loaded')
            self.info = info
            return info

        info.load(self._load_parameters, raise_on_error=True)

        self._owned = True
        self.info = ModuleInfo.from_name(self.name)
        verification_cause: BaseException | None = None
        try:
            verified = self.info is not None and self.info.loaded and self._parameters_verified(self.info)
        except BaseException as error:
            verification_cause = error
            verified = False
            if isinstance(error, Exception):
                logger.warning(f'Failed unreadable-parameter postcondition for {self.name}', exc_info=True)

        if not verified:
            verification_error = self._verification_error()
            try:
                self._rollback_unverified_load(verification_error, verification_cause)
            finally:
                self.info = None
        assert self.info is not None
        return self.info

    def release(self, timeout: float = DEFAULT_TIMEOUT) -> bool:
        """Unload an owned module within a bounded retry window.

        Args:
            timeout: Maximum number of seconds to retry unload or confirm absence.
        """
        if not self._owned:
            return True

        info = ModuleInfo.from_name(self.name)
        if info is None:
            info = ModuleInfo(name=self.name).discover()
        if info.state_observed and not info.loaded:
            self._owned = False
            return True

        deadline = time.monotonic() + timeout
        unload_succeeded = False
        while True:
            if unload_succeeded:
                info = info.discover()
            else:
                try:
                    unload_succeeded = info.unload()
                except (ModuleInUseError, ModuleUnloadError, OSError):
                    unload_succeeded = False

            if info.state_observed and not info.loaded:
                self._owned = False
                return True

            remaining = deadline - time.monotonic()
            if remaining <= 0:
                logger.error(f'Failed to unload owned module {self.name}')
                return False
            time.sleep(min(MODULE_RETRY_INTERVAL, remaining))

    def __enter__(self) -> ModuleInfo:
        """Acquire the module for a ``with`` block."""
        return self.acquire()

    def __exit__(self, _exc_type: object, exc_value: BaseException | None, _traceback: object) -> None:
        """Release the module acquired for a ``with`` block or surface cleanup failure."""
        try:
            released = self.release()
        except BaseException as cleanup_error:
            if exc_value is not None:
                raise BaseExceptionGroup(
                    f'Module {self.name} context body and cleanup failed',
                    [exc_value, cleanup_error],
                ) from None
            raise
        if released:
            return
        cleanup_error = ModuleUnloadError(f'Failed to release owned module {self.name}')
        if exc_value is not None:
            raise BaseExceptionGroup(
                f'Module {self.name} context body and cleanup failed',
                [exc_value, cleanup_error],
            ) from None
        raise cleanup_error

    def _parameters_match(self, info: ModuleInfo) -> bool:
        """Return whether requested parameters were read back from sysfs."""
        return parameters_match(info.parameters, self.parameters)

    def _parameters_verified(self, info: ModuleInfo) -> bool:
        """Verify parameter readback or an explicit all-unreadable fallback."""
        if self._parameters_match(info):
            return True
        if (
            not self.parameters
            or self.unreadable_parameters_postcondition is None
            or any(key in info.parameters for key in self.parameters)
        ):
            return False
        return self.unreadable_parameters_postcondition()

    def _verification_error(self) -> ModuleVerificationError:
        """Describe the failed post-load state without hiding its requested values."""
        if self.info is None:
            return ModuleVerificationError(f'Module {self.name} could not be rediscovered after loading')
        if not self.info.loaded:
            return ModuleVerificationError(f'Module {self.name} is not loaded after successful modprobe')
        requested = ', '.join(f'{key}={value}' for key, value in self._load_parameters.items()) or 'module loaded state'
        return ModuleVerificationError(f'Failed to verify module {self.name} postcondition: {requested}')

    def _rollback_unverified_load(
        self,
        verification_error: ModuleVerificationError,
        verification_cause: BaseException | None,
    ) -> None:
        """Roll back an owned load, then surface the typed verification failure."""
        cleanup_error = None if self.release() else ModuleUnloadError(f'Failed to roll back owned module {self.name}')

        if cleanup_error is not None:
            if verification_cause is not None and not isinstance(verification_cause, Exception):
                raise BaseExceptionGroup(
                    f'Module {self.name} parameter verification and rollback failed',
                    [verification_cause, verification_error, cleanup_error],
                )
            if verification_cause is not None:
                verification_error.__cause__ = verification_cause
            raise cleanup_error from verification_error

        if verification_cause is not None and not isinstance(verification_cause, Exception):
            raise BaseExceptionGroup(
                f'Module {self.name} parameter verification failed',
                [verification_cause, verification_error],
            )
        raise verification_error from verification_cause

owns_load property

Whether this lifecycle currently owns the module load.

__enter__()

Acquire the module for a with block.

Source code in sts_libs/src/sts/utils/modules.py 
434
435
436
def __enter__(self) -> ModuleInfo:
    """Acquire the module for a ``with`` block."""
    return self.acquire()

__exit__(_exc_type, exc_value, _traceback)

Release the module acquired for a with block or surface cleanup failure.

Source code in sts_libs/src/sts/utils/modules.py 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
def __exit__(self, _exc_type: object, exc_value: BaseException | None, _traceback: object) -> None:
    """Release the module acquired for a ``with`` block or surface cleanup failure."""
    try:
        released = self.release()
    except BaseException as cleanup_error:
        if exc_value is not None:
            raise BaseExceptionGroup(
                f'Module {self.name} context body and cleanup failed',
                [exc_value, cleanup_error],
            ) from None
        raise
    if released:
        return
    cleanup_error = ModuleUnloadError(f'Failed to release owned module {self.name}')
    if exc_value is not None:
        raise BaseExceptionGroup(
            f'Module {self.name} context body and cleanup failed',
            [exc_value, cleanup_error],
        ) from None
    raise cleanup_error

acquire()

Make the module available or raise an explicit acquisition outcome.

Missing modules and exclusive conflicts are typed pre-mutation outcomes. Load and post-load verification failures are typed setup errors. Any failed rollback retains ownership for a later :meth:release retry.

Source code in sts_libs/src/sts/utils/modules.py 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
def acquire(self) -> ModuleInfo:
    """Make the module available or raise an explicit acquisition outcome.

    Missing modules and exclusive conflicts are typed pre-mutation outcomes.
    Load and post-load verification failures are typed setup errors. Any
    failed rollback retains ownership for a later :meth:`release` retry.
    """
    info = ModuleInfo.from_name(self.name)
    if info is None:
        raise ModuleUnavailableError(f'Module {self.name} is unavailable')
    if info.loaded:
        if self.exclusive:
            raise ModuleAlreadyLoadedError(f'Module {self.name} is already loaded')
        self.info = info
        return info

    info.load(self._load_parameters, raise_on_error=True)

    self._owned = True
    self.info = ModuleInfo.from_name(self.name)
    verification_cause: BaseException | None = None
    try:
        verified = self.info is not None and self.info.loaded and self._parameters_verified(self.info)
    except BaseException as error:
        verification_cause = error
        verified = False
        if isinstance(error, Exception):
            logger.warning(f'Failed unreadable-parameter postcondition for {self.name}', exc_info=True)

    if not verified:
        verification_error = self._verification_error()
        try:
            self._rollback_unverified_load(verification_error, verification_cause)
        finally:
            self.info = None
    assert self.info is not None
    return self.info

release(timeout=DEFAULT_TIMEOUT)

Unload an owned module within a bounded retry window.

Parameters:

Name Type Description Default
timeout float

Maximum number of seconds to retry unload or confirm absence.

 DEFAULT_TIMEOUT
Source code in sts_libs/src/sts/utils/modules.py 
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
def release(self, timeout: float = DEFAULT_TIMEOUT) -> bool:
    """Unload an owned module within a bounded retry window.

    Args:
        timeout: Maximum number of seconds to retry unload or confirm absence.
    """
    if not self._owned:
        return True

    info = ModuleInfo.from_name(self.name)
    if info is None:
        info = ModuleInfo(name=self.name).discover()
    if info.state_observed and not info.loaded:
        self._owned = False
        return True

    deadline = time.monotonic() + timeout
    unload_succeeded = False
    while True:
        if unload_succeeded:
            info = info.discover()
        else:
            try:
                unload_succeeded = info.unload()
            except (ModuleInUseError, ModuleUnloadError, OSError):
                unload_succeeded = False

        if info.state_observed and not info.loaded:
            self._owned = False
            return True

        remaining = deadline - time.monotonic()
        if remaining <= 0:
            logger.error(f'Failed to unload owned module {self.name}')
            return False
        time.sleep(min(MODULE_RETRY_INTERVAL, remaining))

ModuleInfo pydantic-model

Bases: StsBaseModel

Kernel module state from /proc/modules and sysfs.

Call discover() after construction to populate from the system.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Kernel module state from /proc/modules and sysfs.\n\nCall ``discover()`` after construction to populate from the system.",
  "properties": {
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "refcount": {
      "default": 0,
      "title": "Refcount",
      "type": "integer"
    },
    "used_by": {
      "items": {
        "type": "string"
      },
      "title": "Used By",
      "type": "array"
    },
    "state": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "State"
    },
    "address": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Address"
    },
    "parameters": {
      "additionalProperties": true,
      "title": "Parameters",
      "type": "object"
    }
  },
  "title": "ModuleInfo",
  "type": "object"
}

Fields:

  • name (str | None)
  • size (int | None)
  • refcount (int)
  • used_by (list[str])
  • state (str | None)
  • address (str | None)
  • parameters (dict[str, Any])
Source code in sts_libs/src/sts/utils/modules.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class ModuleInfo(StsBaseModel):
    """Kernel module state from /proc/modules and sysfs.

    Call ``discover()`` after construction to populate from the system.
    """

    name: str | None = None
    size: int | None = None
    refcount: int = 0
    used_by: list[str] = Field(default_factory=list)
    state: str | None = None
    address: str | None = None
    parameters: dict[str, Any] = Field(default_factory=dict)

    _state_observed: bool = PrivateAttr(default=False)

    def discover(self) -> Self:
        """Load module state from /proc/modules and sysfs; returns self for chaining."""
        self._refresh_state()
        return self

    def _refresh_state(self) -> None:
        """Read module information from /proc/modules and /sys/module.

        Called from discover() and after load/unload operations.
        """
        self._state_observed = False
        self.size = None
        self.refcount = 0
        self.used_by = []
        self.state = None
        self.address = None
        self.parameters = {}

        module_lines = self._read_module_lines()

        if not self.name and module_lines:
            parts = module_lines[0].split()
            if parts:
                self.name = parts[0]

        if not self.name:
            return
        if module_lines is not None:
            self._populate_state(module_lines)
        self._refresh_parameters()

    def _read_module_lines(self) -> list[str] | None:
        """Read the kernel module list without conflating failure with emptiness."""
        try:
            return Path('/proc/modules').read_text().splitlines()
        except OSError:
            logger.warning(f'Failed to get module info for {self.name}', exc_info=True)
            return None

    def _populate_state(self, module_lines: list[str]) -> None:
        """Populate this module from one successfully read kernel snapshot."""
        try:
            for line in module_lines:
                parts = line.split(maxsplit=5)
                if not parts or parts[0] != self.name:
                    continue
                self.size = int(parts[1])
                self.refcount = int(parts[2]) if len(parts) > 2 else 0
                if parts[3] != '-':
                    self.used_by = parts[3].rstrip(',').split(',')
                self.state = parts[4] if len(parts) > 4 else None
                self.address = parts[5] if len(parts) > 5 else None
                break
        except (IndexError, ValueError):
            logger.warning(f'Failed to parse module info for {self.name}', exc_info=True)
        else:
            self._state_observed = True

    def _refresh_parameters(self) -> None:
        """Read currently exposed sysfs parameters for the named module."""
        if not self.name:
            return
        param_path = Path('/sys/module') / self.name / 'parameters'
        if not param_path.is_dir():
            return
        try:
            for param in param_path.iterdir():
                if param.is_file():
                    self.parameters[param.name] = param.read_text().strip()
        except OSError:
            logger.warning(f'Failed to get parameters for {self.name}', exc_info=True)

    @property
    def state_observed(self) -> bool:
        """Whether the last discovery read a valid ``/proc/modules`` snapshot."""
        return self._state_observed

    @property
    def exists(self) -> bool:
        """True if modinfo can find this module."""
        if self.name:
            return run_argv(['modinfo', self.name, '-n']).succeeded
        return False

    @property
    def loaded(self) -> bool:
        """True if the module is currently loaded."""
        return bool(self.state)

    def get_parameter_names(self) -> set[str]:
        """Return module parameter names advertised by ``modinfo``."""
        if not self.name:
            raise ModuleError('Module name required to query advertised parameters')

        result = run_argv(['modinfo', '-p', self.name]).assert_ok(
            f'Failed to query advertised parameters for module {self.name}'
        )
        return {line.split(':', maxsplit=1)[0].strip() for line in result.stdout.splitlines() if line.strip()}

    def load(
        self,
        parameters: str | Mapping[str, str | int | bool] | None = None,
        *,
        raise_on_error: bool = False,
    ) -> bool:
        """Load module via modprobe.

        Args:
            parameters: Module parameters as a shell-like string or mapping. They
                are always passed to ``modprobe`` as separate arguments.
            raise_on_error: Raise :class:`ModuleLoadError` with command details
                instead of returning ``False``.
        """
        if not self.name:
            error = ModuleLoadError('Module name required')
            if raise_on_error:
                raise error
            logger.error(str(error))
            return False

        cmd = ['modprobe', self.name]
        if isinstance(parameters, str):
            cmd.extend(shlex.split(parameters))
        elif parameters:
            cmd.extend(f'{key}={int(value) if isinstance(value, bool) else value}' for key, value in parameters.items())

        result = run_argv(cmd)
        if result.failed:
            detail = result.stderr.strip() or f'command exited with status {result.rc}'
            error = ModuleLoadError(f'Failed to load module {self.name}: {detail}')
            if raise_on_error:
                raise error
            logger.error(str(error))
            return False

        # Update module information
        self._refresh_state()
        return True

    def unload(self) -> bool:
        """Unload module via ``modprobe -r``."""
        if not self.name:
            logger.error('Module name required')
            return False

        result = run_argv(['modprobe', '-r', self.name])
        if result.failed:
            if f'modprobe: FATAL: Module {self.name} is in use.' in result.stderr:
                raise ModuleInUseError(self.name)
            raise ModuleUnloadError(result.stderr)

        # Update module information
        self._refresh_state()
        return True

    def unload_with_dependencies(self) -> bool:
        """Recursively unload this module and all modules that depend on it."""
        if not self.name:
            logger.error('Module name required')
            return False

        if self.used_by:
            logger.debug(f'Removing modules dependent on {self.name}')
            for module in self.used_by:
                if (info := ModuleInfo(name=module).discover()) and not info.unload_with_dependencies():
                    logger.error('Failed to unload dependent modules')
                    return False

        return self.unload()

    @classmethod
    def from_name(cls, name: str) -> ModuleInfo | None:
        """Get module info by name, or None if the module doesn't exist."""
        info = cls(name=name).discover()
        return info if info.exists else None

exists property

True if modinfo can find this module.

loaded property

True if the module is currently loaded.

state_observed property

Whether the last discovery read a valid /proc/modules snapshot.

discover()

Load module state from /proc/modules and sysfs; returns self for chaining.

Source code in sts_libs/src/sts/utils/modules.py 
70
71
72
73
def discover(self) -> Self:
    """Load module state from /proc/modules and sysfs; returns self for chaining."""
    self._refresh_state()
    return self

from_name(name) classmethod

Get module info by name, or None if the module doesn't exist.

Source code in sts_libs/src/sts/utils/modules.py 
240
241
242
243
244
@classmethod
def from_name(cls, name: str) -> ModuleInfo | None:
    """Get module info by name, or None if the module doesn't exist."""
    info = cls(name=name).discover()
    return info if info.exists else None

get_parameter_names()

Return module parameter names advertised by modinfo.

Source code in sts_libs/src/sts/utils/modules.py 
159
160
161
162
163
164
165
166
167
def get_parameter_names(self) -> set[str]:
    """Return module parameter names advertised by ``modinfo``."""
    if not self.name:
        raise ModuleError('Module name required to query advertised parameters')

    result = run_argv(['modinfo', '-p', self.name]).assert_ok(
        f'Failed to query advertised parameters for module {self.name}'
    )
    return {line.split(':', maxsplit=1)[0].strip() for line in result.stdout.splitlines() if line.strip()}

load(parameters=None, *, raise_on_error=False)

Load module via modprobe.

Parameters:

Name Type Description Default
parameters str | Mapping[str, str | int | bool] | None

Module parameters as a shell-like string or mapping. They are always passed to modprobe as separate arguments.

 None
raise_on_error bool

Raise :class:ModuleLoadError with command details instead of returning False.

 False
Source code in sts_libs/src/sts/utils/modules.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def load(
    self,
    parameters: str | Mapping[str, str | int | bool] | None = None,
    *,
    raise_on_error: bool = False,
) -> bool:
    """Load module via modprobe.

    Args:
        parameters: Module parameters as a shell-like string or mapping. They
            are always passed to ``modprobe`` as separate arguments.
        raise_on_error: Raise :class:`ModuleLoadError` with command details
            instead of returning ``False``.
    """
    if not self.name:
        error = ModuleLoadError('Module name required')
        if raise_on_error:
            raise error
        logger.error(str(error))
        return False

    cmd = ['modprobe', self.name]
    if isinstance(parameters, str):
        cmd.extend(shlex.split(parameters))
    elif parameters:
        cmd.extend(f'{key}={int(value) if isinstance(value, bool) else value}' for key, value in parameters.items())

    result = run_argv(cmd)
    if result.failed:
        detail = result.stderr.strip() or f'command exited with status {result.rc}'
        error = ModuleLoadError(f'Failed to load module {self.name}: {detail}')
        if raise_on_error:
            raise error
        logger.error(str(error))
        return False

    # Update module information
    self._refresh_state()
    return True

unload()

Unload module via modprobe -r.

Source code in sts_libs/src/sts/utils/modules.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def unload(self) -> bool:
    """Unload module via ``modprobe -r``."""
    if not self.name:
        logger.error('Module name required')
        return False

    result = run_argv(['modprobe', '-r', self.name])
    if result.failed:
        if f'modprobe: FATAL: Module {self.name} is in use.' in result.stderr:
            raise ModuleInUseError(self.name)
        raise ModuleUnloadError(result.stderr)

    # Update module information
    self._refresh_state()
    return True

unload_with_dependencies()

Recursively unload this module and all modules that depend on it.

Source code in sts_libs/src/sts/utils/modules.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def unload_with_dependencies(self) -> bool:
    """Recursively unload this module and all modules that depend on it."""
    if not self.name:
        logger.error('Module name required')
        return False

    if self.used_by:
        logger.debug(f'Removing modules dependent on {self.name}')
        for module in self.used_by:
            if (info := ModuleInfo(name=module).discover()) and not info.unload_with_dependencies():
                logger.error('Failed to unload dependent modules')
                return False

    return self.unload()

ModuleManager

Higher-level kernel module operations with timeout-based state waiting.

Source code in sts_libs/src/sts/utils/modules.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
class ModuleManager:
    """Higher-level kernel module operations with timeout-based state waiting."""

    def __init__(self) -> None:
        self.modules_path = Path('/proc/modules')
        self.parameters_path = Path('/sys/module')

    def get_all(self) -> list[ModuleInfo]:
        """Get all currently loaded modules."""
        modules: list[ModuleInfo] = []
        try:
            for line in self.modules_path.read_text().splitlines():
                parts = line.split(maxsplit=4)
                info = ModuleInfo(name=parts[0]).discover()
                if info.exists:
                    modules.append(info)
        except (OSError, IndexError):
            logger.exception('Failed to get module list')
            return []

        return modules

    @staticmethod
    def get_parameters(name: str) -> dict[str, str]:
        """Get module parameters from sysfs."""
        if info := ModuleInfo(name=name).discover():
            return info.parameters
        return {}

    def load(
        self,
        name: str,
        parameters: str | Mapping[str, str | int | bool] | None = None,
        timeout: int = DEFAULT_TIMEOUT,
    ) -> bool:
        """Load module and wait for it to appear (no-op if already loaded)."""
        info = ModuleInfo(name=name).discover()
        if info.loaded:
            return True

        success = info.load(parameters)

        return success and self._wait_for_module_state(name, expected_state=True, timeout=timeout)

    def unload(self, name: str, timeout: int = DEFAULT_TIMEOUT) -> bool:
        """Unload module and wait for it to disappear (no-op if not loaded)."""
        info = ModuleInfo(name=name).discover()
        if not info.loaded:
            return True

        success = info.unload()

        return success and self._wait_for_module_state(name, expected_state=False, timeout=timeout)

    def unload_with_dependencies(self, name: str, timeout: int = DEFAULT_TIMEOUT) -> bool:
        """Recursively unload module and its dependents."""
        info = ModuleInfo(name=name).discover()
        if not info.loaded:
            return True

        success = info.unload_with_dependencies()

        return success and self._wait_for_module_state(name, expected_state=False, timeout=timeout)

    def _wait_for_module_state(self, module_name: str, *, expected_state: bool, timeout: int = DEFAULT_TIMEOUT) -> bool:
        """Poll until module reaches *expected_state* or *timeout* expires."""
        start_time = time.monotonic()
        while time.monotonic() - start_time < timeout:
            info = ModuleInfo(name=module_name).discover()
            current_state = info.loaded
            if current_state == expected_state:
                return True
            time.sleep(0.1)
        return False

get_all()

Get all currently loaded modules.

Source code in sts_libs/src/sts/utils/modules.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def get_all(self) -> list[ModuleInfo]:
    """Get all currently loaded modules."""
    modules: list[ModuleInfo] = []
    try:
        for line in self.modules_path.read_text().splitlines():
            parts = line.split(maxsplit=4)
            info = ModuleInfo(name=parts[0]).discover()
            if info.exists:
                modules.append(info)
    except (OSError, IndexError):
        logger.exception('Failed to get module list')
        return []

    return modules

get_parameters(name) staticmethod

Get module parameters from sysfs.

Source code in sts_libs/src/sts/utils/modules.py 
269
270
271
272
273
274
@staticmethod
def get_parameters(name: str) -> dict[str, str]:
    """Get module parameters from sysfs."""
    if info := ModuleInfo(name=name).discover():
        return info.parameters
    return {}

load(name, parameters=None, timeout=DEFAULT_TIMEOUT)

Load module and wait for it to appear (no-op if already loaded).

Source code in sts_libs/src/sts/utils/modules.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def load(
    self,
    name: str,
    parameters: str | Mapping[str, str | int | bool] | None = None,
    timeout: int = DEFAULT_TIMEOUT,
) -> bool:
    """Load module and wait for it to appear (no-op if already loaded)."""
    info = ModuleInfo(name=name).discover()
    if info.loaded:
        return True

    success = info.load(parameters)

    return success and self._wait_for_module_state(name, expected_state=True, timeout=timeout)

unload(name, timeout=DEFAULT_TIMEOUT)

Unload module and wait for it to disappear (no-op if not loaded).

Source code in sts_libs/src/sts/utils/modules.py 
291
292
293
294
295
296
297
298
299
def unload(self, name: str, timeout: int = DEFAULT_TIMEOUT) -> bool:
    """Unload module and wait for it to disappear (no-op if not loaded)."""
    info = ModuleInfo(name=name).discover()
    if not info.loaded:
        return True

    success = info.unload()

    return success and self._wait_for_module_state(name, expected_state=False, timeout=timeout)

unload_with_dependencies(name, timeout=DEFAULT_TIMEOUT)

Recursively unload module and its dependents.

Source code in sts_libs/src/sts/utils/modules.py 
301
302
303
304
305
306
307
308
309
def unload_with_dependencies(self, name: str, timeout: int = DEFAULT_TIMEOUT) -> bool:
    """Recursively unload module and its dependents."""
    info = ModuleInfo(name=name).discover()
    if not info.loaded:
        return True

    success = info.unload_with_dependencies()

    return success and self._wait_for_module_state(name, expected_state=False, timeout=timeout)

parameters_match(actual, expected)

Return whether sysfs parameter readback matches the requested configuration.

Kernel boolean parameters may read back as Y/N while the caller passes True/False or 1/0. Both forms are accepted.

Source code in sts_libs/src/sts/utils/modules.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def parameters_match(actual: Mapping[str, object], expected: Mapping[str, str | int | bool]) -> bool:
    """Return whether sysfs parameter readback matches the requested configuration.

    Kernel boolean parameters may read back as ``Y``/``N`` while the caller
    passes ``True``/``False`` or ``1``/``0``.  Both forms are accepted.
    """
    for key, value in expected.items():
        actual_value = str(actual.get(key))
        if isinstance(value, bool) or (isinstance(value, int) and value in (0, 1)):
            accepted = {'Y', '1'} if value else {'N', '0'}
            if actual_value not in accepted:
                return False
        elif actual_value != str(value):
            return False
    return True

File Operations

sts.utils.files

File/directory operations, mount/umount, mkfs, checksum helpers.

DirAccessError

Bases: DirectoryError

Directory cannot be accessed.

Source code in sts_libs/src/sts/utils/files.py 
42
43
class DirAccessError(DirectoryError):
    """Directory cannot be accessed."""

DirNotFoundError

Bases: DirectoryError

Directory does not exist.

Source code in sts_libs/src/sts/utils/files.py 
34
35
class DirNotFoundError(DirectoryError):
    """Directory does not exist."""

DirTypeError

Bases: DirectoryError

Path exists but is not a directory.

Source code in sts_libs/src/sts/utils/files.py 
38
39
class DirTypeError(DirectoryError):
    """Path exists but is not a directory."""

Directory pydantic-model

Bases: StsBaseModel

Directory wrapper with optional auto-creation.

Attributes:

Name Type Description
create bool

If True, create the directory (with parents) on construction.
Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Directory wrapper with optional auto-creation.\n\nAttributes:\n    create: If True, create the directory (with parents) on construction.",
  "properties": {
    "path": {
      "format": "path",
      "title": "Path",
      "type": "string"
    },
    "create": {
      "default": false,
      "title": "Create",
      "type": "boolean"
    },
    "mode": {
      "default": 493,
      "title": "Mode",
      "type": "integer"
    }
  },
  "title": "Directory",
  "type": "object"
}

Fields:

  • path (Path)
  • create (bool)
  • mode (int)

Validators:

  • _create_if_requested
Source code in sts_libs/src/sts/utils/files.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class Directory(StsBaseModel):
    """Directory wrapper with optional auto-creation.

    Attributes:
        create: If True, create the directory (with parents) on construction.
    """

    path: Path = Field(default_factory=Path.cwd)
    create: bool = False
    mode: int = 0o755

    @model_validator(mode='after')
    def _create_if_requested(self) -> Self:
        """Create directory if requested."""
        if self.create and not self.exists:
            try:
                self.path.mkdir(mode=self.mode, parents=True, exist_ok=True)
            except OSError:
                logger.exception('Failed to create directory')
        return self

    @property
    def exists(self) -> bool:
        """True if path exists and is a directory."""
        return self.path.is_dir()

    def validate_exists(self) -> None:
        """Raise DirNotFoundError / DirTypeError if path is not a valid directory."""
        if not self.path.exists():
            raise DirNotFoundError(f'Directory not found: {self.path}')
        if not self.exists:
            raise DirTypeError(f'Not a directory: {self.path}')

    def iter_files(self, *, recursive: bool = False) -> Iterator[Path]:
        """Yield Path objects for each file in the directory."""
        try:
            if recursive:
                for item in self.path.rglob('*'):
                    if item.is_file():
                        yield item
            else:
                for item in self.path.iterdir():
                    if item.is_file():
                        yield item
        except PermissionError as e:
            logger.exception(f'Permission denied accessing {self.path}')
            raise DirAccessError(f'Permission denied: {self.path}') from e
        except OSError as e:
            logger.exception(f'Error accessing {self.path}')
            raise DirAccessError(f'Error accessing directory: {e}') from e

    @staticmethod
    def should_remove_file_with_pattern(file: Path, pattern: str) -> bool:
        """True if file contents contain *pattern*."""
        try:
            content = file.read_text()
        except (OSError, UnicodeDecodeError):
            logger.exception(f'Error reading {file}')
            return False
        return pattern in content

    @staticmethod
    def should_remove_file_without_pattern(file: Path, pattern: str) -> bool:
        """True if file contents do NOT contain *pattern*."""
        try:
            content = file.read_text()
        except (OSError, UnicodeDecodeError):
            logger.exception(f'Error reading {file}')
            return False
        return pattern not in content

    @staticmethod
    def remove_file(file: Path) -> None:
        """Remove file, logging errors on failure."""
        try:
            file.unlink()
        except OSError:
            logger.exception(f'Error removing {file}')

    def count_files(self) -> int:
        """Count files in directory (excluding subdirectories)."""
        self.validate_exists()
        return sum(1 for _ in self.iter_files())

    def rm_files_containing(self, pattern: str, *, invert: bool = False) -> None:
        """Delete files whose contents match (or don't match) *pattern*.

        Args:
            pattern: Substring to search for in file contents.
            invert: If True, delete files that do NOT contain the pattern.
        """
        self.validate_exists()
        check_func = self.should_remove_file_without_pattern if invert else self.should_remove_file_with_pattern
        for file in self.iter_files():
            if check_func(file, pattern):
                self.remove_file(file)

    def remove_dir(self) -> None:
        """Remove directory and all its contents."""
        self.validate_exists()
        try:
            rmtree(self.path)
        except (OSError, PermissionError):
            logger.exception(f'Error removing {self.path}')

exists property

True if path exists and is a directory.

count_files()

Count files in directory (excluding subdirectories).

Source code in sts_libs/src/sts/utils/files.py 
125
126
127
128
def count_files(self) -> int:
    """Count files in directory (excluding subdirectories)."""
    self.validate_exists()
    return sum(1 for _ in self.iter_files())

iter_files(*, recursive=False)

Yield Path objects for each file in the directory.

Source code in sts_libs/src/sts/utils/files.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def iter_files(self, *, recursive: bool = False) -> Iterator[Path]:
    """Yield Path objects for each file in the directory."""
    try:
        if recursive:
            for item in self.path.rglob('*'):
                if item.is_file():
                    yield item
        else:
            for item in self.path.iterdir():
                if item.is_file():
                    yield item
    except PermissionError as e:
        logger.exception(f'Permission denied accessing {self.path}')
        raise DirAccessError(f'Permission denied: {self.path}') from e
    except OSError as e:
        logger.exception(f'Error accessing {self.path}')
        raise DirAccessError(f'Error accessing directory: {e}') from e

remove_dir()

Remove directory and all its contents.

Source code in sts_libs/src/sts/utils/files.py 
143
144
145
146
147
148
149
def remove_dir(self) -> None:
    """Remove directory and all its contents."""
    self.validate_exists()
    try:
        rmtree(self.path)
    except (OSError, PermissionError):
        logger.exception(f'Error removing {self.path}')

remove_file(file) staticmethod

Remove file, logging errors on failure.

Source code in sts_libs/src/sts/utils/files.py 
117
118
119
120
121
122
123
@staticmethod
def remove_file(file: Path) -> None:
    """Remove file, logging errors on failure."""
    try:
        file.unlink()
    except OSError:
        logger.exception(f'Error removing {file}')

rm_files_containing(pattern, *, invert=False)

Delete files whose contents match (or don't match) pattern.

Parameters:

Name Type Description Default
pattern str

Substring to search for in file contents.

 required
invert bool

If True, delete files that do NOT contain the pattern.

 False
Source code in sts_libs/src/sts/utils/files.py 
130
131
132
133
134
135
136
137
138
139
140
141
def rm_files_containing(self, pattern: str, *, invert: bool = False) -> None:
    """Delete files whose contents match (or don't match) *pattern*.

    Args:
        pattern: Substring to search for in file contents.
        invert: If True, delete files that do NOT contain the pattern.
    """
    self.validate_exists()
    check_func = self.should_remove_file_without_pattern if invert else self.should_remove_file_with_pattern
    for file in self.iter_files():
        if check_func(file, pattern):
            self.remove_file(file)

should_remove_file_with_pattern(file, pattern) staticmethod

True if file contents contain pattern.

Source code in sts_libs/src/sts/utils/files.py 
 97
 98
 99
100
101
102
103
104
105
@staticmethod
def should_remove_file_with_pattern(file: Path, pattern: str) -> bool:
    """True if file contents contain *pattern*."""
    try:
        content = file.read_text()
    except (OSError, UnicodeDecodeError):
        logger.exception(f'Error reading {file}')
        return False
    return pattern in content

should_remove_file_without_pattern(file, pattern) staticmethod

True if file contents do NOT contain pattern.

Source code in sts_libs/src/sts/utils/files.py 
107
108
109
110
111
112
113
114
115
@staticmethod
def should_remove_file_without_pattern(file: Path, pattern: str) -> bool:
    """True if file contents do NOT contain *pattern*."""
    try:
        content = file.read_text()
    except (OSError, UnicodeDecodeError):
        logger.exception(f'Error reading {file}')
        return False
    return pattern not in content

validate_exists()

Raise DirNotFoundError / DirTypeError if path is not a valid directory.

Source code in sts_libs/src/sts/utils/files.py 
72
73
74
75
76
77
def validate_exists(self) -> None:
    """Raise DirNotFoundError / DirTypeError if path is not a valid directory."""
    if not self.path.exists():
        raise DirNotFoundError(f'Directory not found: {self.path}')
    if not self.exists:
        raise DirTypeError(f'Not a directory: {self.path}')

DirectoryError

Bases: STSError

Base class for directory-related errors.

Source code in sts_libs/src/sts/utils/files.py 
30
31
class DirectoryError(STSError):
    """Base class for directory-related errors."""

change_directory(path)

Temporarily change working directory, restoring on exit.

Source code in sts_libs/src/sts/utils/files.py 
152
153
154
155
156
157
158
159
160
@contextmanager
def change_directory(path: Path) -> Generator[None, None, None]:
    """Temporarily change working directory, restoring on exit."""
    original_cwd = Path.cwd()
    try:
        os.chdir(path)
        yield
    finally:
        os.chdir(original_cwd)

checksum(path)

Compute the SHA-256 hex digest of a file, or None on failure.

Source code in sts_libs/src/sts/utils/files.py 
399
400
401
402
403
404
405
406
407
408
409
def checksum(path: str | Path) -> str | None:
    """Compute the SHA-256 hex digest of a file, or None on failure."""
    hasher = hashlib.sha256()
    try:
        with Path(path).open('rb') as f:
            for chunk in iter(lambda: f.read(1024 * 1024), b''):
                hasher.update(chunk)
    except OSError:
        logger.exception(f'Failed to compute checksum for {path}')
        return None
    return hasher.hexdigest()

count_files(directory=None)

Count files in directory (defaults to cwd).

Source code in sts_libs/src/sts/utils/files.py 
163
164
165
166
def count_files(directory: str | Path | None = None) -> int:
    """Count files in directory (defaults to cwd)."""
    path = Path(directory) if directory else Path.cwd()
    return Directory(path=path).count_files()

fallocate(path, *args, **kwargs)

Preallocate or deallocate space for a file.

Source code in sts_libs/src/sts/utils/files.py 
311
312
313
314
315
316
317
318
319
320
def fallocate(path: str | Path, *args: str, **kwargs: str) -> bool:
    """Preallocate or deallocate space for a file."""
    cmd = ['fallocate']
    if args:
        cmd.extend(arg for arg in args if arg)
    if kwargs:
        cmd.extend(f'--{k.replace("_", "-")}={v}' for k, v in kwargs.items() if v)
    cmd.append(str(path))
    result = run_argv(cmd)
    return result.succeeded

get_free_space(path=None)

Get free space in bytes for the filesystem containing path.

Source code in sts_libs/src/sts/utils/files.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def get_free_space(path: str | Path | None = None) -> int | None:
    """Get free space in bytes for the filesystem containing *path*."""
    path_str = str(path) if path else '.'
    result = run_argv(['df', '-B', '1', path_str])
    if result.failed:
        logger.error('Failed to get free space')
        return None

    # Parse output like:
    # Filesystem     1B-blocks       Used   Available Use% Mounted on
    # /dev/sda1    1073741824   10485760  1063256064   1% /mnt
    if match := re.search(r'\S+\s+\d+\s+\d+\s+(\d+)', result.stdout):
        return int(match.group(1))

    return None

is_mounted(device=None, mountpoint=None)

Check whether a device is mounted or a path is an exact mountpoint.

Source code in sts_libs/src/sts/utils/files.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def is_mounted(device: str | Path | None = None, mountpoint: str | Path | None = None) -> bool:
    """Check whether a device is mounted or a path is an exact mountpoint."""
    if not device and not mountpoint:
        return False

    cmd = ['findmnt', '--noheadings']
    if device:
        cmd.extend(['--source', str(device)])
    if mountpoint:
        cmd.extend(['--mountpoint', str(mountpoint)])

    result = run_argv(cmd)
    if result.succeeded:
        return True
    if mountpoint:
        if result.rc == 1 and not result.stderr and not result.timed_out:
            return False
        result.assert_ok(f'Failed to check whether {mountpoint} is a mountpoint')
    return False

mkfs(device=None, fs_type=None, *args, force=False)

Create filesystem on device.

Parameters:

Name Type Description Default
device str | Path | None

Block device path.

 None
fs_type str | None

Filesystem type (e.g. 'ext4', 'xfs').

 None
force bool

Add the filesystem-specific force flag (-F for ext*, -f for xfs/btrfs).

 False
Source code in sts_libs/src/sts/utils/files.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def mkfs(device: str | Path | None = None, fs_type: str | None = None, *args: str, force: bool = False) -> bool:
    """Create filesystem on device.

    Args:
        device: Block device path.
        fs_type: Filesystem type (e.g. 'ext4', 'xfs').
        force: Add the filesystem-specific force flag (``-F`` for ext*, ``-f`` for xfs/btrfs).
    """
    force_options: dict[str, str] = {
        'xfs': '-f',
        'btrfs': '-f',
    }

    if not device or not fs_type:
        logger.error('Device and filesystem type required')
        return False

    device_str = str(device)
    cmd = [f'mkfs.{fs_type}']

    if force:
        cmd.append(force_options.get(fs_type, '-F'))

    if args:
        cmd.extend(arg for arg in args if arg)
    cmd.append(device_str)

    result = run_argv(cmd)
    if result.failed:
        logger.error(f'Failed to create {fs_type} filesystem on {device_str}: {result.stderr}')
        return False
    return True

mount(device=None, mountpoint=None, fs_type=None, options=None)

Mount device at mountpoint (creates mountpoint directory if needed).

Source code in sts_libs/src/sts/utils/files.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def mount(
    device: str | Path | None = None,
    mountpoint: str | Path | None = None,
    fs_type: str | None = None,
    options: str | None = None,
) -> bool:
    """Mount device at mountpoint (creates mountpoint directory if needed)."""
    cmd = ['mount']
    if fs_type:
        cmd.extend(['-t', fs_type])
    if options:
        cmd.extend(['-o', options])
    if device:
        cmd.append(str(device))
    if mountpoint:
        mountpoint_path = Path(mountpoint)
        Directory(path=mountpoint_path, create=True)
        cmd.append(str(mountpoint_path))

    result = run_argv(cmd)
    if result.failed:
        logger.error(f'Failed to mount device: {result.stderr}')
        return False
    return True

mounted(device, mountpoint, *, fs_type=None, options=None)

Mount a device for a block and always attempt to unmount it afterwards.

Source code in sts_libs/src/sts/utils/files.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
@contextmanager
def mounted(
    device: str | Path,
    mountpoint: str | Path,
    *,
    fs_type: str | None = None,
    options: str | None = None,
) -> Generator[Path, None, None]:
    """Mount a device for a block and always attempt to unmount it afterwards."""
    mountpoint_path = Path(mountpoint)
    if not mount(device, mountpoint_path, fs_type=fs_type, options=options):
        msg = f'Failed to mount {device} at {mountpoint_path}'
        raise DirectoryError(msg)
    try:
        yield mountpoint_path
    except BaseException as body_error:
        try:
            unmount(mountpoint_path).assert_ok(f'Failed to unmount {mountpoint_path}')
        except STSError as cleanup_error:
            raise BaseExceptionGroup('Mount body and cleanup failed', [body_error, cleanup_error]) from None
        raise
    else:
        unmount(mountpoint_path).assert_ok(f'Failed to unmount {mountpoint_path}')

rm_files_containing(directory=None, pattern='', *, invert=False)

Delete files whose contents match (or don't match) pattern.

Source code in sts_libs/src/sts/utils/files.py 
169
170
171
172
def rm_files_containing(directory: str | Path | None = None, pattern: str = '', *, invert: bool = False) -> None:
    """Delete files whose contents match (or don't match) *pattern*."""
    path = Path(directory) if directory else Path.cwd()
    Directory(path=path).rm_files_containing(pattern, invert=invert)

unmount(mountpoint, /, *, force=False, missing_ok=True)

Unmount an exact mountpoint.

Parameters:

Name Type Description Default
mountpoint str | Path

Mounted path to unmount.

 required
force bool

Pass -f to umount.

 False
missing_ok bool

Treat an already-unmounted mountpoint as success.

 True

Returns:

Type Description
CommandResult

The real command result, or a successful no-op result when allowed.
Source code in sts_libs/src/sts/utils/files.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def unmount(
    mountpoint: str | Path,
    /,
    *,
    force: bool = False,
    missing_ok: bool = True,
) -> CommandResult:
    """Unmount an exact mountpoint.

    Args:
        mountpoint: Mounted path to unmount.
        force: Pass ``-f`` to ``umount``.
        missing_ok: Treat an already-unmounted mountpoint as success.

    Returns:
        The real command result, or a successful no-op result when allowed.

    """
    mountpoint_str = str(mountpoint)
    cmd = ['umount']
    if force:
        cmd.append('-f')
    cmd.append(mountpoint_str)

    if missing_ok and not is_mounted(mountpoint=mountpoint_str):
        return _already_unmounted_result(cmd, mountpoint_str)

    result = run_argv(cmd)
    if result.failed and missing_ok and not is_mounted(mountpoint=mountpoint_str):
        return _already_unmounted_result(cmd, mountpoint_str)
    return result

verify_checksum(path, expected)

Verify a file's SHA-256 checksum matches expected.

Source code in sts_libs/src/sts/utils/files.py 
412
413
414
415
416
417
418
419
420
421
def verify_checksum(path: str | Path, expected: str) -> bool:
    """Verify a file's SHA-256 checksum matches *expected*."""
    actual = checksum(path)
    if actual is None:
        return False
    if actual != expected:
        logger.error(f'Checksum mismatch for {path}: expected {expected}, got {actual}')
        return False
    logger.debug(f'{path}: checksum verified')
    return True

write_data(target, source, *, sync=True, **kwargs)

Write data using dd.

Parameters:

Name Type Description Default
target str | Path

Target file/device path.

 required
source str | Path

Source path (e.g. '/dev/urandom', '/dev/zero').

 required
sync bool

Run sync after writing.

 True
**kwargs str | int

Passed as key=value dd options (e.g. bs='1M', count=10).

 {}
Source code in sts_libs/src/sts/utils/files.py 
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
def write_data(
    target: str | Path,
    source: str | Path,
    *,
    sync: bool = True,
    **kwargs: str | int,
) -> bool:
    """Write data using dd.

    Args:
        target: Target file/device path.
        source: Source path (e.g. '/dev/urandom', '/dev/zero').
        sync: Run ``sync`` after writing.
        **kwargs: Passed as ``key=value`` dd options (e.g. ``bs='1M'``, ``count=10``).
    """
    # Build dd command with defaults and kwargs
    cmd_parts = ['dd']

    cmd_parts.extend((f'if={source!s}', f'of={target!s}'))

    # Add any remaining kwargs as dd options
    for key, value in kwargs.items():
        # Handle special cases where key might be a Python keyword
        dd_key = key.rstrip('_')  # Remove trailing underscore if present
        cmd_parts.append(f'{dd_key}={value}')

    result = run_argv(cmd_parts)

    if result.failed:
        logger.error(f'dd command failed: {result.stderr}')
        logger.error(f'Command: {cmd_parts}')
        return False

    if sync:
        sync_result = run_argv(['sync'])
        if sync_result.failed:
            logger.warning('Sync operation failed but file write succeeded')

    return True

write_zeroes(target, *, sync=True, **kwargs)

Write zeroes using dd (convenience wrapper around write_data).

Source code in sts_libs/src/sts/utils/files.py 
364
365
366
367
368
369
370
371
def write_zeroes(
    target: str | Path,
    *,
    sync: bool = True,
    **kwargs: str | int,
) -> bool:
    """Write zeroes using dd (convenience wrapper around ``write_data``)."""
    return write_data(target, '/dev/zero', sync=sync, **kwargs)

Configuration Files

sts.utils.config

Key=value configuration file management with comment preservation.

Config

Base configuration class for managing configuration files.

Source code in sts_libs/src/sts/utils/config.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class Config:
    """Base configuration class for managing configuration files."""

    def __init__(self, config_path: Path) -> None:
        """Load configuration from *config_path* (if it exists)."""
        self.config_path = config_path
        self.parameters: dict[str, str] = {}
        self._removed_keys: set[str] = set()
        if self.config_path.exists():
            self._load_config()

    def _load_config(self) -> None:
        """Parse ``Key = Value`` lines, skipping comments."""
        try:
            lines = self.config_path.read_text().splitlines()
            for line in lines:
                stripped_line = line.strip()
                if stripped_line and not stripped_line.startswith('#') and '=' in stripped_line:
                    key, value = map(str.strip, stripped_line.split('=', 1))
                    self.parameters[key] = value
        except OSError:
            logger.exception('Failed to load config')

    def set_parameters(self, parameters: dict[str, str]) -> None:
        """Update multiple parameters at once."""
        self.parameters.update(parameters)

    def remove_parameters(self, keys: Iterable[str]) -> None:
        """Mark configuration parameters for removal.

        Removed keys are dropped from `parameters` immediately and their
        lines are omitted from the file the next time `save()` runs.

        Args:
            keys: Parameter names to remove
        """
        for key in keys:
            self.parameters.pop(key, None)
            self._removed_keys.add(key)

    def get_parameter(self, name: str) -> str | None:
        """Get parameter value, or None if not set."""
        return self.parameters.get(name)

    def save(self) -> bool:
        """Write parameters back to the file, preserving comments and ordering."""
        try:
            lines = self.config_path.read_text().splitlines() if self.config_path.exists() else []
            updated_lines: list[str] = []

            for line in lines:
                if line.strip().startswith('#') or '=' not in line:
                    updated_lines.append(line)
                    continue

                key = line.split('=', 1)[0].strip()
                if key in self._removed_keys:
                    continue
                if key in self.parameters:
                    updated_lines.append(f'{key} = {self.parameters[key]}')
                    del self.parameters[key]
                else:
                    updated_lines.append(line)

            for key, value in self.parameters.items():
                updated_lines.append(f'{key} = {value}')

            self.config_path.write_text('\n'.join(updated_lines) + '\n')
        except OSError:
            logger.exception('Failed to save config')
            return False
        self._removed_keys.clear()
        return True

__init__(config_path)

Load configuration from config_path (if it exists).

Source code in sts_libs/src/sts/utils/config.py 
21
22
23
24
25
26
27
def __init__(self, config_path: Path) -> None:
    """Load configuration from *config_path* (if it exists)."""
    self.config_path = config_path
    self.parameters: dict[str, str] = {}
    self._removed_keys: set[str] = set()
    if self.config_path.exists():
        self._load_config()

get_parameter(name)

Get parameter value, or None if not set.

Source code in sts_libs/src/sts/utils/config.py 
58
59
60
def get_parameter(self, name: str) -> str | None:
    """Get parameter value, or None if not set."""
    return self.parameters.get(name)

remove_parameters(keys)

Mark configuration parameters for removal.

Removed keys are dropped from parameters immediately and their lines are omitted from the file the next time save() runs.

Parameters:

Name Type Description Default
keys Iterable[str]

Parameter names to remove

 required
Source code in sts_libs/src/sts/utils/config.py 
45
46
47
48
49
50
51
52
53
54
55
56
def remove_parameters(self, keys: Iterable[str]) -> None:
    """Mark configuration parameters for removal.

    Removed keys are dropped from `parameters` immediately and their
    lines are omitted from the file the next time `save()` runs.

    Args:
        keys: Parameter names to remove
    """
    for key in keys:
        self.parameters.pop(key, None)
        self._removed_keys.add(key)

save()

Write parameters back to the file, preserving comments and ordering.

Source code in sts_libs/src/sts/utils/config.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def save(self) -> bool:
    """Write parameters back to the file, preserving comments and ordering."""
    try:
        lines = self.config_path.read_text().splitlines() if self.config_path.exists() else []
        updated_lines: list[str] = []

        for line in lines:
            if line.strip().startswith('#') or '=' not in line:
                updated_lines.append(line)
                continue

            key = line.split('=', 1)[0].strip()
            if key in self._removed_keys:
                continue
            if key in self.parameters:
                updated_lines.append(f'{key} = {self.parameters[key]}')
                del self.parameters[key]
            else:
                updated_lines.append(line)

        for key, value in self.parameters.items():
            updated_lines.append(f'{key} = {value}')

        self.config_path.write_text('\n'.join(updated_lines) + '\n')
    except OSError:
        logger.exception('Failed to save config')
        return False
    self._removed_keys.clear()
    return True

set_parameters(parameters)

Update multiple parameters at once.

Source code in sts_libs/src/sts/utils/config.py 
41
42
43
def set_parameters(self, parameters: dict[str, str]) -> None:
    """Update multiple parameters at once."""
    self.parameters.update(parameters)

Fstab Management

sts.utils.fstab

/etc/fstab entry management with backup/restore and a temporary_entry context manager.

The temporary_entry context manager is useful for tests that need a mount point registered in fstab (e.g., snapm mount manager tests).

Fstab

Manage /etc/fstab entries.

Provides static methods for querying mount information and managing fstab entries with backup/restore support.

Source code in sts_libs/src/sts/utils/fstab.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class Fstab:
    """Manage /etc/fstab entries.

    Provides static methods for querying mount information and
    managing fstab entries with backup/restore support.
    """

    @staticmethod
    def get_device(mount_point: Path) -> str | None:
        """Get the source device for a mounted filesystem via findmnt."""
        result = run(f'findmnt -n -o SOURCE {mount_point}')
        if result.succeeded and result.stdout.strip():
            return result.stdout.strip()
        return None

    @staticmethod
    def get_fstype(mount_point: Path) -> str | None:
        """Get the filesystem type for a mounted filesystem via findmnt."""
        result = run(f'findmnt -n -o FSTYPE {mount_point}')
        if result.succeeded and result.stdout.strip():
            return result.stdout.strip()
        return None

    @staticmethod
    def has_entry(mount_point: Path) -> bool:
        """Check whether a mount point already has an fstab entry."""
        mp_str = str(mount_point)
        try:
            for line in FSTAB_PATH.read_text().splitlines():
                stripped = line.strip()
                if stripped and not stripped.startswith('#'):
                    fields = stripped.split()
                    if len(fields) >= 2 and fields[1] == mp_str:
                        return True
        except OSError:
            logger.exception('Failed to read fstab')
        return False

    @staticmethod
    def add_entry(
        device: Path,
        mount_point: Path,
        fs_type: str,
        options: str = 'defaults',
        dump: int = 0,
        fsck_pass: int = 0,
    ) -> bool:
        """Append an entry to /etc/fstab."""
        entry = f'{device} {mount_point} {fs_type} {options} {dump} {fsck_pass}\n'
        try:
            if _ends_without_newline(FSTAB_PATH):
                entry = f'\n{entry}'
            with FSTAB_PATH.open('a') as f:
                f.write(entry)
        except OSError:
            logger.exception('Failed to add fstab entry')
            return False
        else:
            logger.debug(f'Added fstab entry: {entry.strip()}')
            return True

    @staticmethod
    def remove_entry(mount_point: Path) -> bool:
        """Remove all fstab entries for a given mount point."""
        mp_str = str(mount_point)
        try:
            lines = FSTAB_PATH.read_text().splitlines(keepends=True)
            filtered: list[str] = []
            for line in lines:
                fields = line.split()
                if line.strip().startswith('#') or not fields or (len(fields) >= 2 and fields[1] != mp_str):
                    filtered.append(line)
            FSTAB_PATH.write_text(''.join(filtered))
        except OSError:
            logger.exception('Failed to remove fstab entry')
            return False
        else:
            logger.debug(f'Removed fstab entries for {mount_point}')
            return True

    @staticmethod
    def backup() -> Path:
        """Create a backup of /etc/fstab and return its path."""
        backup_path = FSTAB_PATH.with_suffix(FSTAB_BACKUP_SUFFIX)
        backup_path.write_bytes(FSTAB_PATH.read_bytes())
        logger.debug(f'Backed up {FSTAB_PATH} to {backup_path}')
        return backup_path

    @staticmethod
    def restore(backup_path: Path | None = None) -> bool:
        """Restore /etc/fstab from a backup (defaults to the standard backup location)."""
        if backup_path is None:
            backup_path = FSTAB_PATH.with_suffix(FSTAB_BACKUP_SUFFIX)

        if not backup_path.exists():
            logger.error(f'Backup file not found: {backup_path}')
            return False

        try:
            FSTAB_PATH.write_bytes(backup_path.read_bytes())
            backup_path.unlink(missing_ok=True)
        except OSError:
            logger.exception('Failed to restore fstab')
            return False
        else:
            logger.debug(f'Restored {FSTAB_PATH} from {backup_path}')
            return True

    @staticmethod
    def _remove_remote_entries() -> None:
        """Remove entries with remote filesystem types from /etc/fstab."""
        lines = FSTAB_PATH.read_text().splitlines(keepends=True)
        filtered: list[str] = []
        removed = 0
        for line in lines:
            stripped = line.strip()
            if stripped.startswith('#') or not stripped:
                filtered.append(line)
                continue
            fields = stripped.split()
            if len(fields) >= 3 and fields[2] in REMOTE_FS_TYPES:
                logger.debug(f'Removing remote fstab entry: {stripped}')
                removed += 1
                continue
            filtered.append(line)
        if removed:
            FSTAB_PATH.write_text(''.join(filtered))
            logger.info(f'Removed {removed} remote filesystem entries from fstab')

    @staticmethod
    @contextmanager
    def without_remote_entries() -> Generator[None, None, None]:
        """Temporarily remove remote filesystem entries from /etc/fstab.

        Backs up fstab, removes entries with network filesystem types
        (NFS, CIFS, etc.), and restores the original on exit.

        Uses a separate backup file from ``backup()`` so this context
        manager can nest safely with ``temporary_entry()``.
        """
        backup_path = FSTAB_PATH.with_suffix('.sts-remote-backup')
        backup_path.write_bytes(FSTAB_PATH.read_bytes())
        try:
            Fstab._remove_remote_entries()
            yield
        finally:
            FSTAB_PATH.write_bytes(backup_path.read_bytes())
            backup_path.unlink(missing_ok=True)
            logger.debug('Restored original fstab (remote entries)')

    @staticmethod
    @contextmanager
    def temporary_entry(
        mount_point: Path,
        options: str = 'defaults',
    ) -> Generator[None, None, None]:
        """Temporarily add a mount point to /etc/fstab with automatic rollback.

        Discovers the device and filesystem type from the live mount via
        findmnt, backs up fstab, adds the entry, and restores the original
        fstab on exit (even if an exception occurs).

        Args:
            mount_point: Path of the currently mounted filesystem to add
            options: Mount options for the fstab entry (default: 'defaults')

        Example:
            ```python
            with Fstab.temporary_entry(Path('/mnt/test')):
                # /mnt/test is now in /etc/fstab
                snapset.mount()  # snapm can now find the entry
            # Original fstab is restored
            ```
        """
        device = Fstab.get_device(mount_point)
        fs_type = Fstab.get_fstype(mount_point)
        assert device, f'Could not determine device for {mount_point}'
        assert fs_type, f'Could not determine fstype for {mount_point}'

        backup_path = Fstab.backup()
        try:
            Fstab.add_entry(Path(device), mount_point, fs_type, options=options)
            yield
        finally:
            Fstab.restore(backup_path)

add_entry(device, mount_point, fs_type, options='defaults', dump=0, fsck_pass=0) staticmethod

Append an entry to /etc/fstab.

Source code in sts_libs/src/sts/utils/fstab.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@staticmethod
def add_entry(
    device: Path,
    mount_point: Path,
    fs_type: str,
    options: str = 'defaults',
    dump: int = 0,
    fsck_pass: int = 0,
) -> bool:
    """Append an entry to /etc/fstab."""
    entry = f'{device} {mount_point} {fs_type} {options} {dump} {fsck_pass}\n'
    try:
        if _ends_without_newline(FSTAB_PATH):
            entry = f'\n{entry}'
        with FSTAB_PATH.open('a') as f:
            f.write(entry)
    except OSError:
        logger.exception('Failed to add fstab entry')
        return False
    else:
        logger.debug(f'Added fstab entry: {entry.strip()}')
        return True

backup() staticmethod

Create a backup of /etc/fstab and return its path.

Source code in sts_libs/src/sts/utils/fstab.py 
123
124
125
126
127
128
129
@staticmethod
def backup() -> Path:
    """Create a backup of /etc/fstab and return its path."""
    backup_path = FSTAB_PATH.with_suffix(FSTAB_BACKUP_SUFFIX)
    backup_path.write_bytes(FSTAB_PATH.read_bytes())
    logger.debug(f'Backed up {FSTAB_PATH} to {backup_path}')
    return backup_path

get_device(mount_point) staticmethod

Get the source device for a mounted filesystem via findmnt.

Source code in sts_libs/src/sts/utils/fstab.py 
50
51
52
53
54
55
56
@staticmethod
def get_device(mount_point: Path) -> str | None:
    """Get the source device for a mounted filesystem via findmnt."""
    result = run(f'findmnt -n -o SOURCE {mount_point}')
    if result.succeeded and result.stdout.strip():
        return result.stdout.strip()
    return None

get_fstype(mount_point) staticmethod

Get the filesystem type for a mounted filesystem via findmnt.

Source code in sts_libs/src/sts/utils/fstab.py 
58
59
60
61
62
63
64
@staticmethod
def get_fstype(mount_point: Path) -> str | None:
    """Get the filesystem type for a mounted filesystem via findmnt."""
    result = run(f'findmnt -n -o FSTYPE {mount_point}')
    if result.succeeded and result.stdout.strip():
        return result.stdout.strip()
    return None

has_entry(mount_point) staticmethod

Check whether a mount point already has an fstab entry.

Source code in sts_libs/src/sts/utils/fstab.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@staticmethod
def has_entry(mount_point: Path) -> bool:
    """Check whether a mount point already has an fstab entry."""
    mp_str = str(mount_point)
    try:
        for line in FSTAB_PATH.read_text().splitlines():
            stripped = line.strip()
            if stripped and not stripped.startswith('#'):
                fields = stripped.split()
                if len(fields) >= 2 and fields[1] == mp_str:
                    return True
    except OSError:
        logger.exception('Failed to read fstab')
    return False

remove_entry(mount_point) staticmethod

Remove all fstab entries for a given mount point.

Source code in sts_libs/src/sts/utils/fstab.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@staticmethod
def remove_entry(mount_point: Path) -> bool:
    """Remove all fstab entries for a given mount point."""
    mp_str = str(mount_point)
    try:
        lines = FSTAB_PATH.read_text().splitlines(keepends=True)
        filtered: list[str] = []
        for line in lines:
            fields = line.split()
            if line.strip().startswith('#') or not fields or (len(fields) >= 2 and fields[1] != mp_str):
                filtered.append(line)
        FSTAB_PATH.write_text(''.join(filtered))
    except OSError:
        logger.exception('Failed to remove fstab entry')
        return False
    else:
        logger.debug(f'Removed fstab entries for {mount_point}')
        return True

restore(backup_path=None) staticmethod

Restore /etc/fstab from a backup (defaults to the standard backup location).

Source code in sts_libs/src/sts/utils/fstab.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@staticmethod
def restore(backup_path: Path | None = None) -> bool:
    """Restore /etc/fstab from a backup (defaults to the standard backup location)."""
    if backup_path is None:
        backup_path = FSTAB_PATH.with_suffix(FSTAB_BACKUP_SUFFIX)

    if not backup_path.exists():
        logger.error(f'Backup file not found: {backup_path}')
        return False

    try:
        FSTAB_PATH.write_bytes(backup_path.read_bytes())
        backup_path.unlink(missing_ok=True)
    except OSError:
        logger.exception('Failed to restore fstab')
        return False
    else:
        logger.debug(f'Restored {FSTAB_PATH} from {backup_path}')
        return True

temporary_entry(mount_point, options='defaults') staticmethod

Temporarily add a mount point to /etc/fstab with automatic rollback.

Discovers the device and filesystem type from the live mount via findmnt, backs up fstab, adds the entry, and restores the original fstab on exit (even if an exception occurs).

Parameters:

Name Type Description Default
mount_point Path

Path of the currently mounted filesystem to add

 required
options str

Mount options for the fstab entry (default: 'defaults')

 'defaults'
Example 
with Fstab.temporary_entry(Path('/mnt/test')):
    # /mnt/test is now in /etc/fstab
    snapset.mount()  # snapm can now find the entry
# Original fstab is restored
Source code in sts_libs/src/sts/utils/fstab.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
@staticmethod
@contextmanager
def temporary_entry(
    mount_point: Path,
    options: str = 'defaults',
) -> Generator[None, None, None]:
    """Temporarily add a mount point to /etc/fstab with automatic rollback.

    Discovers the device and filesystem type from the live mount via
    findmnt, backs up fstab, adds the entry, and restores the original
    fstab on exit (even if an exception occurs).

    Args:
        mount_point: Path of the currently mounted filesystem to add
        options: Mount options for the fstab entry (default: 'defaults')

    Example:
        ```python
        with Fstab.temporary_entry(Path('/mnt/test')):
            # /mnt/test is now in /etc/fstab
            snapset.mount()  # snapm can now find the entry
        # Original fstab is restored
        ```
    """
    device = Fstab.get_device(mount_point)
    fs_type = Fstab.get_fstype(mount_point)
    assert device, f'Could not determine device for {mount_point}'
    assert fs_type, f'Could not determine fstype for {mount_point}'

    backup_path = Fstab.backup()
    try:
        Fstab.add_entry(Path(device), mount_point, fs_type, options=options)
        yield
    finally:
        Fstab.restore(backup_path)

without_remote_entries() staticmethod

Temporarily remove remote filesystem entries from /etc/fstab.

Backs up fstab, removes entries with network filesystem types (NFS, CIFS, etc.), and restores the original on exit.

Uses a separate backup file from backup() so this context manager can nest safely with temporary_entry().

Source code in sts_libs/src/sts/utils/fstab.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@staticmethod
@contextmanager
def without_remote_entries() -> Generator[None, None, None]:
    """Temporarily remove remote filesystem entries from /etc/fstab.

    Backs up fstab, removes entries with network filesystem types
    (NFS, CIFS, etc.), and restores the original on exit.

    Uses a separate backup file from ``backup()`` so this context
    manager can nest safely with ``temporary_entry()``.
    """
    backup_path = FSTAB_PATH.with_suffix('.sts-remote-backup')
    backup_path.write_bytes(FSTAB_PATH.read_bytes())
    try:
        Fstab._remove_remote_entries()
        yield
    finally:
        FSTAB_PATH.write_bytes(backup_path.read_bytes())
        backup_path.unlink(missing_ok=True)
        logger.debug('Restored original fstab (remote entries)')

NFS Exports

sts.utils.nfs

/etc/exports management -- add/remove NFS exports and refresh exportfs.

NfsExports

Manage /etc/exports entries.

Provides static methods for adding/removing NFS export entries and refreshing the kernel's export table to match.

Source code in sts_libs/src/sts/utils/nfs.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class NfsExports:
    """Manage /etc/exports entries.

    Provides static methods for adding/removing NFS export entries and
    refreshing the kernel's export table to match.
    """

    @staticmethod
    def add(path: str | Path, options: str = DEFAULT_EXPORT_OPTIONS, clients: str = '*') -> bool:
        """Add an export entry to /etc/exports and refresh exportfs."""
        export_line = f'{path} {clients}({options})\n'
        try:
            if _ends_without_newline(EXPORTS_PATH):
                export_line = f'\n{export_line}'
            with EXPORTS_PATH.open('a') as f:
                f.write(export_line)
        except OSError:
            logger.exception(f'Failed to add NFS export for {path}')
            return False

        logger.debug(f'Added NFS export: {export_line.strip()}')
        return NfsExports.refresh()

    @staticmethod
    def remove(path: str | Path) -> bool:
        """Remove all export entries for *path* from /etc/exports and refresh exportfs."""
        path_str = str(path)
        if EXPORTS_PATH.exists():
            try:
                lines = EXPORTS_PATH.read_text().splitlines(keepends=True)
                lines = [line for line in lines if path_str not in line]
                EXPORTS_PATH.write_text(''.join(lines))
            except OSError:
                logger.exception(f'Failed to remove NFS export for {path}')
                return False

        logger.debug(f'Removed NFS export: {path}')
        return NfsExports.refresh()

    @staticmethod
    def refresh() -> bool:
        """Run ``exportfs -ra`` to re-export all directories."""
        result = run('exportfs -ra')
        if result.failed:
            logger.error(f'exportfs -ra failed: {result.stderr}')
            return False
        return True

add(path, options=DEFAULT_EXPORT_OPTIONS, clients='*') staticmethod

Add an export entry to /etc/exports and refresh exportfs.

Source code in sts_libs/src/sts/utils/nfs.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@staticmethod
def add(path: str | Path, options: str = DEFAULT_EXPORT_OPTIONS, clients: str = '*') -> bool:
    """Add an export entry to /etc/exports and refresh exportfs."""
    export_line = f'{path} {clients}({options})\n'
    try:
        if _ends_without_newline(EXPORTS_PATH):
            export_line = f'\n{export_line}'
        with EXPORTS_PATH.open('a') as f:
            f.write(export_line)
    except OSError:
        logger.exception(f'Failed to add NFS export for {path}')
        return False

    logger.debug(f'Added NFS export: {export_line.strip()}')
    return NfsExports.refresh()

refresh() staticmethod

Run exportfs -ra to re-export all directories.

Source code in sts_libs/src/sts/utils/nfs.py 
72
73
74
75
76
77
78
79
@staticmethod
def refresh() -> bool:
    """Run ``exportfs -ra`` to re-export all directories."""
    result = run('exportfs -ra')
    if result.failed:
        logger.error(f'exportfs -ra failed: {result.stderr}')
        return False
    return True

remove(path) staticmethod

Remove all export entries for path from /etc/exports and refresh exportfs.

Source code in sts_libs/src/sts/utils/nfs.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@staticmethod
def remove(path: str | Path) -> bool:
    """Remove all export entries for *path* from /etc/exports and refresh exportfs."""
    path_str = str(path)
    if EXPORTS_PATH.exists():
        try:
            lines = EXPORTS_PATH.read_text().splitlines(keepends=True)
            lines = [line for line in lines if path_str not in line]
            EXPORTS_PATH.write_text(''.join(lines))
        except OSError:
            logger.exception(f'Failed to remove NFS export for {path}')
            return False

    logger.debug(f'Removed NFS export: {path}')
    return NfsExports.refresh()

Process Management

sts.utils.processes

Process discovery and control via /proc.

ProcessInfo pydantic-model

Bases: StsBaseModel

Process state from /proc. Call discover() after construction.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Process state from /proc. Call ``discover()`` after construction.",
  "properties": {
    "pid": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Pid"
    },
    "name": {
      "default": "",
      "title": "Name",
      "type": "string"
    },
    "status": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Status"
    },
    "cmdline": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Cmdline"
    }
  },
  "title": "ProcessInfo",
  "type": "object"
}

Fields:

  • pid (int | None)
  • name (str)
  • status (str | None)
  • cmdline (str | None)
Source code in sts_libs/src/sts/utils/processes.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class ProcessInfo(StsBaseModel):
    """Process state from /proc. Call ``discover()`` after construction."""

    # Optional parameters
    pid: int | None = None
    name: str = ''
    status: str | None = None
    cmdline: str | None = None

    def _find_first_available_pid(self) -> None:
        """Find the PID of the first available process."""
        try:
            proc_entries = [entry for entry in Path('/proc').iterdir() if entry.is_dir()]
        except OSError:
            return

        # Find first numeric directory
        for entry in proc_entries:
            if entry.name.isdigit():
                self.pid = int(entry.name)
                break

    def _find_pid_for_name(self) -> None:
        """Find PID for a process matching self.name."""
        try:
            proc_entries = [entry for entry in Path('/proc').iterdir() if entry.is_dir()]
        except OSError:
            return

        # Find process with matching name
        for entry in proc_entries:
            if not entry.name.isdigit():
                continue
            comm_path = entry / 'comm'
            try:
                if comm_path.exists() and comm_path.read_text().strip() == self.name:
                    self.pid = int(entry.name)
                    break
            except (OSError, ValueError):
                continue

    def _populate_info_from_pid(self) -> None:
        """Populate name, status, and cmdline from self.pid."""
        if not self.pid:
            return

        proc_path = Path('/proc') / str(self.pid)

        try:
            # Get process name if not provided
            if not self.name:
                comm_path = proc_path / 'comm'
                if comm_path.exists():
                    self.name = comm_path.read_text().strip()

            # Get process status
            status_path = proc_path / 'status'
            if status_path.exists():
                status_content = status_path.read_text().splitlines()
                for line in status_content:
                    if line.startswith('State:'):
                        self.status = line.split(':')[1].strip()
                        break

            # Get process command line
            cmdline_path = proc_path / 'cmdline'
            if cmdline_path.exists():
                self.cmdline = cmdline_path.read_text().strip('\x00').replace('\x00', ' ')

        except OSError:
            logger.exception(f'Failed to get process info for PID {self.pid}')

    def discover(self) -> Self:
        """Populate from /proc; returns self for chaining."""
        # If no parameters provided, get first available process
        if not any([self.pid, self.name]):
            self._find_first_available_pid()
        # If name provided but no pid, find matching process
        elif self.name and not self.pid:
            self._find_pid_for_name()

        # If pid found or provided, get other information
        self._populate_info_from_pid()
        return self

    @property
    def exists(self) -> bool:
        """True if the /proc/<pid> directory exists."""
        return bool(self.pid and Path('/proc', str(self.pid)).exists())

    @property
    def running(self) -> bool:
        """True if the process is alive (via ``kill(0)``)."""
        if not self.pid:
            return False

        try:
            os.kill(self.pid, 0)
        except ProcessLookupError:
            return False
        except PermissionError:
            # Process exists but we don't have permission to send signals
            return True
        return True

    def kill(self, timeout: float = 1.0) -> bool:
        """Send SIGTERM, then SIGKILL after *timeout* seconds if still alive."""
        if not self.pid:
            return False

        if not self.running:
            return True

        try:
            # Try SIGTERM first
            os.kill(self.pid, signal.SIGTERM)
            time.sleep(timeout)

            # If still running, use SIGKILL
            if self.running:
                os.kill(self.pid, signal.SIGKILL)
                time.sleep(timeout)

                # Check if process is still running
                if self.running:
                    logger.error(f'Failed to kill process {self.pid}')
                    return False

        except ProcessLookupError:
            # Process already terminated
            pass
        except PermissionError:
            logger.exception(f'Permission denied killing process {self.pid}')
            return False

        return True

    @classmethod
    def from_pid(cls, pid: int) -> Self:
        """Create and discover by PID. Check ``.exists`` to confirm it was found."""
        return cls(pid=pid).discover()

    @classmethod
    def from_name(cls, name: str) -> Self:
        """Create and discover by name. Check ``.exists`` to confirm it was found."""
        return cls(name=name).discover()

exists property

True if the /proc/ directory exists.

running property

True if the process is alive (via kill(0)).

discover()

Populate from /proc; returns self for chaining.

Source code in sts_libs/src/sts/utils/processes.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def discover(self) -> Self:
    """Populate from /proc; returns self for chaining."""
    # If no parameters provided, get first available process
    if not any([self.pid, self.name]):
        self._find_first_available_pid()
    # If name provided but no pid, find matching process
    elif self.name and not self.pid:
        self._find_pid_for_name()

    # If pid found or provided, get other information
    self._populate_info_from_pid()
    return self

from_name(name) classmethod

Create and discover by name. Check .exists to confirm it was found.

Source code in sts_libs/src/sts/utils/processes.py 
166
167
168
169
@classmethod
def from_name(cls, name: str) -> Self:
    """Create and discover by name. Check ``.exists`` to confirm it was found."""
    return cls(name=name).discover()

from_pid(pid) classmethod

Create and discover by PID. Check .exists to confirm it was found.

Source code in sts_libs/src/sts/utils/processes.py 
161
162
163
164
@classmethod
def from_pid(cls, pid: int) -> Self:
    """Create and discover by PID. Check ``.exists`` to confirm it was found."""
    return cls(pid=pid).discover()

kill(timeout=1.0)

Send SIGTERM, then SIGKILL after timeout seconds if still alive.

Source code in sts_libs/src/sts/utils/processes.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def kill(self, timeout: float = 1.0) -> bool:
    """Send SIGTERM, then SIGKILL after *timeout* seconds if still alive."""
    if not self.pid:
        return False

    if not self.running:
        return True

    try:
        # Try SIGTERM first
        os.kill(self.pid, signal.SIGTERM)
        time.sleep(timeout)

        # If still running, use SIGKILL
        if self.running:
            os.kill(self.pid, signal.SIGKILL)
            time.sleep(timeout)

            # Check if process is still running
            if self.running:
                logger.error(f'Failed to kill process {self.pid}')
                return False

    except ProcessLookupError:
        # Process already terminated
        pass
    except PermissionError:
        logger.exception(f'Permission denied killing process {self.pid}')
        return False

    return True

ProcessManager

Bulk process operations (list, find by name, kill all).

Source code in sts_libs/src/sts/utils/processes.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
class ProcessManager:
    """Bulk process operations (list, find by name, kill all)."""

    def __init__(self) -> None:
        self.proc_path = Path('/proc')

    def get_all(self) -> list[ProcessInfo]:
        """Get all running processes."""
        processes: list[ProcessInfo] = []

        try:
            # Get list of process directories
            proc_entries = [entry for entry in self.proc_path.iterdir() if entry.is_dir()]

            # Filter numeric directories and create processes
            for entry in proc_entries:
                if entry.name.isdigit():
                    pid = int(entry.name)
                    info = ProcessInfo(pid=pid).discover()
                    if info.exists:
                        processes.append(info)

        except OSError:
            logger.exception('Failed to get process list')
            return []

        return processes

    def get_by_name(self, name: str) -> list[ProcessInfo]:
        """Get all processes matching *name*."""
        return [p for p in self.get_all() if p.name == name]

    def kill_all(self, name: str, timeout: float = 1.0) -> bool:
        """Kill all processes matching *name* via ``killall``."""
        result = run(f'killall {name}')
        if result.failed:
            return False

        # Wait for processes to finish
        time.sleep(timeout)

        # Check if any processes still running
        return not bool(self.get_by_name(name))

get_all()

Get all running processes.

Source code in sts_libs/src/sts/utils/processes.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def get_all(self) -> list[ProcessInfo]:
    """Get all running processes."""
    processes: list[ProcessInfo] = []

    try:
        # Get list of process directories
        proc_entries = [entry for entry in self.proc_path.iterdir() if entry.is_dir()]

        # Filter numeric directories and create processes
        for entry in proc_entries:
            if entry.name.isdigit():
                pid = int(entry.name)
                info = ProcessInfo(pid=pid).discover()
                if info.exists:
                    processes.append(info)

    except OSError:
        logger.exception('Failed to get process list')
        return []

    return processes

get_by_name(name)

Get all processes matching name.

Source code in sts_libs/src/sts/utils/processes.py 
200
201
202
def get_by_name(self, name: str) -> list[ProcessInfo]:
    """Get all processes matching *name*."""
    return [p for p in self.get_all() if p.name == name]

kill_all(name, timeout=1.0)

Kill all processes matching name via killall.

Source code in sts_libs/src/sts/utils/processes.py 
204
205
206
207
208
209
210
211
212
213
214
def kill_all(self, name: str, timeout: float = 1.0) -> bool:
    """Kill all processes matching *name* via ``killall``."""
    result = run(f'killall {name}')
    if result.failed:
        return False

    # Wait for processes to finish
    time.sleep(timeout)

    # Check if any processes still running
    return not bool(self.get_by_name(name))

Size Handling

sts.utils.size

Size parsing and conversion between human-readable strings and bytes.

Size pydantic-model

Bases: ReportModel

Immutable size value with unit, supporting parsing and conversion.

Large byte values are auto-scaled to the most readable unit on construction.

Show JSON schema: 
{
  "$defs": {
    "Unit": {
      "description": "Size units.",
      "enum": [
        "B",
        "KiB",
        "MiB",
        "GiB",
        "TiB",
        "PiB",
        "EiB",
        "ZiB",
        "YiB"
      ],
      "title": "Unit",
      "type": "string"
    }
  },
  "description": "Immutable size value with unit, supporting parsing and conversion.\n\nLarge byte values are auto-scaled to the most readable unit on construction.",
  "properties": {
    "value": {
      "default": 0.0,
      "title": "Value",
      "type": "number"
    },
    "unit": {
      "$ref": "#/$defs/Unit",
      "default": "B"
    }
  },
  "title": "Size",
  "type": "object"
}

Fields:

  • value (float)
  • unit (Unit)

Validators:

  • _normalize_unit
Source code in sts_libs/src/sts/utils/size.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class Size(ReportModel):
    """Immutable size value with unit, supporting parsing and conversion.

    Large byte values are auto-scaled to the most readable unit on construction.
    """

    # Optional parameters with defaults
    value: float = 0.0
    unit: Unit = Unit.B

    # Regular expression for parsing human-readable sizes
    SIZE_PATTERN: ClassVar[re.Pattern[str]] = re.compile(
        r'^([\-0-9\.]+)(Ki|Mi|Gi|Ti|Pi|Ei|Zi|Yi)?B$',
    )

    # Unit multipliers (powers of 1024)
    MULTIPLIERS: ClassVar[dict[Unit, int]] = {
        Unit.B: BYTE,
        Unit.KIB: KIB,
        Unit.MIB: MIB,
        Unit.GIB: GIB,
        Unit.TIB: TIB,
        Unit.PIB: PIB,
        Unit.EIB: EIB,
        Unit.ZIB: ZIB,
        Unit.YIB: YIB,
    }

    @model_validator(mode='before')
    @classmethod
    def _normalize_unit(cls, data: Any) -> Any:
        """Auto-scale large byte values to the most readable unit."""
        d: dict[str, Any] | str | int | None = data
        if isinstance(d, dict):
            value: Any = d.get('value', 0.0)
            unit: Any = d.get('unit', Unit.B)
            # Normalize Unit from string if needed
            if isinstance(unit, str):
                unit = Unit(unit)
            if unit == Unit.B and value >= KIB:
                for next_unit in list(Unit)[1:]:
                    if value < KIB:
                        break
                    value /= 1024
                    unit = next_unit
                data = {**d, 'value': value, 'unit': unit}
        return data

    @classmethod
    def from_string(cls, size: str) -> Size | None:
        """Parse a human-readable size string (e.g. '1KiB'), or None if invalid."""
        if not size:
            return None

        # Handle pure numbers as bytes
        if size.isdigit():
            return cls(value=float(size), unit=Unit.B)

        # Parse size with unit
        match = cls.SIZE_PATTERN.match(size)
        if not match:
            logger.error(f'Invalid size format: {size}')
            return None

        try:
            value = float(match.group(1))
            unit_str = match.group(2)
            unit = Unit.B if not unit_str else Unit(f'{unit_str}B')
            return cls(value=value, unit=unit)
        except (ValueError, KeyError):
            logger.exception('Failed to parse size')
            return None

    def to_bytes(self) -> int:
        """Convert to bytes."""
        return int(self.value * self.MULTIPLIERS[self.unit])

    @classmethod
    def from_bytes(cls, bytes_: int) -> Size:
        """Create a Size from a byte count, auto-scaling to the best unit."""
        if bytes_ < KIB:
            return cls(value=float(bytes_), unit=Unit.B)

        value = float(bytes_)
        unit = Unit.B  # Default unit
        for next_unit in list(Unit)[1:]:  # Skip B
            if value < KIB:
                break
            value /= 1024
            unit = next_unit

        return cls(value=value, unit=unit)

    def __str__(self) -> str:
        # Remove decimal part if whole number
        if self.value.is_integer():
            return f'{int(self.value)}{self.unit}'
        return f'{self.value:.1f}{self.unit}'

from_bytes(bytes_) classmethod

Create a Size from a byte count, auto-scaling to the best unit.

Source code in sts_libs/src/sts/utils/size.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@classmethod
def from_bytes(cls, bytes_: int) -> Size:
    """Create a Size from a byte count, auto-scaling to the best unit."""
    if bytes_ < KIB:
        return cls(value=float(bytes_), unit=Unit.B)

    value = float(bytes_)
    unit = Unit.B  # Default unit
    for next_unit in list(Unit)[1:]:  # Skip B
        if value < KIB:
            break
        value /= 1024
        unit = next_unit

    return cls(value=value, unit=unit)

from_string(size) classmethod

Parse a human-readable size string (e.g. '1KiB'), or None if invalid.

Source code in sts_libs/src/sts/utils/size.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@classmethod
def from_string(cls, size: str) -> Size | None:
    """Parse a human-readable size string (e.g. '1KiB'), or None if invalid."""
    if not size:
        return None

    # Handle pure numbers as bytes
    if size.isdigit():
        return cls(value=float(size), unit=Unit.B)

    # Parse size with unit
    match = cls.SIZE_PATTERN.match(size)
    if not match:
        logger.error(f'Invalid size format: {size}')
        return None

    try:
        value = float(match.group(1))
        unit_str = match.group(2)
        unit = Unit.B if not unit_str else Unit(f'{unit_str}B')
        return cls(value=value, unit=unit)
    except (ValueError, KeyError):
        logger.exception('Failed to parse size')
        return None

to_bytes()

Convert to bytes.

Source code in sts_libs/src/sts/utils/size.py 
124
125
126
def to_bytes(self) -> int:
    """Convert to bytes."""
    return int(self.value * self.MULTIPLIERS[self.unit])

Unit

Bases: StrEnum

Size units.

Source code in sts_libs/src/sts/utils/size.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class Unit(StrEnum):
    """Size units."""

    def __str__(self) -> str:
        return self.value

    def __format__(self, format_spec: str) -> str:
        return self.value.__format__(format_spec)

    B = 'B'
    KIB = 'KiB'
    MIB = 'MiB'
    GIB = 'GiB'
    TIB = 'TiB'
    PIB = 'PiB'
    EIB = 'EiB'
    ZIB = 'ZiB'
    YIB = 'YiB'

size_bytes_2_size_human(bytes_)

Convert bytes to human-readable size string, or None if invalid.

Source code in sts_libs/src/sts/utils/size.py 
163
164
165
166
167
168
169
170
171
172
173
def size_bytes_2_size_human(bytes_: int | str | None) -> str | None:
    """Convert bytes to human-readable size string, or None if invalid."""
    if not bytes_:
        return None

    try:
        size = Size.from_bytes(int(bytes_))
        return str(size)
    except (ValueError, TypeError):
        logger.exception('Invalid bytes value')
        return None

size_human_2_size_bytes(size)

Convert human-readable size string to bytes, or None if invalid.

Source code in sts_libs/src/sts/utils/size.py 
156
157
158
159
160
def size_human_2_size_bytes(size: str) -> int | None:
    """Convert human-readable size string to bytes, or None if invalid."""
    if size_obj := Size.from_string(size):
        return size_obj.to_bytes()
    return None

size_human_check(size)

Check if a human-readable size string is valid.

Source code in sts_libs/src/sts/utils/size.py 
151
152
153
def size_human_check(size: str) -> bool:
    """Check if a human-readable size string is valid."""
    return Size.from_string(size) is not None

String Utilities

sts.utils.string_extras

String manipulation utilities.

none_to_empty(value)

Convert None to empty string, otherwise pass through.

Source code in sts_libs/src/sts/utils/string_extras.py 
18
19
20
def none_to_empty(value: str | None) -> str:
    """Convert None to empty string, otherwise pass through."""
    return '' if value is None else value

rand_string(length=8, chars=None)

Generate a random string (default: 8 chars from lowercase + digits).

Source code in sts_libs/src/sts/utils/string_extras.py 
12
13
14
15
def rand_string(length: int = 8, chars: str | None = None) -> str:
    """Generate a random string (default: 8 chars from lowercase + digits)."""
    chars = chars or string.ascii_lowercase + string.digits
    return ''.join(random.choices(chars, k=length))

Test Marks

sts.marks

Pytest skip/xfail helpers for distribution, version, and package gating.

Three tiers:

  • Imperative helpers (requires_*) — call inside fixtures to pytest.skip().
  • Marker factory (requires_package) — returns pytest.MarkDecorator for decorators and pytestmark.
  • Boolean predicates (is_*) — for pytest.mark.skipif / xfail conditions and pytestmark.

is_distro(name)

True if running on the given distribution.

Source code in sts_libs/src/sts/marks.py 
87
88
89
def is_distro(name: str) -> bool:
    """True if running on the given distribution."""
    return SystemInfo.get_current().distribution == name

is_rhel_below(version)

True if running on RHEL below the given version.

Source code in sts_libs/src/sts/marks.py 
81
82
83
84
def is_rhel_below(version: int | VersionInfo) -> bool:
    """True if running on RHEL below the given version."""
    info = SystemInfo.get_current()
    return info.distribution == 'rhel' and info.version < _to_version(version)

requires_package(name, min_version, reason='')

Skipif marker: skip if installed package is below minimum version.

Parameters:

Name Type Description Default
name str

Package name (as known to rpm).

 required
min_version VersionInfo

Minimum required version.

 required
reason str

Custom skip reason.

 ''
Source code in sts_libs/src/sts/marks.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def requires_package(name: str, min_version: VersionInfo, reason: str = '') -> pytest.MarkDecorator:
    """Skipif marker: skip if installed package is below minimum version.

    Args:
        name: Package name (as known to rpm).
        min_version: Minimum required version.
        reason: Custom skip reason.
    """
    installed = get_package_version(name) or VersionInfo(0, 0, 0)
    return pytest.mark.skipif(
        installed < min_version,
        reason=reason or f'requires {name} >= {min_version}',
    )

requires_rhel(min_version, reason='')

Skip the current test if on RHEL below the given version.

Parameters:

Name Type Description Default
min_version int | VersionInfo

Minimum RHEL version. int is shorthand for major-only.

 required
reason str

Custom skip reason.

 ''
Source code in sts_libs/src/sts/marks.py 
31
32
33
34
35
36
37
38
39
40
41
def requires_rhel(min_version: int | VersionInfo, reason: str = '') -> None:
    """Skip the current test if on RHEL below the given version.

    Args:
        min_version: Minimum RHEL version. ``int`` is shorthand for major-only.
        reason: Custom skip reason.
    """
    info = SystemInfo.get_current()
    version = _to_version(min_version)
    if info.distribution == 'rhel' and info.version < version:
        pytest.skip(reason or f'requires RHEL {version}+')

requires_tool(tool, min_version, reason='')

Skip the current test if a tool's version is below minimum.

Expects tool to have a version attribute comparable to VersionInfo.

Parameters:

Name Type Description Default
tool object

Object with a version attribute.

 required
min_version VersionInfo

Minimum required version.

 required
reason str

Custom skip reason.

 ''
Source code in sts_libs/src/sts/marks.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def requires_tool(tool: object, min_version: VersionInfo, reason: str = '') -> None:
    """Skip the current test if a tool's version is below minimum.

    Expects ``tool`` to have a ``version`` attribute comparable to VersionInfo.

    Args:
        tool: Object with a ``version`` attribute.
        min_version: Minimum required version.
        reason: Custom skip reason.
    """
    tool_version = getattr(tool, 'version', None)
    if tool_version is not None and tool_version < min_version:
        tool_name = type(tool).__name__
        pytest.skip(reason or f'requires {tool_name} >= {min_version}')

Version Handling

sts.utils.version

Semantic version parsing and comparison.

VersionInfo

Bases: NamedTuple

Parsed version with major.minor.micro.patch-build components.

Supports comparison via tuple ordering and parsing via from_string().

Source code in sts_libs/src/sts/utils/version.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class VersionInfo(NamedTuple):
    """Parsed version with major.minor.micro.patch-build components.

    Supports comparison via tuple ordering and parsing via ``from_string()``.
    """

    major: int
    minor: int = 0
    micro: int = 0
    patch: int = 0
    build: int = 0

    def __str__(self) -> str:
        """Format as dotted version string, trimming trailing zeros.

        Examples:
            >>> str(VersionInfo(9))
            '9'
            >>> str(VersionInfo(9, 4))
            '9.4'
            >>> str(VersionInfo(0, 5, 1))
            '0.5.1'
        """
        parts = [self.major, self.minor, self.micro, self.patch]
        last_nonzero = 0
        for i, part in enumerate(parts):
            if part != 0:
                last_nonzero = i
        version = '.'.join(str(p) for p in parts[: last_nonzero + 1])
        if self.build:
            version += f'-{self.build}'
        return version

    @classmethod
    def from_string(cls, version_string: str) -> VersionInfo:
        """Parse ``major[.minor[.micro[.patch]]][-build]`` into a VersionInfo.

        Raises:
            ValueError: If the string is empty, malformed, or contains negative components.
        """
        if not version_string or not version_string.strip():
            raise ValueError('Version string cannot be empty')

        version_string = version_string.strip()

        # Check for supported format
        if not cls._is_valid_format(version_string):
            raise ValueError(
                f'Unsupported version format: "{version_string}". Expected format: "major[.minor[.micro]][-build]"'
            )

        # Split on hyphen first to separate build number
        hyphen_parts = version_string.split('-', 1)
        main_version = hyphen_parts[0]
        build = 0
        if len(hyphen_parts) > 1:
            build_match = re.match(r'\d+', hyphen_parts[1])
            if build_match:
                build = int(build_match.group())

        parts = main_version.split('.')

        # Ensure we have at least a major version
        if len(parts) < 1 or not parts[0]:
            raise ValueError('Version string must contain at least a major version')

        major = int(parts[0])
        minor = int(parts[1]) if len(parts) > 1 and parts[1] else 0
        micro = int(parts[2]) if len(parts) > 2 and parts[2] else 0
        patch = int(parts[3]) if len(parts) > 3 and parts[3] else 0

        # Ensure all components are non-negative
        if any(component < 0 for component in (major, minor, micro, patch, build)):
            raise ValueError('Version components must be non-negative integers')

        return cls(major=major, minor=minor, micro=micro, patch=patch, build=build)

    @staticmethod
    def _is_valid_format(version_string: str) -> bool:
        """Validate that version_string matches the supported format."""
        # Split on hyphen to check build part
        hyphen_parts = version_string.split('-')

        # Can have at most one hyphen
        if len(hyphen_parts) > 2:
            return False

        main_version = hyphen_parts[0]
        build_part = hyphen_parts[1] if len(hyphen_parts) > 1 else None

        # Validate main version part (major.minor.micro.patch)
        pattern = r'^[0-9]+(\.[0-9]+){0,3}$'
        if not re.match(pattern, main_version):
            return False

        # Check main version components
        parts = main_version.split('.')
        if len(parts) > 4:  # Maximum 4 components
            return False

        # Validate build part if present - must start with digits,
        # optionally followed by a dist tag (e.g., "2.el10", "6.el9")
        return build_part is None or bool(re.match(r'^\d+', build_part))

__str__()

Format as dotted version string, trimming trailing zeros.

Examples:

>>> str(VersionInfo(9))
'9'
>>> str(VersionInfo(9, 4))
'9.4'
>>> str(VersionInfo(0, 5, 1))
'0.5.1'
Source code in sts_libs/src/sts/utils/version.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __str__(self) -> str:
    """Format as dotted version string, trimming trailing zeros.

    Examples:
        >>> str(VersionInfo(9))
        '9'
        >>> str(VersionInfo(9, 4))
        '9.4'
        >>> str(VersionInfo(0, 5, 1))
        '0.5.1'
    """
    parts = [self.major, self.minor, self.micro, self.patch]
    last_nonzero = 0
    for i, part in enumerate(parts):
        if part != 0:
            last_nonzero = i
    version = '.'.join(str(p) for p in parts[: last_nonzero + 1])
    if self.build:
        version += f'-{self.build}'
    return version

from_string(version_string) classmethod

Parse major[.minor[.micro[.patch]]][-build] into a VersionInfo.

Raises:

Type Description
ValueError

If the string is empty, malformed, or contains negative components.
Source code in sts_libs/src/sts/utils/version.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@classmethod
def from_string(cls, version_string: str) -> VersionInfo:
    """Parse ``major[.minor[.micro[.patch]]][-build]`` into a VersionInfo.

    Raises:
        ValueError: If the string is empty, malformed, or contains negative components.
    """
    if not version_string or not version_string.strip():
        raise ValueError('Version string cannot be empty')

    version_string = version_string.strip()

    # Check for supported format
    if not cls._is_valid_format(version_string):
        raise ValueError(
            f'Unsupported version format: "{version_string}". Expected format: "major[.minor[.micro]][-build]"'
        )

    # Split on hyphen first to separate build number
    hyphen_parts = version_string.split('-', 1)
    main_version = hyphen_parts[0]
    build = 0
    if len(hyphen_parts) > 1:
        build_match = re.match(r'\d+', hyphen_parts[1])
        if build_match:
            build = int(build_match.group())

    parts = main_version.split('.')

    # Ensure we have at least a major version
    if len(parts) < 1 or not parts[0]:
        raise ValueError('Version string must contain at least a major version')

    major = int(parts[0])
    minor = int(parts[1]) if len(parts) > 1 and parts[1] else 0
    micro = int(parts[2]) if len(parts) > 2 and parts[2] else 0
    patch = int(parts[3]) if len(parts) > 3 and parts[3] else 0

    # Ensure all components are non-negative
    if any(component < 0 for component in (major, minor, micro, patch, build)):
        raise ValueError('Version components must be non-negative integers')

    return cls(major=major, minor=minor, micro=micro, patch=patch, build=build)

Multihost Synchronization

sts.utils.sync

Multihost test synchronization via rstrnt-sync and TMT topology.

Three barrier helpers cover the typical multihost lifecycle:

  • barrier_start: setter announces readiness, no linger
  • barrier: mid-test sync point, no linger
  • barrier_finish: setter lingers to keep daemon alive for the peer

barrier(label, *, setter_role, timeout=3600)

Mid-test barrier -- both hosts continue, no linger.

Use between test phases when both hosts need to synchronize but neither is about to exit.

Parameters:

Name Type Description Default
label str

Barrier / state name (e.g. 'DATAWRITTEN')

 required
setter_role str

Role of the host that announces the state

 required
timeout int

Seconds the blocker waits before giving up

 3600
Example 
barrier('DATAWRITTEN', setter_role='server')
Source code in sts_libs/src/sts/utils/sync.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
def barrier(
    label: str,
    *,
    setter_role: str,
    timeout: int = 3600,
) -> None:
    """Mid-test barrier -- both hosts continue, no linger.

    Use between test phases when both hosts need to synchronize but
    neither is about to exit.

    Args:
        label: Barrier / state name (e.g. ``'DATAWRITTEN'``)
        setter_role: Role of the host that announces the state
        timeout: Seconds the blocker waits before giving up

    Example:
        ```python
        barrier('DATAWRITTEN', setter_role='server')
        ```
    """
    _do_barrier(label, setter_role=setter_role, timeout=timeout)

barrier_finish(label, *, setter_role, linger=DEFAULT_LINGER, timeout=3600)

Final barrier -- setter lingers so the peer can detect the state.

Use when the setter is about to exit (e.g. the client signalling completion). After announcing the state the setter sleeps for linger seconds (default 120) to keep the rstrnt-sync daemon alive for the peer to poll.

Parameters:

Name Type Description Default
label str

Barrier / state name (e.g. 'CLIENTDONE')

 required
setter_role str

Role of the host that announces the state

 required
linger int

Seconds to sleep after announcing

 DEFAULT_LINGER
timeout int

Seconds the blocker waits before giving up

 3600
Example 
barrier_finish('CLIENTDONE', setter_role='client')
Source code in sts_libs/src/sts/utils/sync.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def barrier_finish(
    label: str,
    *,
    setter_role: str,
    linger: int = DEFAULT_LINGER,
    timeout: int = 3600,
) -> None:
    """Final barrier -- setter lingers so the peer can detect the state.

    Use when the setter is about to exit (e.g. the client signalling
    completion).  After announcing the state the setter sleeps for
    *linger* seconds (default 120) to keep the rstrnt-sync daemon
    alive for the peer to poll.

    Args:
        label: Barrier / state name (e.g. ``'CLIENTDONE'``)
        setter_role: Role of the host that announces the state
        linger: Seconds to sleep after announcing
        timeout: Seconds the blocker waits before giving up

    Example:
        ```python
        barrier_finish('CLIENTDONE', setter_role='client')
        ```
    """
    _do_barrier(label, setter_role=setter_role, linger=linger, timeout=timeout)

barrier_start(label, *, setter_role, timeout=3600)

Initial barrier -- setter keeps running, no linger.

Use at the beginning of the test when the setter (typically the server) announces readiness and continues with its own work.

Parameters:

Name Type Description Default
label str

Barrier / state name (e.g. 'SERVERREADY')

 required
setter_role str

Role of the host that announces the state

 required
timeout int

Seconds the blocker waits before giving up

 3600
Example 
barrier_start('SERVERREADY', setter_role='server')
Source code in sts_libs/src/sts/utils/sync.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
def barrier_start(
    label: str,
    *,
    setter_role: str,
    timeout: int = 3600,
) -> None:
    """Initial barrier -- setter keeps running, no linger.

    Use at the beginning of the test when the setter (typically the
    server) announces readiness and continues with its own work.

    Args:
        label: Barrier / state name (e.g. ``'SERVERREADY'``)
        setter_role: Role of the host that announces the state
        timeout: Seconds the blocker waits before giving up

    Example:
        ```python
        barrier_start('SERVERREADY', setter_role='server')
        ```
    """
    _do_barrier(label, setter_role=setter_role, timeout=timeout)

check_sync_daemon()

Log rstrnt-sync process status and port 6776 listener (diagnostic only).

Source code in sts_libs/src/sts/utils/sync.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
def check_sync_daemon() -> None:
    """Log rstrnt-sync process status and port 6776 listener (diagnostic only)."""
    ps = run('ps aux | grep rstrnt-sync | grep -v grep')
    if ps.succeeded and ps.stdout.strip():
        logger.debug(f'rstrnt-sync process:\n{ps.stdout.strip()}')
    else:
        logger.warning('rstrnt-sync process NOT found')

    ss = run(f'ss -tlnp | grep {SYNC_PORT}')
    if ss.succeeded and ss.stdout.strip():
        logger.debug(f'Port {SYNC_PORT} listener:\n{ss.stdout.strip()}')
    else:
        logger.warning(f'Nothing listening on port {SYNC_PORT}')

get_my_fqdn()

Get this host's FQDN (from topology or socket.getfqdn()).

Source code in sts_libs/src/sts/utils/sync.py 
118
119
120
121
122
123
124
125
126
def get_my_fqdn() -> str:
    """Get this host's FQDN (from topology or ``socket.getfqdn()``)."""
    topo = load_topology()
    guest: dict[str, Any] | str | None = topo.get('guest')
    if isinstance(guest, dict):
        hostname: str | None = guest.get('hostname')
        if hostname:
            return str(hostname)
    return socket.getfqdn()

get_peer_fqdn(role)

Get a peer's FQDN by role from the topology (falls back to <ROLE>_IP env).

Raises:

Type Description
RuntimeError

If no hostname can be determined for the role.
Source code in sts_libs/src/sts/utils/sync.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def get_peer_fqdn(role: str) -> str:
    """Get a peer's FQDN by role from the topology (falls back to ``<ROLE>_IP`` env).

    Raises:
        RuntimeError: If no hostname can be determined for the role.
    """
    topo = load_topology()
    guests = topo.get('guests', {})
    for guest_name, props in guests.items():
        p: dict[str, Any] | str | None = props
        if isinstance(p, dict) and p.get('role') == role:
            hostname: str | None = p.get('hostname')
            if hostname:
                logger.debug(f'Peer FQDN for role "{role}" -> guest "{guest_name}" -> {hostname}')
                return hostname

    fallback = getenv(f'{role.upper()}_IP')
    if fallback:
        logger.debug(f'Peer FQDN for role "{role}" via {role.upper()}_IP -> {fallback}')
        return fallback

    msg = (
        f'Hostname for role "{role}" not found. '
        f'TMT_TOPOLOGY_BASH={getenv("TMT_TOPOLOGY_BASH", "<unset>")}, '
        f'{role.upper()}_IP is also unset.'
    )
    raise RuntimeError(msg)

get_role()

Get this host's role (from topology, TMT_ROLE env, or default 'server').

Source code in sts_libs/src/sts/utils/sync.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def get_role() -> str:
    """Get this host's role (from topology, TMT_ROLE env, or default 'server')."""
    topo = load_topology()
    guest: dict[str, Any] | str | None = topo.get('guest')
    if isinstance(guest, dict):
        role: str | None = guest.get('role')
        if role:
            logger.debug(f'Role from topology guest.role: {role}')
            return str(role)

    role = getenv('TMT_ROLE', 'server')
    logger.debug(f'Role from TMT_ROLE env var (fallback): {role}')
    return role

init_sync()

Parse topology and set Beaker-compatible env vars. Call once before using barriers.

Source code in sts_libs/src/sts/utils/sync.py 
214
215
216
217
def init_sync() -> None:
    """Parse topology and set Beaker-compatible env vars. Call once before using barriers."""
    topo = load_topology()
    _setup_sync_env(get_my_fqdn(), topo.get('guests', {}))

is_client()

True if this host's role is 'client'.

Source code in sts_libs/src/sts/utils/sync.py 
113
114
115
def is_client() -> bool:
    """True if this host's role is 'client'."""
    return get_role() == 'client'

is_server()

True if this host's role is 'server'.

Source code in sts_libs/src/sts/utils/sync.py 
108
109
110
def is_server() -> bool:
    """True if this host's role is 'server'."""
    return get_role() == 'server'

load_topology()

Parse the TMT_TOPOLOGY_BASH file into a dict (empty if unavailable).

Source code in sts_libs/src/sts/utils/sync.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
def load_topology() -> dict[str, Any]:
    """Parse the TMT_TOPOLOGY_BASH file into a dict (empty if unavailable)."""
    topo_path = getenv('TMT_TOPOLOGY_BASH')
    if not topo_path:
        return {}
    path = Path(topo_path)
    if not path.is_file():
        logger.warning(f'TMT_TOPOLOGY_BASH points to missing file: {topo_path}')
        return {}

    text = path.read_text()
    logger.debug(f'Loaded tmt topology from {topo_path}')
    return _parse_topology_bash(text)

resolve_ipv4(hostname)

Resolve hostname to IPv4 (returns hostname as-is on failure).

Useful for protocols that do not support IPv6 (e.g. iSCSI).

Source code in sts_libs/src/sts/utils/sync.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def resolve_ipv4(hostname: str) -> str:
    """Resolve hostname to IPv4 (returns hostname as-is on failure).

    Useful for protocols that do not support IPv6 (e.g. iSCSI).
    """
    try:
        results = socket.getaddrinfo(hostname, None, socket.AF_INET)
        if results:
            ipv4 = str(results[0][4][0])
            logger.debug(f'Resolved {hostname} -> {ipv4} (IPv4)')
            return ipv4
    except socket.gaierror:
        logger.warning(f'IPv4 resolution failed for {hostname}, using as-is')
    return hostname

sync_block(state, host, *, timeout=3600, retry=60)

Block until a remote host has announced a state.

Calls rstrnt-sync-block -s <state> <host> and polls every retry seconds until the state is found or timeout is exceeded.

Parameters:

Name Type Description Default
state str

State label to wait for

 required
host str

FQDN of the host that will announce the state

 required
timeout int

Seconds before giving up

 3600
retry int

Seconds between polls

 60
Example 
sync_block('SERVERREADY', 'server.example.com')
Source code in sts_libs/src/sts/utils/sync.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
def sync_block(
    state: str,
    host: str,
    *,
    timeout: int = 3600,
    retry: int = 60,
) -> None:
    """Block until a remote host has announced a state.

    Calls ``rstrnt-sync-block -s <state> <host>`` and polls every
    *retry* seconds until the state is found or *timeout* is exceeded.

    Args:
        state: State label to wait for
        host: FQDN of the host that will announce the state
        timeout: Seconds before giving up
        retry: Seconds between polls

    Example:
        ```python
        sync_block('SERVERREADY', 'server.example.com')
        ```
    """
    check_sync_daemon()

    logger.info(f'rstrnt-sync-block: waiting for "{state}" on {host} (timeout={timeout}s, retry={retry}s)')
    result = run(f'rstrnt-sync-block -s "{state}" {host} --timeout {timeout} --retry {retry}')
    assert result.succeeded, f'rstrnt-sync-block -s "{state}" {host} failed: {result.stderr}'
    logger.info(f'rstrnt-sync-block: "{state}" reached by {host}')

sync_set(state, *, linger=0)

Announce that this host has reached a state.

Calls rstrnt-sync-set -s <state>. When the process is about to exit, set linger to a non-zero value so the daemon stays alive long enough for the peer to poll it.

Parameters:

Name Type Description Default
state str

State label to announce (e.g. 'SERVERREADY')

 required
linger int

Seconds to sleep after setting the state

 0
Example 
sync_set('SERVERREADY')
sync_set('CLIENTDONE', linger=120)
Source code in sts_libs/src/sts/utils/sync.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def sync_set(state: str, *, linger: int = 0) -> None:
    """Announce that this host has reached a state.

    Calls ``rstrnt-sync-set -s <state>``.  When the process is about
    to exit, set *linger* to a non-zero value so the daemon stays
    alive long enough for the peer to poll it.

    Args:
        state: State label to announce (e.g. ``'SERVERREADY'``)
        linger: Seconds to sleep after setting the state

    Example:
        ```python
        sync_set('SERVERREADY')
        sync_set('CLIENTDONE', linger=120)
        ```
    """
    logger.info(f'rstrnt-sync-set: setting state "{state}"')
    result = run(f'rstrnt-sync-set -s "{state}"')
    logger.debug(f'rstrnt-sync-set: rc={result.rc}, stdout={result.stdout}, stderr={result.stderr}')
    assert result.succeeded, f'rstrnt-sync-set -s "{state}" failed: {result.stderr}'

    check_sync_daemon()

    if linger > 0:
        logger.info(f'rstrnt-sync-set: sleeping {linger}s to keep daemon alive for peer')
        time.sleep(linger)
        check_sync_daemon()

System Checks

sts.utils.syscheck

Post-test system health checks: kernel taint, dmesg, abrt, kdump.

abrt_check()

Check abrt for issues (True if clean, not installed, or daemon not running).

Source code in sts_libs/src/sts/utils/syscheck.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def abrt_check() -> bool:
    """Check abrt for issues (True if clean, not installed, or daemon not running)."""
    logger.info('Checking abrt for errors')

    # Check if abrt is installed
    result = run('rpm -q abrt', log_level=logging.DEBUG)
    if result.failed:
        logger.debug('abrt not installed, skipping check')
        return True

    # Check if abrtd is running
    result = run('pidof abrtd', log_level=logging.DEBUG)
    if result.failed:
        logger.debug('abrtd not running, skipping check')
        return True

    # Check for issues
    result = run('abrt-cli list')
    if result.failed:
        logger.error('abrt-cli failed')
        return False

    # Parse output for directories
    error = False
    for line in result.stdout.splitlines():
        if match := re.match(r'Directory:\s+(\S+)', line):
            directory = match.group(1)
            filename = f'{directory.replace(":", "-")}.tar.gz'
            logger.info(f'Found abrt issue: {filename}')

            # Archive directory
            run(f'tar cfzP {filename} {directory}')
            target = LOG_PATH / filename
            Path(filename).rename(target)

            # Remove from abrt to avoid affecting next test
            run(f'abrt-cli rm {directory}')
            error = True

    if error:
        logger.error('Found abrt errors')
        return False

    logger.debug('No abrt errors found')
    return True

check_all()

Run all system health checks; save logs and sosreport on failure.

Under tmt, only kernel taint and abrt are checked (tmt covers the rest).

Source code in sts_libs/src/sts/utils/syscheck.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def check_all() -> bool:
    """Run all system health checks; save logs and sosreport on failure.

    Under tmt, only kernel taint and abrt are checked (tmt covers the rest).
    """
    logger.info('Checking for errors on the system')
    error_count = 0

    if _running_under_tmt():
        logger.info('Running under tmt, checking only kernel taint and abrt')
        checks = [kernel_check, abrt_check]
    else:
        checks = [
            kernel_check,
            abrt_check,
            messages_dump_check,
            dmesg_check,
            kdump_check,
        ]

    for check in checks:
        if not check():
            error_count += 1

    if error_count and not _running_under_tmt():
        # Save system logs
        messages = Path('/var/log/messages')
        if messages.is_file():
            logger.info('Saving /var/log/messages')
            target = LOG_PATH / 'messages.log'
            target.write_bytes(messages.read_bytes())

        # Generate sosreport if available
        if get_package('sos').is_installed and (sos_file := system.generate_sosreport()):
            logger.info(f'Generated sosreport: {sos_file}')

    return error_count == 0

dmesg_check()

Check dmesg for segfaults and call traces.

Source code in sts_libs/src/sts/utils/syscheck.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def dmesg_check() -> bool:
    """Check dmesg for segfaults and call traces."""
    logger.info('Checking dmesg for errors')

    result = run('dmesg', log_level=logging.DEBUG)
    if result.failed:
        logger.warning('Failed to read dmesg')
        return True

    for pattern in ERROR_PATTERNS:
        if pattern.search(result.stdout):
            logger.error(f'Found error in dmesg matching: {pattern.pattern}')
            target = LOG_PATH / 'dmesg.log'
            target.write_text(result.stdout)
            return False

    run('dmesg -c', log_level=logging.DEBUG)
    logger.debug('No errors found in dmesg')
    return True

kdump_check()

Check /var/crash for recent kdump crashes (last 24 hours).

Source code in sts_libs/src/sts/utils/syscheck.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def kdump_check() -> bool:
    """Check /var/crash for recent kdump crashes (last 24 hours)."""
    logger.info('Checking for kdump crashes')

    # Get hostname
    result = run('hostname', log_level=logging.DEBUG)
    if result.failed:
        logger.error('Failed to get hostname')
        return False

    hostname = result.stdout.strip()
    crash_dir = Path('/var/crash') / hostname

    if not crash_dir.exists():
        logger.debug('No kdump directory found')
        return True

    # Get crash timestamps
    crashes: list[str] = []
    for crash in crash_dir.iterdir():
        if match := re.match(r'.*?-(.*)', crash.name):
            date = match.group(1).replace('.', '-')
            index = date.rfind('-')
            date = f'{date[:index]} {date[index + 1 :]}'
            crashes.append(date)

    if not crashes:
        logger.debug('No kdump crashes found')
        return True

    # Check crash times
    now = datetime.now(UTC).timestamp()
    for crash in crashes:
        result = run(f'date --date="{crash}" +%s')
        if result.failed:
            logger.warning(f'Failed to parse crash date: {crash}')
            continue

        crash_time = int(result.stdout)
        if crash_time > now - 86400:  # Last 24 hours
            logger.error(f'Found recent crash: {crash}')
            return False

    logger.debug('No recent kdump crashes found')
    return True

kernel_check()

Check kernel taint status (True if clean or unreadable).

Source code in sts_libs/src/sts/utils/syscheck.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def kernel_check() -> bool:
    """Check kernel taint status (True if clean or unreadable)."""
    logger.info('Checking for tainted kernel')

    # Get current tainted value
    try:
        tainted = int(Path('/proc/sys/kernel/tainted').read_text().strip())
    except (PermissionError, FileNotFoundError, OSError) as exc:
        logger.warning(f'Cannot check kernel taint: {exc}')
        return True

    if tainted == 0:
        return True

    logger.warning('Kernel is tainted!')

    # Check tainted bits
    bit = 0
    value = tainted
    while value:
        if value & 1:
            logger.debug(f'TAINT bit {bit} is set')
        bit += 1
        value >>= 1

    # List tainted module definitions
    result = run('cat /usr/src/kernels/`uname -r`/include/linux/kernel.h | grep TAINT_', log_level=logging.DEBUG)
    if result.succeeded:
        logger.debug('Tainted bit definitions:')
        logger.debug(result.stdout)

    # Check for tainted modules
    result = run("cat /proc/modules | grep -e '(.*)' | cut -d' ' -f1", log_level=logging.DEBUG)
    if result.failed:
        return False

    # Skip certain modules
    ignore_modules = {'ocrdma', 'nvme_fc', 'nvmet_fc', 'kvdo', 'uds'}  # Tech Preview modules
    found_issue = False

    for module in result.stdout.splitlines():
        if not module:
            continue

        logger.info(f'Found tainted module: {module}')
        run(f'modinfo {module}', log_level=logging.DEBUG)

        if module in ignore_modules:
            logger.debug(f'Ignoring known tainted module: {module}')
            continue

        found_issue = True

    return not found_issue

messages_dump_check()

Check /var/log/messages for kernel dump traces.

Source code in sts_libs/src/sts/utils/syscheck.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def messages_dump_check() -> bool:
    """Check /var/log/messages for kernel dump traces."""
    logger.info('Checking for kernel dumps')

    messages = Path('/var/log/messages')
    if not messages.is_file():
        logger.warning('No messages file found')
        return True

    # Read messages file
    try:
        content = messages.read_text()
    except UnicodeDecodeError:
        content = messages.read_text(encoding='latin-1')

    # Search for dumps
    begin = r'\[ cut here \]'
    end = r'\[ end trace '
    pattern = f'{begin}(.*?){end}'

    if dumps := re.findall(pattern, content, re.MULTILINE):
        logger.error('Found kernel dumps:')
        for dump in dumps:
            logger.error(dump)
        return False

    logger.debug('No kernel dumps found')
    return True

Error Handling

sts.utils.errors

Storage Test Suite error hierarchy.

Convention for subsystem errors:

  1. Every subsystem defines <Module>Error(STSError) as its base. Exception: DmError inherits DeviceError because Device Mapper is the Linux device layer.

  2. Domain subclasses are optional — add them only when callers need to catch distinct failure modes (e.g. FIOConfigError vs FIOExecutionError, or Stratis pool/fs/blockdev errors).

  3. Command failures go through CommandResult.assert_ok() which raises STSError. There are no per-module *CommandError classes.

DeviceError

Bases: STSError

Base class for device-related errors.

Source code in sts_libs/src/sts/utils/errors.py 
27
28
class DeviceError(STSError):
    """Base class for device-related errors."""

DeviceNotFoundError

Bases: DeviceError

Device does not exist.

Source code in sts_libs/src/sts/utils/errors.py 
31
32
class DeviceNotFoundError(DeviceError):
    """Device does not exist."""

DeviceTypeError

Bases: DeviceError

Device is not of expected type.

Source code in sts_libs/src/sts/utils/errors.py 
35
36
class DeviceTypeError(DeviceError):
    """Device is not of expected type."""

ModuleAlreadyLoadedError

Bases: ModuleError

Exclusive acquisition conflicts with a pre-existing module load.

Source code in sts_libs/src/sts/utils/errors.py 
51
52
class ModuleAlreadyLoadedError(ModuleError):
    """Exclusive acquisition conflicts with a pre-existing module load."""

ModuleError

Bases: STSError

Base class for kernel module errors.

Source code in sts_libs/src/sts/utils/errors.py 
39
40
class ModuleError(STSError):
    """Base class for kernel module errors."""

ModuleInUseError

Bases: ModuleError

Module cannot be unloaded because it is in use.

Source code in sts_libs/src/sts/utils/errors.py 
63
64
class ModuleInUseError(ModuleError):
    """Module cannot be unloaded because it is in use."""

ModuleLoadError

Bases: ModuleError

Failed to load kernel module.

Source code in sts_libs/src/sts/utils/errors.py 
43
44
class ModuleLoadError(ModuleError):
    """Failed to load kernel module."""

ModuleUnavailableError

Bases: ModuleError

Kernel module is absent before setup mutates module state.

Source code in sts_libs/src/sts/utils/errors.py 
47
48
class ModuleUnavailableError(ModuleError):
    """Kernel module is absent before setup mutates module state."""

ModuleUnloadError

Bases: ModuleError

Failed to unload kernel module.

Source code in sts_libs/src/sts/utils/errors.py 
59
60
class ModuleUnloadError(ModuleError):
    """Failed to unload kernel module."""

ModuleVerificationError

Bases: ModuleError

Loaded module state does not satisfy its requested postconditions.

Source code in sts_libs/src/sts/utils/errors.py 
55
56
class ModuleVerificationError(ModuleError):
    """Loaded module state does not satisfy its requested postconditions."""

PackageError

Bases: STSError

Base class for package-related errors.

Source code in sts_libs/src/sts/utils/errors.py 
67
68
class PackageError(STSError):
    """Base class for package-related errors."""

PackageInstallError

Bases: PackageError

Failed to install package.

Source code in sts_libs/src/sts/utils/errors.py 
75
76
class PackageInstallError(PackageError):
    """Failed to install package."""

PackageNotFoundError

Bases: PackageError

Package does not exist.

Source code in sts_libs/src/sts/utils/errors.py 
71
72
class PackageNotFoundError(PackageError):
    """Package does not exist."""

STSError

Bases: Exception

Base class for all STS exceptions.

Source code in sts_libs/src/sts/utils/errors.py 
23
24
class STSError(Exception):
    """Base class for all STS exceptions."""

SysError

Bases: STSError

Base class for system-related errors.

Source code in sts_libs/src/sts/utils/errors.py 
79
80
class SysError(STSError):
    """Base class for system-related errors."""

SysNotSupportedError

Bases: SysError

Operation not supported on this system.

Source code in sts_libs/src/sts/utils/errors.py 
83
84
class SysNotSupportedError(SysError):
    """Operation not supported on this system."""

FMF Markers Plugin

sts.utils.fmf_markers

Pytest plugin: apply fmf metadata (tags, tier) as pytest markers.

Reads main.fmf files via the fmf library, resolves tag and tier inheritance, and applies them as pytest markers during collection. This enables pytest-native test selection that mirrors tmt plan filters.

Usage::

pytest tests/ -m ci              # same tests as the sts-ci plan
pytest tests/ -m tier1           # all tier-1 tests
pytest tests/ -m "dm and tier1"  # tier-1 device-mapper tests
pytest tests/ -m "not hw_fc"     # exclude FC-hardware tests

Requires the fmf library (available wherever tmt is installed). When fmf is not importable the plugin silently does nothing.

FmfMarkerPlugin

Apply fmf tags and tier as pytest markers during collection.

Source code in sts_libs/src/sts/utils/fmf_markers.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class FmfMarkerPlugin:
    """Apply fmf tags and tier as pytest markers during collection."""

    def __init__(self, tree: Any) -> None:
        self._metadata = _build_test_metadata(tree)
        self._ci_excluded_tags, self._ci_required_tier = _parse_ci_filters(tree)

    def _is_ci(self, node: _FmfNode) -> bool:
        return node.tier == self._ci_required_tier and not (node.tags & self._ci_excluded_tags)

    def pytest_collection_modifyitems(  # noqa: D102
        self,
        config: pytest.Config,
        items: list[pytest.Item],
    ) -> None:
        del config
        for item in items:
            test_file = Path(str(item.fspath)).resolve()
            nodes = self._metadata.get(test_file)
            if nodes is None:
                continue

            all_tags: set[str] = set()
            tiers: set[int] = set()
            ci_eligible = False

            for node in nodes:
                all_tags |= node.tags
                if node.tier is not None:
                    tiers.add(node.tier)
                if self._is_ci(node):
                    ci_eligible = True

            for tag in all_tags:
                item.add_marker(_normalize_tag(tag))

            for tier in tiers:
                item.add_marker(f'tier{tier}')

            if ci_eligible:
                item.add_marker('ci')

pytest_configure(config)

Register fmf-derived markers if the fmf library is available.

Source code in sts_libs/src/sts/utils/fmf_markers.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def pytest_configure(config: pytest.Config) -> None:
    """Register fmf-derived markers if the fmf library is available."""
    try:
        import fmf  # noqa: PLC0415
    except ImportError:
        logger.warning('fmf_markers: fmf library not installed, markers disabled (pip install fmf)')
        return

    rootdir = Path(str(config.rootdir))  # pyright: ignore[reportAttributeAccessIssue]
    tree_root = rootdir
    while tree_root != tree_root.parent:
        if (tree_root / '.fmf' / 'version').exists():
            break
        tree_root = tree_root.parent
    else:
        logger.warning('fmf_markers: no .fmf tree found above %s, markers disabled', rootdir)
        return

    config.addinivalue_line('markers', 'ci: test is safe for CI (derived from sts-ci plan filters)')
    for tier in range(4):
        config.addinivalue_line('markers', f'tier{tier}: fmf tier {tier}')

    try:
        tree = fmf.Tree(str(tree_root))

        all_tags: set[str] = set()
        for node in tree.prune():
            for tag in node.data.get('tag') or []:
                all_tags.add(tag)
        for tag in sorted(all_tags):
            config.addinivalue_line('markers', f'{_normalize_tag(tag)}: fmf tag "{tag}"')

        config.pluginmanager.register(FmfMarkerPlugin(tree), name=PLUGIN_NAME)
    except Exception as exc:  # noqa: BLE001
        logger.warning('fmf_markers: failed to load fmf tree from %s, markers disabled: %s', tree_root, exc)

tmt Pytest Integration

sts.utils.tmt_pytest

Pytest plugin: report individual test outcomes as tmt subresults.

When running under tmt (TMT_TEST_PIDFILE is set), writes $TMT_TEST_DATA/tmt-report-results.yaml at session end so that tmt's shell framework reduces outcomes into a single parent result and keeps each pytest item as a subresult.

This deliberately writes tmt-report-results.yaml (the file normally created by the tmt-report-result command) rather than results.yaml (which requires result: custom in fmf metadata and produces separate top-level results, not subresults).

TmtReportPlugin

Collect pytest outcomes and write them for tmt at session end.

Source code in sts_libs/src/sts/utils/tmt_pytest.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class TmtReportPlugin:
    """Collect pytest outcomes and write them for tmt at session end."""

    def __init__(self) -> None:
        self.results: dict[str, TmtSubresult] = {}

    def pytest_runtest_logreport(self, report: pytest.TestReport) -> None:
        """Record one setup/call/teardown report."""
        apply_report(self.results, report)

    def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int) -> None:
        """Write results at session end."""
        del session, exitstatus
        self._flush()

    def _flush(self) -> None:
        if not self.results:
            return
        test_data = getenv('TMT_TEST_DATA')
        if not test_data:
            logger.warning('TMT_TEST_PIDFILE set but TMT_TEST_DATA missing; skipping result write')
            return
        write_results(Path(test_data) / RESULTS_FILENAME, list(self.results.values()))

pytest_runtest_logreport(report)

Record one setup/call/teardown report.

Source code in sts_libs/src/sts/utils/tmt_pytest.py 
126
127
128
def pytest_runtest_logreport(self, report: pytest.TestReport) -> None:
    """Record one setup/call/teardown report."""
    apply_report(self.results, report)

pytest_sessionfinish(session, exitstatus)

Write results at session end.

Source code in sts_libs/src/sts/utils/tmt_pytest.py 
130
131
132
133
def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int) -> None:
    """Write results at session end."""
    del session, exitstatus
    self._flush()

TmtSubresult

One pytest item mapped to a tmt subresult entry.

Source code in sts_libs/src/sts/utils/tmt_pytest.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
class TmtSubresult:
    """One pytest item mapped to a tmt subresult entry."""

    __slots__ = ('name', 'note', 'result')

    name: str
    result: TmtOutcome
    note: list[str]

    def __init__(self, name: str, result: TmtOutcome) -> None:
        self.name = name
        self.result = result
        self.note = []

apply_report(results, report)

Fold a pytest setup/call/teardown report into one subresult per item.

Source code in sts_libs/src/sts/utils/tmt_pytest.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def apply_report(results: dict[str, TmtSubresult], report: pytest.TestReport) -> None:
    """Fold a pytest setup/call/teardown report into one subresult per item."""
    outcome, note = _phase_outcome(report)
    item = results.get(report.nodeid)
    if item is None:
        item = TmtSubresult(name=_tmt_name(report.nodeid), result=outcome)
        results[report.nodeid] = item
    else:
        item.result = _MERGE[item.result, outcome]
    if note and note not in item.note:
        item.note.append(note)

pytest_configure(config)

Register the tmt reporter only inside a tmt test invocation.

Source code in sts_libs/src/sts/utils/tmt_pytest.py 
145
146
147
148
def pytest_configure(config: pytest.Config) -> None:
    """Register the tmt reporter only inside a tmt test invocation."""
    if getenv('TMT_TEST_PIDFILE'):
        config.pluginmanager.register(TmtReportPlugin(), name=PLUGIN_NAME)

write_results(path, items)

Write tmt-report-results.yaml (name + result required, note optional).

Source code in sts_libs/src/sts/utils/tmt_pytest.py 
106
107
108
109
110
111
112
113
114
115
116
117
def write_results(path: Path, items: Sequence[TmtSubresult]) -> None:
    """Write tmt-report-results.yaml (name + result required, note optional)."""
    # ponytail: hand-rolled YAML — json.dumps handles escaping; use pyyaml if schema grows
    lines: list[str] = []
    for item in items:
        lines.append(f'- name: {json.dumps(item.name, ensure_ascii=False)}')
        lines.append(f'  result: {json.dumps(item.result)}')
        if item.note:
            lines.append('  note:')
            lines.extend(f'    - {json.dumps(n, ensure_ascii=False)}' for n in item.note)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text('\n'.join(lines) + '\n')

Test State Persistence

sts.utils.teststate

Cross-phase test state persistence.

Provides JSON state storage for tests that span multiple phases (e.g. reboot tests). The storage directory is resolved in order:

  1. TMT_TEST_DATA environment variable (always preferred under tmt)
  2. Explicit data_dir passed by the caller
  3. /var/tmp/sts/ fallback

Snapshot-boot tests must pass a data_dir that lives outside the snapshotted root filesystem (e.g. /boot/sts), otherwise the state file will not be visible after booting the snapshot. Use :func:on_different_device to verify the path is on a separate filesystem from /.

TestState

JSON state file for persisting data across test phases.

Parameters:

Name Type Description Default
name str

Base name for the state file (without .json suffix).

 required
data_dir Path | None

Explicit storage directory used when TMT_TEST_DATA is not set. Snapshot-boot tests should pass a path outside the snapshotted root (e.g. Path('/boot/sts')).

 None
Source code in sts_libs/src/sts/utils/teststate.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class TestState:
    """JSON state file for persisting data across test phases.

    Args:
        name: Base name for the state file (without ``.json`` suffix).
        data_dir: Explicit storage directory used when ``TMT_TEST_DATA``
            is not set.  Snapshot-boot tests should pass a path outside
            the snapshotted root (e.g. ``Path('/boot/sts')``).
    """

    __test__ = False

    def __init__(self, name: str, *, data_dir: Path | None = None) -> None:
        tmt_data = getenv('TMT_TEST_DATA')
        if tmt_data:
            parent = Path(tmt_data)
        elif data_dir is not None:
            parent = data_dir
        else:
            parent = _FALLBACK_DIR
        self.path = parent / f'{name}.json'

    def save(self, data: dict[str, Any]) -> None:
        """Write *data* as JSON to the state file."""
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.path.write_text(json.dumps(data))
        logger.info(f'Saved test state to {self.path}')

    def load(self) -> dict[str, Any] | None:
        """Read state from the file, or return ``None`` if absent."""
        if not self.path.exists():
            return None
        return json.loads(self.path.read_text())

    def remove(self) -> None:
        """Delete the state file if it exists."""
        self.path.unlink(missing_ok=True)

load()

Read state from the file, or return None if absent.

Source code in sts_libs/src/sts/utils/teststate.py 
74
75
76
77
78
def load(self) -> dict[str, Any] | None:
    """Read state from the file, or return ``None`` if absent."""
    if not self.path.exists():
        return None
    return json.loads(self.path.read_text())

remove()

Delete the state file if it exists.

Source code in sts_libs/src/sts/utils/teststate.py 
80
81
82
def remove(self) -> None:
    """Delete the state file if it exists."""
    self.path.unlink(missing_ok=True)

save(data)

Write data as JSON to the state file.

Source code in sts_libs/src/sts/utils/teststate.py 
68
69
70
71
72
def save(self, data: dict[str, Any]) -> None:
    """Write *data* as JSON to the state file."""
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self.path.write_text(json.dumps(data))
    logger.info(f'Saved test state to {self.path}')

on_different_device(path, reference=Path('/'))

Check whether path resides on a different filesystem than reference.

Useful for snapshot-boot tests that need state stored outside the snapshotted root. Compares st_dev of the two paths.

Parameters:

Name Type Description Default
path Path

Path to check (must exist).

 required
reference Path

Path to compare against (default: /).

 Path('/')
Source code in sts_libs/src/sts/utils/teststate.py 
33
34
35
36
37
38
39
40
41
42
43
def on_different_device(path: Path, reference: Path = Path('/')) -> bool:
    """Check whether *path* resides on a different filesystem than *reference*.

    Useful for snapshot-boot tests that need state stored outside the
    snapshotted root.  Compares ``st_dev`` of the two paths.

    Args:
        path: Path to check (must exist).
        reference: Path to compare against (default: ``/``).
    """
    return path.stat().st_dev != reference.stat().st_dev

CLI Tool Base

sts.utils.cli

Shared base for CLI tool wrappers.

Provides argv-based command construction using run_argv (no shell).

CliTool pydantic-model

Bases: StsBaseModel

Base for CLI tool wrappers.

Subclasses set CLI_NAME and optionally override _global_args().

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Base for CLI tool wrappers.\n\nSubclasses set CLI_NAME and optionally override _global_args().",
  "properties": {},
  "title": "CliTool",
  "type": "object"
}
Source code in sts_libs/src/sts/utils/cli.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class CliTool(StsBaseModel):
    """Base for CLI tool wrappers.

    Subclasses set CLI_NAME and optionally override _global_args().
    """

    CLI_NAME: ClassVar[str]

    def _global_args(self) -> list[str]:
        """Return global CLI arguments inserted after CLI_NAME.

        Override in subclasses that have config-driven global flags
        (e.g. stratis --unhyphenated_uuids).
        """
        return []

    def run_command(
        self,
        subcommand: str | Sequence[str] | None = None,
        action: str | Sequence[str] | None = None,
        options: CliOptions | None = None,
        positional_args: list[str] | None = None,
    ) -> CommandResult:
        """Build and run a CLI command via run_argv (no shell).

        Args:
            subcommand: Primary subcommand -- string or list for multi-word
                (e.g. ``['debug', 'get-object-path']``).
            action: Action verb -- string or list for multi-word
                (e.g. ``['bind', 'keyring']``).
            options: CLI options dict. None value = bare flag, str = ``--key value``.
            positional_args: Trailing positional arguments.
        """
        argv: list[str] = [self.CLI_NAME]
        argv.extend(self._global_args())

        if subcommand is not None:
            if isinstance(subcommand, str):
                argv.append(subcommand)
            else:
                argv.extend(subcommand)
        if action is not None:
            if isinstance(action, str):
                argv.append(action)
            else:
                argv.extend(action)
        if options is not None:
            for key, value in options.items():
                argv.append(key)
                if value is not None:
                    argv.append(value)
        if positional_args:
            argv.extend(positional_args)

        return run_argv(argv)

run_command(subcommand=None, action=None, options=None, positional_args=None)

Build and run a CLI command via run_argv (no shell).

Parameters:

Name Type Description Default
subcommand str | Sequence[str] | None

Primary subcommand -- string or list for multi-word (e.g. ['debug', 'get-object-path']).

 None
action str | Sequence[str] | None

Action verb -- string or list for multi-word (e.g. ['bind', 'keyring']).

 None
options CliOptions | None

CLI options dict. None value = bare flag, str = --key value.

 None
positional_args list[str] | None

Trailing positional arguments.

 None
Source code in sts_libs/src/sts/utils/cli.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def run_command(
    self,
    subcommand: str | Sequence[str] | None = None,
    action: str | Sequence[str] | None = None,
    options: CliOptions | None = None,
    positional_args: list[str] | None = None,
) -> CommandResult:
    """Build and run a CLI command via run_argv (no shell).

    Args:
        subcommand: Primary subcommand -- string or list for multi-word
            (e.g. ``['debug', 'get-object-path']``).
        action: Action verb -- string or list for multi-word
            (e.g. ``['bind', 'keyring']``).
        options: CLI options dict. None value = bare flag, str = ``--key value``.
        positional_args: Trailing positional arguments.
    """
    argv: list[str] = [self.CLI_NAME]
    argv.extend(self._global_args())

    if subcommand is not None:
        if isinstance(subcommand, str):
            argv.append(subcommand)
        else:
            argv.extend(subcommand)
    if action is not None:
        if isinstance(action, str):
            argv.append(action)
        else:
            argv.extend(action)
    if options is not None:
        for key, value in options.items():
            argv.append(key)
            if value is not None:
                argv.append(value)
    if positional_args:
        argv.extend(positional_args)

    return run_argv(argv)