gns3-server/tests/compute/docker/test_docker_vm.py

# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import asyncio
import pytest
import uuid
import sys
import os
from tests.utils import asyncio_patch, AsyncioMagicMock

from gns3server.ubridge.ubridge_error import UbridgeNamespaceError
from gns3server.compute.docker.docker_vm import DockerVM
from gns3server.compute.docker.docker_error import *
from gns3server.compute.docker import Docker
from gns3server.utils.get_resource import get_resource


from unittest.mock import patch, MagicMock, PropertyMock, call
from gns3server.config import Config


@pytest.fixture()
def manager(port_manager):
    m = Docker.instance()
    m.port_manager = port_manager
    return m


@pytest.fixture(scope="function")
def vm(project, manager):
    vm = DockerVM("test", str(uuid.uuid4()), project, manager, "ubuntu:latest")
    vm._cid = "e90e34656842"
    vm.allocate_aux = False
    return vm


def test_json(vm, project):
    assert vm.__json__() == {
        'container_id': 'e90e34656842',
        'image': 'ubuntu:latest',
        'name': 'test',
        'project_id': project.id,
        'node_id': vm.id,
        'adapters': 1,
        'console': vm.console,
        'console_type': 'telnet',
        'console_resolution': '1024x768',
        'console_http_port': 80,
        'console_http_path': '/',
        'aux': vm.aux,
        'start_command': vm.start_command,
        'environment': vm.environment,
        'node_directory': vm.working_dir,
        'status': 'stopped'
    }


def test_start_command(vm):

    vm.start_command = "hello"
    assert vm.start_command == "hello"
    vm.start_command = " "
    assert vm.start_command is None


def test_create(loop, project, manager):

    response = {
        "Id": "e90e34656806",
        "Warnings": []
    }
    with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "ubuntu"}]) as mock_list_images:
        with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
            vm = DockerVM("test", str(uuid.uuid4()), project, manager, "ubuntu:latest")
            loop.run_until_complete(asyncio.async(vm.create()))
            mock.assert_called_with("POST", "containers/create", data={
                "Tty": True,
                "OpenStdin": True,
                "StdinOnce": False,
                "HostConfig":
                    {
                        "CapAdd": ["ALL"],
                        "Binds": [
                            "{}:/gns3:ro".format(get_resource("compute/docker/resources")),
                            "{}:/gns3volumes/etc/network:rw".format(os.path.join(vm.working_dir, "etc", "network"))
                        ],
                        "Privileged": True
                    },
                "Volumes": {},
                "NetworkDisabled": True,
                "Name": "test",
                "Hostname": "test",
                "Image": "ubuntu:latest",
                "Env": [
                    "container=docker",
                    "GNS3_MAX_ETHERNET=eth0",
                    "GNS3_VOLUMES=/etc/network"
                    ],
                "Entrypoint": ["/gns3/init.sh"],
                "Cmd": ["/bin/sh"]
            })
        assert vm._cid == "e90e34656806"


def test_create_with_tag(loop, project, manager):

    response = {
        "Id": "e90e34656806",
        "Warnings": []
    }
    with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "ubuntu"}]) as mock_list_images:
        with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
            vm = DockerVM("test", str(uuid.uuid4()), project, manager, "ubuntu:16.04")
            loop.run_until_complete(asyncio.async(vm.create()))
            mock.assert_called_with("POST", "containers/create", data={
                "Tty": True,
                "OpenStdin": True,
                "StdinOnce": False,
                "HostConfig":
                    {
                        "CapAdd": ["ALL"],
                        "Binds": [
                            "{}:/gns3:ro".format(get_resource("compute/docker/resources")),
                            "{}:/gns3volumes/etc/network:rw".format(os.path.join(vm.working_dir, "etc", "network"))
                        ],
                        "Privileged": True
                    },
                "Volumes": {},
                "NetworkDisabled": True,
                "Name": "test",
                "Hostname": "test",
                "Image": "ubuntu:16.04",
                "Env": [
                    "container=docker",
                    "GNS3_MAX_ETHERNET=eth0",
                    "GNS3_VOLUMES=/etc/network"
                    ],
                "Entrypoint": ["/gns3/init.sh"],
                "Cmd": ["/bin/sh"]
            })
        assert vm._cid == "e90e34656806"


