1
0
mirror of https://github.com/GNS3/gns3-server synced 2024-12-24 15:58:08 +00:00
gns3-server/tests/compute/test_project.py
2021-04-17 23:34:28 +09:30

243 lines
7.2 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# 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/>.
import os
import sys
import uuid
import pytest
from uuid import uuid4
from unittest.mock import patch
from tests.utils import asyncio_patch
from gns3server.compute.project import Project
from gns3server.compute.notification_manager import NotificationManager
from gns3server.compute.compute_error import ComputeError, ComputeForbiddenError
from gns3server.compute.vpcs import VPCS, VPCSVM
from gns3server.config import Config
@pytest.fixture(scope="function")
@pytest.mark.asyncio
async def manager(port_manager):
m = VPCS.instance()
m.port_manager = port_manager
return m
@pytest.fixture(scope="function")
@pytest.mark.asyncio
async def node(compute_project, manager):
node = manager.create_node("test", compute_project.id, "00010203-0405-0607-0809-0a0b0c0d0e0f")
return await node
@pytest.mark.asyncio
async def test_affect_uuid():
p = Project(project_id='00010203-0405-0607-0809-0a0b0c0d0e0f')
assert p.id == '00010203-0405-0607-0809-0a0b0c0d0e0f'
@pytest.mark.asyncio
async def test_clean_tmp_directory():
"""
The tmp directory should be clean at project open and close
"""
p = Project(project_id='00010203-0405-0607-0809-0a0b0c0d0e0f')
path = p.tmp_working_directory()
os.makedirs(path)
await p.close()
assert not os.path.exists(path)
os.makedirs(path)
p = Project(project_id='00010203-0405-0607-0809-0a0b0c0d0e0f')
assert not os.path.exists(path)
@pytest.mark.asyncio
async def test_path(projects_dir):
directory = projects_dir
with patch("gns3server.compute.project.Project.is_local", return_value=True):
with patch("gns3server.utils.path.get_default_project_directory", return_value=directory):
p = Project(project_id=str(uuid4()))
assert p.path == os.path.join(directory, p.id)
assert os.path.exists(os.path.join(directory, p.id))
@pytest.mark.asyncio
async def test_init_path(tmpdir):
with patch("gns3server.compute.project.Project.is_local", return_value=True):
p = Project(path=str(tmpdir), project_id=str(uuid4()))
assert p.path == str(tmpdir)
@pytest.mark.asyncio
async def test_changing_path_not_allowed(tmpdir):
with patch("gns3server.compute.project.Project.is_local", return_value=False):
with pytest.raises(ComputeForbiddenError):
p = Project(project_id=str(uuid4()))
p.path = str(tmpdir)
@pytest.mark.asyncio
async def test_variables():
variables = [{"name": "VAR1", "value": "VAL1"}]
p = Project(project_id=str(uuid4()), variables=variables)
assert p.variables == variables
@pytest.mark.asyncio
async def test_json():
p = Project(project_id=str(uuid4()))
assert p.asdict() == {
"name": p.name,
"project_id": p.id,
"variables": None
}
@pytest.mark.asyncio
async def test_json_with_variables():
variables = [{"name": "VAR1", "value": "VAL1"}]
p = Project(project_id=str(uuid4()), variables=variables)
assert p.asdict() == {
"name": p.name,
"project_id": p.id,
"variables": variables
}
@pytest.mark.asyncio
async def test_node_working_directory(node, projects_dir):
directory = projects_dir
with patch("gns3server.compute.project.Project.is_local", return_value=True):
p = Project(project_id=str(uuid4()))
assert p.node_working_directory(node) == os.path.join(directory, p.id, 'project-files', node.module_name, node.id)
assert os.path.exists(p.node_working_directory(node))
@pytest.mark.asyncio
async def test_node_working_path(node, projects_dir):
directory = projects_dir
with patch("gns3server.compute.project.Project.is_local", return_value=True):
p = Project(project_id=str(uuid4()))
assert p.node_working_path(node) == os.path.join(directory, p.id, 'project-files', node.module_name, node.id)
# after this execution directory structure should not be created
assert not os.path.exists(p.node_working_path(node))
@pytest.mark.asyncio
async def test_project_delete():
project = Project(project_id=str(uuid4()))
directory = project.path
assert os.path.exists(directory)
await project.delete()
assert os.path.exists(directory) is False
@pytest.mark.skipif(not sys.platform.startswith("win") and os.getuid() == 0, reason="Root can delete any project")
@pytest.mark.asyncio
async def test_project_delete_permission_issue():
project = Project(project_id=str(uuid4()))
directory = project.path
assert os.path.exists(directory)
os.chmod(directory, 0)
with pytest.raises(ComputeError):
await project.delete()
os.chmod(directory, 700)
@pytest.mark.asyncio
async def test_project_add_node(manager):
project = Project(project_id=str(uuid4()))
node = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
project.add_node(node)
assert len(project.nodes) == 1
@pytest.mark.asyncio
async def test_project_close(node, compute_project):
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM.close") as mock:
await compute_project.close()
assert mock.called
assert node.id not in node.manager._nodes
@pytest.mark.asyncio
async def test_list_files(tmpdir, config):
config.settings.Server.projects_path = str(tmpdir)
project = Project(project_id=str(uuid4()))
path = project.path
os.makedirs(os.path.join(path, "vm-1", "dynamips"))
with open(os.path.join(path, "vm-1", "dynamips", "test.bin"), "w+") as f:
f.write("test")
open(os.path.join(path, "vm-1", "dynamips", "test.ghost"), "w+").close()
with open(os.path.join(path, "test.txt"), "w+") as f:
f.write("test2")
files = await project.list_files()
assert files == [
{
"path": "test.txt",
"md5sum": "ad0234829205b9033196ba818f7a872b"
},
{
"path": os.path.join("vm-1", "dynamips", "test.bin"),
"md5sum": "098f6bcd4621d373cade4e832627b4f6"
}
]
@pytest.mark.asyncio
async def test_emit():
with NotificationManager.instance().queue() as queue:
await queue.get(0.5) #  Ping
project = Project(project_id=str(uuid4()))
project.emit("test", {})
(action, event, context) = await queue.get(0.5)
assert action == "test"
assert context["project_id"] == project.id
@pytest.mark.asyncio
async def test_update_project():
variables = [{"name": "TEST", "value": "VAL"}]
project = Project(project_id=str(uuid.uuid4()))
await project.update(variables=variables)
assert project.variables == variables