arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-11-24 17:28:08 +00:00
Code Issues Releases Wiki Activity

Support multiple images location

Browse Source 
Ref #546
This commit is contained in:
Julien Duponchelle 2016-06-02 15:19:34 +02:00
parent cb7dbfb256
commit a702155230
No known key found for this signature in database
GPG Key ID: CE8B29639E07F5E8
3 changed files with 99 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
gns3server/compute/base_manager.py
View File

@ -389,6 +389,24 @@ class BaseManager:
assert nio is not None
return nio
def images_directories(self):
"""
Return all directory where we will look for images
by priority
"""
server_config = self.config.get_section_config("Server")
paths = []
img_directory = self.get_images_directory()
os.makedirs(img_directory, exist_ok=True)
paths.append(img_directory)
for directory in server_config.get("additional_images_path", "").split(":"):
paths.append(directory)
# Compatibility with old topologies we look in parent directory
paths.append(os.path.normpath(os.path.join(self.get_images_directory(), '..')))
# Return only the existings paths
return [force_unix_path(p) for p in paths if os.path.exists(p)]
def get_abs_image_path(self, path):
"""
Get the absolute path of an image
@ -400,6 +418,7 @@ class BaseManager:
if not path:
return ""
server_config = self.config.get_section_config("Server")
img_directory = self.get_images_directory()
# Windows path should not be send to a unix server
@ -408,27 +427,25 @@ class BaseManager:
raise NodeError("{} is not allowed on this remote server. Please use only a filename in {}.".format(path, img_directory))
if not os.path.isabs(path):
orig_path = path
s = os.path.split(path)
path = os.path.normpath(os.path.join(img_directory, *s))
# Compatibility with old topologies we look in parent directory
# We look at first in new location
if not os.path.exists(path):
old_path = os.path.normpath(os.path.join(img_directory, '..', *s))
if os.path.exists(old_path):
return force_unix_path(old_path)
for directory in self.images_directories():
path = os.path.normpath(os.path.join(directory, *s))
if os.path.exists(path):
return force_unix_path(path)
# Not found we return the default directory
return force_unix_path(os.path.join(self.get_images_directory(), *s))
# For non local server we disallow using absolute path outside image directory
if server_config.get("local", False) is True:
return force_unix_path(path)
else:
# For non local server we disallow using absolute path outside image directory
if Config.instance().get_section_config("Server").get("local", False) is False:
img_directory = self.config.get_section_config("Server").get("images_path", os.path.expanduser("~/GNS3/images"))
img_directory = force_unix_path(img_directory)
path = force_unix_path(path)
if len(os.path.commonprefix([img_directory, path])) < len(img_directory):
raise NodeError("{} is not allowed on this remote server. Please use only a filename in {}.".format(path, img_directory))
return force_unix_path(path)
path = force_unix_path(path)
for directory in self.images_directories():
if os.path.commonprefix([directory, path]) == directory:
return path
raise NodeError("{} is not allowed on this remote server. Please use only a filename in {}.".format(path, self.get_images_directory()))
def get_relative_image_path(self, path):
"""
@ -442,10 +459,10 @@ class BaseManager:
if not path:
return ""
img_directory = force_unix_path(self.get_images_directory())
path = force_unix_path(self.get_abs_image_path(path))
if os.path.commonprefix([img_directory, path]) == img_directory:
return os.path.relpath(path, img_directory)
for directory in self.images_directories():
if os.path.commonprefix([directory, path]) == directory:
return os.path.relpath(path, directory)
return path
@asyncio.coroutine

8
tests/compute/qemu/test_qemu_vm.py
View File

@ -510,10 +510,10 @@ def test_hda_disk_image(vm, tmpdir):
vm.manager.config.set("Server", "images_path", str(tmpdir))
vm.hda_disk_image = str(tmpdir / "test")
assert vm.hda_disk_image == force_unix_path(str(tmpdir / "test"))
vm.hda_disk_image = "test"
assert vm.hda_disk_image == force_unix_path(str(tmpdir / "QEMU" / "test"))
vm.hda_disk_image = str(tmpdir / "test1")
assert vm.hda_disk_image == force_unix_path(str(tmpdir / "test1"))
vm.hda_disk_image = "test2"
assert vm.hda_disk_image == force_unix_path(str(tmpdir / "QEMU" / "test2"))
def test_hda_disk_image_ova(vm, tmpdir):