def test_create_vnc(loop, project, manager):

    response = {
        "Id": "e90e34656806",
        "Warnings": []
    }

    with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "ubuntu"}]) as mock_list_images:
        with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
            vm = DockerVM("test", str(uuid.uuid4()), project, manager, "ubuntu", console_type="vnc", console=5900)
            vm._start_vnc = MagicMock()
            vm._display = 42
            loop.run_until_complete(asyncio.async(vm.create()))
            mock.assert_called_with("POST", "containers/create", data={
                "Tty": True,
                "OpenStdin": True,
                "StdinOnce": False,
                "HostConfig":
                    {
                        "CapAdd": ["ALL"],
                        "Binds": [
                            "{}:/gns3:ro".format(get_resource("compute/docker/resources")),
                            "{}:/gns3volumes/etc/network:rw".format(os.path.join(vm.working_dir, "etc", "network")),
                            '/tmp/.X11-unix/:/tmp/.X11-unix/'
                        ],
                        "Privileged": True
                    },
                "Volumes": {},
                "NetworkDisabled": True,
                "Name": "test",
                "Hostname": "test",
                "Image": "ubuntu:latest",
                "Env": [
                    "container=docker",
                    "GNS3_MAX_ETHERNET=eth0",
                    "GNS3_VOLUMES=/etc/network",
                    "QT_GRAPHICSSYSTEM=native",
                    "DISPLAY=:42"
                    ],
                "Entrypoint": ["/gns3/init.sh"],
                "Cmd": ["/bin/sh"]
            })
        assert vm._start_vnc.called
        assert vm._cid == "e90e34656806"
        assert vm._console_type == "vnc"


def test_create_start_cmd(loop, project, manager):

    response = {
        "Id": "e90e34656806",
        "Warnings": []
    }
    with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "ubuntu"}]) as mock_list_images:
        with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
            vm = DockerVM("test", str(uuid.uuid4()), project, manager, "ubuntu:latest")
            vm._start_command = "/bin/ls"
            loop.run_until_complete(asyncio.async(vm.create()))
            mock.assert_called_with("POST", "containers/create", data={
                "Tty": True,
                "OpenStdin": True,
                "StdinOnce": False,
                "HostConfig":
                    {
                        "CapAdd": ["ALL"],
                        "Binds": [
                            "{}:/gns3:ro".format(get_resource("compute/docker/resources")),
                            "{}:/gns3volumes/etc/network:rw".format(os.path.join(vm.working_dir, "etc", "network"))
                        ],
                        "Privileged": True
                    },
                "Volumes": {},
                "Entrypoint": ["/gns3/init.sh"],
                "Cmd": ["/bin/ls"],
                "NetworkDisabled": True,
                "Name": "test",
                "Hostname": "test",
                "Image": "ubuntu:latest",
                "Env": [
                    "container=docker",
                    "GNS3_MAX_ETHERNET=eth0",
                    "GNS3_VOLUMES=/etc/network"
                    ]
            })
        assert vm._cid == "e90e34656806"


