Infrastructure Testing

Checking Docker container health and Linux services over SSH with Pytest and Paramiko.

UI and API tests tell you whether the application behaves correctly. They cannot tell you why it stopped working when a container is crash-looping, a system service never started, or a port is not listening. Infrastructure tests check the environment itself. This article builds a small suite with Pytest and Paramiko that connects to a Linux host over SSH and verifies Docker container health and system services.

What We Will Test

The target is the host that runs Summit Trust Bank, my sample banking application (a web UI, a REST API, and a Postgres database, each running in Docker). The suite answers questions like these:

  • Are the app and database containers running, and do they report healthy?
  • Is Postgres accepting connections, and does the API health endpoint respond?
  • Are the required Linux services (docker, ssh, cron) active and enabled on boot?
  • Are there any failed systemd units, and are the expected ports listening?

Why Pytest and Paramiko?

  • Pytest brings fixtures, parametrization, markers, and readable failure output. You get a real test runner and JUnit reporting for CI without extra tooling.
  • Paramiko is a pure-Python SSH client. It runs commands on remote hosts, so the tests need no agent installed on the target. It works the same on a developer laptop and in a CI job.

As in my Playwright article, I keep the reusable framework code apart from the tests. Tests state the expectation, and the framework knows how to run commands and parse results.

Project Setup

pip install pytest paramiko
infra-tests/
  infra/                     # the framework
    ssh.py                   # Paramiko wrapper
    docker.py                # container helpers
    systemd.py               # service helpers
  tests/                     # the tests
    test_docker_health.py
    test_system_services.py
  conftest.py                # fixtures
  pytest.ini
  requirements.txt

A small SSH wrapper

The wrapper opens one connection, runs commands, and returns a result object with the exit code and output. Tests then assert on that result instead of juggling Paramiko streams.

# infra/ssh.py
from __future__ import annotations

from dataclasses import dataclass

import paramiko


@dataclass(frozen=True)
class CommandResult:
    command: str
    exit_code: int
    stdout: str
    stderr: str

    @property
    def ok(self) -> bool:
        return self.exit_code == 0


class SshHost:
    """A thin wrapper around a Paramiko SSH connection."""

    def __init__(self, hostname: str, username: str, key_filename: str, port: int = 22):
        self._client = paramiko.SSHClient()
        # Trust only hosts already listed in known_hosts; unknown hosts are rejected
        self._client.load_system_host_keys()
        self._client.set_missing_host_key_policy(paramiko.RejectPolicy())
        self._client.connect(
            hostname,
            port=port,
            username=username,
            key_filename=key_filename,
            timeout=10,
        )

    def run(self, command: str, timeout: float = 30) -> CommandResult:
        _, stdout, stderr = self._client.exec_command(command, timeout=timeout)
        out = stdout.read().decode().strip()
        err = stderr.read().decode().strip()
        exit_code = stdout.channel.recv_exit_status()
        return CommandResult(command, exit_code, out, err)

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> SshHost:
        return self

    def __exit__(self, *exc_info) -> None:
        self.close()

Two details are worth calling out:

  • Host key checking stays on. It is tempting to use AutoAddPolicy to make connection errors go away, but that accepts any server and defeats the purpose of SSH host verification. Add the host to known_hosts instead.
  • Read the output before asking for the exit status. Calling recv_exit_status() first can hang when a command produces a lot of output. These checks print little, so reading stdout then stderr is fine. For commands with large output on both streams, read them with a channel loop.

Fixtures

A session-scoped fixture opens the SSH connection once for the whole run. Connection details come from environment variables, so no host names or keys live in the repository.

# conftest.py
import os

import pytest

from infra.docker import DockerHost
from infra.ssh import SshHost
from infra.systemd import Systemd


def _env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        pytest.fail(f"Set the {name} environment variable", pytrace=False)
    return value


@pytest.fixture(scope="session")
def ssh():
    with SshHost(
        hostname=_env("INFRA_HOST"),
        username=os.environ.get("INFRA_USER", "deploy"),
        key_filename=_env("INFRA_SSH_KEY"),
    ) as host:
        yield host


@pytest.fixture(scope="session")
def docker(ssh):
    return DockerHost(ssh)


@pytest.fixture(scope="session")
def systemd(ssh):
    return Systemd(ssh)

