Add generic function to install resource files

gns3server/controller/__init__.py

@@ -281,6 +281,24 @@ class Controller:
         except OSError as e:
             log.error(str(e))
 
+    @staticmethod
+    def install_resource_files(dst_path, resource_name):
+        """
+        Install files from resources to user's file system
+        """
+
+        if hasattr(sys, "frozen") and sys.platform.startswith("win"):
+            resource_path = os.path.normpath(os.path.join(os.path.dirname(sys.executable), resource_name))
+            for filename in os.listdir(resource_path):
+                if not os.path.exists(os.path.join(dst_path, filename)):
+                    shutil.copy(os.path.join(resource_path, filename), os.path.join(dst_path, filename))
+        else:
+            for entry in importlib_resources.files(f'gns3server.{resource_name}').iterdir():
+                full_path = os.path.join(dst_path, entry.name)
+                if entry.is_file() and not os.path.exists(full_path):
+                    log.debug(f'Installing {resource_name} resource file "{entry.name}" to "{full_path}"')
+                    shutil.copy(str(entry), os.path.join(dst_path, entry.name))
+
     def _install_base_configs(self):
         """
         At startup we copy base configs to the user location to allow
@@ -290,17 +308,7 @@ class Controller:
         dst_path = self.configs_path()
         log.info(f"Installing base configs in '{dst_path}'")
         try:
-            if hasattr(sys, "frozen") and sys.platform.startswith("win"):
-                resource_path = os.path.normpath(os.path.join(os.path.dirname(sys.executable), "configs"))
-                for filename in os.listdir(resource_path):
-                    if not os.path.exists(os.path.join(dst_path, filename)):
-                        shutil.copy(os.path.join(resource_path, filename), os.path.join(dst_path, filename))
-            else:
-                for entry in importlib_resources.files('gns3server.configs').iterdir():
-                    full_path = os.path.join(dst_path, entry.name)
-                    if entry.is_file() and not os.path.exists(full_path):
-                        log.debug(f"Installing base config file {entry.name} to {full_path}")
-                        shutil.copy(str(entry), os.path.join(dst_path, entry.name))
+            Controller.install_resource_files(dst_path, "configs")
         except OSError as e:
             log.error(f"Could not install base config files to {dst_path}: {e}")
 
@@ -313,17 +321,7 @@ class Controller:
         dst_path = self.disks_path()
         log.info(f"Installing built-in disks in '{dst_path}'")
         try:
-            if hasattr(sys, "frozen") and sys.platform.startswith("win"):
-                resource_path = os.path.normpath(os.path.join(os.path.dirname(sys.executable), "disks"))
-                for filename in os.listdir(resource_path):
-                    if not os.path.exists(os.path.join(dst_path, filename)):
-                        shutil.copy(os.path.join(resource_path, filename), os.path.join(dst_path, filename))
-            else:
-                for entry in importlib_resources.files('gns3server.disks').iterdir():
-                    full_path = os.path.join(dst_path, entry.name)
-                    if entry.is_file() and not os.path.exists(full_path):
-                        log.debug(f"Installing disk file {entry.name} to {full_path}")
-                        shutil.copy(str(entry), os.path.join(dst_path, entry.name))
+            Controller.install_resource_files(dst_path, "disks")
         except OSError as e:
             log.error(f"Could not install disk files to {dst_path}: {e}")
 

gns3server/controller/appliance_manager.py

@@ -100,17 +100,9 @@ class ApplianceManager:
 
         dst_path = self._builtin_appliances_path(delete_first=True)
         log.info(f"Installing built-in appliances in '{dst_path}'")
+        from . import Controller
         try:
-            if hasattr(sys, "frozen") and sys.platform.startswith("win"):
-                resource_path = os.path.normpath(os.path.join(os.path.dirname(sys.executable), "appliances"))
-                for filename in os.listdir(resource_path):
-                    shutil.copy(os.path.join(resource_path, filename), os.path.join(dst_path, filename))
-            else:
-                for entry in importlib_resources.files('gns3server.appliances').iterdir():
-                    full_path = os.path.join(dst_path, entry.name)
-                    if entry.is_file():
-                        log.debug(f"Installing built-in appliance file {entry.name} to {full_path}")
-                        shutil.copy(str(entry), os.path.join(dst_path, entry.name))
+            Controller.instance().install_resource_files(dst_path, "appliances")
         except OSError as e:
             log.error(f"Could not install built-in appliance files to {dst_path}: {e}")
 

tests/controller/test_controller.py

@@ -395,6 +395,31 @@ async def test_install_base_configs(controller, config, tmpdir):
         assert f.read() == 'test'
 
 
+@pytest.mark.parametrize(
+    "builtin_disk",
+    [
+        "empty8G.qcow2",
+        "empty10G.qcow2",
+        "empty20G.qcow2",
+        "empty30G.qcow2",
+        "empty40G.qcow2",
+        "empty50G.qcow2",
+        "empty100G.qcow2",
+        "empty150G.qcow2",
+        "empty200G.qcow2",
+        "empty250G.qcow2",
+        "empty500G.qcow2",
+        "empty1T.qcow2"
+    ]
+)
+async def test_install_builtin_disks(controller, config, tmpdir, builtin_disk):
+
+    config.set_section_config("Server", {"images_path": str(tmpdir)})
+    controller._install_builtin_disks()
+    # we only install Qemu empty disks at this time
+    assert os.path.exists(str(tmpdir / "QEMU" / builtin_disk))
+
+
 def test_appliances(controller, tmpdir):
 
     my_appliance = {