Skip to content

Utilities

Helper modules used across sts-libs.

Command Line

sts.utils.cmdline

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

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 
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
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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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 
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
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 
151
152
153
def exists(cmd: str) -> bool:
    """Check if command exists in PATH."""
    return shutil.which(cmd) is not None

run(cmd, msg=None, timeout=600)

Run a shell command and return a CommandResult.

Source code in sts_libs/src/sts/utils/cmdline.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
123
124
def run(cmd: str, msg: str | None = None, timeout: int | None = 600) -> CommandResult:
    """Run a shell command and return a ``CommandResult``."""
    msg = msg or 'Running'
    logger.info(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,
        )
    if result.returncode != 0:
        logger.warning(f"Command failed (rc={result.returncode}): '{cmd}'")
        if result.stderr:
            logger.warning(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)

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

Source code in sts_libs/src/sts/utils/cmdline.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def run_argv(argv: Sequence[str], msg: str | None = None, timeout: int | None = 600) -> CommandResult:
    """Run a command from an argument vector (no shell) and return a ``CommandResult``."""
    cmd_str = shlex.join(argv)
    logger.info(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 '')
        )
    if result.returncode != 0:
        logger.warning(f"Command failed (rc={result.returncode}): '{cmd_str}'")
        if result.stderr:
            logger.warning(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)

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')
            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
225
226
227
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()

    for package in packages:
        if not get_package(package).is_installed:
            return all(pm.install(package) for package in packages)
    return True

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}')
    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 