Checking Docker Container Health

Docker reports container state through docker inspect, which prints JSON. If the image or Compose file defines a healthcheck, the state also includes a Health section whose status is starting, healthy, or unhealthy. Summit Trust Bank defines health checks for both its app and database containers.

The Docker helper

Right after a deployment a container can legitimately be starting, so the helper can also poll until the container becomes healthy or a timeout expires.

# infra/docker.py
from __future__ import annotations

import json
import shlex
import time

from infra.ssh import SshHost


class DockerHost:
    def __init__(self, ssh: SshHost):
        self._ssh = ssh

    def inspect(self, container: str) -> dict:
        result = self._ssh.run(f"docker inspect {shlex.quote(container)}")
        assert result.ok, f"docker inspect {container} failed: {result.stderr}"
        return json.loads(result.stdout)[0]

    def health_status(self, container: str) -> str:
        state = self.inspect(container)["State"]
        return state.get("Health", {}).get("Status", "none")

    def wait_until_healthy(self, container: str, timeout: float = 60, interval: float = 2) -> None:
        deadline = time.monotonic() + timeout
        status = self.health_status(container)
        while status != "healthy" and time.monotonic() < deadline:
            time.sleep(interval)
            status = self.health_status(container)
        assert status == "healthy", f"{container} is '{status}' after {timeout}s, expected 'healthy'"

Container names come from Docker Compose. With the project name summit-trust-bank, the services app and db become summit-trust-bank-app-1 and summit-trust-bank-db-1. Always pass names through shlex.quote when they are interpolated into a shell command.

The container tests

# tests/test_docker_health.py
import pytest

pytestmark = pytest.mark.infra

CONTAINERS = ["summit-trust-bank-app-1", "summit-trust-bank-db-1"]


@pytest.mark.parametrize("container", CONTAINERS)
def test_container_is_running(docker, container):
    state = docker.inspect(container)["State"]
    assert state["Status"] == "running", f"{container} is {state['Status']}"
    assert not state["OOMKilled"], f"{container} was killed for running out of memory"


@pytest.mark.parametrize("container", CONTAINERS)
def test_container_becomes_healthy(docker, container):
    docker.wait_until_healthy(container, timeout=60)


def test_no_unhealthy_containers(ssh):
    result = ssh.run("docker ps --filter health=unhealthy --format '{{.Names}}'")
    assert result.ok, result.stderr
    assert result.stdout == "", f"Unhealthy containers: {result.stdout.splitlines()}"

The first two tests are parametrized, so each container gets its own pass or fail line. The last test is a catch-all that finds any unhealthy container on the host, including ones you did not list.

Going one level deeper: data and API health

A container can be running while the software inside is not usable. These tests run checks that follow the data path: first the database, then the API that sits in front of it.

def test_postgres_accepts_connections(ssh):
    result = ssh.run("docker exec summit-trust-bank-db-1 pg_isready -U bank -d bank")
    assert result.ok, f"pg_isready failed: {result.stdout} {result.stderr}"


def test_api_health_endpoint(ssh):
    result = ssh.run("curl --fail --silent --max-time 5 http://localhost:8000/api/health")
    assert result.ok, f"Health endpoint failed (exit {result.exit_code}): {result.stderr}"

pg_isready ships with Postgres and exits 0 when the server accepts connections. The curl --fail flag makes curl return a non-zero exit code for HTTP errors, so the exit code alone tells you whether the endpoint is healthy.

Checking Linux System Services

On a systemd-based host, systemctl reports service state. systemctl is-active prints active (or inactive, failed, activating) and systemctl is-enabled reports whether the service starts on boot. Services that are running now but not enabled are a common source of "it worked until the reboot" outages, so I check both.

The systemd helper

The helper also fetches recent journal entries, so a failing test shows why the service is down without anyone having to log in.

# infra/systemd.py
from __future__ import annotations

import shlex

from infra.ssh import SshHost


class Systemd:
    def __init__(self, ssh: SshHost):
        self._ssh = ssh

    def state(self, unit: str) -> str:
        result = self._ssh.run(f"systemctl is-active {shlex.quote(unit)}")
        return result.stdout or "unknown"

    def is_enabled(self, unit: str) -> bool:
        result = self._ssh.run(f"systemctl is-enabled {shlex.quote(unit)}")
        return result.stdout == "enabled"

    def recent_logs(self, unit: str, lines: int = 20) -> str:
        result = self._ssh.run(f"journalctl -u {shlex.quote(unit)} -n {lines} --no-pager")
        return result.stdout or result.stderr