def test_create_environment(loop, project, manager):

    response = {
        "Id": "e90e34656806",
        "Warnings": []
    }
    with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "ubuntu"}]) as mock_list_images:
        with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
            vm = DockerVM("test", str(uuid.uuid4()), project, manager, "ubuntu")
            vm.environment = "YES=1\nNO=0"
            loop.run_until_complete(asyncio.async(vm.create()))
            mock.assert_called_with("POST", "containers/create", data={
                "Tty": True,
                "OpenStdin": True,
                "StdinOnce": False,
                "HostConfig":
                    {
                        "CapAdd": ["ALL"],
                        "Binds": [
                            "{}:/gns3:ro".format(get_resource("compute/docker/resources")),
                            "{}:/gns3volumes/etc/network:rw".format(os.path.join(vm.working_dir, "etc", "network"))
                        ],
                        "Privileged": True
                    },
                "Env": [
                    "container=docker",
                    "GNS3_MAX_ETHERNET=eth0",
                    "GNS3_VOLUMES=/etc/network",
                    "YES=1",
                    "NO=0"
                ],
                "Volumes": {},
                "NetworkDisabled": True,
                "Name": "test",
                "Hostname": "test",
                "Image": "ubuntu:latest",
                "Entrypoint": ["/gns3/init.sh"],
                "Cmd": ["/bin/sh"]
            })
        assert vm._cid == "e90e34656806"


def test_create_image_not_available(loop, project, manager):

    call = 0

    @asyncio.coroutine
    def information():
        nonlocal call
        if call == 0:
            call += 1
            raise DockerHttp404Error("missing")
        else:
            return {}

    response = {
        "Id": "e90e34656806",
        "Warnings": []
    }

    vm = DockerVM("test", str(uuid.uuid4()), project, manager, "ubuntu")
    vm._get_image_information = MagicMock()
    vm._get_image_information.side_effect = information
    with asyncio_patch("gns3server.compute.docker.DockerVM.pull_image", return_value=True) as mock_pull:
        with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
            loop.run_until_complete(asyncio.async(vm.create()))
            mock.assert_called_with("POST", "containers/create", data={
                "Tty": True,
                "OpenStdin": True,
                "StdinOnce": False,
                "HostConfig":
                    {
                        "CapAdd": ["ALL"],
                        "Binds": [
                            "{}:/gns3:ro".format(get_resource("compute/docker/resources")),
                            "{}:/gns3volumes/etc/network:rw".format(os.path.join(vm.working_dir, "etc", "network"))
                        ],
                        "Privileged": True
                    },
                "Volumes": {},
                "NetworkDisabled": True,
                "Name": "test",
                "Hostname": "test",
                "Image": "ubuntu:latest",
                "Env": [
                    "container=docker",
                    "GNS3_MAX_ETHERNET=eth0",
                    "GNS3_VOLUMES=/etc/network"
                    ],
                "Entrypoint": ["/gns3/init.sh"],
                "Cmd": ["/bin/sh"]
            })
        assert vm._cid == "e90e34656806"
        mock_pull.assert_called_with("ubuntu:latest")


def test_get_container_state(loop, vm):
    response = {
        "State": {
            "Error": "",
            "ExitCode": 9,
            "FinishedAt": "2015-01-06T15:47:32.080254511Z",
            "OOMKilled": False,
            "Paused": False,
            "Pid": 0,
            "Restarting": False,
            "Running": True,
            "StartedAt": "2015-01-06T15:47:32.072697474Z"
        }
    }
    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
        assert loop.run_until_complete(asyncio.async(vm._get_container_state())) == "running"

    response["State"]["Running"] = False
    response["State"]["Paused"] = True
    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
        assert loop.run_until_complete(asyncio.async(vm._get_container_state())) == "paused"

    response["State"]["Running"] = False
    response["State"]["Paused"] = False
    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
        assert loop.run_until_complete(asyncio.async(vm._get_container_state())) == "exited"


def test_is_running(loop, vm):
    response = {
        "State": {
            "Running": False,
            "Paused": False
        }
    }
    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
        assert loop.run_until_complete(asyncio.async(vm.is_running())) is False

    response["State"]["Running"] = True
    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
        assert loop.run_until_complete(asyncio.async(vm.is_running())) is True


def test_pause(loop, vm):

    with asyncio_patch("gns3server.compute.docker.Docker.query") as mock:
        loop.run_until_complete(asyncio.async(vm.pause()))

    mock.assert_called_with("POST", "containers/e90e34656842/pause")
    assert vm.status == "suspended"


