1
0
mirror of https://github.com/GNS3/gns3-server synced 2024-12-04 22:28:12 +00:00
gns3-server/tests/api/routes/controller/test_computes.py

193 lines
7.3 KiB
Python
Raw Normal View History

#!/usr/bin/env python
#
# Copyright (C) 2020 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/>.
2021-04-05 04:51:41 +00:00
import uuid
2020-10-02 06:37:50 +00:00
import pytest
from fastapi import FastAPI, status
from httpx import AsyncClient
2021-04-15 09:00:22 +00:00
from gns3server.schemas.controller.computes import Compute
pytestmark = pytest.mark.asyncio
import unittest
from tests.utils import asyncio_patch
2021-04-05 04:51:41 +00:00
class TestComputeRoutes:
async def test_compute_create(self, app: FastAPI, client: AsyncClient) -> None:
params = {
"protocol": "http",
"host": "localhost",
"port": 84,
"user": "julien",
"password": "secure"}
response = await client.post(app.url_path_for("create_compute"), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["compute_id"] is not None
del params["password"]
for param, value in params.items():
assert response.json()[param] == value
async def test_compute_create_with_id(self, app: FastAPI, client: AsyncClient) -> None:
compute_id = str(uuid.uuid4())
params = {
"compute_id": compute_id,
"protocol": "http",
"host": "localhost",
"port": 84,
"user": "julien",
"password": "secure"}
response = await client.post(app.url_path_for("create_compute"), json=params)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["compute_id"] == compute_id
del params["password"]
for param, value in params.items():
assert response.json()[param] == value
async def test_compute_list(self, app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_computes"))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) > 0
async def test_compute_get(self, app: FastAPI, client: AsyncClient, test_compute: Compute) -> None:
response = await client.get(app.url_path_for("get_compute", compute_id=test_compute.compute_id))
assert response.status_code == status.HTTP_200_OK
assert response.json()["compute_id"] == str(test_compute.compute_id)
async def test_compute_update(self, app: FastAPI, client: AsyncClient, test_compute: Compute) -> None:
params = {
"protocol": "http",
"host": "localhost",
"port": 42,
2021-04-05 04:51:41 +00:00
"user": "julien",
"password": "secure"
}
response = await client.post(app.url_path_for("create_compute"), json=params)
assert response.status_code == status.HTTP_201_CREATED
compute_id = response.json()["compute_id"]
params["protocol"] = "https"
response = await client.put(app.url_path_for("update_compute", compute_id=compute_id), json=params)
assert response.status_code == status.HTTP_200_OK
del params["password"]
for param, value in params.items():
assert response.json()[param] == value
async def test_compute_delete(self, app: FastAPI, client: AsyncClient, test_compute: Compute) -> None:
response = await client.delete(app.url_path_for("delete_compute", compute_id=test_compute.compute_id))
assert response.status_code == status.HTTP_204_NO_CONTENT
class TestComputeFeatures:
async def test_compute_list_images(self, app: FastAPI, client: AsyncClient) -> None:
params = {
"protocol": "http",
"host": "localhost",
"port": 84,
"user": "julien",
"password": "secure"
}
response = await client.post(app.url_path_for("create_compute"), json=params)
assert response.status_code == status.HTTP_201_CREATED
compute_id = response.json()["compute_id"]
with asyncio_patch("gns3server.controller.compute.Compute.images", return_value=[{"filename": "linux.qcow2"}, {"filename": "asav.qcow2"}]) as mock:
response = await client.get(app.url_path_for("delete_compute", compute_id=compute_id) + "/qemu/images")
assert response.json() == [{"filename": "linux.qcow2"}, {"filename": "asav.qcow2"}]
mock.assert_called_with("qemu")
async def test_compute_list_vms(self, app: FastAPI, client: AsyncClient) -> None:
params = {
"protocol": "http",
"host": "localhost",
"port": 4242,
2021-04-05 04:51:41 +00:00
"user": "julien",
"password": "secure"
}
response = await client.post(app.url_path_for("get_computes"), json=params)
assert response.status_code == status.HTTP_201_CREATED
compute_id = response.json()["compute_id"]
with asyncio_patch("gns3server.controller.compute.Compute.forward", return_value=[]) as mock:
response = await client.get(app.url_path_for("get_compute", compute_id=compute_id) + "/virtualbox/vms")
mock.assert_called_with("GET", "virtualbox", "vms")
assert response.json() == []
async def test_compute_create_img(self, app: FastAPI, client: AsyncClient) -> None:
params = {
"protocol": "http",
"host": "localhost",
"port": 4284,
2021-04-05 04:51:41 +00:00
"user": "julien",
"password": "secure"
}
response = await client.post(app.url_path_for("get_computes"), json=params)
assert response.status_code == status.HTTP_201_CREATED
compute_id = response.json()["compute_id"]
params = {"path": "/test"}
with asyncio_patch("gns3server.controller.compute.Compute.forward", return_value=[]) as mock:
response = await client.post(app.url_path_for("get_compute", compute_id=compute_id) + "/qemu/img", json=params)
assert response.json() == []
mock.assert_called_with("POST", "qemu", "img", data=unittest.mock.ANY)
# async def test_compute_autoidlepc(self, app: FastAPI, client: AsyncClient) -> None:
#
# params = {
# "protocol": "http",
# "host": "localhost",
# "port": 84,
# "user": "julien",
# "password": "secure"
# }
#
# response = await client.post(app.url_path_for("get_computes"), json=params)
# assert response.status_code == status.HTTP_201_CREATED
# compute_id = response.json()["compute_id"]
#
# params = {
# "platform": "c7200",
# "image": "test.bin",
# "ram": 512
# }
#
# with asyncio_patch("gns3server.controller.Controller.autoidlepc", return_value={"idlepc": "0x606de20c"}) as mock:
# response = await client.post(app.url_path_for("autoidlepc", compute_id=compute_id) + "/auto_idlepc", json=params)
# assert mock.called
# assert response.status_code == status.HTTP_200_OK