230
231
232
233
234
235
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 
238
239
240
241
242
243
244
245
246
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.

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 
 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
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)

    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.
        """
        # If no name provided, get first loaded module
        if not self.name:
            try:
                line = Path('/proc/modules').read_text().splitlines()[0]
                self.name = line.split()[0]
            except (OSError, IndexError):
                return

        # Get module information if name is available
        if self.name:
            # Reset mutable state before re-reading
            self.used_by = []
            self.parameters = {}
            try:
                for line in Path('/proc/modules').read_text().splitlines():
                    parts = line.split(maxsplit=5)
                    if parts[0] == self.name:
                        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 (OSError, IndexError, ValueError):
                logger.warning(f'Failed to get module info for {self.name}', exc_info=True)

            # Get module parameters
            param_path = Path('/sys/module') / self.name / 'parameters'
            if param_path.is_dir():
                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 exists(self) -> bool:
        """True if modinfo can find this module."""
        if self.name:
            return run(f'modinfo {self.name} -n').succeeded
        return False

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

    def load(self, parameters: str | None = None) -> bool:
        """Load module via modprobe.

        Args:
            parameters: Module parameters string passed to modprobe.
        """
        if not self.name:
            logger.error('Module name required')
            return False

        cmd = ['modprobe', self.name]
        if parameters:
            cmd.append(parameters)

        result = run(' '.join(cmd))
        if result.failed:
            logger.error(f'Failed to load module: {result.stderr}')
            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(f'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.

discover()

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

Source code in sts_libs/src/sts/utils/modules.py 
40
41
42
43
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 
153
154
155
156
157
@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

load(parameters=None)

Load module via modprobe.

Parameters:

Name Type Description Default
parameters str | None

Module parameters string passed to modprobe.

 None
Source code in sts_libs/src/sts/utils/modules.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def load(self, parameters: str | None = None) -> bool:
    """Load module via modprobe.

    Args:
        parameters: Module parameters string passed to modprobe.
    """
    if not self.name:
        logger.error('Module name required')
        return False

    cmd = ['modprobe', self.name]
    if parameters:
        cmd.append(parameters)

    result = run(' '.join(cmd))
    if result.failed:
        logger.error(f'Failed to load module: {result.stderr}')
        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 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def unload(self) -> bool:
    """Unload module via ``modprobe -r``."""
    if not self.name:
        logger.error('Module name required')
        return False

    result = run(f'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 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
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 
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
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 | 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.time()
        while time.time() - 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 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
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 
182
183
184
185
186
187
@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 
189
190
191
192
193
194
195
196
197
def load(self, name: str, parameters: str | 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 
199
200
201
202
203
204
205
206
207
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 
209
210
211
212
213
214
215
216
217
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)

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 
41
42
class DirAccessError(DirectoryError):
    """Directory cannot be accessed."""

DirNotFoundError

Bases: DirectoryError

Directory does not exist.

Source code in sts_libs/src/sts/utils/files.py 
33
34
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 
37
38
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 
 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
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 
124
125
126
127
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 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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 
142
143
144
145
146
147
148
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 
116
117
118
119
120
121
122
@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 
129
130
131
132
133
134
135
136
137
138
139
140
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 
 96
 97
 98
 99
100
101
102
103
104
@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 
106
107
108
109
110
111
112
113
114
@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 
71
72
73
74
75
76
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 
29
30
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 
151
152
153
154
155
156
157
158
159
@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 
349
350
351
352
353
354
355
356
357
358
359
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 
162
163
164
165
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 
285
286
287
288
289
290
291
292
293
294
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(' '.join(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 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
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(f'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 if a device or mountpoint is currently mounted.

Source code in sts_libs/src/sts/utils/files.py 
174
175
176
177
178
179
180
def is_mounted(device: str | Path | None = None, mountpoint: str | Path | None = None) -> bool:
    """Check if a device or mountpoint is currently mounted."""
    if device:
        return run(f'mount | grep {device!s}').succeeded
    if mountpoint:
        return run(f'mount | grep {mountpoint!s}').succeeded
    return False

mkfs(device=None, fs_type=None, *args, **kwargs)

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
**kwargs str | bool

Extra mkfs options; force=True adds -F (ext) or -f (xfs).

 {}
Source code in sts_libs/src/sts/utils/files.py 
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
def mkfs(device: str | Path | None = None, fs_type: str | None = None, *args: str, **kwargs: str | bool) -> bool:
    """Create filesystem on device.

    Args:
        device: Block device path.
        fs_type: Filesystem type (e.g. 'ext4', 'xfs').
        **kwargs: Extra mkfs options; ``force=True`` adds ``-F`` (ext) or ``-f`` (xfs).
    """
    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 kwargs.pop('force', False):
        force_option = '-F' if fs_type != 'xfs' else '-f'
        cmd.append(force_option)

    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(device_str)

    result = run(' '.join(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 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
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(' '.join(cmd))
    if result.failed:
        logger.error(f'Failed to mount device: {result.stderr}')
        return False
    return True

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 
168
169
170
171
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)

umount(device=None, mountpoint=None, *, force=False)

Unmount device or mountpoint (no-op if already unmounted).

Source code in sts_libs/src/sts/utils/files.py 
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
def umount(
    device: str | Path | None = None,
    mountpoint: str | Path | None = None,
    *,
    force: bool = False,
) -> bool:
    """Unmount device or mountpoint (no-op if already unmounted)."""
    if device and not is_mounted(device=str(device)):
        return True
    if mountpoint and not is_mounted(mountpoint=str(mountpoint)):
        return True

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

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

verify_checksum(path, expected)

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

Source code in sts_libs/src/sts/utils/files.py 
362
363
364
365
366
367
368
369
370
371
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 
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
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}')

    cmd = ' '.join(cmd_parts)
    result = run(cmd)

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

    if sync:
        sync_result = run('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 
339
340
341
342
343
344
345
346
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 
 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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
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
    @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 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@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 
122
123
124
125
126
127
128
@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 
49
50
51
52
53
54
55
@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 
57
58
59
60
61
62
63
@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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@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 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@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 
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
@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)

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))

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
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

    @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))

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 
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
@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')
    if result.failed:
        logger.warning('abrt not installed, skipping check')
        return True

    # Check if abrtd is running
    result = run('pidof abrtd')
    if result.failed:
        logger.warning('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')
    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')  # Clear dmesg
    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')
    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_')
    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")
    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}')

        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."""

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 
51
52
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."""

ModuleUnloadError

Bases: ModuleError

Failed to unload kernel module.

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

PackageError

Bases: STSError

Base class for package-related errors.

Source code in sts_libs/src/sts/utils/errors.py 
55
56
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 
63
64
class PackageInstallError(PackageError):
    """Failed to install package."""

PackageNotFoundError

Bases: PackageError

Package does not exist.

Source code in sts_libs/src/sts/utils/errors.py 
59
60
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 
67
68
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 
71
72
class SysNotSupportedError(SysError):
    """Operation not supported on this system."""

tmt Integration

sts.utils.tmt

TMT (Test Management Tool) custom result submission and log gathering.

See: https://tmt.readthedocs.io/en/stable/spec/tests.html#result

CustomResults

Bases: TypedDict

Schema for a single TMT custom test result entry.

Attributes:

Name Type Description
name str

Result name path (e.g. '/step-1' or '/setup/iscsi/target').
Source code in sts_libs/src/sts/utils/tmt.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class CustomResults(TypedDict):
    """Schema for a single TMT custom test result entry.

    Attributes:
        name: Result name path (e.g. '/step-1' or '/setup/iscsi/target').
    """

    name: str
    result: TmtResult
    note: str | None
    log: list[str] | None
    serialnumber: int | None
    guest: GuestType | None
    duration: str | None
    ids: dict[str, str] | None

GuestType

Bases: TypedDict

Guest information for custom results.

Source code in sts_libs/src/sts/utils/tmt.py 
62
63
64
65
66
class GuestType(TypedDict):
    """Guest information for custom results."""

    name: str | None
    role: str | None

Results

Accumulate TMT custom results, then submit() as results.json.

Example 
results = Results()
results.add(name='setup', result='pass')
results.add(name='test', result='pass', log=['test.log'])
results.submit()
Source code in sts_libs/src/sts/utils/tmt.py 
 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 Results:
    """Accumulate TMT custom results, then ``submit()`` as results.json.

    Example:
        ```python
        results = Results()
        results.add(name='setup', result='pass')
        results.add(name='test', result='pass', log=['test.log'])
        results.submit()
        ```
    """

    def __init__(self) -> None:
        self.results: list[dict[str, Any]] = []
        self.timestamp = timestamp()

    def add(
        self,
        name: str = '/',
        result: TmtResult = 'pass',
        note: str | None = None,
        log: list[str] | None = None,
        errors: list[str] | None = None,
    ) -> None:
        """Add a result entry. If *errors* is non-empty, result is forced to 'fail'."""
        if not name.startswith('/'):
            name = f'/{name}'
        if errors:
            result = 'fail'

        # Calculate duration
        new_timestamp = timestamp()
        duration = calculate_duration(self.timestamp, new_timestamp)
        self.timestamp = new_timestamp

        # Create result
        result_to_add = CustomResults(
            name=name,
            result=result,
            note=note,
            log=log,
            duration=duration,
            ids=None,
            serialnumber=None,
            guest=None,
        )

        self.results.append(remove_nones(result_to_add))

    def submit(self) -> None:
        """Write accumulated results to TMT_TEST_DATA/results.json."""
        file = Path(TMT_TEST_DATA / 'results.json')
        with file.open('w') as f:
            json.dump(self.results, f)

add(name='/', result='pass', note=None, log=None, errors=None)

Add a result entry. If errors is non-empty, result is forced to 'fail'.

Source code in sts_libs/src/sts/utils/tmt.py 
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
def add(
    self,
    name: str = '/',
    result: TmtResult = 'pass',
    note: str | None = None,
    log: list[str] | None = None,
    errors: list[str] | None = None,
) -> None:
    """Add a result entry. If *errors* is non-empty, result is forced to 'fail'."""
    if not name.startswith('/'):
        name = f'/{name}'
    if errors:
        result = 'fail'

    # Calculate duration
    new_timestamp = timestamp()
    duration = calculate_duration(self.timestamp, new_timestamp)
    self.timestamp = new_timestamp

    # Create result
    result_to_add = CustomResults(
        name=name,
        result=result,
        note=note,
        log=log,
        duration=duration,
        ids=None,
        serialnumber=None,
        guest=None,
    )

    self.results.append(remove_nones(result_to_add))

submit()

Write accumulated results to TMT_TEST_DATA/results.json.

Source code in sts_libs/src/sts/utils/tmt.py 
144
145
146
147
148
def submit(self) -> None:
    """Write accumulated results to TMT_TEST_DATA/results.json."""
    file = Path(TMT_TEST_DATA / 'results.json')
    with file.open('w') as f:
        json.dump(self.results, f)

calculate_duration(start, end)

Format elapsed time between two timestamps as 'hh:mm:ss'.

Source code in sts_libs/src/sts/utils/tmt.py 
56
57
58
59
def calculate_duration(start: float, end: float) -> str:
    """Format elapsed time between two timestamps as 'hh:mm:ss'."""
    secs = int(end - start)
    return f'{secs // 3600:02d}:{secs % 3600 // 60:02d}:{secs % 60:02d}'

gather_logs_from_dir(logs_path, name)

Archive a log directory into TMT_TEST_DATA as a tarball.

Source code in sts_libs/src/sts/utils/tmt.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def gather_logs_from_dir(logs_path: str, name: str | None) -> Path | None:
    """Archive a log directory into TMT_TEST_DATA as a tarball."""
    path = Path(logs_path)
    if not path.is_dir():
        return None

    # Generate tarfile name
    if not name:
        name = str(path).replace('/', '_')
    if '.tar' not in name:
        name = f'{name}.tar'

    # Create tarfile
    tarfile_path = f'{TMT_TEST_DATA}/{name}'
    with tarfile.open(tarfile_path, 'w') as tar:
        tar.add(path, recursive=True)
    return Path(tarfile_path)

remove_nones(cr)

Strip None values from a CustomResults dict.

Source code in sts_libs/src/sts/utils/tmt.py 
90
91
92
def remove_nones(cr: CustomResults) -> dict[str, Any]:
    """Strip None values from a CustomResults dict."""
    return {k: v for k, v in cr.items() if v is not None}

timestamp()

Current time in seconds since epoch.

Source code in sts_libs/src/sts/utils/tmt.py 
51
52
53
def timestamp() -> float:
    """Current time in seconds since epoch."""
    return time.time()