Skip to content

Add a wait_for_healthcheck method to DockerCompose #241

Description

@csikb

It's possible to check if the healthcheck passed trought the cli.
docker ps --format json will list all the services running and their corresponding healthchecks.

Proposed method signature: compose.wait_for_healthcheck(sec: int = 10)

High level description how it should work:
Iterate trough services check if healthcheck is healthy.
If not: wait and repeat until timeout.

testcontaners java and node has the option to wait for healthcheck. It would be nice if testcontainers-python would have the same funtionality

Activity

  1. rhoban13 commented on Jan 17, 2024

    @rhoban13
    Contributor

    Ideally this should work for even the DockerContainer as well.

  2. lukasz-matter commented on Sep 19, 2024

    @lukasz-matter

    A very simple implementation for a DockerContainer can look like this:

        with container:
            start_time = time.time()
            underlying_container = container.get_wrapped_container()
            while time.time() - start_time < 10:
                underlying_container.reload()
                if underlying_container.health == 'healthy':
                    print(f"Container is healthy!")
                    break
                time.sleep(.1)
            else:
                raise TimeoutError(
                    f"Container status is not healthy after 10s - it's {underlying_container.health}!"
                )
            yield container
  3. aligator commented on Apr 2, 2025

    @aligator

    .health didn't work for me somehow...

    I got it working with using the attributes directly.

    Could something like this wait_for_healthy be added to testcontainers-python?

    from testcontainers.core.container import DockerContainer
    from testcontainers.core.waiting_utils import wait_container_is_ready
    
    class UnhealthyError(Exception):
       pass
       
    @wait_container_is_ready(UnhealthyError)
    def wait_for_healthy(container: DockerContainer):
      if container._container is None:
        raise UnhealthyError
      
      container._container.reload()
      if container._container.attrs.get('State', {}).get('Health', {}).get('Status', 'unknown') != "healthy":
        raise UnhealthyError
    
    def create_rabbit_container() -> DockerContainer:
        container = (DockerContainer(image="rabbitmq:3")
            .with_env("RABBITMQ_DEFAULT_USER", "test")
            .with_env("RABBITMQ_DEFAULT_PASS", "test")
            .with_exposed_ports("5672")
            .with_kwargs(
                healthcheck={
                    "test": ["CMD", "rabbitmq-diagnostics", "-q", "ping"],
                    "interval": 30000000000,  # 30s in nanoseconds
                    "timeout": 5000000000,    # 5s in nanoseconds
                    "retries": 3,
                    "start_period": 60000000000  # 60s in nanoseconds
                }
            ))
    
        container.start()
    
        wait_for_healthy(container)
        
        return container
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions