How to use aiodocker - 10 common examples

params=httpize(params),
                headers=headers,
                data=data,
                timeout=timeout,
                chunked=chunked,
            )
        except asyncio.TimeoutError:
            raise
        if (response.status // 100) in [4, 5]:
            what = await response.read()
            content_type = response.headers.get("content-type", "")
            response.close()
            if content_type == "application/json":
                raise DockerError(response.status, json.loads(what.decode("utf8")))
            else:
                raise DockerError(response.status, {"message": what.decode("utf8")})
        return response

async def test_connect_envvar(monkeypatch):
    monkeypatch.setenv("DOCKER_HOST", "unix:///var/run/does-not-exist-docker.sock")
    docker = Docker()
    assert isinstance(docker.connector, aiohttp.connector.UnixConnector)
    assert docker.docker_host == "unix://localhost"
    with pytest.raises(aiohttp.ClientOSError):
        await docker.containers.list()
    await docker.close()

    monkeypatch.setenv("DOCKER_HOST", "http://localhost:9999")
    docker = Docker()
    assert isinstance(docker.connector, aiohttp.TCPConnector)
    assert docker.docker_host == "http://localhost:9999"
    with pytest.raises(aiohttp.ClientOSError):
        await docker.containers.list()
    await docker.close()

def test_wait_timeout():
    docker = Docker()

    yield from docker.pull("debian:jessie")

    config = {
        "Cmd":["sh"],
        "Image":"debian:jessie",
         "AttachStdin":True,
         "AttachStdout":True,
         "AttachStderr":True,
         "Tty":False,
         "OpenStdin":True,
         "StdinOnce":True,
    }

    container = yield from docker.containers.create_or_replace(config=config, name='testing')
    yield from container.start()

def test_stdio_stdin():
    docker = Docker()

    yield from docker.pull("debian:jessie")

    config = {
        "Cmd":["sh"],
        "Image":"debian:jessie",
         "AttachStdin":True,
         "AttachStdout":True,
         "AttachStderr":True,
         "Tty":False,
         "OpenStdin":True,
         "StdinOnce":True,
    }


    container = yield from docker.containers.create_or_replace(config=config, name='testing')

def test_image_push():
    docker = Docker()
    yield from docker.pull("debian")
    yield from docker.images.tag("debian", repo=os.environ['DOCKER_REGISTRY']+"/debian")
    #push_results = yield from docker.images.push(os.environ['DOCKER_REGISTRY']+"/debian")

def test_events():
    docker = Docker()
    queue = yield from docker.events.query()

    yield from docker.pull("debian:jessie")

    config = {
        "Cmd":["sh"],
        "Image":"debian:jessie",
    }

    container = yield from docker.containers.create_or_replace(config=config, name='testing')
    #print("put archive response:", result)
    yield from container.start()

    i = yield from queue.__aiter__()
    while True:
        try:

async def test_auto_pull_digest_when_missing(agent, mocker):
    behavior = AutoPullBehavior.DIGEST
    inspect_mock = AsyncMock(
        side_effect=DockerError(status=404,
                                data={'message': 'Simulated missing image'}))
    mocker.patch.object(agent.docker.images, 'inspect', new=inspect_mock)
    pull = await agent.check_image(imgref, query_digest, behavior)
    assert pull
    inspect_mock.assert_called_with(imgref.canonical)

async def test_run_container_with_missing_image(docker):
    name = "python:latest"
    try:
        await docker.images.delete(name)
    except DockerError as e:
        if e.status == 404:
            pass  # already missing, pass
        else:
            raise

    # should automatically pull the image
    container = await docker.containers.run(
        config={"Cmd": ["-c", "print('hello')"], "Entrypoint": "python", "Image": name}
    )

    await _validate_hello(container)

def test_clean_mapping():
    dirty_dict = {"a": None, "b": {}, "c": [], "d": 1}
    clean_dict = {"b": {}, "c": [], "d": 1}
    result = utils.clean_map(dirty_dict)
    assert result == clean_dict

try:
        yield kernel_info
    finally:
        if kernel_info['id'] in agent.container_registry:
            # Container id may be changed (e.g. restarting kernel), so we
            # should not rely on the initial value of the container_id.
            container_info = agent.container_registry[kernel_info['id']]
            container_id = container_info['container_id']
        else:
            # If fallback to initial container_id if kernel is deleted.
            container_id = kernel_info['container_id']
        try:
            container = docker.containers.container(container_id)
            cinfo = await container.show() if container else None
        except aiodocker.exceptions.DockerError:
            cinfo = None
        if cinfo and cinfo['State']['Status'] != 'removing':
            await container.delete(force=True)