def test_unpause(loop, vm):

    with asyncio_patch("gns3server.compute.docker.Docker.query") as mock:
        loop.run_until_complete(asyncio.async(vm.unpause()))

    mock.assert_called_with("POST", "containers/e90e34656842/unpause")


def test_start(loop, vm, manager, free_console_port):

    assert vm.status != "started"
    vm.adapters = 1

    vm.allocate_aux = True
    vm._start_aux = AsyncioMagicMock()

    vm._get_container_state = AsyncioMagicMock(return_value="stopped")
    vm._start_ubridge = AsyncioMagicMock()
    vm._get_namespace = AsyncioMagicMock(return_value=42)
    vm._add_ubridge_connection = AsyncioMagicMock()
    vm._start_console = AsyncioMagicMock()

    nio = manager.create_nio({"type": "nio_udp", "lport": free_console_port, "rport": free_console_port, "rhost": "127.0.0.1"})
    loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, nio)))

    with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query:
        loop.run_until_complete(asyncio.async(vm.start()))

    mock_query.assert_called_with("POST", "containers/e90e34656842/start")
    vm._add_ubridge_connection.assert_called_once_with(nio, 0)
    assert vm._start_ubridge.called
    assert vm._start_console.called
    assert vm._start_aux.called
    assert vm.status == "started"


def test_start_namespace_failed(loop, vm, manager, free_console_port):

    assert vm.status != "started"
    vm.adapters = 1

    nio = manager.create_nio({"type": "nio_udp", "lport": free_console_port, "rport": free_console_port, "rhost": "127.0.0.1"})
    loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, nio)))

    with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="stopped"):
        with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query:
            with asyncio_patch("gns3server.compute.docker.DockerVM._start_ubridge") as mock_start_ubridge:
                with asyncio_patch("gns3server.compute.docker.DockerVM._get_namespace", return_value=42) as mock_namespace:
                    with asyncio_patch("gns3server.compute.docker.DockerVM._add_ubridge_connection", side_effect=UbridgeNamespaceError()) as mock_add_ubridge_connection:
                        with asyncio_patch("gns3server.compute.docker.DockerVM._get_log", return_value='Hello not available') as mock_log:

                            with pytest.raises(DockerError):
                                loop.run_until_complete(asyncio.async(vm.start()))

    mock_query.assert_any_call("POST", "containers/e90e34656842/start")
    mock_add_ubridge_connection.assert_called_once_with(nio, 0)
    assert mock_start_ubridge.called
    assert vm.status == "stopped"


def test_start_without_nio(loop, vm, manager, free_console_port):
    """
    If no nio exists we will create one.
    """

    assert vm.status != "started"
    vm.adapters = 1

    with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="stopped"):
        with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query:
            with asyncio_patch("gns3server.compute.docker.DockerVM._start_ubridge") as mock_start_ubridge:
                with asyncio_patch("gns3server.compute.docker.DockerVM._get_namespace", return_value=42) as mock_namespace:
                    with asyncio_patch("gns3server.compute.docker.DockerVM._add_ubridge_connection") as mock_add_ubridge_connection:
                        with asyncio_patch("gns3server.compute.docker.DockerVM._start_console") as mock_start_console:
                            loop.run_until_complete(asyncio.async(vm.start()))

    mock_query.assert_called_with("POST", "containers/e90e34656842/start")
    assert mock_add_ubridge_connection.called
    assert mock_start_ubridge.called
    assert mock_start_console.called
    assert vm.status == "started"


def test_start_unpause(loop, vm, manager, free_console_port):

    with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="paused"):
        with asyncio_patch("gns3server.compute.docker.DockerVM.unpause", return_value="paused") as mock:
            loop.run_until_complete(asyncio.async(vm.start()))
    assert mock.called
    assert vm.status == "started"


def test_restart(loop, vm):

    with asyncio_patch("gns3server.compute.docker.Docker.query") as mock:
        loop.run_until_complete(asyncio.async(vm.restart()))

    mock.assert_called_with("POST", "containers/e90e34656842/restart")


