feat: forbid deleting templates and images still used by projects

Deleting a template with prune_images, deleting an image, or pruning
orphan images only checked template references — a project node still
pointing at the image (e.g. via hda_disk_image_backing_file) was left
with a dangling reference and the project could no longer be opened.

- Add controller helpers scanning every known project (opened projects
  via in-memory nodes, closed projects via their .gns3 file) for
  template and image usage; unreadable topologies are skipped
- Guard DELETE /templates/{id}, DELETE /templates/{id}?prune_images,
  DELETE /images/{path} and /images/prune with a 409 listing the
  project names
- Run all template-delete checks before any mutation so a refused
  deletion cannot leave the template gone while its images survive

tests/api/routes/controller/test_images.py

@@ -16,6 +16,8 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
+import json
+import uuid
 import pytest
 import hashlib
 
@@ -282,6 +284,99 @@ class TestImageRoutes:
         images_in_db = await images_repo.get_images()
         assert len(images_in_db) == 0
 
+    @staticmethod
+    async def _add_closed_project_with_node(controller: Controller, name: str, node: dict):
+        """
+        Register a closed project whose .gns3 contains a single node,
+        mirroring how the controller picks up existing projects from disk.
+        """
+
+        project_dir = os.path.join(controller.projects_directory(), name)
+        os.makedirs(project_dir)
+        project_id = str(uuid.uuid4())
+        topology = {
+            "name": name,
+            "project_id": project_id,
+            "topology": {"computes": [], "links": [], "drawings": [], "nodes": [node]},
+        }
+        with open(os.path.join(project_dir, f"{name}.gns3"), "w+") as f:
+            json.dump(topology, f)
+        return await controller.add_project(
+            project_id=project_id, name=name, path=project_dir,
+            filename=f"{name}.gns3", status="closed",
+        )
+
+    async def test_image_delete_used_by_project(
+            self,
+            app: FastAPI,
+            client: AsyncClient,
+            controller: Controller,
+            db_session: AsyncSession,
+            tmpdir: str,
+    ) -> None:
+        """
+        An image referenced by a node in a project must not be deleted,
+        even when the project is closed and no template uses the image.
+        """
+
+        image_path = os.path.join(tmpdir, "used.qcow2")
+        with open(image_path, "wb+") as f:
+            f.write(b'\x42\x42\x42\x42')
+
+        images_repo = ImagesRepository(db_session)
+        await images_repo.add_image("used.qcow2", "qemu", 42, image_path, "e342eb86c1229b6c154367a5476969b5", "md5")
+
+        guarded_project = await self._add_closed_project_with_node(
+            controller, "Guarded",
+            {"node_id": str(uuid.uuid4()), "node_type": "qemu", "compute_id": "local",
+             "name": "n1", "properties": {"hda_disk_image_backing_file": "used.qcow2"}},
+        )
+
+        response = await client.delete(app.url_path_for("delete_image", image_path="used.qcow2"))
+        assert response.status_code == status.HTTP_409_CONFLICT
+        assert "Guarded" in response.json()["message"]
+        assert os.path.exists(image_path)
+
+        # once no project uses it anymore the deletion goes through
+        controller.remove_project(guarded_project)
+        response = await client.delete(app.url_path_for("delete_image", image_path="used.qcow2"))
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert not os.path.exists(image_path)
+
+    async def test_prune_images_keeps_project_referenced(
+            self,
+            app: FastAPI,
+            client: AsyncClient,
+            controller: Controller,
+            db_session: AsyncSession,
+            tmpdir: str,
+    ) -> None:
+        """
+        Pruning must keep images referenced by a project node while
+        removing the unreferenced ones.
+        """
+
+        images_repo = ImagesRepository(db_session)
+        for filename in ("used.qcow2", "unused.qcow2"):
+            image_path = os.path.join(tmpdir, filename)
+            with open(image_path, "wb+") as f:
+                f.write(b'\x42\x42\x42\x42')
+            await images_repo.add_image(filename, "qemu", 42, image_path, "e342eb86c1229b6c154367a5476969b5", "md5")
+
+        await self._add_closed_project_with_node(
+            controller, "Guarded",
+            {"node_id": str(uuid.uuid4()), "node_type": "qemu", "compute_id": "local",
+             "name": "n1", "properties": {"hda_disk_image_backing_file": "used.qcow2"}},
+        )
+
+        response = await client.delete(app.url_path_for("prune_images"))
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        assert await images_repo.get_image(os.path.join(tmpdir, "used.qcow2")) is not None
+        assert os.path.exists(os.path.join(tmpdir, "used.qcow2"))
+        assert await images_repo.get_image(os.path.join(tmpdir, "unused.qcow2")) is None
+        assert not os.path.exists(os.path.join(tmpdir, "unused.qcow2"))
+
     async def test_image_upload_create_appliance(
             self, app: FastAPI,
             client: AsyncClient,

tests/api/routes/controller/test_templates.py

@@ -16,6 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
+import json
 import shutil
 
 import pytest
@@ -270,6 +271,128 @@ class TestTemplateRoutes:
         images = await images_repo.get_images()
         assert len(images) == 0
 
+    @staticmethod
+    async def _add_closed_project_with_node(controller: Controller, name: str, node: dict):
+        """
+        Register a closed project whose .gns3 contains a single node,
+        mirroring how the controller picks up existing projects from disk.
+        """
+
+        project_dir = os.path.join(controller.projects_directory(), name)
+        os.makedirs(project_dir)
+        project_id = str(uuid.uuid4())
+        topology = {
+            "name": name,
+            "project_id": project_id,
+            "topology": {"computes": [], "links": [], "drawings": [], "nodes": [node]},
+        }
+        with open(os.path.join(project_dir, f"{name}.gns3"), "w+") as f:
+            json.dump(topology, f)
+        # an explicit project_id marks this as an existing on-disk project
+        # (same path load_project takes at startup)
+        return await controller.add_project(
+            project_id=project_id, name=name, path=project_dir,
+            filename=f"{name}.gns3", status="closed",
+        )
+
+    async def test_template_delete_used_by_project(
+            self,
+            app: FastAPI,
+            client: AsyncClient,
+            controller: Controller,
+    ) -> None:
+        """
+        A template referenced by a node in a project must not be deleted,
+        even when the project is closed.
+        """
+
+        template_id = str(uuid.uuid4())
+        params = {"template_id": template_id,
+                  "name": "VPCS_GUARDED",
+                  "compute_id": "local",
+                  "template_type": "vpcs"}
+
+        response = await client.post(app.url_path_for("create_template"), json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+
+        guarded_project = await self._add_closed_project_with_node(
+            controller, "Guarded",
+            {"node_id": str(uuid.uuid4()), "node_type": "vpcs", "compute_id": "local",
+             "name": "n1", "template_id": template_id, "properties": {}},
+        )
+
+        response = await client.delete(app.url_path_for("delete_template", template_id=template_id))
+        assert response.status_code == status.HTTP_409_CONFLICT
+        assert "Guarded" in response.json()["message"]
+
+        # the template survived the refused deletion
+        response = await client.get(app.url_path_for("get_template", template_id=template_id))
+        assert response.status_code == status.HTTP_200_OK
+
+        # once no project uses it anymore the deletion goes through
+        controller.remove_project(guarded_project)
+        response = await client.delete(app.url_path_for("delete_template", template_id=template_id))
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+    async def test_template_delete_with_prune_images_used_by_project(
+            self,
+            app: FastAPI,
+            client: AsyncClient,
+            controller: Controller,
+            db_session: AsyncSession,
+            tmpdir: str,
+    ) -> None:
+        """
+        Pruning an image still referenced by a project node must be refused
+        before any mutation: the template and the image file both survive.
+        """
+
+        image_path = os.path.join(tmpdir, "used.qcow2")
+        with open(image_path, "wb+") as f:
+            f.write(b'\x42\x42\x42\x42')
+
+        images_repo = ImagesRepository(db_session)
+        await images_repo.add_image("used.qcow2", "qemu", 42, image_path, "e342eb86c1229b6c154367a5476969b5", "md5")
+
+        template_id = str(uuid.uuid4())
+        params = {"template_id": template_id,
+                  "name": "QEMU_GUARDED",
+                  "compute_id": "local",
+                  "hda_disk_image": "used.qcow2",
+                  "template_type": "qemu"}
+
+        response = await client.post(app.url_path_for("create_template"), json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+
+        # the node references the image but not the template: the template
+        # check passes, the image check must still refuse the prune
+        guarded_project = await self._add_closed_project_with_node(
+            controller, "Guarded",
+            {"node_id": str(uuid.uuid4()), "node_type": "qemu", "compute_id": "local",
+             "name": "n1", "properties": {"hda_disk_image_backing_file": "used.qcow2"}},
+        )
+
+        response = await client.delete(
+            app.url_path_for("delete_template", template_id=template_id),
+            params={"prune_images": True}
+        )
+        assert response.status_code == status.HTTP_409_CONFLICT
+        assert "Guarded" in response.json()["message"]
+
+        # neither the template nor the image file was touched
+        response = await client.get(app.url_path_for("get_template", template_id=template_id))
+        assert response.status_code == status.HTTP_200_OK
+        assert os.path.exists(image_path)
+
+        controller.remove_project(guarded_project)
+        response = await client.delete(
+            app.url_path_for("delete_template", template_id=template_id),
+            params={"prune_images": True}
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert not os.path.exists(image_path)
+        assert await images_repo.get_image(image_path) is None
+
     # async def test_create_node_from_template(self, controller_api, controller, project):
     #
     #     id = str(uuid.uuid4())

tests/controller/test_controller.py

@@ -534,3 +534,74 @@ async def test_autoidlepc(controller):
         await controller.autoidlepc("local", "c7200", "test.bin", 512)
     assert node_mock.dynamips_auto_idlepc.called
     assert len(controller.projects) == 0
+
+
+@pytest.mark.asyncio
+async def test_find_projects_using_template_and_images(controller):
+    """
+    The template/image usage checks must see nodes of opened projects
+    (in-memory Node objects) as well as nodes of closed projects (raw
+    node dicts read back from the .gns3 file).
+    """
+
+    compute = MagicMock()
+    response = MagicMock()
+    response.json = {"console": 2048}
+    compute.post = AsyncioMagicMock(return_value=response)
+
+    project1 = await controller.add_project(name="Test1")
+    project2 = await controller.add_project(name="Test2")
+
+    template_id = str(uuid.uuid4())
+    await project1.add_node(
+        compute,
+        "n1",
+        None,
+        node_type="vpcs",
+        template_id=template_id,
+        properties={"hda_disk_image": "/tmp/images/disk.qcow2", "hda_disk_image_backing_file": "base.qcow2"},
+    )
+    await project2.add_node(compute, "n2", None, node_type="vpcs", properties={})
+
+    # opened projects
+    assert controller.find_projects_using_template(template_id) == ["Test1"]
+    assert controller.find_projects_using_template(str(uuid.uuid4())) == []
+    # image references are matched on file name whether stored as a bare
+    # name or as an absolute path
+    assert controller.find_projects_using_image("base.qcow2") == ["Test1"]
+    assert controller.find_projects_using_image("disk.qcow2") == ["Test1"]
+    assert controller.find_projects_using_image("unknown.qcow2") == []
+    referenced = controller.collect_referenced_image_filenames()
+    assert "base.qcow2" in referenced
+    assert "disk.qcow2" in referenced
+
+    # closed projects: same answers from the .gns3 file on disk
+    await project1.close()
+    await project2.close()
+    assert controller.find_projects_using_template(template_id) == ["Test1"]
+    assert controller.find_projects_using_image("base.qcow2") == ["Test1"]
+    assert controller.find_projects_using_image("disk.qcow2") == ["Test1"]
+
+
+@pytest.mark.asyncio
+async def test_find_projects_skips_unreadable_topology(controller):
+    """
+    A project whose topology cannot be read must not break the usage checks.
+    """
+
+    compute = MagicMock()
+    response = MagicMock()
+    response.json = {"console": 2048}
+    compute.post = AsyncioMagicMock(return_value=response)
+
+    project = await controller.add_project(name="Broken")
+    await project.add_node(
+        compute, "n1", None, node_type="vpcs",
+        template_id=str(uuid.uuid4()), properties={"hda_disk_image": "lost.qcow2"},
+    )
+    await project.close()
+    os.remove(project.topology_file)
+
+    assert controller.find_projects_using_template(str(uuid.uuid4())) == []
+    assert controller.find_projects_using_image("lost.qcow2") == []
+    assert controller.collect_referenced_image_filenames() == set()