66
tests/compute/test_manager.py
View File

@ -88,6 +88,28 @@ def test_create_node_old_topology(loop, project, tmpdir, vpcs):
assert f.read() == "1"
def test_images_directories(qemu, tmpdir):
path1 = tmpdir / "images1" / "QEMU" / "test1.bin"
path1.write("1", ensure=True)
path1 = force_unix_path(str(path1))
path2 = tmpdir / "images2" / "test2.bin"
path2.write("1", ensure=True)
path2 = force_unix_path(str(path2))
with patch("gns3server.config.Config.get_section_config", return_value={
"images_path": str(tmpdir / "images1"),
"additional_images_path": "/tmp/null24564:{}".format(tmpdir / "images2"),
"local": False}):
# /tmp/null24564 is ignored because doesn't exists
res = qemu.images_directories()
assert res[0] == str(tmpdir / "images1" / "QEMU")
assert res[1] == str(tmpdir / "images2")
assert res[2] == str(tmpdir / "images1")
assert len(res) == 3
def test_get_abs_image_path(qemu, tmpdir):
os.makedirs(str(tmpdir / "QEMU"))
path1 = force_unix_path(str(tmpdir / "test1.bin"))
@ -125,20 +147,50 @@ def test_get_abs_image_path_non_local(qemu, tmpdir):
assert qemu.get_abs_image_path(path2) == path2
def test_get_abs_image_additional_image_paths(qemu, tmpdir):
path1 = tmpdir / "images1" / "QEMU" / "test1.bin"
path1.write("1", ensure=True)
path1 = force_unix_path(str(path1))
path2 = tmpdir / "images2" / "test2.bin"
path2.write("1", ensure=True)
path2 = force_unix_path(str(path2))
with patch("gns3server.config.Config.get_section_config", return_value={
"images_path": str(tmpdir / "images1"),
"additional_images_path": "/tmp/null24564:{}".format(tmpdir / "images2"),
"local": False}):
assert qemu.get_abs_image_path("test1.bin") == path1
assert qemu.get_abs_image_path("test2.bin") == path2
# Absolute path
assert qemu.get_abs_image_path(str(path2)) == path2
# If not found return the default path
assert qemu.get_abs_image_path("test4.bin") == os.path.join(qemu.get_images_directory(), "test4.bin")
def test_get_relative_image_path(qemu, tmpdir):
os.makedirs(str(tmpdir / "QEMU"))
path1 = force_unix_path(str(tmpdir / "test1.bin"))
os.makedirs(str(tmpdir / "images1" / "QEMU"))
path1 = force_unix_path(str(tmpdir / "images1" / "test1.bin"))
open(path1, 'w+').close()
path2 = force_unix_path(str(tmpdir / "QEMU" / "test2.bin"))
path2 = force_unix_path(str(tmpdir / "images1" / "QEMU" / "test2.bin"))
open(path2, 'w+').close()
with patch("gns3server.config.Config.get_section_config", return_value={"images_path": str(tmpdir)}):
assert qemu.get_relative_image_path(path1) == path1
assert qemu.get_relative_image_path("test1.bin") == path1
os.makedirs(str(tmpdir / "images2"))
path3 = force_unix_path(str(tmpdir / "images2" / "test3.bin"))
open(path3, 'w+').close()
with patch("gns3server.config.Config.get_section_config", return_value={
"images_path": str(tmpdir / "images1"),
"additional_images_path": str(tmpdir / "images2")
}):
assert qemu.get_relative_image_path(path1) == "test1.bin"
assert qemu.get_relative_image_path("test1.bin") == "test1.bin"
assert qemu.get_relative_image_path(path2) == "test2.bin"
assert qemu.get_relative_image_path("test2.bin") == "test2.bin"
assert qemu.get_relative_image_path("../test1.bin") == path1
assert qemu.get_relative_image_path("../test1.bin") == "test1.bin"
assert qemu.get_relative_image_path("test3.bin") == "test3.bin"
def test_get_relative_image_path_ova(qemu, tmpdir):