diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 44af109..aeb2498 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,7 +8,7 @@ jobs: strategy: fail-fast: false matrix: - python: ['3.11', '3.12', '3.13', '3.14'] + python: ['3.11', '3.12', '3.13', '3.14', '3.14t'] runs-on: ubuntu-latest steps: - name: Checkout repository @@ -42,6 +42,7 @@ jobs: pystemd examples tests + e2e - name: Run isort uses: isort/isort-action@v1 diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml new file mode 100644 index 0000000..bf80a73 --- /dev/null +++ b/.github/workflows/e2e-tests.yml @@ -0,0 +1,86 @@ +name: E2E Tests + +on: + push: + branches: [ main ] + pull_request: + branches: [ main ] + +jobs: + e2e-tests: + name: Python ${{ matrix.python-version }} + runs-on: ubuntu-latest + + strategy: + matrix: + python-version: ['3.11', '3.12', '3.13', '3.14', '3.14t'] + + env: + PYTHON_VERSION: ${{ matrix.python-version }} + UNIT_NAME: pystemd-e2e-py${{ matrix.python-version }}.service + MACHINE_NAME: pystemd-test-py${{ matrix.python-version }} + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up latest Python + uses: actions/setup-python@v4 + with: + python-version: '3.14' + + - name: setup-mkosi + uses: systemd/mkosi@v26 + + - name: Install systemd-container + run: | + sudo apt-get update + sudo apt-get install -y systemd-container + + - name: Generate mkosi keys + run: | + sudo mkosi genkey + + - name: Build mkosi test image + run: | + sudo mkosi -E "$PYTHON_VERSION" build + + - name: Boot container + run: | + sudo systemd-run --unit "$UNIT_NAME" --same-dir \ + systemd-nspawn \ + --machine="$MACHINE_NAME" \ + --boot \ + --directory=pystemd-test \ + --bind-ro=${{ github.workspace }}/e2e:/opt/pystemd/e2e + + # Wait for container to be ready + for i in {1..30}; do + if sudo systemd-run --machine="$MACHINE_NAME" --wait --pipe /bin/true 2>/dev/null; then + echo "Container is ready" + break + fi + echo "Waiting for container to start... ($i/30)" + sleep 1 + done + sudo journalctl -u "$UNIT_NAME" + sleep 1 + sudo systemd-run --machine="$MACHINE_NAME" --wait --pipe /bin/echo 'hello world' || exit 1 + + - name: Run E2E tests + run: | + sudo systemd-run \ + --machine="$MACHINE_NAME" \ + --wait \ + --pipe \ + --setenv=PYSTEMD_E2E_CONTAINER=1\ + --property=PrivateTmp=true \ + -- \ + /opt/pystemd/venv/bin/pytest \ + -o cache_dir=/tmp/pytest_cache \ + /opt/pystemd/e2e/ -v + + - name: Stop container + if: always() + run: | + sudo machinectl terminate "$MACHINE_NAME" || true diff --git a/.gitignore b/.gitignore index 185921b..25eb30f 100644 --- a/.gitignore +++ b/.gitignore @@ -98,3 +98,10 @@ pystemd/RELEASE # not a fan of keeping uv lock files around uv.lock + +# mkosi generated files +pystemd-test/ +pystemd-test-*/ +.#pystemd-test*.lck +mkosi.key +mkosi.crt diff --git a/E2E_TESTING.md b/E2E_TESTING.md new file mode 100644 index 0000000..2f8d3ee --- /dev/null +++ b/E2E_TESTING.md @@ -0,0 +1,185 @@ +# pystemd E2E Testing Setup + +## Overview + +This setup provides comprehensive end-to-end testing for pystemd using mkosi and systemd-nspawn. Tests run in a real systemd environment inside a container to ensure accurate testing of systemd integration. + +## Quick Start + +```bash +# Install dependencies (one-time setup) +sudo dnf install mkosi systemd-container # Fedora/RHEL +# or +sudo apt install mkosi systemd-container # Debian/Ubuntu + +# generate keys if they dont exists +mkosi genkey + +# build container, there steps do not need to be run as root. +mkosi clean && mkosi build + +# Start container +UNIT_NAME=pystemd-e2e-local.service +MACHINE_NAME=pystemd-test-local +sudo systemd-run --unit "$UNIT_NAME" --same-dir \ + systemd-nspawn \ + --machine="$MACHINE_NAME" \ + --boot \ + --directory=pystemd-test \ + --bind-ro=`pwd`/e2e:/opt/pystemd/e2e + +# Run all E2E tests +sudo systemd-run \ + --machine="$MACHINE_NAME" \ + --wait \ + --pipe \ + --setenv=PYSTEMD_E2E_CONTAINER=1\ + --property=PrivateTmp=true \ + -- \ + /opt/pystemd/venv/bin/pytest \ + -o cache_dir=/tmp/pytest_cache \ + /opt/pystemd/e2e/ -v +``` + + +## How It Works + +### Manual Container Management + +The E2E testing workflow uses a straightforward approach: + +1. **Build the container image** using mkosi - this creates a Fedora environment with pystemd installed +2. **Boot the container** using systemd-nspawn as a background service +3. **Run tests inside the container** using `systemd-run --machine` +4. **Stop the container** when done + +The container runs with `--boot` which starts a full systemd init inside, providing a realistic systemd environment for testing. + +### Container Boot Command + +The container is booted as a systemd service using systemd-nspawn: + +```bash +UNIT_NAME=pystemd-e2e-local.service +MACHINE_NAME=pystemd-test + +sudo systemd-run --unit "$UNIT_NAME" --same-dir \ + systemd-nspawn \ + --machine="$MACHINE_NAME" \ + --boot \ + --directory=pystemd-test \ + --bind-ro=`pwd`/e2e:/opt/pystemd/e2e +``` + +Tests are then executed inside the container using: + +```bash +sudo systemd-run \ + --machine="$MACHINE_NAME" \ + --wait \ + --pipe \ + --setenv=PYSTEMD_E2E_CONTAINER=1 \ + --property=PrivateTmp=true \ + -- \ + /opt/pystemd/venv/bin/pytest \ + -o cache_dir=/tmp/pytest_cache \ + /opt/pystemd/e2e/ -v +``` + + +## Test Suite + +Tests are located in the `e2e/` directory: + +- **`test_pystemd_run.py`** - Tests for `pystemd.run()` +- **`test_manager.py`** - Tests for systemd Manager API +- **`test_unit.py`** - Tests for Unit operations +- **`test_transient_units.py`** - Tests for transient unit creation +- **`test_dbus.py`** - Tests for D-Bus connections + +## Adding New Tests + +Create a new test file in the `e2e/` directory: + +```python +# e2e/test_my_feature.py +import pystemd.run + +def test_my_feature(): + """Test my new feature""" + unit = pystemd.run([b'/bin/echo', b'hello'], wait=True) + assert unit.Service.ExecMainStatus == 0 +``` + +```bash +sudo systemd-run \ + --machine="$MACHINE_NAME" \ + --wait \ + --pipe \ + --setenv=PYSTEMD_E2E_CONTAINER=1 \ + --property=PrivateTmp=true \ + -- \ + /opt/pystemd/venv/bin/pytest \ + -o cache_dir=/tmp/pytest_cache \ + /opt/pystemd/e2e/test_my_feature.py -v +``` + + +And run it using + +## CI/CD Integration + +The GitHub Actions workflow (`.github/workflows/e2e-tests.yml`) automates E2E testing: + +**Triggers:** +- Pushes to `main` or `develop` branches +- Pull requests targeting `main` + +**Matrix Testing:** +- Tests across multiple Python versions: 3.11, 3.12, 3.13, 3.14, 3.14t (free-threaded) +- Each Python version runs in its own container instance + + +## Troubleshooting + +### Container fails to start +```bash +# Check if another container is running +sudo machinectl list + +# SSH into the container +sudo machinectl shell pystemd-test + +# Terminate stale container +sudo machinectl terminate pystemd-test +``` + +### Image not found +```bash +# Rebuild the mkosi image +sudo mkosi --force build +``` + +### Tests hang +```bash +# Check container status +sudo machinectl status pystemd-test + +# View container logs +sudo journalctl -M pystemd-test +``` + +### Building with a different Python version + +By default, the container is built with Python 3.14. To build with a different Python version, set the `PYTHON_VERSION` environment variable before building: + +```bash +# Build with Python 3.12 +mkosi clean && mkosi -E PYTHON_VERSION=3.12 build + +# Build with Python 3.14 free-threaded +mkosi clean && mkosi-E PYTHON_VERSION=3.14t build +``` + +The `-E` flag passes the `PYTHON_VERSION` environment variable to mkosi, which is then used by `mkosi.build.chroot` to install the specified Python version using `uv python install`. + diff --git a/e2e/test_dbus.py b/e2e/test_dbus.py new file mode 100644 index 0000000..c8386ee --- /dev/null +++ b/e2e/test_dbus.py @@ -0,0 +1,9 @@ +"""E2E tests for D-Bus integration""" + +from pystemd.dbuslib import DBus + + +def test_dbus_connection(): + """Test basic D-Bus connection""" + with DBus() as bus: + assert bus is not None diff --git a/e2e/test_manager.py b/e2e/test_manager.py new file mode 100644 index 0000000..13f5ac2 --- /dev/null +++ b/e2e/test_manager.py @@ -0,0 +1,48 @@ +"""E2E tests for pystemd Manager functionality""" + +from pystemd.systemd1 import Manager + + +def test_manager_version(): + """Test getting systemd version""" + with Manager() as manager: + version = manager.Manager.Version + assert version is not None + assert isinstance(version, bytes) + + +def test_manager_architecture(): + """Test getting system architecture""" + with Manager() as manager: + arch = manager.Manager.Architecture + assert arch is not None + assert isinstance(arch, bytes) + + +def test_list_units(): + """Test listing units""" + with Manager() as manager: + units = manager.Manager.ListUnits() + assert len(units) > 0 + # Each unit should be a tuple with multiple fields + assert isinstance(units[0], tuple) + + +def test_list_unit_files(): + """Test listing unit files""" + with Manager() as manager: + unit_files = manager.Manager.ListUnitFiles() + assert len(unit_files) > 0 + # Each should be (name, state) tuple + for name, state in unit_files: + assert isinstance(name, bytes) + assert isinstance(state, bytes) + + +def test_get_unit(): + """Test getting a unit by name""" + with Manager() as manager: + # Get a unit that should always exist + unit_path = manager.Manager.GetUnit(b"dbus.service") + assert unit_path is not None + assert isinstance(unit_path, bytes) diff --git a/e2e/test_pystemd_run.py b/e2e/test_pystemd_run.py new file mode 100644 index 0000000..2cc4d14 --- /dev/null +++ b/e2e/test_pystemd_run.py @@ -0,0 +1,125 @@ +"""E2E tests for pystemd.run functionality""" + +import os +import time + +import pytest + +import pystemd.run +from pystemd.exceptions import PystemdRunError + + +def test_simple_command(): + """Test running a simple command""" + unit = pystemd.run([b"/bin/true"], remain_after_exit=True) + assert unit is not None + assert unit.Service.ExecMainStatus == 0 + + +def test_command_with_args(): + """Test running command with arguments""" + unit = pystemd.run([b"/bin/sleep", b"1"], remain_after_exit=True, wait=True) + assert unit is not None + assert unit.Service.MainPID == 0 # Should have exited + assert unit.Service.ExecMainStatus == 0 + + +def test_command_with_env(): + """Test running command with environment variables""" + unit = pystemd.run( + [b"/bin/sh", b"-c", b'test "$MY_VAR" = "test_value"'], + env={b"MY_VAR": b"test_value"}, + remain_after_exit=True, + wait=True, + ) + assert unit.Service.ExecMainStatus == 0 + + +def test_command_with_cwd(): + """Test running command with custom working directory""" + unit = pystemd.run( + [b"/bin/pwd"], + cwd=b"/tmp", + remain_after_exit=True, + wait=True, + ) + assert unit.Service.ExecMainStatus == 0 + + +def test_command_with_user(): + """Test running command as different user (requires root)""" + if os.geteuid() != 0: + pytest.skip("Requires root privileges") + + unit = pystemd.run( + [b"/bin/id", b"-u"], + user=b"nobody", + remain_after_exit=True, + wait=True, + ) + assert unit.Service.ExecMainStatus == 0 + + +def test_wait_for_activation(): + """Test wait_for_activation parameter""" + unit = pystemd.run( + [b"/bin/sleep", b"10"], + wait_for_activation=True, + remain_after_exit=True, + ) + # Should return quickly after activation + assert unit.Service.MainPID != 0 # Still running + # Clean up + unit.Unit.Stop(b"replace") + + +def test_raise_on_fail(): + """Test raise_on_fail parameter""" + with pytest.raises(PystemdRunError): + pystemd.run( + [b"/bin/false"], + wait=True, + raise_on_fail=True, + ) + + +def test_runtime_max_sec(): + """Test runtime_max_sec timeout""" + start = time.time() + unit = pystemd.run( + [b"/bin/sleep", b"100"], + runtime_max_sec=2, + remain_after_exit=True, + wait=True, + ) + elapsed = time.time() - start + # Should have been killed after ~2 seconds + assert elapsed < 10 + assert unit.Service.ExecMainStatus != 0 # Should have non-zero exit + + +def test_service_type_oneshot(): + """Test service_type parameter""" + unit = pystemd.run( + [b"/bin/true"], + service_type=b"oneshot", + remain_after_exit=True, + wait=True, + ) + assert unit.Service.ExecMainStatus == 0 + + +def test_stop_cmd(): + """Test stop_cmd parameter""" + unit = pystemd.run( + [b"/bin/sleep", b"100"], + stop_cmd=[b"/bin/echo", b"stopping"], + remain_after_exit=True, + ) + # Give it a moment to start + time.sleep(1) + assert unit.Service.MainPID != 0 + # Stop it + unit.Unit.Stop(b"replace") + time.sleep(1) + assert unit.Service.MainPID == 0 diff --git a/e2e/test_transient_units.py b/e2e/test_transient_units.py new file mode 100644 index 0000000..8b034aa --- /dev/null +++ b/e2e/test_transient_units.py @@ -0,0 +1,57 @@ +"""E2E tests for creating and managing transient units""" + +import os +import time + +import pytest + +import pystemd.run +from pystemd.dbuslib import DBus +from pystemd.systemd1 import Manager, Unit + + +def test_create_transient_unit_via_manager(): + """Test creating transient unit via Manager API""" + if os.geteuid() != 0: + pytest.skip("Requires root privileges") + + unit_name = f"test-{int(time.time())}.service".encode() + + unit_properties = { + b"Description": b"E2E test transient unit", + b"ExecStart": [(b"/bin/sleep", (b"/bin/sleep", b"5"), False)], + b"RemainAfterExit": True, + } + + with DBus() as bus, Manager(bus=bus) as manager: + job_path = manager.Manager.StartTransientUnit( + unit_name, b"fail", unit_properties + ) + assert job_path is not None + + # Get the unit and check it + with Unit(unit_name, bus=bus) as unit: + time.sleep(1) + assert unit.Service.MainPID != 0 + # Clean up + unit.Unit.Stop(b"replace") + + +def test_transient_unit_with_dependencies(): + """Test creating transient unit with dependencies""" + if os.geteuid() != 0: + pytest.skip("Requires root privileges") + + unit = pystemd.run( + [b"/bin/sleep", b"5"], + extra={ + b"After": [b"network.target"], + }, + remain_after_exit=True, + ) + + # Check dependencies are set + assert b"network.target" in unit.Unit.After + + # Cleanup + unit.Unit.Stop(b"replace") diff --git a/e2e/test_unit.py b/e2e/test_unit.py new file mode 100644 index 0000000..e9784b9 --- /dev/null +++ b/e2e/test_unit.py @@ -0,0 +1,60 @@ +"""E2E tests for pystemd Unit functionality""" + +import os +import time + +import pytest + +import pystemd.run + + +def test_unit_start_stop(): + """Test starting and stopping a transient unit""" + if os.geteuid() != 0: + pytest.skip("Requires root privileges") + + # Create a transient unit + unit = pystemd.run( + [b"/bin/sleep", b"60"], + remain_after_exit=True, + ) + + # Wait a moment for it to start + time.sleep(1) + + # Check it's running + assert unit.Service.MainPID != 0 + assert unit.Unit.ActiveState == b"active" + + # Stop it + unit.Unit.Stop(b"replace") + + # Wait for it to stop + time.sleep(1) + assert unit.Service.MainPID == 0 + + +def test_unit_restart(): + """Test restarting a unit""" + if os.geteuid() != 0: + pytest.skip("Requires root privileges") + + unit = pystemd.run( + [b"/bin/sleep", b"60"], + remain_after_exit=True, + ) + + time.sleep(1) + first_pid = unit.Service.MainPID + assert first_pid != 0 + + # Restart + unit.Unit.Restart(b"replace") + time.sleep(1) + + second_pid = unit.Service.MainPID + assert second_pid != 0 + assert second_pid != first_pid # Should be different PID + + # Cleanup + unit.Unit.Stop(b"replace") diff --git a/mkosi.build.chroot b/mkosi.build.chroot new file mode 100755 index 0000000..25bfd98 --- /dev/null +++ b/mkosi.build.chroot @@ -0,0 +1,29 @@ +#!/bin/bash +# mkosi build script - builds and installs pystemd in the test image + +set -ex + +PYTHON_VERSION=${PYTHON_VERSION:-3.14} +PYSTEMD_ROOT=/opt/pystemd +TEST_USER=${TEST_USER:-pystemd-test-user} + +export UV_PYTHON_INSTALL_DIR="$PYSTEMD_ROOT/uvpython" +export UV_PROJECT_ENVIRONMENT="$PYSTEMD_ROOT/venv" + +# Create test user for running tests as non-root +useradd -m -s /bin/bash "$TEST_USER" + +# Install Python and sync dependencies +uv python install "$PYTHON_VERSION" +uv sync --python "$PYTHON_VERSION" --all-extras + +# Install pystemd as non-editable +"$UV_PROJECT_ENVIRONMENT/bin/python3" -m ensurepip +"$UV_PROJECT_ENVIRONMENT/bin/python3" -m pip install . --no-deps + +# Copy to destination image +mkdir -p "$DESTDIR/opt" +cp -r "$PYSTEMD_ROOT" "$DESTDIR/opt/" + +# Create mount point for e2e tests (will be bind-mounted at runtime) +mkdir -p "$DESTDIR$PYSTEMD_ROOT/e2e" diff --git a/mkosi.conf b/mkosi.conf new file mode 100644 index 0000000..4ff25a1 --- /dev/null +++ b/mkosi.conf @@ -0,0 +1,54 @@ +# mkosi configuration for pystemd E2E testing +# This creates a Fedora image with systemd and Python for testing + +[Distribution] +Distribution=fedora +Release=43 + +[Output] +ImageId=pystemd-test +Format=directory + +[Content] +Ssh=no +Autologin=yes +RootPassword=root +Locale=en_US.UTF-8 +Timezone=UTC +Hostname=pystemd-test +Packages= + systemd + systemd-container + systemd-devel + systemd-resolved + dbus-devel + gcc + libxml2-devel + libxslt-devel + pkgconf-pkg-config + openssl + openssh-server + +# Install build dependencies +BuildPackages= + rpm + uv + +# Build script will be in mkosi.build +WithDocs=no + +[Build] +# Build and install pystemd from source +BuildSources=. + +# Enable network access during build for pip and uv +WithNetwork=yes + +[Config] +# Pass PYTHON_VERSION from host environment +PassEnvironment=PYTHON_VERSION + +[Runtime] +# Run in ephemeral mode - changes are discarded after shutdown +Ephemeral=yes +VSock=yes diff --git a/pyproject.toml b/pyproject.toml index a45ae76..25a4c1c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,9 +29,8 @@ classifiers = [ "Programming Language :: Python :: 3.14", "Development Status :: 5 - Production/Stable", "Topic :: Utilities", - "License :: OSI Approved :: GNU Lesser General Public License v2 or later (LGPLv2+)", ] -license = { text = "LGPL-2.1+" } +license = "LGPL-2.1+" keywords = ["systemd", "linux", "dbus"] [project.optional-dependencies] diff --git a/tests/test_futures.py b/tests/test_futures.py index a0f25b8..15b208e 100644 --- a/tests/test_futures.py +++ b/tests/test_futures.py @@ -69,9 +69,18 @@ def test_enter( @patch.object(pystemd.futures, "enter_unit", autospec=True) @patch.object(pystemd.futures, "TransientUnitContext", autospec=True) -@patch("pystemd.utils.random_unit_name", autospec=True, return_value="pystemd-future-test.service") +@patch( + "pystemd.utils.random_unit_name", + autospec=True, + return_value="pystemd-future-test.service", +) class TestTransientUnitProcess(unittest.TestCase): - def test_pre_run(self, random_unit_name: MagicMock, TransientUnitContext: MagicMock, enter_unit: Mock): + def test_pre_run( + self, + random_unit_name: MagicMock, + TransientUnitContext: MagicMock, + enter_unit: Mock, + ): properties = {b"foo": b"bar"} target = Mock() p = pystemd.futures.TransientUnitProcess(properties=properties, target=target)