def test_stop(loop, vm):
    mock = MagicMock()
    vm._ubridge_hypervisor = mock
    vm._ubridge_hypervisor.is_running.return_value = True
    vm._fix_permissions = MagicMock()

    with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="running"):
        with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query:
            loop.run_until_complete(asyncio.async(vm.stop()))
            mock_query.assert_called_with("POST", "containers/e90e34656842/stop", params={"t": 5})
    assert mock.stop.called
    assert vm._ubridge_hypervisor is None
    assert vm._fix_permissions.called


def test_stop_paused_container(loop, vm):

    with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="paused"):
        with asyncio_patch("gns3server.compute.docker.DockerVM.unpause") as mock_unpause:
            with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query:
                loop.run_until_complete(asyncio.async(vm.stop()))
                mock_query.assert_called_with("POST", "containers/e90e34656842/stop", params={"t": 5})
                assert mock_unpause.called


def test_update(loop, vm):

    response = {
        "Id": "e90e34656806",
        "Warnings": []
    }

    original_console = vm.console
    original_aux = vm.aux

    with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "ubuntu"}]) as mock_list_images:
        with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="stopped"):
            with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock_query:
                loop.run_until_complete(asyncio.async(vm.update()))

    mock_query.assert_any_call("DELETE", "containers/e90e34656842", params={"force": 1, "v": 1})
    mock_query.assert_any_call("POST", "containers/create", data={
        "Tty": True,
        "OpenStdin": True,
        "StdinOnce": False,
        "HostConfig":
        {
            "CapAdd": ["ALL"],
            "Binds": [
                "{}:/gns3:ro".format(get_resource("compute/docker/resources")),
                "{}:/gns3volumes/etc/network:rw".format(os.path.join(vm.working_dir, "etc", "network"))
            ],
            "Privileged": True
        },
        "Volumes": {},
        "NetworkDisabled": True,
        "Name": "test",
        "Hostname": "test",
        "Image": "ubuntu:latest",
        "Env": [
            "container=docker",
            "GNS3_MAX_ETHERNET=eth0",
            "GNS3_VOLUMES=/etc/network"
        ],
        "Entrypoint": ["/gns3/init.sh"],
        "Cmd": ["/bin/sh"]
    })
    assert vm.console == original_console
    assert vm.aux == original_aux


def test_update_vnc(loop, vm):

    response = {
        "Id": "e90e34656806",
        "Warnings": []
    }

    vm._console_type = "vnc"
    vm._console = 5900
    vm._display = "display"
    original_console = vm.console
    original_aux = vm.aux

    with asyncio_patch("gns3server.compute.docker.DockerVM._start_vnc"):
        with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "ubuntu"}]) as mock_list_images:
            with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="stopped"):
                with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock_query:
                    loop.run_until_complete(asyncio.async(vm.update()))

    assert vm.console == original_console
    assert vm.aux == original_aux


def test_update_running(loop, vm):

    response = {
        "Id": "e90e34656806",
        "Warnings": []
    }

    original_console = vm.console
    vm.start = MagicMock()

    with asyncio_patch("gns3server.compute.docker.Docker.list_images", return_value=[{"image": "ubuntu"}]) as mock_list_images:
        with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="running"):
            with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock_query:
                loop.run_until_complete(asyncio.async(vm.update()))

    mock_query.assert_any_call("DELETE", "containers/e90e34656842", params={"force": 1, "v": 1})
    mock_query.assert_any_call("POST", "containers/create", data={
        "Tty": True,
        "OpenStdin": True,
        "StdinOnce": False,
        "HostConfig":
        {
            "CapAdd": ["ALL"],
            "Binds": [
                "{}:/gns3:ro".format(get_resource("compute/docker/resources")),
                "{}:/gns3volumes/etc/network:rw".format(os.path.join(vm.working_dir, "etc", "network"))
            ],
            "Privileged": True
        },
        "Volumes": {},
        "NetworkDisabled": True,
        "Name": "test",
        "Hostname": "test",
        "Image": "ubuntu:latest",
        "Env": [
            "container=docker",
            "GNS3_MAX_ETHERNET=eth0",
            "GNS3_VOLUMES=/etc/network"
        ],
        "Entrypoint": ["/gns3/init.sh"],
        "Cmd": ["/bin/sh"]
    })

    assert vm.console == original_console
    assert vm.start.called


