diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 919c1e7..9b77c5b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -87,7 +87,7 @@ jobs: shell: bash run: | set -euo pipefail - mapfile -t ALL < <(git ls-files 'tests/**/test_*.py' \ + mapfile -t ALL < <(git ls-files ':(glob)tests/**/test_*.py' \ | grep -v '^tests/sim/' | grep -v '^tests/e2e/' | sort) SEL=() for i in "${!ALL[@]}"; do diff --git a/tests/conftest.py b/tests/conftest.py index 2ccb965..f5da5a2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,11 +5,12 @@ import subprocess import uuid from pathlib import Path -from typing import Literal import numpy as np import pytest +from systemd_probe import classify_systemd_probe + SEQ_LEN = 1024 SAMPLES_PER_SHARD = 128 @@ -115,22 +116,6 @@ def corpus_dir(tmp_path_factory: pytest.TempPathFactory) -> Path: return d -# Messages meaning "this user/host has no usable system manager", not "the probe broke". -_SYSTEMD_UNAVAILABLE = ( - "Interactive authentication required", - "Access denied", - "Failed to connect to bus", - "System has not been booted with systemd", - "Permission denied", -) - - -def _classify_systemd_probe(returncode: int, stderr: str) -> Literal["ok", "unavailable", "error"]: - if returncode == 0: - return "ok" - return "unavailable" if any(m in stderr for m in _SYSTEMD_UNAVAILABLE) else "error" - - @pytest.fixture(scope="session") def systemd_manager() -> None: """Skip unless this user may start transient units on the system manager, as the product does. @@ -158,7 +143,7 @@ def systemd_manager() -> None: timeout=30, check=False, ) - verdict = _classify_systemd_probe(probe.returncode, probe.stderr) + verdict = classify_systemd_probe(probe.returncode, probe.stderr) reason = f"rc={probe.returncode} {probe.stderr.strip()[-500:]}" except FileNotFoundError as exc: verdict, reason = "unavailable", repr(exc) diff --git a/tests/systemd_probe.py b/tests/systemd_probe.py new file mode 100644 index 0000000..8ed7231 --- /dev/null +++ b/tests/systemd_probe.py @@ -0,0 +1,20 @@ +"""Classification of the `systemd-run` probe result (imported by conftest and its unit test).""" + +from __future__ import annotations + +from typing import Literal + +# Messages meaning "this user/host has no usable system manager", not "the probe broke". +_SYSTEMD_UNAVAILABLE = ( + "Interactive authentication required", + "Access denied", + "Failed to connect to bus", + "System has not been booted with systemd", + "Permission denied", +) + + +def classify_systemd_probe(returncode: int, stderr: str) -> Literal["ok", "unavailable", "error"]: + if returncode == 0: + return "ok" + return "unavailable" if any(m in stderr for m in _SYSTEMD_UNAVAILABLE) else "error" diff --git a/tests/test_systemd_probe_classify.py b/tests/test_systemd_probe_classify.py index 59c8bf3..07b4efa 100644 --- a/tests/test_systemd_probe_classify.py +++ b/tests/test_systemd_probe_classify.py @@ -2,7 +2,7 @@ import pytest -from conftest import _classify_systemd_probe +from systemd_probe import classify_systemd_probe DENIED = [ "Failed to start transient service unit: Interactive authentication required.", @@ -14,12 +14,12 @@ def test_ok() -> None: - assert _classify_systemd_probe(0, "") == "ok" + assert classify_systemd_probe(0, "") == "ok" @pytest.mark.parametrize("stderr", DENIED) def test_unavailable(stderr: str) -> None: - assert _classify_systemd_probe(1, stderr) == "unavailable" + assert classify_systemd_probe(1, stderr) == "unavailable" @pytest.mark.parametrize( @@ -27,4 +27,4 @@ def test_unavailable(stderr: str) -> None: ["Failed to start transient service unit: Resource temporarily unavailable (EAGAIN)", ""], ) def test_error(stderr: str) -> None: - assert _classify_systemd_probe(1, stderr) == "error" + assert classify_systemd_probe(1, stderr) == "error"