Reading the journal requires the SSH user to be root or a member of the systemd-journal group.

The service tests

# tests/test_system_services.py
import pytest

pytestmark = pytest.mark.infra

# Unit names vary by distribution (for example, ssh on Debian/Ubuntu and sshd on RHEL)
REQUIRED_SERVICES = ["docker", "ssh", "cron"]


@pytest.mark.parametrize("service", REQUIRED_SERVICES)
def test_service_is_active(systemd, service):
    state = systemd.state(service)
    assert state == "active", f"{service} is {state}\n{systemd.recent_logs(service)}"


@pytest.mark.parametrize("service", REQUIRED_SERVICES)
def test_service_starts_on_boot(systemd, service):
    assert systemd.is_enabled(service), f"{service} is not enabled to start on boot"


def test_no_failed_units(ssh):
    result = ssh.run("systemctl --failed --no-legend --plain")
    assert result.stdout == "", f"Failed units:\n{result.stdout}"


@pytest.mark.parametrize("port", [8000, 5433])
def test_port_is_listening(ssh, port):
    result = ssh.run(f"ss --listening --tcp --numeric --no-header 'sport = :{port}'")
    assert result.ok and result.stdout, f"Nothing is listening on TCP port {port}"

Ports 8000 and 5433 are the host ports Docker publishes for the app and the database. If the cron service were stopped, the failure output would look like this:

FAILED tests/test_system_services.py::test_service_is_active[cron] - AssertionError: cron is inactive
Jul 01 02:14:09 qa-host systemd[1]: Stopped Regular background program processing daemon.

The second line is illustrative journal output, shown here to demonstrate the kind of detail the assertion message carries.

Configuring and Running the Suite

A custom infra marker lets you run these slow, network-bound tests separately from fast unit tests. Registering the marker in pytest.ini avoids warnings, and the JUnit report feeds CI dashboards.

# pytest.ini
[pytest]
testpaths = tests
markers =
    infra: tests that connect to a remote host over SSH
addopts = -ra --junitxml=results/junit.xml
export INFRA_HOST=qa-host.example.com
export INFRA_USER=deploy
export INFRA_SSH_KEY=~/.ssh/id_ed25519

pytest -m infra

Running it in CI

In GitLab CI, store the private key and the host's known_hosts entry as masked or protected File variables. GitLab writes a File variable to a temporary file and sets the variable to its path, which matches what INFRA_SSH_KEY expects. This is a sketch, so adjust it for your runners.

infra-tests:
  stage: verify
  image: python:3.12-slim
  variables:
    INFRA_HOST: qa-host.example.com
    INFRA_USER: deploy
  before_script:
    - pip install -r requirements.txt
    - mkdir -p ~/.ssh
    - cp "$INFRA_KNOWN_HOSTS" ~/.ssh/known_hosts
  script:
    - pytest -m infra
  artifacts:
    when: always
    reports:
      junit: results/junit.xml

Running this job after each deployment gives you an automated smoke test of the environment, and it fails before the UI and API suites waste time against a broken host.

Security Considerations

  • Use a dedicated, least-privilege user. Membership in the docker group is effectively root access on the host. If the checks only need read access, consider a restricted user or a read-only Docker socket proxy.
  • Authenticate with keys, not passwords, and keep private keys in your CI secret store rather than in the repository.
  • Keep host key verification enabled and distribute known_hosts entries deliberately.
  • Quote anything interpolated into a command. Use shlex.quote for names and keep checks read-only.

Takeaways

  • Infrastructure tests catch environment problems that UI and API tests report only as confusing failures.
  • Paramiko plus a thin wrapper turns SSH commands into plain Python results that Pytest can assert on.
  • Parametrize over containers, services, and ports so each failure points at one specific thing.
  • Put diagnostics in the assertion message, such as journal logs or the last health status, so failures are actionable.
  • Keep the framework (SSH, Docker, systemd helpers) separate from the tests, and keep credentials and host details out of the code.

For the full reference, see the Paramiko documentation and the Pytest documentation.

Back to Articles