def test_delete(loop, vm):

    with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="stopped"):
        with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query:
            loop.run_until_complete(asyncio.async(vm.delete()))
        mock_query.assert_called_with("DELETE", "containers/e90e34656842", params={"force": 1, "v": 1})


def test_close(loop, vm, port_manager):
    nio = {"type": "nio_udp",
           "lport": 4242,
           "rport": 4343,
           "rhost": "127.0.0.1"}
    nio = vm.manager.create_nio(nio)
    loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, nio)))

    with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="stopped"):
        with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query:
            loop.run_until_complete(asyncio.async(vm.close()))
        mock_query.assert_called_with("DELETE", "containers/e90e34656842", params={"force": 1, "v": 1})

    assert vm._closed is True
    assert "4242" not in port_manager.udp_ports


def test_close_vnc(loop, vm, port_manager):

    vm._console_type = "vnc"
    vm._x11vnc_process = MagicMock()
    vm._xvfb_process = MagicMock()

    with asyncio_patch("gns3server.compute.docker.DockerVM._get_container_state", return_value="stopped"):
        with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query:
            loop.run_until_complete(asyncio.async(vm.close()))
        mock_query.assert_called_with("DELETE", "containers/e90e34656842", params={"force": 1, "v": 1})

    assert vm._closed is True
    assert vm._xvfb_process.terminate.called


def test_get_namespace(loop, vm):
    response = {
        "State": {
            "Pid": 42
        }
    }
    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock_query:
        assert loop.run_until_complete(asyncio.async(vm._get_namespace())) == 42
    mock_query.assert_called_with("GET", "containers/e90e34656842/json")


def test_add_ubridge_connection(loop, vm):

    nio = {"type": "nio_udp",
           "lport": 4242,
           "rport": 4343,
           "rhost": "127.0.0.1"}
    nio = vm.manager.create_nio(nio)
    nio.startPacketCapture("/tmp/capture.pcap")
    vm._ubridge_hypervisor = MagicMock()
    vm._namespace = 42

    loop.run_until_complete(asyncio.async(vm._add_ubridge_connection(nio, 0)))

    calls = [
        call.send('bridge create bridge0'),
        call.send("bridge add_nio_tap bridge0 tap-gns3-e0"),
        call.send('docker move_to_ns tap-gns3-e0 42 eth0'),
        call.send('bridge add_nio_udp bridge0 4242 127.0.0.1 4343'),
        call.send('bridge start_capture bridge0 "/tmp/capture.pcap"'),
        call.send('bridge start bridge0')
    ]
    # We need to check any_order ortherwise mock is confused by asyncio
    vm._ubridge_hypervisor.assert_has_calls(calls, any_order=True)


def test_add_ubridge_connection_none_nio(loop, vm):

    nio = None
    vm._ubridge_hypervisor = MagicMock()
    vm._namespace = 42

    loop.run_until_complete(asyncio.async(vm._add_ubridge_connection(nio, 0)))

    calls = [
        call.send('bridge create bridge0'),
        call.send("bridge add_nio_tap bridge0 tap-gns3-e0"),
        call.send('docker move_to_ns tap-gns3-e0 42 eth0'),

    ]
    # We need to check any_order ortherwise mock is confused by asyncio
    vm._ubridge_hypervisor.assert_has_calls(calls, any_order=True)


def test_add_ubridge_connection_invalid_adapter_number(loop, vm):

    nio = {"type": "nio_udp",
           "lport": 4242,
           "rport": 4343,
           "rhost": "127.0.0.1"}
    nio = vm.manager.create_nio(nio)
    with pytest.raises(DockerError):
        loop.run_until_complete(asyncio.async(vm._add_ubridge_connection(nio, 12)))


def test_add_ubridge_connection_no_free_interface(loop, vm):

    nio = {"type": "nio_udp",
           "lport": 4242,
           "rport": 4343,
           "rhost": "127.0.0.1"}
    nio = vm.manager.create_nio(nio)
    with pytest.raises(DockerError):

        # We create fake ethernet interfaces for docker
        interfaces = ["tap-gns3-e{}".format(index) for index in range(4096)]

        with patch("psutil.net_if_addrs", return_value=interfaces):
            loop.run_until_complete(asyncio.async(vm._add_ubridge_connection(nio, 0)))


def test_adapter_add_nio_binding(vm, loop):
    nio = {"type": "nio_udp",
           "lport": 4242,
           "rport": 4343,
           "rhost": "127.0.0.1"}
    nio = vm.manager.create_nio(nio)
    loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, nio)))
    assert vm._ethernet_adapters[0].get_nio(0) == nio


def test_adapter_add_nio_binding_invalid_adapter(vm, loop):
    nio = {"type": "nio_udp",
           "lport": 4242,
           "rport": 4343,
           "rhost": "127.0.0.1"}
    nio = vm.manager.create_nio(nio)
    with pytest.raises(DockerError):
        loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(12, nio)))


def test_adapter_remove_nio_binding(vm, loop):
    vm.ubridge = MagicMock()
    vm.ubridge.is_running.return_value = True

    nio = {"type": "nio_udp",
           "lport": 4242,
           "rport": 4343,
           "rhost": "127.0.0.1"}
    nio = vm.manager.create_nio(nio)
    loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, nio)))

    with asyncio_patch("gns3server.compute.docker.DockerVM._ubridge_send") as delete_ubridge_mock:
        loop.run_until_complete(asyncio.async(vm.adapter_remove_nio_binding(0)))
        assert vm._ethernet_adapters[0].get_nio(0) is None
        delete_ubridge_mock.assert_any_call('bridge stop bridge0')
        delete_ubridge_mock.assert_any_call('bridge remove_nio_udp bridge0 4242 127.0.0.1 4343')


def test_adapter_remove_nio_binding_invalid_adapter(vm, loop):
    with pytest.raises(DockerError):
        loop.run_until_complete(asyncio.async(vm.adapter_remove_nio_binding(12)))


def test_start_capture(vm, tmpdir, manager, free_console_port, loop):

    output_file = str(tmpdir / "test.pcap")
    nio = manager.create_nio({"type": "nio_udp", "lport": free_console_port, "rport": free_console_port, "rhost": "127.0.0.1"})
    loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, nio)))
    loop.run_until_complete(asyncio.async(vm.start_capture(0, output_file)))
    assert vm._ethernet_adapters[0].get_nio(0).capturing


def test_stop_capture(vm, tmpdir, manager, free_console_port, loop):

    output_file = str(tmpdir / "test.pcap")
    nio = manager.create_nio({"type": "nio_udp", "lport": free_console_port, "rport": free_console_port, "rhost": "127.0.0.1"})
    loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, nio)))
    loop.run_until_complete(vm.start_capture(0, output_file))
    assert vm._ethernet_adapters[0].get_nio(0).capturing
    loop.run_until_complete(asyncio.async(vm.stop_capture(0)))
    assert vm._ethernet_adapters[0].get_nio(0).capturing is False


def test_get_log(loop, vm):
    @asyncio.coroutine
    def read():
        return b'Hello\nWorld'

    mock_query = MagicMock()
    mock_query.read = read

    with asyncio_patch("gns3server.compute.docker.Docker.http_query", return_value=mock_query) as mock:
        images = loop.run_until_complete(asyncio.async(vm._get_log()))
        mock.assert_called_with("GET", "containers/e90e34656842/logs", params={"stderr": 1, "stdout": 1}, data={})


def test_get_image_informations(project, manager, loop):
    response = {
    }
    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
        vm = DockerVM("test", str(uuid.uuid4()), project, manager, "ubuntu")
        loop.run_until_complete(asyncio.async(vm._get_image_information()))
        mock.assert_called_with("GET", "images/ubuntu:latest/json")


@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
def test_mount_binds(vm, tmpdir):
    image_infos = {
        "Config": {
            "Volumes": {
                "/test/experimental": {}
            }
        }
    }

    dst = os.path.join(vm.working_dir, "test/experimental")
    assert vm._mount_binds(image_infos) == [
        "{}:/gns3:ro".format(get_resource("compute/docker/resources")),
        "{}:/gns3volumes/etc/network:rw".format(os.path.join(vm.working_dir, "etc", "network")),
        "{}:/gns3volumes{}".format(dst, "/test/experimental")
    ]

    assert vm._volumes == ["/etc/network", "/test/experimental"]
    assert os.path.exists(dst)


def test_start_vnc(vm, loop):
    vm.console_resolution = "1280x1024"
    with patch("shutil.which", return_value="/bin/x"):
        with asyncio_patch("gns3server.compute.docker.docker_vm.wait_for_file_creation") as mock_wait:
            with asyncio_patch("asyncio.create_subprocess_exec") as mock_exec:
                loop.run_until_complete(asyncio.async(vm._start_vnc()))
    assert vm._display is not None
    mock_exec.assert_any_call("Xvfb", "-nolisten", "tcp", ":{}".format(vm._display), "-screen", "0", "1280x1024x16")
    mock_exec.assert_any_call("x11vnc", "-forever", "-nopw", "-shared", "-geometry", "1280x1024", "-display", "WAIT:{}".format(vm._display), "-rfbport", str(vm.console), "-rfbportv6", str(vm.console), "-noncache", "-listen", "127.0.0.1")
    mock_wait.assert_called_with("/tmp/.X11-unix/X{}".format(vm._display))


def test_start_vnc_xvfb_missing(vm, loop):
    with pytest.raises(DockerError):
        loop.run_until_complete(asyncio.async(vm._start_vnc()))


def test_start_aux(vm, loop):

    with asyncio_patch("asyncio.subprocess.create_subprocess_exec", return_value=MagicMock()) as mock_exec:
        loop.run_until_complete(asyncio.async(vm._start_aux()))
    mock_exec.assert_called_with('docker', 'exec', '-i', 'e90e34656842', '/gns3/bin/busybox', 'script', '-qfc', 'while true; do TERM=vt100 /gns3/bin/busybox sh; done', '/dev/null', stderr=asyncio.subprocess.STDOUT, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE)


def test_create_network_interfaces(vm):

    vm.adapters = 5
    network_config = vm._create_network_config()
    assert os.path.exists(os.path.join(network_config, "interfaces"))
    assert os.path.exists(os.path.join(network_config, "if-up.d"))

    with open(os.path.join(network_config, "interfaces")) as f:
        content = f.read()
    assert "eth0" in content
    assert "eth4" in content
    assert "eth5" not in content


@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
def test_fix_permission(vm, loop):
    vm._volumes = ["/etc"]
    process = MagicMock()
    with asyncio_patch("asyncio.subprocess.create_subprocess_exec", return_value=process) as mock_exec:
        loop.run_until_complete(vm._fix_permissions())
    mock_exec.assert_called_with('docker', 'exec', 'e90e34656842', '/gns3/bin/busybox', 'sh', '-c', '(/gns3/bin/busybox find "/etc" -depth -print0 | /gns3/bin/busybox xargs -0 /gns3/bin/busybox stat -c \'%a:%u:%g:%n\' > "/etc/.gns3_perms") && /gns3/bin/busybox chmod -R u+rX "/etc" && /gns3/bin/busybox chown {}:{} -R "/etc"'.format(os.getuid(), os.getgid()))
    assert process.wait.called